batty_cli/team/

events.rs

//! Structured JSONL event stream for external consumers.

use std::fs::{self, OpenOptions};
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};

use super::DEFAULT_EVENT_LOG_MAX_BYTES;
use super::merge::MergeMode;

/// Bundled parameters for merge-confidence scoring events.
pub struct MergeConfidenceInfo<'a> {
    pub engineer: &'a str,
    pub task: &'a str,
    pub confidence: f64,
    pub files_changed: usize,
    pub lines_changed: usize,
    pub has_migrations: bool,
    pub has_config_changes: bool,
    pub rename_count: usize,
}

pub struct AutoMergeDecisionInfo<'a> {
    pub engineer: &'a str,
    pub task: &'a str,
    pub action_type: &'a str,
    pub confidence: f64,
    pub reason: &'a str,
    pub details: &'a str,
}

pub struct VerificationPhaseChangeInfo<'a> {
    pub engineer: &'a str,
    pub task: &'a str,
    pub from_phase: &'a str,
    pub to_phase: &'a str,
    pub iteration: u32,
}

pub struct QualityMetricsInfo<'a> {
    pub backend: &'a str,
    pub role: &'a str,
    pub task: &'a str,
    pub narration_ratio: f64,
    pub commit_frequency: f64,
    pub first_pass_test_rate: f64,
    pub retry_rate: f64,
    pub time_to_completion_secs: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]

pub struct TeamEvent {
    pub event: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub action_type: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub version: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub git_ref: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub role: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub task: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub recipient: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub from: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub to: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub restart: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub restart_count: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub load: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub working_members: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub total_members: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_running: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub step: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub details: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub uptime_secs: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_size_bytes: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub output_bytes: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub filename: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub content_hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub backend: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub success: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub merge_mode: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub narration_ratio: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub commit_frequency: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub first_pass_test_rate: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub retry_rate: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub time_to_completion_secs: Option<u64>,
    pub ts: u64,
}

impl TeamEvent {
    fn now() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs()
    }

    fn base(event: &str) -> Self {
        Self {
            event: event.into(),
            action_type: None,
            version: None,
            git_ref: None,
            role: None,
            task: None,
            recipient: None,
            from: None,
            to: None,
            restart: None,
            restart_count: None,
            load: None,
            working_members: None,
            total_members: None,
            session_running: None,
            reason: None,
            step: None,
            error: None,
            details: None,
            uptime_secs: None,
            session_size_bytes: None,
            output_bytes: None,
            filename: None,
            content_hash: None,
            backend: None,
            success: None,
            merge_mode: None,
            narration_ratio: None,
            commit_frequency: None,
            first_pass_test_rate: None,
            retry_rate: None,
            time_to_completion_secs: None,
            ts: Self::now(),
        }
    }

    pub fn daemon_started() -> Self {
        Self {
            action_type: Some("session_lifecycle".into()),
            session_running: Some(true),
            ..Self::base("daemon_started")
        }
    }

    pub fn daemon_reloading() -> Self {
        Self {
            action_type: Some("session_lifecycle".into()),
            session_running: Some(true),
            ..Self::base("daemon_reloading")
        }
    }

    pub fn daemon_reloaded() -> Self {
        Self {
            action_type: Some("session_lifecycle".into()),
            session_running: Some(true),
            ..Self::base("daemon_reloaded")
        }
    }

    pub fn daemon_stopped() -> Self {
        Self {
            action_type: Some("session_lifecycle".into()),
            session_running: Some(false),
            ..Self::base("daemon_stopped")
        }
    }

    pub fn daemon_stopped_with_reason(reason: &str, uptime_secs: u64) -> Self {
        Self {
            action_type: Some("session_lifecycle".into()),
            session_running: Some(false),
            reason: Some(reason.into()),
            uptime_secs: Some(uptime_secs),
            ..Self::base("daemon_stopped")
        }
    }

    pub fn daemon_heartbeat(uptime_secs: u64) -> Self {
        Self {
            uptime_secs: Some(uptime_secs),
            ..Self::base("daemon_heartbeat")
        }
    }

    pub fn context_exhausted(
        role: &str,
        task: Option<u32>,
        session_size_bytes: Option<u64>,
    ) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|task_id| task_id.to_string()),
            session_size_bytes,
            ..Self::base("context_exhausted")
        }
    }

    pub fn loop_step_error(step: &str, error: &str) -> Self {
        Self {
            step: Some(step.into()),
            error: Some(error.into()),
            ..Self::base("loop_step_error")
        }
    }

    pub fn daemon_panic(reason: &str) -> Self {
        Self {
            reason: Some(reason.into()),
            ..Self::base("daemon_panic")
        }
    }

    pub fn task_assigned(role: &str, task: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            ..Self::base("task_assigned")
        }
    }

    pub fn dispatch_fallback_used(role: &str, task: &str, recipient: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            recipient: Some(recipient.into()),
            reason: Some(reason.into()),
            ..Self::base("dispatch_fallback_used")
        }
    }

    pub fn dispatch_overlap_prevented(task_id: u32, blocked_by: &[u32], details: &str) -> Self {
        Self {
            task: Some(task_id.to_string()),
            reason: Some(
                blocked_by
                    .iter()
                    .map(|task_id| task_id.to_string())
                    .collect::<Vec<_>>()
                    .join(","),
            ),
            details: Some(details.into()),
            ..Self::base("dispatch_overlap_prevented")
        }
    }

    pub fn dispatch_overlap_skipped(
        candidate_task_id: u32,
        conflicting_task_id: &str,
        conflicting_files: &[String],
    ) -> Self {
        Self {
            task: Some(candidate_task_id.to_string()),
            reason: Some(conflicting_task_id.to_string()),
            details: Some(conflicting_files.join(",")),
            ..Self::base("dispatch_overlap_skipped")
        }
    }

    pub fn cwd_corrected(role: &str, path: &str) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(path.into()),
            ..Self::base("cwd_corrected")
        }
    }

    pub fn review_nudge_sent(role: &str, task: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            ..Self::base("review_nudge_sent")
        }
    }

    pub fn review_escalated(task: &str, reason: &str) -> Self {
        Self {
            task: Some(task.into()),
            reason: Some(reason.into()),
            ..Self::base("review_escalated")
        }
    }

    pub fn state_reconciliation(role: Option<&str>, task: Option<&str>, correction: &str) -> Self {
        Self {
            role: role.map(str::to_string),
            task: task.map(str::to_string),
            reason: Some(correction.into()),
            ..Self::base("state_reconciliation")
        }
    }

    pub fn task_escalated(role: &str, task: &str, reason: Option<&str>) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: reason.map(|r| r.into()),
            ..Self::base("task_escalated")
        }
    }

    pub fn task_stale(role: &str, task: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(reason.into()),
            ..Self::base("task_stale")
        }
    }

    pub fn task_aged(task: &str, reason: &str) -> Self {
        Self {
            task: Some(task.into()),
            reason: Some(reason.into()),
            ..Self::base("task_aged")
        }
    }

    pub fn review_stale(task: &str, reason: &str) -> Self {
        Self {
            task: Some(task.into()),
            reason: Some(reason.into()),
            ..Self::base("review_stale")
        }
    }

    pub fn task_unblocked(role: &str, task: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            ..Self::base("task_unblocked")
        }
    }

    pub fn task_claim_created(role: &str, task: &str, ttl_secs: u64, expires_at: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(format!("ttl_secs={ttl_secs} expires_at={expires_at}")),
            ..Self::base("task_claim_created")
        }
    }

    pub fn task_claim_progress(role: &str, task: &str, progress_type: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(progress_type.into()),
            ..Self::base("task_claim_progress")
        }
    }

    pub fn task_claim_warning(role: &str, task: &str, expires_in_secs: u64) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(format!("expires_in_secs={expires_in_secs}")),
            ..Self::base("task_claim_warning")
        }
    }

    pub fn task_claim_expired(
        role: &str,
        task: &str,
        reclaimed: bool,
        time_held_secs: Option<u64>,
    ) -> Self {
        let mut reason = format!("reclaimed={reclaimed}");
        if let Some(time_held_secs) = time_held_secs {
            reason.push_str(&format!(" time_held_secs={time_held_secs}"));
        }
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(reason),
            ..Self::base("task_claim_expired")
        }
    }

    pub fn task_claim_extended(role: &str, task: &str, new_expires_at: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(format!("new_expires_at={new_expires_at}")),
            ..Self::base("task_claim_extended")
        }
    }

    pub fn scope_fence_violation(role: &str, task_id: u32, details: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task_id.to_string()),
            reason: Some(details.into()),
            ..Self::base("scope_fence_violation")
        }
    }

    pub fn board_task_archived(task: &str, role: Option<&str>) -> Self {
        Self {
            role: role.map(str::to_string),
            task: Some(task.into()),
            ..Self::base("board_task_archived")
        }
    }

    pub fn auto_doctor_action(
        action_type: &str,
        task_id: Option<u32>,
        engineer: Option<&str>,
        details: &str,
    ) -> Self {
        Self {
            action_type: Some(action_type.into()),
            task: task_id.map(|task_id| task_id.to_string()),
            role: engineer.map(str::to_string),
            details: Some(details.into()),
            ..Self::base("auto_doctor_action")
        }
    }

    pub fn performance_regression(task: &str, reason: &str) -> Self {
        Self {
            task: Some(task.into()),
            reason: Some(reason.into()),
            ..Self::base("performance_regression")
        }
    }

    pub fn task_completed(role: &str, task: Option<&str>) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|t| t.into()),
            ..Self::base("task_completed")
        }
    }

    pub fn quality_metrics_recorded(info: &QualityMetricsInfo<'_>) -> Self {
        Self {
            role: Some(info.role.into()),
            task: Some(info.task.into()),
            backend: Some(info.backend.into()),
            narration_ratio: Some(info.narration_ratio),
            commit_frequency: Some(info.commit_frequency),
            first_pass_test_rate: Some(info.first_pass_test_rate),
            retry_rate: Some(info.retry_rate),
            time_to_completion_secs: Some(info.time_to_completion_secs),
            ..Self::base("quality_metrics_recorded")
        }
    }

    pub fn standup_generated(recipient: &str) -> Self {
        Self {
            recipient: Some(recipient.into()),
            ..Self::base("standup_generated")
        }
    }

    pub fn supervisory_digest_emitted(
        role: &str,
        notice_count: u32,
        suppressed_duplicates: u32,
    ) -> Self {
        Self {
            role: Some(role.into()),
            details: Some(format!(
                "notice_count={notice_count} suppressed_duplicates={suppressed_duplicates}"
            )),
            ..Self::base("supervisory_digest_emitted")
        }
    }

    pub fn retro_generated() -> Self {
        Self::base("retro_generated")
    }

    pub fn pattern_detected(pattern_type: &str, frequency: u32) -> Self {
        Self {
            reason: Some(format!("{pattern_type}:{frequency}")),
            ..Self::base("pattern_detected")
        }
    }

    pub fn disk_hygiene_cleanup(summary: &str) -> Self {
        Self {
            details: Some(summary.into()),
            ..Self::base("disk_hygiene_cleanup")
        }
    }

    pub fn member_crashed(role: &str, restart: bool) -> Self {
        Self {
            role: Some(role.into()),
            restart: Some(restart),
            ..Self::base("member_crashed")
        }
    }

    pub fn pane_death(role: &str) -> Self {
        Self {
            role: Some(role.into()),
            ..Self::base("pane_death")
        }
    }

    pub fn pane_respawned(role: &str) -> Self {
        Self {
            role: Some(role.into()),
            ..Self::base("pane_respawned")
        }
    }

    pub fn shim_disconnect(role: &str, reason: &str, details: &str, expected: bool) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(reason.into()),
            details: Some(details.into()),
            success: Some(expected),
            ..Self::base("shim_disconnect")
        }
    }

    pub fn narration_detected(role: &str, task: Option<u32>) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|id| id.to_string()),
            ..Self::base("narration_detected")
        }
    }

    pub fn narration_nudged(role: &str, task: Option<u32>) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|id| id.to_string()),
            ..Self::base("narration_nudged")
        }
    }

    pub fn narration_restart(role: &str, task: Option<u32>) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|id| id.to_string()),
            ..Self::base("narration_restart")
        }
    }

    pub fn narration_rejection(role: &str, task_id: u32, rejection_count: u32) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task_id.to_string()),
            reason: Some(format!("rejection_count={rejection_count}")),
            ..Self::base("narration_rejection")
        }
    }

    pub fn tact_cycle_triggered(role: &str, idle_engineers: u32, board_summary: &str) -> Self {
        Self {
            role: Some(role.into()),
            working_members: Some(idle_engineers),
            reason: Some(board_summary.into()),
            ..Self::base("tact_cycle_triggered")
        }
    }

    pub fn tact_tasks_created(
        role: &str,
        tasks_created: u32,
        latency_secs: u64,
        success: bool,
        error: Option<&str>,
    ) -> Self {
        Self {
            role: Some(role.into()),
            restart_count: Some(tasks_created),
            uptime_secs: Some(latency_secs),
            reason: Some(if success { "success" } else { "failure" }.into()),
            error: error.map(str::to_string),
            ..Self::base("tact_tasks_created")
        }
    }

    pub fn context_pressure_warning(
        role: &str,
        task: Option<u32>,
        pressure_score: u64,
        threshold: u64,
        output_bytes: u64,
    ) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|id| id.to_string()),
            load: Some(pressure_score as f64),
            output_bytes: Some(output_bytes),
            reason: Some(format!("threshold={threshold}")),
            ..Self::base("context_pressure_warning")
        }
    }

    pub fn stall_detected(role: &str, task: Option<u32>, stall_duration_secs: u64) -> Self {
        Self::stall_detected_with_reason(role, task, stall_duration_secs, None)
    }

    pub fn stall_detected_with_reason(
        role: &str,
        task: Option<u32>,
        stall_duration_secs: u64,
        reason: Option<&str>,
    ) -> Self {
        Self {
            role: Some(role.into()),
            task: task.map(|id| id.to_string()),
            uptime_secs: Some(stall_duration_secs),
            reason: reason.map(str::to_string),
            ..Self::base("stall_detected")
        }
    }

    /// Record a backend health state change for an agent.
    ///
    /// `reason` encodes the transition, e.g. "healthy→unreachable".
    pub fn health_changed(role: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(reason.into()),
            ..Self::base("health_changed")
        }
    }

    pub fn message_routed(from: &str, to: &str) -> Self {
        Self {
            from: Some(from.into()),
            to: Some(to.into()),
            ..Self::base("message_routed")
        }
    }

    pub fn discord_event_sent(channel: &str, source_event: &str) -> Self {
        Self {
            action_type: Some("notification".into()),
            recipient: Some(channel.into()),
            reason: Some(source_event.into()),
            ..Self::base("discord_event_sent")
        }
    }

    pub fn notification_delivery_sample(
        from: &str,
        to: &str,
        latency_secs: u64,
        isolation_mode: &str,
    ) -> Self {
        Self {
            action_type: Some(isolation_mode.into()),
            from: Some(from.into()),
            to: Some(to.into()),
            uptime_secs: Some(latency_secs),
            ..Self::base("notification_delivery_sample")
        }
    }

    pub fn inbox_message_expired(target: &str, from: &str, message_age_secs: u64) -> Self {
        Self {
            recipient: Some(target.into()),
            from: Some(from.into()),
            uptime_secs: Some(message_age_secs),
            ..Self::base("inbox_message_expired")
        }
    }

    pub fn inbox_message_deduplicated(target: &str, from: &str, signature: u64) -> Self {
        Self {
            recipient: Some(target.into()),
            from: Some(from.into()),
            content_hash: Some(format!("{signature:016x}")),
            ..Self::base("inbox_message_deduplicated")
        }
    }

    pub fn inbox_batch_delivered(
        target: &str,
        message_count: usize,
        escalation_count: usize,
    ) -> Self {
        Self {
            recipient: Some(target.into()),
            restart_count: Some(message_count as u32),
            reason: Some(format!("escalation_count={escalation_count}")),
            ..Self::base("inbox_batch_delivered")
        }
    }

    pub fn backend_quota_exhausted(role: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(reason.into()),
            ..Self::base("backend_quota_exhausted")
        }
    }

    pub fn agent_spawned(role: &str) -> Self {
        Self {
            role: Some(role.into()),
            ..Self::base("agent_spawned")
        }
    }

    pub fn agent_restarted(role: &str, task: &str, reason: &str, restart_count: u32) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(reason.into()),
            restart_count: Some(restart_count),
            ..Self::base("agent_restarted")
        }
    }

    pub fn agent_handoff(role: &str, task: &str, reason: &str, success: bool) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(reason.into()),
            success: Some(success),
            ..Self::base("agent_handoff")
        }
    }

    pub fn handoff_injected(role: &str, task: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(reason.into()),
            success: Some(true),
            ..Self::base("handoff_injected")
        }
    }

    pub fn task_resumed(role: &str, task: &str, reason: &str, restart_count: u32) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            reason: Some(reason.into()),
            restart_count: Some(restart_count),
            ..Self::base("task_resumed")
        }
    }

    pub fn delivery_failed(role: &str, from: &str, reason: &str) -> Self {
        Self::delivery_failed_with_details(role, from, reason, None)
    }

    pub fn delivery_failed_with_details(
        role: &str,
        from: &str,
        reason: &str,
        details: Option<&str>,
    ) -> Self {
        Self {
            role: Some(role.into()),
            from: Some(from.into()),
            reason: Some(reason.into()),
            details: details.map(str::to_string),
            ..Self::base("delivery_failed")
        }
    }

    pub fn task_auto_merged(
        engineer: &str,
        task: &str,
        confidence: f64,
        files_changed: usize,
        lines_changed: usize,
    ) -> Self {
        Self::task_auto_merged_with_mode(
            engineer,
            task,
            confidence,
            files_changed,
            lines_changed,
            None,
        )
    }

    pub fn task_auto_merged_with_mode(
        engineer: &str,
        task: &str,
        confidence: f64,
        files_changed: usize,
        lines_changed: usize,
        merge_mode: Option<MergeMode>,
    ) -> Self {
        Self {
            role: Some(engineer.into()),
            task: Some(task.into()),
            load: Some(confidence),
            reason: Some(format!("files={} lines={}", files_changed, lines_changed)),
            success: Some(true),
            merge_mode: merge_mode.map(|mode| mode.as_str().to_string()),
            ..Self::base("task_auto_merged")
        }
    }

    pub fn auto_merge_decision_recorded(info: &AutoMergeDecisionInfo<'_>) -> Self {
        Self {
            role: Some(info.engineer.into()),
            task: Some(info.task.into()),
            action_type: Some(info.action_type.into()),
            load: Some(info.confidence),
            reason: Some(info.reason.into()),
            details: Some(info.details.into()),
            ..Self::base("auto_merge_decision_recorded")
        }
    }

    pub fn auto_merge_post_verify_result(
        engineer: &str,
        task: &str,
        success: Option<bool>,
        reason: &str,
        details: Option<&str>,
    ) -> Self {
        Self {
            role: Some(engineer.into()),
            task: Some(task.into()),
            success,
            reason: Some(reason.into()),
            details: details.map(str::to_string),
            ..Self::base("auto_merge_post_verify_result")
        }
    }

    pub fn task_manual_merged(task: &str) -> Self {
        Self::task_manual_merged_with_mode(task, None)
    }

    pub fn task_manual_merged_with_mode(task: &str, merge_mode: Option<MergeMode>) -> Self {
        Self {
            task: Some(task.into()),
            success: Some(true),
            merge_mode: merge_mode.map(|mode| mode.as_str().to_string()),
            ..Self::base("task_manual_merged")
        }
    }

    pub fn task_merge_failed(
        engineer: &str,
        task: &str,
        merge_mode: Option<MergeMode>,
        details: &str,
    ) -> Self {
        Self {
            role: Some(engineer.into()),
            task: Some(task.into()),
            success: Some(false),
            merge_mode: merge_mode.map(|mode| mode.as_str().to_string()),
            reason: Some("merge_failed".into()),
            details: Some(details.into()),
            ..Self::base("task_merge_failed")
        }
    }

    /// Emitted for every completed task to record its merge confidence score.
    pub fn merge_confidence_scored(info: &MergeConfidenceInfo<'_>) -> Self {
        let detail = format!(
            "files={} lines={} migrations={} config={} renames={}",
            info.files_changed,
            info.lines_changed,
            info.has_migrations,
            info.has_config_changes,
            info.rename_count
        );
        Self {
            role: Some(info.engineer.into()),
            task: Some(info.task.into()),
            load: Some(info.confidence),
            reason: Some(detail),
            ..Self::base("merge_confidence_scored")
        }
    }

    pub fn verification_phase_changed(info: &VerificationPhaseChangeInfo<'_>) -> Self {
        Self {
            role: Some(info.engineer.into()),
            task: Some(info.task.into()),
            step: Some(info.to_phase.into()),
            reason: Some(format!(
                "from={} to={} iteration={}",
                info.from_phase, info.to_phase, info.iteration
            )),
            restart_count: Some(info.iteration),
            ..Self::base("verification_phase_changed")
        }
    }

    pub fn verification_evidence_collected(
        engineer: &str,
        task: &str,
        evidence_kind: &str,
        detail: &str,
    ) -> Self {
        Self {
            role: Some(engineer.into()),
            task: Some(task.into()),
            step: Some(evidence_kind.into()),
            reason: Some(detail.into()),
            ..Self::base("verification_evidence_collected")
        }
    }

    pub fn verification_max_iterations_reached(
        engineer: &str,
        task: &str,
        iteration: u32,
        escalated_to: &str,
    ) -> Self {
        Self {
            role: Some(engineer.into()),
            task: Some(task.into()),
            recipient: Some(escalated_to.into()),
            reason: Some(format!("iteration={iteration}")),
            restart_count: Some(iteration),
            ..Self::base("verification_max_iterations_reached")
        }
    }

    pub fn review_escalated_by_role(role: &str, task: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            ..Self::base("review_escalated")
        }
    }

    pub fn pipeline_starvation_detected(idle_engineers: usize, todo_tasks: usize) -> Self {
        Self {
            reason: Some(format!(
                "idle_engineers={idle_engineers} todo_tasks={todo_tasks}"
            )),
            ..Self::base("pipeline_starvation_detected")
        }
    }

    pub fn task_reworked(role: &str, task: &str) -> Self {
        Self {
            role: Some(role.into()),
            task: Some(task.into()),
            ..Self::base("task_reworked")
        }
    }

    pub fn task_recycled(task_id: u32, cron_expr: &str) -> Self {
        Self {
            task: Some(format!("#{task_id}")),
            reason: Some(cron_expr.into()),
            ..Self::base("task_recycled")
        }
    }

    pub fn barrier_artifact_created(role: &str, filename: &str, content_hash: &str) -> Self {
        Self {
            role: Some(role.into()),
            filename: Some(filename.into()),
            content_hash: Some(content_hash.into()),
            ..Self::base("barrier_artifact_created")
        }
    }

    pub fn barrier_artifact_read(role: &str, filename: &str, content_hash: &str) -> Self {
        Self {
            role: Some(role.into()),
            filename: Some(filename.into()),
            content_hash: Some(content_hash.into()),
            ..Self::base("barrier_artifact_read")
        }
    }

    pub fn barrier_violation_attempt(role: &str, filename: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            filename: Some(filename.into()),
            reason: Some(reason.into()),
            ..Self::base("barrier_violation_attempt")
        }
    }

    pub fn load_snapshot(working_members: u32, total_members: u32, session_running: bool) -> Self {
        let load = if total_members == 0 {
            0.0
        } else {
            working_members as f64 / total_members as f64
        };
        Self {
            load: Some(load),
            working_members: Some(working_members),
            total_members: Some(total_members),
            session_running: Some(session_running),
            ..Self::base("load_snapshot")
        }
    }

    pub fn parity_updated(summary: &crate::team::parity::ParitySummary) -> Self {
        Self {
            load: Some(summary.overall_parity_pct as f64 / 100.0),
            reason: Some(format!(
                "total={} spec={} tests={} implementation={} verified_pass={} verified_fail={}",
                summary.total_behaviors,
                summary.spec_complete,
                summary.tests_complete,
                summary.implementation_complete,
                summary.verified_pass,
                summary.verified_fail
            )),
            ..Self::base("parity_updated")
        }
    }

    pub fn release_succeeded(
        version: &str,
        git_ref: &str,
        tag: &str,
        notes_path: Option<&str>,
    ) -> Self {
        Self {
            action_type: Some("release".into()),
            version: Some(version.into()),
            git_ref: Some(git_ref.into()),
            task: Some(tag.into()),
            reason: Some("tag_created".into()),
            success: Some(true),
            details: notes_path.map(str::to_string),
            ..Self::base("release_succeeded")
        }
    }

    pub fn release_failed(
        version: Option<&str>,
        git_ref: Option<&str>,
        tag: Option<&str>,
        reason: &str,
        details: Option<&str>,
    ) -> Self {
        Self {
            action_type: Some("release".into()),
            version: version.map(str::to_string),
            git_ref: git_ref.map(str::to_string),
            task: tag.map(str::to_string),
            reason: Some(reason.into()),
            details: details.map(str::to_string),
            success: Some(false),
            ..Self::base("release_failed")
        }
    }

    pub fn worktree_reconciled(role: &str, branch: &str) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(format!("branch '{branch}' merged into main")),
            ..Self::base("worktree_reconciled")
        }
    }

    pub fn worktree_refreshed(role: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(reason.into()),
            ..Self::base("worktree_refreshed")
        }
    }

    /// Emitted when the daemon reconciles a topology change.
    ///
    /// `reason` contains a human-readable summary (e.g. "+2 added, -1 removed").
    pub fn topology_changed(added: u32, removed: u32, reason: &str) -> Self {
        Self {
            working_members: Some(added),
            total_members: Some(removed),
            reason: Some(reason.into()),
            ..Self::base("topology_changed")
        }
    }

    /// Emitted when an agent is removed during a scale-down.
    pub fn agent_removed(role: &str, reason: &str) -> Self {
        Self {
            role: Some(role.into()),
            reason: Some(reason.into()),
            ..Self::base("agent_removed")
        }
    }
}

// ---------------------------------------------------------------------------
// Structured lifecycle events (typed, schema-versioned)
// ---------------------------------------------------------------------------

/// Agent session lifecycle phases.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionLifecycle {
    Spawning,
    Ready,
    Working,
    Blocked,
    Idle,
    Finished,
    Failed,
}

/// Task lifecycle phases.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TaskLifecycle {
    Claimed,
    InProgress,
    TestsRunning,
    TestsPassed,
    TestsFailed,
    Review,
    Merged,
    Rejected,
}

/// Merge lifecycle phases.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MergeLifecycle {
    Started,
    Conflict,
    Success,
}

/// Wrapper enum over all lifecycle event types.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "category", content = "phase")]
pub enum LifecycleEventType {
    #[serde(rename = "session")]
    Session(SessionLifecycle),
    #[serde(rename = "task")]
    Task(TaskLifecycle),
    #[serde(rename = "merge")]
    Merge(MergeLifecycle),
}

/// A typed, schema-versioned lifecycle event written to events.jsonl.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LifecycleEvent {
    pub schema_version: u32,
    pub ts: u64,
    pub event_type: LifecycleEventType,
    pub agent_name: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub task_id: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub details: Option<String>,
}

impl LifecycleEvent {
    /// Current schema version for lifecycle events.
    pub const SCHEMA_VERSION: u32 = 1;

    /// Create a new lifecycle event with the current timestamp.
    pub fn new(
        event_type: LifecycleEventType,
        agent_name: &str,
        task_id: Option<u32>,
        details: Option<String>,
    ) -> Self {
        Self {
            schema_version: Self::SCHEMA_VERSION,
            ts: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs(),
            event_type,
            agent_name: agent_name.to_string(),
            task_id,
            details,
        }
    }
}

/// Write a lifecycle event as a JSON line to the given events.jsonl path.
pub fn emit_lifecycle_event(path: &Path, event: &LifecycleEvent) -> Result<()> {
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let mut file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .with_context(|| format!("failed to open lifecycle event log: {}", path.display()))?;
    let json = serde_json::to_string(event)?;
    writeln!(file, "{json}")?;
    file.flush()?;
    Ok(())
}

pub struct EventSink {
    writer: Box<dyn Write + Send>,
    path: PathBuf,
    max_bytes: Option<u64>,
}

impl EventSink {
    pub fn new(path: &Path) -> Result<Self> {
        Self::new_with_max_bytes(path, DEFAULT_EVENT_LOG_MAX_BYTES)
    }

    pub fn new_with_max_bytes(path: &Path, max_bytes: u64) -> Result<Self> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        rotate_event_log_if_needed(path, max_bytes, 0)?;
        let file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
            .with_context(|| format!("failed to open event sink: {}", path.display()))?;
        Ok(Self {
            writer: Box::new(BufWriter::new(file)),
            path: path.to_path_buf(),
            max_bytes: Some(max_bytes),
        })
    }

    #[cfg(test)]
    pub(crate) fn from_writer(path: &Path, writer: impl Write + Send + 'static) -> Self {
        Self {
            writer: Box::new(writer),
            path: path.to_path_buf(),
            max_bytes: None,
        }
    }

    pub fn emit(&mut self, event: TeamEvent) -> Result<()> {
        let json = serde_json::to_string(&event)?;
        self.rotate_if_needed((json.len() + 1) as u64)?;
        writeln!(self.writer, "{json}")?;
        self.writer.flush()?;
        Ok(())
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    fn rotate_if_needed(&mut self, next_entry_bytes: u64) -> Result<()> {
        let Some(max_bytes) = self.max_bytes else {
            return Ok(());
        };
        self.writer.flush()?;
        if rotate_event_log_if_needed(&self.path, max_bytes, next_entry_bytes)? {
            self.writer = Box::new(BufWriter::new(
                OpenOptions::new()
                    .create(true)
                    .append(true)
                    .open(&self.path)
                    .with_context(|| {
                        format!("failed to reopen event sink: {}", self.path.display())
                    })?,
            ));
        }
        Ok(())
    }
}

fn rotated_event_log_path(path: &Path) -> PathBuf {
    PathBuf::from(format!("{}.1", path.display()))
}

fn rotate_event_log_if_needed(path: &Path, max_bytes: u64, next_entry_bytes: u64) -> Result<bool> {
    let len = match fs::metadata(path) {
        Ok(metadata) => metadata.len(),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(error) => {
            return Err(error).with_context(|| format!("failed to stat {}", path.display()));
        }
    };

    if len == 0 {
        return Ok(false);
    }

    if len.saturating_add(next_entry_bytes) <= max_bytes {
        return Ok(false);
    }

    let rotated = rotated_event_log_path(path);
    if rotated.exists() {
        fs::remove_file(&rotated)
            .with_context(|| format!("failed to remove {}", rotated.display()))?;
    }
    fs::rename(path, &rotated).with_context(|| {
        format!(
            "failed to rotate event log {} to {}",
            path.display(),
            rotated.display()
        )
    })?;
    Ok(true)
}

pub fn read_events(path: &Path) -> Result<Vec<TeamEvent>> {
    if !path.exists() {
        return Ok(Vec::new());
    }
    let content = fs::read_to_string(path).context("failed to read event log")?;
    let mut events = Vec::new();
    for line in content.lines() {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        if let Ok(event) = serde_json::from_str::<TeamEvent>(line) {
            events.push(event);
        }
    }
    Ok(events)
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};
    use std::thread;

    use super::*;

    #[test]
    fn event_serializes_to_json() {
        let event = TeamEvent::task_assigned("eng-1-1", "fix auth bug");
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"event\":\"task_assigned\""));
        assert!(json.contains("\"role\":\"eng-1-1\""));
        assert!(json.contains("\"task\":\"fix auth bug\""));
        assert!(json.contains("\"ts\":"));
    }

    #[test]
    fn optional_fields_omitted() {
        let event = TeamEvent::daemon_started();
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("\"role\""));
        assert!(!json.contains("\"task\""));
    }

    #[test]
    fn event_sink_writes_jsonl() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let mut sink = EventSink::new(&path).unwrap();
        sink.emit(TeamEvent::daemon_started()).unwrap();
        sink.emit(TeamEvent::task_assigned("eng-1", "fix bug"))
            .unwrap();
        sink.emit(TeamEvent::daemon_stopped()).unwrap();

        let content = std::fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = content.lines().collect();
        assert_eq!(lines.len(), 3);
        assert!(lines[0].contains("daemon_started"));
        assert!(lines[1].contains("task_assigned"));
        assert!(lines[2].contains("daemon_stopped"));
    }

    #[test]
    fn all_event_variants_serialize_with_correct_event_field() {
        let variants: Vec<(&str, TeamEvent)> = vec![
            ("daemon_started", TeamEvent::daemon_started()),
            ("daemon_reloading", TeamEvent::daemon_reloading()),
            ("daemon_reloaded", TeamEvent::daemon_reloaded()),
            ("daemon_stopped", TeamEvent::daemon_stopped()),
            (
                "daemon_stopped",
                TeamEvent::daemon_stopped_with_reason("signal", 3600),
            ),
            ("daemon_heartbeat", TeamEvent::daemon_heartbeat(120)),
            (
                "context_exhausted",
                TeamEvent::context_exhausted("eng-1", Some(42), Some(1_024)),
            ),
            (
                "loop_step_error",
                TeamEvent::loop_step_error("poll_watchers", "tmux error"),
            ),
            (
                "daemon_panic",
                TeamEvent::daemon_panic("index out of bounds"),
            ),
            ("task_assigned", TeamEvent::task_assigned("eng-1", "task")),
            (
                "cwd_corrected",
                TeamEvent::cwd_corrected("eng-1", "/tmp/worktree"),
            ),
            (
                "task_escalated",
                TeamEvent::task_escalated("eng-1", "task", None),
            ),
            ("task_unblocked", TeamEvent::task_unblocked("eng-1", "task")),
            (
                "performance_regression",
                TeamEvent::performance_regression("42", "runtime_ms=1300 avg_ms=1000 pct=30"),
            ),
            (
                "task_completed",
                TeamEvent::task_completed("eng-1", Some("42")),
            ),
            ("standup_generated", TeamEvent::standup_generated("manager")),
            ("retro_generated", TeamEvent::retro_generated()),
            (
                "pattern_detected",
                TeamEvent::pattern_detected("merge_conflict_recurrence", 5),
            ),
            ("member_crashed", TeamEvent::member_crashed("eng-1", true)),
            ("pane_death", TeamEvent::pane_death("eng-1")),
            ("pane_respawned", TeamEvent::pane_respawned("eng-1")),
            (
                "context_pressure_warning",
                TeamEvent::context_pressure_warning("eng-1", Some(42), 88, 100, 400_000),
            ),
            (
                "tact_cycle_triggered",
                TeamEvent::tact_cycle_triggered("architect", 3, "todo=0, backlog=1"),
            ),
            (
                "tact_tasks_created",
                TeamEvent::tact_tasks_created("architect", 2, 14, true, None),
            ),
            (
                "board_task_archived",
                TeamEvent::board_task_archived("42", Some("eng-1")),
            ),
            ("message_routed", TeamEvent::message_routed("a", "b")),
            ("agent_spawned", TeamEvent::agent_spawned("eng-1")),
            (
                "agent_restarted",
                TeamEvent::agent_restarted("eng-1", "42", "context_exhausted", 1),
            ),
            (
                "delivery_failed",
                TeamEvent::delivery_failed("eng-1", "manager", "message marker missing"),
            ),
            (
                "task_auto_merged",
                TeamEvent::task_auto_merged_with_mode(
                    "eng-1",
                    "42",
                    0.95,
                    2,
                    30,
                    Some(MergeMode::IsolatedIntegration),
                ),
            ),
            (
                "task_manual_merged",
                TeamEvent::task_manual_merged_with_mode("42", Some(MergeMode::DirectRoot)),
            ),
            (
                "task_merge_failed",
                TeamEvent::task_merge_failed(
                    "eng-1",
                    "43",
                    Some(MergeMode::IsolatedIntegration),
                    "isolated merge path failed: boom",
                ),
            ),
            (
                "merge_confidence_scored",
                TeamEvent::merge_confidence_scored(&MergeConfidenceInfo {
                    engineer: "eng-1",
                    task: "42",
                    confidence: 0.85,
                    files_changed: 3,
                    lines_changed: 50,
                    has_migrations: false,
                    has_config_changes: false,
                    rename_count: 0,
                }),
            ),
            (
                "review_nudge_sent",
                TeamEvent::review_nudge_sent("manager", "42"),
            ),
            (
                "review_escalated",
                TeamEvent::review_escalated("42", "stale review"),
            ),
            (
                "pipeline_starvation_detected",
                TeamEvent::pipeline_starvation_detected(3, 0),
            ),
            (
                "state_reconciliation",
                TeamEvent::state_reconciliation(Some("eng-1"), Some("42"), "adopt"),
            ),
            ("task_reworked", TeamEvent::task_reworked("eng-1", "42")),
            ("load_snapshot", TeamEvent::load_snapshot(2, 5, true)),
            (
                "parity_updated",
                TeamEvent::parity_updated(&crate::team::parity::ParitySummary {
                    total_behaviors: 10,
                    spec_complete: 8,
                    tests_complete: 6,
                    implementation_complete: 5,
                    verified_pass: 4,
                    verified_fail: 1,
                    overall_parity_pct: 40,
                }),
            ),
            (
                "release_succeeded",
                TeamEvent::release_succeeded(
                    "0.10.0",
                    "abc123",
                    "v0.10.0",
                    Some("/tmp/v0.10.0.md"),
                ),
            ),
            (
                "release_failed",
                TeamEvent::release_failed(
                    Some("0.10.0"),
                    Some("abc123"),
                    Some("v0.10.0"),
                    "verification failed",
                    None,
                ),
            ),
            (
                "worktree_reconciled",
                TeamEvent::worktree_reconciled("eng-1", "eng-1/42"),
            ),
        ];
        for (expected_event, event) in &variants {
            let json = serde_json::to_string(event).unwrap();
            let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
            assert_eq!(parsed["event"].as_str().unwrap(), *expected_event);
            assert!(parsed["ts"].as_u64().is_some());
        }
    }

    #[test]
    fn load_snapshot_serializes_optional_metrics() {
        let event = TeamEvent::load_snapshot(3, 7, false);
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"event\":\"load_snapshot\""));
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["load"].as_f64().unwrap(), 3.0 / 7.0);
        assert_eq!(parsed["working_members"].as_u64().unwrap(), 3);
        assert_eq!(parsed["total_members"].as_u64().unwrap(), 7);
        assert!(!parsed["session_running"].as_bool().unwrap());
    }

    #[test]
    fn parity_updated_serializes_summary_metrics() {
        let event = TeamEvent::parity_updated(&crate::team::parity::ParitySummary {
            total_behaviors: 10,
            spec_complete: 8,
            tests_complete: 6,
            implementation_complete: 5,
            verified_pass: 4,
            verified_fail: 1,
            overall_parity_pct: 40,
        });
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "parity_updated");
        assert_eq!(parsed["load"].as_f64().unwrap(), 0.4);
        let reason = parsed["reason"].as_str().unwrap();
        assert!(reason.contains("total=10"));
        assert!(reason.contains("spec=8"));
        assert!(reason.contains("verified_pass=4"));
    }

    #[test]
    fn release_succeeded_serializes_version_and_git_ref() {
        let event =
            TeamEvent::release_succeeded("0.10.0", "abc123", "v0.10.0", Some("/tmp/v0.10.0.md"));
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "release_succeeded");
        assert_eq!(parsed["version"].as_str().unwrap(), "0.10.0");
        assert_eq!(parsed["git_ref"].as_str().unwrap(), "abc123");
        assert_eq!(parsed["task"].as_str().unwrap(), "v0.10.0");
        assert!(parsed["success"].as_bool().unwrap());
    }

    #[test]
    fn release_failed_serializes_reason_and_attempted_tag() {
        let event = TeamEvent::release_failed(
            Some("0.10.0"),
            Some("abc123"),
            Some("v0.10.0"),
            "verification_failed",
            Some("suite::it_breaks"),
        );
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "release_failed");
        assert_eq!(parsed["version"].as_str().unwrap(), "0.10.0");
        assert_eq!(parsed["git_ref"].as_str().unwrap(), "abc123");
        assert_eq!(parsed["task"].as_str().unwrap(), "v0.10.0");
        assert_eq!(parsed["reason"].as_str().unwrap(), "verification_failed");
        assert_eq!(parsed["details"].as_str().unwrap(), "suite::it_breaks");
        assert!(!parsed["success"].as_bool().unwrap());
    }

    #[test]
    fn read_events_parses_all_known_lines() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let mut sink = EventSink::new(&path).unwrap();
        sink.emit(TeamEvent::daemon_started()).unwrap();
        sink.emit(TeamEvent::load_snapshot(1, 4, true)).unwrap();
        sink.emit(TeamEvent::load_snapshot(2, 4, true)).unwrap();

        let events = read_events(&path).unwrap();
        assert_eq!(events.len(), 3);
        assert_eq!(events[1].event, "load_snapshot");
        assert_eq!(events[1].working_members, Some(1));
        assert_eq!(events[2].total_members, Some(4));
    }

    #[test]
    fn event_sink_appends_to_existing_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");

        // Write one event and close the sink
        {
            let mut sink = EventSink::new(&path).unwrap();
            sink.emit(TeamEvent::daemon_started()).unwrap();
        }

        // Open again and append another
        {
            let mut sink = EventSink::new(&path).unwrap();
            sink.emit(TeamEvent::daemon_stopped()).unwrap();
        }

        let content = std::fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = content.lines().collect();
        assert_eq!(lines.len(), 2);
        assert!(lines[0].contains("daemon_started"));
        assert!(lines[1].contains("daemon_stopped"));
    }

    #[test]
    fn event_with_special_characters_in_task() {
        let event = TeamEvent::task_assigned("eng-1", "fix: \"quotes\" & <angles> / \\ newline\n");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        let task_val = parsed["task"].as_str().unwrap();
        assert!(task_val.contains("\"quotes\""));
        assert!(task_val.contains("<angles>"));
    }

    #[test]
    fn task_escalated_serializes_role_and_task() {
        let event = TeamEvent::task_escalated("eng-1-1", "42", Some("tests_failed"));
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"event\":\"task_escalated\""));
        assert!(json.contains("\"role\":\"eng-1-1\""));
        assert!(json.contains("\"task\":\"42\""));
        assert!(json.contains("\"reason\":\"tests_failed\""));
    }

    #[test]
    fn cwd_corrected_serializes_role_and_reason() {
        let event = TeamEvent::cwd_corrected("eng-1-1", "/tmp/worktree");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "cwd_corrected");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-1");
        assert_eq!(parsed["reason"].as_str().unwrap(), "/tmp/worktree");
    }

    #[test]
    fn pane_death_serializes_role() {
        let event = TeamEvent::pane_death("eng-1-1");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "pane_death");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-1");
    }

    #[test]
    fn pane_respawned_serializes_role() {
        let event = TeamEvent::pane_respawned("eng-1-1");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "pane_respawned");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-1");
    }

    #[test]
    fn agent_restarted_includes_reason_task_and_count() {
        let event = TeamEvent::agent_restarted("eng-1-2", "67", "context_exhausted", 2);
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "agent_restarted");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-2");
        assert_eq!(parsed["task"].as_str().unwrap(), "67");
        assert_eq!(parsed["reason"].as_str().unwrap(), "context_exhausted");
        assert_eq!(parsed["restart_count"].as_u64().unwrap(), 2);
    }

    #[test]
    fn context_pressure_warning_includes_threshold_and_output_bytes() {
        let event = TeamEvent::context_pressure_warning("eng-1-2", Some(67), 91, 100, 420_000);
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();

        assert_eq!(
            parsed["event"].as_str().unwrap(),
            "context_pressure_warning"
        );
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-2");
        assert_eq!(parsed["task"].as_str().unwrap(), "67");
        assert_eq!(parsed["load"].as_f64().unwrap(), 91.0);
        assert_eq!(parsed["output_bytes"].as_u64().unwrap(), 420_000);
        assert_eq!(parsed["reason"].as_str().unwrap(), "threshold=100");
    }

    #[test]
    fn tact_tasks_created_includes_latency_status_and_created_count() {
        let event = TeamEvent::tact_tasks_created("architect", 4, 19, false, Some("parse failed"));
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed["event"].as_str().unwrap(), "tact_tasks_created");
        assert_eq!(parsed["role"].as_str().unwrap(), "architect");
        assert_eq!(parsed["restart_count"].as_u64().unwrap(), 4);
        assert_eq!(parsed["uptime_secs"].as_u64().unwrap(), 19);
        assert_eq!(parsed["reason"].as_str().unwrap(), "failure");
        assert_eq!(parsed["error"].as_str().unwrap(), "parse failed");
    }

    #[test]
    fn board_task_archived_includes_task_and_role() {
        let event = TeamEvent::board_task_archived("88", Some("eng-1-2"));
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed["event"].as_str().unwrap(), "board_task_archived");
        assert_eq!(parsed["task"].as_str().unwrap(), "88");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-2");
    }

    #[test]
    fn delivery_failed_includes_role_sender_and_reason() {
        let event = TeamEvent::delivery_failed("eng-1-2", "manager", "message marker missing");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "delivery_failed");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-2");
        assert_eq!(parsed["from"].as_str().unwrap(), "manager");
        assert_eq!(parsed["reason"].as_str().unwrap(), "message marker missing");
    }

    #[test]
    fn delivery_failed_with_details_serializes_actionable_context() {
        let event = TeamEvent::delivery_failed_with_details(
            "eng-1-2",
            "manager",
            "recipient has no live shim handle",
            Some("classification=missing_shim queue_action=cleared"),
        );
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(
            parsed["details"].as_str().unwrap(),
            "classification=missing_shim queue_action=cleared"
        );
    }

    #[test]
    fn task_unblocked_serializes_role_and_task() {
        let event = TeamEvent::task_unblocked("eng-1-1", "42");
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"event\":\"task_unblocked\""));
        assert!(json.contains("\"role\":\"eng-1-1\""));
        assert!(json.contains("\"task\":\"42\""));
    }

    #[test]
    fn merge_confidence_scored_includes_all_fields() {
        let event = TeamEvent::merge_confidence_scored(&MergeConfidenceInfo {
            engineer: "eng-1-1",
            task: "42",
            confidence: 0.85,
            files_changed: 3,
            lines_changed: 50,
            has_migrations: true,
            has_config_changes: false,
            rename_count: 1,
        });
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "merge_confidence_scored");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-1");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
        assert!((parsed["load"].as_f64().unwrap() - 0.85).abs() < 0.001);
        let reason = parsed["reason"].as_str().unwrap();
        assert!(reason.contains("files=3"));
        assert!(reason.contains("lines=50"));
        assert!(reason.contains("migrations=true"));
        assert!(reason.contains("config=false"));
        assert!(reason.contains("renames=1"));
    }

    #[test]
    fn auto_merge_decision_recorded_serializes_reason_and_details() {
        let event = TeamEvent::auto_merge_decision_recorded(&AutoMergeDecisionInfo {
            engineer: "eng-1-1",
            task: "42",
            action_type: "manual_review",
            confidence: 0.55,
            reason: "routed to manual review: confidence 0.55; 4 files, 90 lines, 3 modules; reasons: touches sensitive paths",
            details: "{\"decision\":\"manual_review\",\"reasons\":[\"touches sensitive paths\"]}",
        });
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(
            parsed["event"].as_str().unwrap(),
            "auto_merge_decision_recorded"
        );
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1-1");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
        assert_eq!(parsed["action_type"].as_str().unwrap(), "manual_review");
        assert!((parsed["load"].as_f64().unwrap() - 0.55).abs() < 0.001);
        assert!(parsed["reason"].as_str().unwrap().contains("manual review"));
        assert!(
            parsed["details"]
                .as_str()
                .unwrap()
                .contains("\"touches sensitive paths\"")
        );
    }

    #[test]
    fn auto_merge_post_verify_result_serializes_success() {
        let event = TeamEvent::auto_merge_post_verify_result(
            "eng-1",
            "42",
            Some(false),
            "failed",
            Some("post-merge verification on main failed"),
        );
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(
            parsed["event"].as_str().unwrap(),
            "auto_merge_post_verify_result"
        );
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
        assert!(!parsed["success"].as_bool().unwrap());
        assert_eq!(parsed["reason"].as_str().unwrap(), "failed");
    }

    #[test]
    fn merge_events_serialize_merge_mode_and_outcome() {
        let auto = TeamEvent::task_auto_merged_with_mode(
            "eng-1",
            "42",
            0.95,
            2,
            30,
            Some(MergeMode::DirectRoot),
        );
        let auto_json: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&auto).unwrap()).unwrap();
        assert_eq!(auto_json["merge_mode"].as_str().unwrap(), "direct_root");
        assert!(auto_json["success"].as_bool().unwrap());

        let manual =
            TeamEvent::task_manual_merged_with_mode("43", Some(MergeMode::IsolatedIntegration));
        let manual_json: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&manual).unwrap()).unwrap();
        assert_eq!(
            manual_json["merge_mode"].as_str().unwrap(),
            "isolated_integration"
        );
        assert!(manual_json["success"].as_bool().unwrap());

        let failed = TeamEvent::task_merge_failed(
            "eng-2",
            "44",
            Some(MergeMode::IsolatedIntegration),
            "isolated merge path failed: broken",
        );
        let failed_json: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&failed).unwrap()).unwrap();
        assert_eq!(
            failed_json["merge_mode"].as_str().unwrap(),
            "isolated_integration"
        );
        assert!(!failed_json["success"].as_bool().unwrap());
        assert_eq!(failed_json["reason"].as_str().unwrap(), "merge_failed");
    }

    #[test]
    fn performance_regression_serializes_task_and_reason() {
        let event = TeamEvent::performance_regression("42", "runtime_ms=1300 avg_ms=1000 pct=30");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "performance_regression");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
        assert_eq!(
            parsed["reason"].as_str().unwrap(),
            "runtime_ms=1300 avg_ms=1000 pct=30"
        );
    }

    #[test]
    fn daemon_stopped_with_reason_includes_fields() {
        let event = TeamEvent::daemon_stopped_with_reason("signal", 7200);
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["reason"].as_str().unwrap(), "signal");
        assert_eq!(parsed["uptime_secs"].as_u64().unwrap(), 7200);
    }

    #[test]
    fn heartbeat_includes_uptime() {
        let event = TeamEvent::daemon_heartbeat(600);
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "daemon_heartbeat");
        assert_eq!(parsed["uptime_secs"].as_u64().unwrap(), 600);
        // No reason/step/error fields
        assert!(parsed.get("reason").is_none());
        assert!(parsed.get("step").is_none());
    }

    #[test]
    fn context_exhausted_includes_role_task_and_session_size() {
        let event = TeamEvent::context_exhausted("eng-1", Some(77), Some(4096));
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "context_exhausted");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1");
        assert_eq!(parsed["task"].as_str().unwrap(), "77");
        assert_eq!(parsed["session_size_bytes"].as_u64().unwrap(), 4096);
    }

    #[test]
    fn loop_step_error_includes_step_and_error() {
        let event = TeamEvent::loop_step_error("poll_watchers", "connection refused");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["step"].as_str().unwrap(), "poll_watchers");
        assert_eq!(parsed["error"].as_str().unwrap(), "connection refused");
    }

    #[test]
    fn daemon_panic_includes_reason() {
        let event = TeamEvent::daemon_panic("index out of bounds");
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "daemon_panic");
        assert_eq!(parsed["reason"].as_str().unwrap(), "index out of bounds");
    }

    #[test]
    fn new_fields_omitted_from_basic_events() {
        let event = TeamEvent::daemon_started();
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("\"reason\""));
        assert!(!json.contains("\"step\""));
        assert!(!json.contains("\"error\""));
        assert!(!json.contains("\"uptime_secs\""));
        assert!(!json.contains("\"restart_count\""));
    }

    #[test]
    fn pattern_detected_includes_reason_payload() {
        let event = TeamEvent::pattern_detected("escalation_cluster", 6);
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "pattern_detected");
        assert_eq!(parsed["reason"].as_str().unwrap(), "escalation_cluster:6");
    }

    #[test]
    fn event_sink_creates_parent_directories() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("deep").join("nested").join("events.jsonl");
        let mut sink = EventSink::new(&path).unwrap();
        sink.emit(TeamEvent::daemon_started()).unwrap();
        assert!(path.exists());
        assert_eq!(sink.path(), path);
    }

    #[test]
    fn event_sink_rotates_oversized_log_on_open() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        fs::write(&path, "0123456789").unwrap();

        let mut sink = EventSink::new_with_max_bytes(&path, 5).unwrap();
        sink.emit(TeamEvent::daemon_started()).unwrap();

        let rotated = rotated_event_log_path(&path);
        assert_eq!(fs::read_to_string(&rotated).unwrap(), "0123456789");
        let current = fs::read_to_string(&path).unwrap();
        assert!(current.contains("daemon_started"));
    }

    #[test]
    fn event_sink_rotates_before_write_that_would_exceed_threshold() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let first_line = "{\"event\":\"first\"}\n";
        fs::write(&path, first_line).unwrap();

        let mut sink = EventSink::new_with_max_bytes(&path, first_line.len() as u64 + 10).unwrap();
        sink.emit(TeamEvent::task_assigned(
            "eng-1",
            "this assignment is long enough to rotate",
        ))
        .unwrap();

        let rotated = rotated_event_log_path(&path);
        assert_eq!(fs::read_to_string(&rotated).unwrap(), first_line);
        let current = fs::read_to_string(&path).unwrap();
        assert!(current.contains("task_assigned"));
        assert!(!current.contains("\"event\":\"first\""));
    }

    #[test]
    fn event_round_trip_preserves_fields_for_agent_restarted() {
        let original = TeamEvent::agent_restarted("eng-1", "42", "context_exhausted", 3);

        let json = serde_json::to_string(&original).unwrap();
        let parsed: TeamEvent = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.event, "agent_restarted");
        assert_eq!(parsed.role.as_deref(), Some("eng-1"));
        assert_eq!(parsed.task.as_deref(), Some("42"));
        assert_eq!(parsed.reason.as_deref(), Some("context_exhausted"));
        assert_eq!(parsed.restart_count, Some(3));
        assert_eq!(parsed.ts, original.ts);
    }

    #[test]
    fn event_round_trip_preserves_fields_for_task_resumed() {
        let original = TeamEvent::task_resumed("eng-1", "42", "context_pressure", 1);

        let json = serde_json::to_string(&original).unwrap();
        let parsed: TeamEvent = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.event, "task_resumed");
        assert_eq!(parsed.role.as_deref(), Some("eng-1"));
        assert_eq!(parsed.task.as_deref(), Some("42"));
        assert_eq!(parsed.reason.as_deref(), Some("context_pressure"));
        assert_eq!(parsed.restart_count, Some(1));
        assert_eq!(parsed.ts, original.ts);
    }

    #[test]
    fn event_round_trip_preserves_fields_for_load_snapshot() {
        let original = TeamEvent::load_snapshot(4, 8, true);

        let json = serde_json::to_string(&original).unwrap();
        let parsed: TeamEvent = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.event, "load_snapshot");
        assert_eq!(parsed.working_members, Some(4));
        assert_eq!(parsed.total_members, Some(8));
        assert_eq!(parsed.session_running, Some(true));
        assert_eq!(parsed.load, Some(0.5));
        assert_eq!(parsed.ts, original.ts);
    }

    #[test]
    fn event_round_trip_preserves_fields_for_delivery_failed() {
        let original = TeamEvent::delivery_failed("eng-2", "manager", "marker missing");

        let json = serde_json::to_string(&original).unwrap();
        let parsed: TeamEvent = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.event, "delivery_failed");
        assert_eq!(parsed.role.as_deref(), Some("eng-2"));
        assert_eq!(parsed.from.as_deref(), Some("manager"));
        assert_eq!(parsed.reason.as_deref(), Some("marker missing"));
        assert!(parsed.details.is_none());
        assert_eq!(parsed.ts, original.ts);
    }

    #[test]
    fn read_events_skips_blank_and_malformed_lines() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        fs::write(
            &path,
            [
                "",
                "{\"event\":\"daemon_started\",\"ts\":1}",
                "not-json",
                "   ",
                "{\"event\":\"daemon_stopped\",\"ts\":2}",
            ]
            .join("\n"),
        )
        .unwrap();

        let events = read_events(&path).unwrap();

        assert_eq!(events.len(), 2);
        assert_eq!(events[0].event, "daemon_started");
        assert_eq!(events[1].event, "daemon_stopped");
    }

    #[test]
    fn rotate_event_log_if_needed_returns_false_for_missing_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");

        let rotated = rotate_event_log_if_needed(&path, 128, 0).unwrap();

        assert!(!rotated);
        assert!(!rotated_event_log_path(&path).exists());
    }

    #[test]
    fn rotate_event_log_if_needed_returns_false_for_empty_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        fs::write(&path, "").unwrap();

        let rotated = rotate_event_log_if_needed(&path, 1, 1).unwrap();

        assert!(!rotated);
        assert!(path.exists());
        assert!(!rotated_event_log_path(&path).exists());
    }

    #[test]
    fn rotate_event_log_if_needed_replaces_existing_rotated_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let rotated_path = rotated_event_log_path(&path);
        fs::write(&path, "current-events").unwrap();
        fs::write(&rotated_path, "old-rotated-events").unwrap();

        let rotated = rotate_event_log_if_needed(&path, 5, 0).unwrap();

        assert!(rotated);
        assert_eq!(fs::read_to_string(&rotated_path).unwrap(), "current-events");
    }

    #[test]
    fn concurrent_event_sinks_append_without_losing_lines() {
        let tmp = tempfile::tempdir().unwrap();
        let path = Arc::new(tmp.path().join("events.jsonl"));
        let ready = Arc::new(std::sync::Barrier::new(5));
        let errors = Arc::new(Mutex::new(Vec::<String>::new()));
        let mut handles = Vec::new();

        for idx in 0..4 {
            let path = Arc::clone(&path);
            let ready = Arc::clone(&ready);
            let errors = Arc::clone(&errors);
            handles.push(thread::spawn(move || {
                ready.wait();
                let result = (|| -> Result<()> {
                    let mut sink = EventSink::new(&path)?;
                    sink.emit(TeamEvent::task_assigned(
                        &format!("eng-{idx}"),
                        &format!("task-{idx}"),
                    ))?;
                    Ok(())
                })();
                if let Err(error) = result {
                    errors.lock().unwrap().push(error.to_string());
                }
            }));
        }

        ready.wait();
        for handle in handles {
            handle.join().unwrap();
        }

        assert!(errors.lock().unwrap().is_empty());
        let events = read_events(&path).unwrap();
        assert_eq!(events.len(), 4);
        for idx in 0..4 {
            assert!(
                events
                    .iter()
                    .any(|event| event.role.as_deref() == Some(&format!("eng-{idx}")))
            );
        }
    }

    #[test]
    fn read_events_handles_large_log_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let mut sink = EventSink::new(&path).unwrap();

        for idx in 0..512 {
            sink.emit(TeamEvent::task_assigned(
                &format!("eng-{idx}"),
                &"x".repeat(128),
            ))
            .unwrap();
        }

        let events = read_events(&path).unwrap();

        assert_eq!(events.len(), 512);
        assert_eq!(events.first().unwrap().event, "task_assigned");
        assert_eq!(events.last().unwrap().event, "task_assigned");
    }

    fn production_unwrap_expect_count(source: &str) -> usize {
        let prod = if let Some(pos) = source.find("\n#[cfg(test)]\nmod tests") {
            &source[..pos]
        } else {
            source
        };
        prod.lines()
            .filter(|line| {
                let trimmed = line.trim();
                !trimmed.starts_with("#[cfg(test)]")
                    && (trimmed.contains(".unwrap(") || trimmed.contains(".expect("))
            })
            .count()
    }

    #[test]
    fn task_completed_includes_task_id() {
        let event = TeamEvent::task_completed("eng-1", Some("42"));
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "task_completed");
        assert_eq!(parsed["role"].as_str().unwrap(), "eng-1");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
    }

    #[test]
    fn task_completed_without_task_id_omits_task_field() {
        let event = TeamEvent::task_completed("eng-1", None);
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"event\":\"task_completed\""));
        assert!(json.contains("\"role\":\"eng-1\""));
        assert!(!json.contains("\"task\""));
    }

    #[test]
    fn task_escalated_without_reason_omits_reason_field() {
        let event = TeamEvent::task_escalated("eng-1", "42", None);
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "task_escalated");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
        assert!(parsed.get("reason").is_none());
    }

    #[test]
    fn task_escalated_with_reason_includes_reason_field() {
        let event = TeamEvent::task_escalated("eng-1", "42", Some("merge_conflict"));
        let json = serde_json::to_string(&event).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["event"].as_str().unwrap(), "task_escalated");
        assert_eq!(parsed["task"].as_str().unwrap(), "42");
        assert_eq!(parsed["reason"].as_str().unwrap(), "merge_conflict");
    }

    #[test]
    fn task_completed_round_trip_preserves_task_id() {
        let original = TeamEvent::task_completed("eng-1", Some("99"));
        let json = serde_json::to_string(&original).unwrap();
        let parsed: TeamEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.event, "task_completed");
        assert_eq!(parsed.role.as_deref(), Some("eng-1"));
        assert_eq!(parsed.task.as_deref(), Some("99"));
    }

    #[test]
    fn task_escalated_round_trip_preserves_reason() {
        let original = TeamEvent::task_escalated("eng-1", "42", Some("context_exhausted"));
        let json = serde_json::to_string(&original).unwrap();
        let parsed: TeamEvent = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed.event, "task_escalated");
        assert_eq!(parsed.role.as_deref(), Some("eng-1"));
        assert_eq!(parsed.task.as_deref(), Some("42"));
        assert_eq!(parsed.reason.as_deref(), Some("context_exhausted"));
    }

    #[test]
    fn production_events_has_no_unwrap_or_expect_calls() {
        let src = include_str!("events.rs");
        assert_eq!(
            production_unwrap_expect_count(src),
            0,
            "production events.rs should avoid unwrap/expect"
        );
    }

    #[test]
    fn stall_detected_event_fields() {
        let event = TeamEvent::stall_detected("eng-1-1", Some(42), 300);
        assert_eq!(event.event, "stall_detected");
        assert_eq!(event.role.as_deref(), Some("eng-1-1"));
        assert_eq!(event.task.as_deref(), Some("42"));
        assert_eq!(event.uptime_secs, Some(300));
    }

    #[test]
    fn stall_detected_event_without_task() {
        let event = TeamEvent::stall_detected("eng-1-1", None, 600);
        assert_eq!(event.event, "stall_detected");
        assert_eq!(event.role.as_deref(), Some("eng-1-1"));
        assert!(event.task.is_none());
        assert_eq!(event.uptime_secs, Some(600));
    }

    #[test]
    fn stall_detected_event_with_reason_records_heuristic() {
        let event = TeamEvent::stall_detected_with_reason(
            "lead",
            None,
            300,
            Some("supervisory_stalled_manager_shim_activity_only"),
        );
        assert_eq!(
            event.reason.as_deref(),
            Some("supervisory_stalled_manager_shim_activity_only")
        );
    }

    #[test]
    fn stall_detected_event_serializes_to_jsonl() {
        let event = TeamEvent::stall_detected("eng-1-1", Some(42), 300);
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"stall_detected\""));
        assert!(json.contains("\"eng-1-1\""));
        assert!(json.contains("\"42\""));
    }

    #[test]
    fn health_changed_event_fields() {
        let event = TeamEvent::health_changed("eng-1-1", "healthy→unreachable");
        assert_eq!(event.event, "health_changed");
        assert_eq!(event.role.as_deref(), Some("eng-1-1"));
        assert_eq!(event.reason.as_deref(), Some("healthy→unreachable"));
    }

    #[test]
    fn health_changed_event_serializes_to_jsonl() {
        let event = TeamEvent::health_changed("eng-1-2", "unreachable→healthy");
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"health_changed\""));
        assert!(json.contains("\"eng-1-2\""));
    }

    // --- Error path and recovery tests (Task #265) ---

    #[test]
    fn event_sink_on_readonly_dir_returns_error() {
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let tmp = tempfile::tempdir().unwrap();
            let readonly_dir = tmp.path().join("readonly");
            fs::create_dir(&readonly_dir).unwrap();
            fs::set_permissions(&readonly_dir, fs::Permissions::from_mode(0o444)).unwrap();

            let path = readonly_dir.join("subdir").join("events.jsonl");
            let result = EventSink::new(&path);
            assert!(result.is_err());

            // Restore permissions for cleanup
            fs::set_permissions(&readonly_dir, fs::Permissions::from_mode(0o755)).unwrap();
        }
    }

    #[test]
    fn read_events_from_nonexistent_file_returns_empty() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("does_not_exist.jsonl");
        let events = read_events(&path).unwrap();
        assert!(events.is_empty());
    }

    #[test]
    fn read_events_all_malformed_lines_returns_empty() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        fs::write(&path, "not json\nalso not json\n{invalid}\n").unwrap();
        let events = read_events(&path).unwrap();
        assert!(events.is_empty());
    }

    #[test]
    fn event_sink_emit_with_failing_writer() {
        struct FailWriter;
        impl Write for FailWriter {
            fn write(&mut self, _buf: &[u8]) -> std::io::Result<usize> {
                Err(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "simulated write failure",
                ))
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Err(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "simulated flush failure",
                ))
            }
        }

        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let mut sink = EventSink::from_writer(&path, FailWriter);
        let result = sink.emit(TeamEvent::daemon_started());
        assert!(result.is_err());
    }

    #[test]
    fn rotate_event_log_replaces_stale_rotated_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        let rotated = rotated_event_log_path(&path);

        fs::write(&path, "current-data-that-is-large").unwrap();
        fs::write(&rotated, "old-rotated-data").unwrap();

        let did_rotate = rotate_event_log_if_needed(&path, 5, 0).unwrap();
        assert!(did_rotate);
        // Old rotated was replaced with current data
        assert_eq!(
            fs::read_to_string(&rotated).unwrap(),
            "current-data-that-is-large"
        );
        // Current file is now gone (rotated away)
        assert!(!path.exists());
    }

    #[test]
    fn event_sink_handles_zero_max_bytes_rotation() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");

        // With max_bytes=0, any existing content triggers rotation, but empty file doesn't
        fs::write(&path, "").unwrap();
        let did_rotate = rotate_event_log_if_needed(&path, 0, 0).unwrap();
        assert!(!did_rotate); // empty file → no rotation

        fs::write(&path, "x").unwrap();
        let did_rotate = rotate_event_log_if_needed(&path, 0, 0).unwrap();
        assert!(did_rotate); // non-empty file at 0-byte limit → rotation
    }

    #[test]
    fn narration_rejection_event_has_correct_fields() {
        let event = TeamEvent::narration_rejection("eng-1-1", 42, 2);
        assert_eq!(event.event, "narration_rejection");
        assert_eq!(event.role.as_deref(), Some("eng-1-1"));
        assert_eq!(event.task.as_deref(), Some("42"));
        assert_eq!(event.reason.as_deref(), Some("rejection_count=2"));
    }

    #[test]
    fn narration_events_have_correct_fields() {
        let detected = TeamEvent::narration_detected("eng-1-4", Some(415));
        assert_eq!(detected.event, "narration_detected");
        assert_eq!(detected.role.as_deref(), Some("eng-1-4"));
        assert_eq!(detected.task.as_deref(), Some("415"));

        let nudged = TeamEvent::narration_nudged("eng-1-4", Some(415));
        assert_eq!(nudged.event, "narration_nudged");
        assert_eq!(nudged.role.as_deref(), Some("eng-1-4"));
        assert_eq!(nudged.task.as_deref(), Some("415"));

        let restarted = TeamEvent::narration_restart("eng-1-4", Some(415));
        assert_eq!(restarted.event, "narration_restart");
        assert_eq!(restarted.role.as_deref(), Some("eng-1-4"));
        assert_eq!(restarted.task.as_deref(), Some("415"));
    }

    #[test]
    fn read_events_partial_json_with_valid_lines_mixed() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");
        // Simulate a truncated write: valid JSON, then partial, then valid
        let content = format!(
            "{}\n{{\"event\":\"trunca\n{}\n",
            r#"{"event":"daemon_started","ts":1}"#, r#"{"event":"daemon_stopped","ts":3}"#
        );
        fs::write(&path, content).unwrap();

        let events = read_events(&path).unwrap();
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].event, "daemon_started");
        assert_eq!(events[1].event, "daemon_stopped");
    }

    // -----------------------------------------------------------------------
    // Lifecycle event tests
    // -----------------------------------------------------------------------

    #[test]
    fn session_lifecycle_serialization_roundtrip() {
        let variants = [
            SessionLifecycle::Spawning,
            SessionLifecycle::Ready,
            SessionLifecycle::Working,
            SessionLifecycle::Blocked,
            SessionLifecycle::Idle,
            SessionLifecycle::Finished,
            SessionLifecycle::Failed,
        ];
        for variant in &variants {
            let json = serde_json::to_string(variant).unwrap();
            let deserialized: SessionLifecycle = serde_json::from_str(&json).unwrap();
            assert_eq!(*variant, deserialized);
        }
    }

    #[test]
    fn task_lifecycle_serialization_roundtrip() {
        let variants = [
            TaskLifecycle::Claimed,
            TaskLifecycle::InProgress,
            TaskLifecycle::TestsRunning,
            TaskLifecycle::TestsPassed,
            TaskLifecycle::TestsFailed,
            TaskLifecycle::Review,
            TaskLifecycle::Merged,
            TaskLifecycle::Rejected,
        ];
        for variant in &variants {
            let json = serde_json::to_string(variant).unwrap();
            let deserialized: TaskLifecycle = serde_json::from_str(&json).unwrap();
            assert_eq!(*variant, deserialized);
        }
    }

    #[test]
    fn merge_lifecycle_serialization_roundtrip() {
        let variants = [
            MergeLifecycle::Started,
            MergeLifecycle::Conflict,
            MergeLifecycle::Success,
        ];
        for variant in &variants {
            let json = serde_json::to_string(variant).unwrap();
            let deserialized: MergeLifecycle = serde_json::from_str(&json).unwrap();
            assert_eq!(*variant, deserialized);
        }
    }

    #[test]
    fn lifecycle_event_serialization_roundtrip() {
        let event = LifecycleEvent::new(
            LifecycleEventType::Session(SessionLifecycle::Ready),
            "eng-1-1",
            Some(42),
            Some("agent is ready".into()),
        );
        let json = serde_json::to_string(&event).unwrap();
        let deserialized: LifecycleEvent = serde_json::from_str(&json).unwrap();

        assert_eq!(deserialized.schema_version, LifecycleEvent::SCHEMA_VERSION);
        assert_eq!(deserialized.agent_name, "eng-1-1");
        assert_eq!(deserialized.task_id, Some(42));
        assert_eq!(deserialized.details.as_deref(), Some("agent is ready"));
        assert_eq!(
            deserialized.event_type,
            LifecycleEventType::Session(SessionLifecycle::Ready)
        );
    }

    #[test]
    fn lifecycle_event_optional_fields_omitted() {
        let event = LifecycleEvent::new(
            LifecycleEventType::Task(TaskLifecycle::Claimed),
            "eng-1-2",
            None,
            None,
        );
        let json = serde_json::to_string(&event).unwrap();
        assert!(!json.contains("task_id"));
        assert!(!json.contains("details"));
        assert!(json.contains("\"agent_name\":\"eng-1-2\""));
    }

    #[test]
    fn lifecycle_event_type_tagged_serialization() {
        // Session variant
        let session = LifecycleEventType::Session(SessionLifecycle::Working);
        let json = serde_json::to_string(&session).unwrap();
        assert!(json.contains("\"category\":\"session\""));
        assert!(json.contains("\"phase\":\"working\""));

        // Task variant
        let task = LifecycleEventType::Task(TaskLifecycle::TestsRunning);
        let json = serde_json::to_string(&task).unwrap();
        assert!(json.contains("\"category\":\"task\""));
        assert!(json.contains("\"phase\":\"tests_running\""));

        // Merge variant
        let merge = LifecycleEventType::Merge(MergeLifecycle::Conflict);
        let json = serde_json::to_string(&merge).unwrap();
        assert!(json.contains("\"category\":\"merge\""));
        assert!(json.contains("\"phase\":\"conflict\""));
    }

    #[test]
    fn emit_lifecycle_event_writes_to_file() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");

        let event = LifecycleEvent::new(
            LifecycleEventType::Session(SessionLifecycle::Spawning),
            "eng-1-1",
            None,
            Some("starting up".into()),
        );
        emit_lifecycle_event(&path, &event).unwrap();

        let content = fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = content.lines().collect();
        assert_eq!(lines.len(), 1);

        let parsed: LifecycleEvent = serde_json::from_str(lines[0]).unwrap();
        assert_eq!(parsed.agent_name, "eng-1-1");
        assert_eq!(
            parsed.event_type,
            LifecycleEventType::Session(SessionLifecycle::Spawning)
        );
    }

    #[test]
    fn emit_lifecycle_event_appends_multiple() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("events.jsonl");

        let events = [
            LifecycleEvent::new(
                LifecycleEventType::Task(TaskLifecycle::Claimed),
                "eng-1-1",
                Some(100),
                None,
            ),
            LifecycleEvent::new(
                LifecycleEventType::Task(TaskLifecycle::InProgress),
                "eng-1-1",
                Some(100),
                None,
            ),
            LifecycleEvent::new(
                LifecycleEventType::Task(TaskLifecycle::TestsPassed),
                "eng-1-1",
                Some(100),
                Some("all 42 tests passed".into()),
            ),
        ];
        for event in &events {
            emit_lifecycle_event(&path, event).unwrap();
        }

        let content = fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = content.lines().filter(|l| !l.is_empty()).collect();
        assert_eq!(lines.len(), 3);

        let parsed: LifecycleEvent = serde_json::from_str(lines[2]).unwrap();
        assert_eq!(
            parsed.event_type,
            LifecycleEventType::Task(TaskLifecycle::TestsPassed)
        );
        assert_eq!(parsed.details.as_deref(), Some("all 42 tests passed"));
    }

    #[test]
    fn emit_lifecycle_event_creates_parent_dirs() {
        let tmp = tempfile::tempdir().unwrap();
        let path = tmp.path().join("nested").join("dir").join("events.jsonl");

        let event = LifecycleEvent::new(
            LifecycleEventType::Merge(MergeLifecycle::Success),
            "eng-1-3",
            Some(200),
            None,
        );
        emit_lifecycle_event(&path, &event).unwrap();
        assert!(path.exists());
    }

    #[test]
    fn lifecycle_event_schema_version_is_one() {
        let event = LifecycleEvent::new(
            LifecycleEventType::Session(SessionLifecycle::Failed),
            "eng-1-1",
            None,
            Some("out of context".into()),
        );
        assert_eq!(event.schema_version, 1);
        let json = serde_json::to_string(&event).unwrap();
        assert!(json.contains("\"schema_version\":1"));
    }

    #[test]
    fn all_session_variants_snake_case_names() {
        let expected = [
            (SessionLifecycle::Spawning, "spawning"),
            (SessionLifecycle::Ready, "ready"),
            (SessionLifecycle::Working, "working"),
            (SessionLifecycle::Blocked, "blocked"),
            (SessionLifecycle::Idle, "idle"),
            (SessionLifecycle::Finished, "finished"),
            (SessionLifecycle::Failed, "failed"),
        ];
        for (variant, name) in &expected {
            let json = serde_json::to_string(variant).unwrap();
            assert_eq!(json, format!("\"{}\"", name));
        }
    }

    #[test]
    fn all_task_variants_snake_case_names() {
        let expected = [
            (TaskLifecycle::Claimed, "claimed"),
            (TaskLifecycle::InProgress, "in_progress"),
            (TaskLifecycle::TestsRunning, "tests_running"),
            (TaskLifecycle::TestsPassed, "tests_passed"),
            (TaskLifecycle::TestsFailed, "tests_failed"),
            (TaskLifecycle::Review, "review"),
            (TaskLifecycle::Merged, "merged"),
            (TaskLifecycle::Rejected, "rejected"),
        ];
        for (variant, name) in &expected {
            let json = serde_json::to_string(variant).unwrap();
            assert_eq!(json, format!("\"{}\"", name));
        }
    }

    #[test]
    fn all_merge_variants_snake_case_names() {
        let expected = [
            (MergeLifecycle::Started, "started"),
            (MergeLifecycle::Conflict, "conflict"),
            (MergeLifecycle::Success, "success"),
        ];
        for (variant, name) in &expected {
            let json = serde_json::to_string(variant).unwrap();
            assert_eq!(json, format!("\"{}\"", name));
        }
    }
}