//! OpenCode connector implementation
//!
//! This module provides a concrete connector implementation that wraps the
//! `opencode_client` library to provide OpenCode.ai integration for Dirigent.
//!
//! # Architecture
//!
//! The OpenCodeConnector:
//! - Wraps `opencode_client::OpenCodeClient` for REST API operations
//! - Manages SSE (Server-Sent Events) subscription for real-time updates
//! - Translates OpenCode events to Dirigent protocol events using `OpenCodeAdapter`
//! - Handles automatic reconnection with exponential backoff
//! - Processes commands (list sessions, send messages, etc.)
//!
//! # Lifecycle
//!
//! 1. Create connector with `OpenCodeConnector::new(id, owner, config)`
//! 2. Start the connector task with `start_task()` which returns a JoinHandle
//! 3. Send commands via the command channel
//! 4. Receive events via the event broadcast channel
//! 5. Stop with `Connector::stop()` which sends Shutdown command
//!
//! # Example
//!
//! ```no_run
//! use dirigent_core::connectors::opencode::{OpenCodeConnector, OpenCodeConfig};
//! use dirigent_core::connectors::{Connector, ConnectorCommand};
//!
//! # async fn example() {
//! let config = OpenCodeConfig {
//!     base_url: "http://localhost:12225".to_string(),
//!     title: "My OpenCode".to_string(),
//!     initial_session: None,
//! };
//!
//! let connector = OpenCodeConnector::new(
//!     "conn-1".to_string(),
//!     uuid::Uuid::now_v7(),
//!     "My OpenCode".to_string(),
//!     config,
//! );
//!
//! // Start the connector task
//! let task_handle = connector.start_task().await;
//!
//! // Subscribe to events
//! let mut events = connector.subscribe();
//!
//! // Send a command
//! let cmd_tx = connector.command_tx();
//! cmd_tx.send(ConnectorCommand::ListSessions).await.ok();
//!
//! // Later: stop the connector
//! connector.stop();
//! task_handle.await.ok();
//! # }
//! ```

use crate::connectors::{Connector, ConnectorCommand};
use crate::sharing::bus::SharingBus;
use crate::types::{ConnectorErrorKind, ConnectorId, ConnectorKind, ConnectorState, UserId};
use dirigent_protocol::adapters::OpenCodeAdapter;
use dirigent_protocol::{Event, TurnCompleteTrigger};
use futures::StreamExt;
use opencode_client::{OpenCodeClient, SseClient};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

/// Configuration for OpenCode connector
///
/// Contains all the necessary information to connect to an OpenCode.ai instance
/// and configure the connector's behavior.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct OpenCodeConfig {
    /// Base URL of the OpenCode API
    ///
    /// Example: "http://localhost:12225" or "https://api.opencode.ai"
    pub base_url: String,

    /// Optional initial session to load on connection
    ///
    /// If provided, the connector will attempt to load this session immediately
    /// after connecting. If None, no initial session is loaded.
    pub initial_session: Option<String>,
}

/// OpenCode connector implementation
///
/// Provides integration with OpenCode.ai by wrapping the `opencode_client` library
/// and implementing the `Connector` trait. This allows the Dirigent core runtime
/// to manage OpenCode connections alongside other connector types.
///
/// # Fields
///
/// The connector maintains both the client libraries and the channels needed for
/// async communication:
/// - REST client for API calls (list sessions, send messages, etc.)
/// - SSE client for real-time event streaming
/// - Command channel for receiving control commands
/// - Event broadcast channel for publishing events to subscribers
/// - Shared state for tracking connector lifecycle
pub struct OpenCodeConnector {
    /// Unique identifier for this connector instance
    id: ConnectorId,

    /// Optional connector UID used for BusEvent routing.
    ///
    /// Populated by the runtime when the connector is registered; legacy
    /// constructor paths default this to `None` and later wiring can fill it in.
    pub connector_uid: Option<Uuid>,

    /// User who owns this connector
    owner: UserId,

    /// Human-readable title for this connector
    ///
    /// Used in UI display and logging. This is stored separately from
    /// OpenCodeConfig as it's an orchestration-level field.
    title: String,

    /// Connector configuration
    config: OpenCodeConfig,

    /// Shared connector state (Initializing, Connecting, Ready, Error, Stopped)
    ///
    /// Protected by RwLock to allow concurrent reads and exclusive writes.
    /// The connector task updates this as it progresses through its lifecycle.
    state: Arc<RwLock<ConnectorState>>,

    /// Structured error classification for the current error state.
    ///
    /// Set alongside `ConnectorState::Error` to provide machine-readable error
    /// categorization. Cleared to `None` when the connector recovers.
    error_kind: Arc<RwLock<Option<ConnectorErrorKind>>>,

    /// Sender for commands to the connector
    ///
    /// Commands sent via this channel are processed by the connector's
    /// background task loop. This is cloneable for sharing across tasks.
    cmd_tx: mpsc::Sender<ConnectorCommand>,

    /// Receiver for commands (internal to the connector task)
    ///
    /// This is consumed by the `start_task()` method and moved into the
    /// background task. It's wrapped in Option so we can take it.
    cmd_rx: Arc<RwLock<Option<mpsc::Receiver<ConnectorCommand>>>>,

    /// Broadcast sender for events
    ///
    /// Events from the OpenCode SSE stream are translated and published here.
    /// Subscribers get receivers via the `subscribe()` method.
    events_tx: broadcast::Sender<Event>,

    /// Shared event bus for direct-to-bus publishes.
    ///
    /// Every event emitted by this connector is published here in addition
    /// to `events_tx`, eliminating the forwarder task tier in the runtime.
    sharing_bus: Arc<SharingBus>,

    /// OpenCode REST client
    ///
    /// Used for API operations like listing sessions, listing messages,
    /// and sending messages.
    client: OpenCodeClient,
}

/// Helper that publishes an event to both the per-connector broadcast
/// channel and the global `SharingBus`. The bus publish goes first so any
/// subscriber reading from both sees the bus event no later than the
/// broadcast event.
async fn emit_event(
    sharing_bus: &Arc<SharingBus>,
    events_tx: &broadcast::Sender<Event>,
    connector_id: &ConnectorId,
    connector_uid: Option<Uuid>,
    event: Event,
) {
    let bus_event = dirigent_protocol::streaming::BusEvent::from_connector_event(
        event.clone(),
        connector_uid,
        connector_id.clone(),
    );
    sharing_bus.publish(bus_event).await;
    let _ = events_tx.send(event);
}

impl OpenCodeConnector {
    /// Create a new OpenCode connector
    ///
    /// Initializes the connector with the given configuration but does not
    /// start it. Call `start_task()` to begin the connector's background task.
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for this connector
    /// * `owner` - User ID of the connector owner
    /// * `title` - Human-readable title for this connector
    /// * `config` - OpenCode configuration (base URL, etc.)
    ///
    /// # Returns
    ///
    /// A new OpenCodeConnector in Initializing state
    ///
    /// # Example
    ///
    /// ```no_run
    /// use dirigent_core::connectors::opencode::{OpenCodeConnector, OpenCodeConfig};
    ///
    /// let config = OpenCodeConfig {
    ///     base_url: "http://localhost:12225".to_string(),
    ///     initial_session: None,
    /// };
    ///
    /// let connector = OpenCodeConnector::new(
    ///     "my-connector".to_string(),
    ///     uuid::Uuid::now_v7(),
    ///     "Local OpenCode".to_string(),
    ///     config,
    /// );
    /// ```
    pub fn new(
        id: ConnectorId,
        owner: UserId,
        title: String,
        config: OpenCodeConfig,
        sharing_bus: Arc<SharingBus>,
    ) -> Self {
        // Create the OpenCode REST client
        let client = OpenCodeClient::new(&config.base_url);

        // Create command channel (capacity 100 for buffering commands)
        let (cmd_tx, cmd_rx) = mpsc::channel(100);

        // Create event broadcast channel (capacity 1000 for event buffering)
        let (events_tx, _) = broadcast::channel(1000);

        info!(
            connector_id = %id,
            owner = %owner,
            title = %title,
            base_url = %config.base_url,
            "Creating OpenCode connector"
        );

        Self {
            id,
            connector_uid: None,
            owner,
            title,
            config,
            state: Arc::new(RwLock::new(ConnectorState::Initializing)),
            error_kind: Arc::new(RwLock::new(None)),
            cmd_tx,
            cmd_rx: Arc::new(RwLock::new(Some(cmd_rx))),
            events_tx,
            sharing_bus,
            client,
        }
    }

    /// Wrap a raw connector `Event` in a `BusEvent` populated with this
    /// connector's routing identity (`connector_uid` + `connector_id`).
    ///
    /// Call sites that currently broadcast a raw `Event` via `events_tx` can
    /// use this helper when migrating to the `BusEvent` pipeline. This is an
    /// additive helper — existing `events_tx.send(event)` emissions remain
    /// unchanged.
    pub fn to_bus_event(
        &self,
        event: dirigent_protocol::Event,
    ) -> dirigent_protocol::streaming::BusEvent {
        dirigent_protocol::streaming::BusEvent::from_connector_event(
            event,
            self.connector_uid,
            self.id.clone(),
        )
    }

    /// Get the events broadcast sender
    ///
    /// This method returns a clone of the broadcast sender for publishing events.
    /// It's useful when creating a ConnectorHandle from this connector.
    ///
    /// # Returns
    ///
    /// A clone of the broadcast sender
    pub fn events_sender(&self) -> broadcast::Sender<Event> {
        self.events_tx.clone()
    }

    /// Get the state Arc
    ///
    /// This method returns a clone of the state Arc, allowing the ConnectorHandle
    /// to share the same state that the connector task updates.
    ///
    /// # Returns
    ///
    /// A clone of the Arc<RwLock<ConnectorState>>
    pub fn state_arc(&self) -> Arc<RwLock<ConnectorState>> {
        Arc::clone(&self.state)
    }

    /// Get the error_kind Arc
    ///
    /// This method returns a clone of the error_kind Arc, allowing the ConnectorHandle
    /// to share the same error classification that the connector task updates.
    pub fn error_kind_arc(&self) -> Arc<RwLock<Option<ConnectorErrorKind>>> {
        Arc::clone(&self.error_kind)
    }

    /// Set connector_id on events that require it
    ///
    /// This helper function injects the connector ID into events that were
    /// translated by the adapter with placeholder empty strings.
    fn set_connector_id(event: &mut Event, connector_id: &str) {
        match event {
            Event::SessionsListed {
                connector_id: cid, ..
            } => *cid = connector_id.to_string(),
            Event::SessionCreated {
                connector_id: cid, ..
            } => *cid = connector_id.to_string(),
            Event::SessionUpdated {
                connector_id: cid, ..
            } => *cid = connector_id.to_string(),
            Event::MessageStarted {
                connector_id: cid, ..
            } => *cid = connector_id.to_string(),
            Event::MessageCompleted {
                connector_id: cid, ..
            } => *cid = connector_id.to_string(),
            Event::SessionUpdate {
                connector_id: cid, ..
            } => *cid = connector_id.to_string(),
            _ => {} // Other events don't need connector_id
        }
    }

    /// Start the connector's background task
    ///
    /// This spawns an async task that:
    /// - Connects to the OpenCode SSE endpoint
    /// - Processes incoming events and translates them to Dirigent events
    /// - Handles commands from the command channel
    /// - Manages automatic reconnection on failures
    ///
    /// # Returns
    ///
    /// A JoinHandle for the background task, which can be used to wait for
    /// the task to complete or to detect if it panics.
    ///
    /// # Panics
    ///
    /// Panics if called more than once (the command receiver can only be taken once).
    ///
    /// # Example
    ///
    /// ```no_run
    /// # use dirigent_core::connectors::opencode::{OpenCodeConnector, OpenCodeConfig};
    /// # async fn example(connector: OpenCodeConnector) {
    /// let task_handle = connector.start_task().await;
    ///
    /// // Do other work...
    ///
    /// // Wait for task to complete
    /// task_handle.await.ok();
    /// # }
    /// ```
    pub async fn start_task(&self) -> JoinHandle<()> {
        let id = self.id.clone();
        let connector_uid = self.connector_uid;
        let config = self.config.clone();
        let state = Arc::clone(&self.state);
        let error_kind = Arc::clone(&self.error_kind);
        let events_tx = self.events_tx.clone();
        let sharing_bus = Arc::clone(&self.sharing_bus);
        let client = self.client.clone();

        // Take the command receiver (this can only be done once)
        let cmd_rx = self
            .cmd_rx
            .write()
            .await
            .take()
            .expect("start_task() called more than once - command receiver already taken");

        info!(connector_id = %id, "Starting OpenCode connector task");

        tokio::spawn(async move {
            Self::run_task(
                id,
                connector_uid,
                config,
                state,
                error_kind,
                events_tx,
                sharing_bus,
                client,
                cmd_rx,
            )
            .await;
        })
    }

    /// Main connector task loop
    ///
    /// This is the core of the connector's async behavior. It manages:
    /// - SSE connection and reconnection
    /// - Event translation and broadcasting
    /// - Command processing
    /// - Error handling and recovery
    ///
    /// This function runs until a Shutdown command is received or an
    /// unrecoverable error occurs.
    async fn run_task(
        id: ConnectorId,
        connector_uid: Option<Uuid>,
        config: OpenCodeConfig,
        state: Arc<RwLock<ConnectorState>>,
        error_kind: Arc<RwLock<Option<ConnectorErrorKind>>>,
        events_tx: broadcast::Sender<Event>,
        sharing_bus: Arc<SharingBus>,
        client: OpenCodeClient,
        mut cmd_rx: mpsc::Receiver<ConnectorCommand>,
    ) {
        info!(connector_id = %id, "OpenCode connector task started");

        // Create the event adapter for translating OpenCode events to Dirigent events
        let adapter = OpenCodeAdapter::new();

        // Reconnection state - Counter 1: connection failures (HTTP/TCP level)
        let mut retry_count = 0;
        let max_retries = 5;
        let retry_delays = [
            Duration::from_secs(1),
            Duration::from_secs(3),
            Duration::from_secs(5),
            Duration::from_secs(5),
            Duration::from_secs(5),
        ];

        // Reconnection state - Counter 2: rapid disconnects (stream dies within stability threshold)
        let mut rapid_disconnect_count: usize = 0;
        let rapid_disconnect_max: usize = 5;
        let rapid_disconnect_delays = [
            Duration::from_secs(2),
            Duration::from_secs(5),
            Duration::from_secs(10),
            Duration::from_secs(30),
            Duration::from_secs(60),
        ];
        let stability_threshold = Duration::from_secs(30);
        let mut connected_at: Option<tokio::time::Instant> = None;
        // Track whether we've ever had a stable connection (for adaptive offline detection)
        // If never stable, declare offline after just 2 rapid disconnects (fast startup detection)
        // If was stable before, use full 5 rapid disconnect patience (transient issues)
        let mut ever_stable = false;

        // Main loop: connect, process events, handle reconnects
        'reconnect_loop: loop {
            // Update state to Connecting
            {
                let mut state_guard = state.write().await;
                *state_guard = ConnectorState::Connecting;
            }
            emit_event(
                &sharing_bus,
                &events_tx,
                &id,
                connector_uid,
                Event::ConnectorStateChanged {
                    connector_id: id.clone(),
                    state: "Connecting".to_string(),
                    error_kind: None,
                },
            )
            .await;

            // Emit a Connecting pseudo-event (not part of protocol, but useful for logging)
            debug!(connector_id = %id, "Attempting to connect to OpenCode SSE");

            // Create SSE client and attempt connection
            let sse_client = SseClient::new(&config.base_url);
            let stream = match sse_client.connect() {
                Ok(stream) => {
                    info!(connector_id = %id, "SSE stream channel acquired, awaiting events");

                    // Reset connection failure count (we did connect successfully)
                    retry_count = 0;
                    // Record when we connected to detect rapid disconnects
                    connected_at = Some(tokio::time::Instant::now());

                    // Update state to Ready and clear error_kind
                    {
                        let mut state_guard = state.write().await;
                        *state_guard = ConnectorState::Ready;
                    }
                    {
                        let mut ek = error_kind.write().await;
                        *ek = None;
                    }

                    // Emit state change + Connected events
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::ConnectorStateChanged {
                            connector_id: id.clone(),
                            state: "Ready".to_string(),
                            error_kind: None,
                        },
                    )
                    .await;
                    emit_event(&sharing_bus, &events_tx, &id, connector_uid, Event::Connected).await;

                    stream
                }
                Err(e) => {
                    error!(
                        connector_id = %id,
                        error = %e,
                        "Failed to connect to OpenCode SSE"
                    );

                    // Update state to Error with ConnectionFailed kind
                    {
                        let mut state_guard = state.write().await;
                        *state_guard =
                            ConnectorState::Error(format!("SSE connection failed: {}", e));
                    }
                    {
                        let mut ek = error_kind.write().await;
                        *ek = Some(ConnectorErrorKind::ConnectionFailed);
                    }

                    // Emit state change + error events
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::ConnectorStateChanged {
                            connector_id: id.clone(),
                            state: format!("Error(SSE connection failed: {})", e),
                            error_kind: Some("connection_failed".to_string()),
                        },
                    )
                    .await;
                    emit_event(
                        &sharing_bus,
                        &events_tx,
                        &id,
                        connector_uid,
                        Event::Error {
                            message: format!("Failed to connect: {}", e),
                        },
                    )
                    .await;

                    // Handle retry logic
                    if retry_count < max_retries {
                        let delay = retry_delays[retry_count.min(retry_delays.len() - 1)];
                        warn!(
                            connector_id = %id,
                            retry_count = retry_count + 1,
                            max_retries = max_retries,
                            delay_secs = delay.as_secs(),
                            "Retrying connection"
                        );

                        retry_count += 1;

                        // Wait before retrying
                        tokio::time::sleep(delay).await;

                        continue 'reconnect_loop;
                    } else {
                        error!(
                            connector_id = %id,
                            "Max retries exceeded, staying in Error state until manual Reconnect"
                        );

                        // Stay in Error state and wait for commands
                        loop {
                            match cmd_rx.recv().await {
                                Some(ConnectorCommand::Reconnect) => {
                                    info!(connector_id = %id, "Received Reconnect command, resetting all retry counters");
                                    retry_count = 0;
                                    rapid_disconnect_count = 0;
                                    {
                                        let mut ek = error_kind.write().await;
                                        *ek = None;
                                    }
                                    continue 'reconnect_loop;
                                }
                                Some(ConnectorCommand::Shutdown) => {
                                    info!(connector_id = %id, "Received Shutdown command");
                                    break 'reconnect_loop;
                                }
                                Some(cmd) => {
                                    warn!(
                                        connector_id = %id,
                                        command = ?cmd,
                                        "Received command while in Error state, ignoring"
                                    );
                                }
                                None => {
                                    error!(connector_id = %id, "Command channel closed");
                                    break 'reconnect_loop;
                                }
                            }
                        }
                    }
                }
            };

            // Convert stream to tokio::select!-compatible type
            let mut stream = Box::pin(stream);

            // Event processing loop
            'event_loop: loop {
                tokio::select! {
                    // Handle incoming SSE events
                    event_result = stream.next() => {
                        match event_result {
                            Some(Ok(oc_event)) => {
                                debug!(connector_id = %id, event = ?oc_event, "Received OpenCode event");

                                // Translate the event using the adapter
                                match adapter.translate_event(oc_event) {
                                    Ok(mut dirigent_event) => {
                                        // Set connector_id on events that need it
                                        Self::set_connector_id(&mut dirigent_event, &id);

                                        // ENHANCEMENT: Fetch user message content if empty
                                        // User messages arrive via SSE with empty content because they don't stream
                                        // We need to fetch the full message with parts from the API
                                        if let Event::MessageCompleted { message, connector_id: _ } = &mut dirigent_event {
                                            if message.role == dirigent_protocol::MessageRole::User && message.content.is_empty() {
                                                debug!(
                                                    connector_id = %id,
                                                    message_id = %message.id,
                                                    session_id = %message.session_id,
                                                    "User message has empty content, fetching from API"
                                                );

                                                // Fetch all messages for this session from the API
                                                match client.list_messages(&message.session_id).await {
                                                    Ok(all_messages) => {
                                                        // Find the specific message we need
                                                        if let Some(msg_with_parts) = all_messages.iter().find(|m| {
                                                            match &m.info {
                                                                opencode_client::types::Message::User(u) => u.id == message.id,
                                                                opencode_client::types::Message::Assistant(a) => a.id == message.id,
                                                            }
                                                        }) {
                                                            // Defensive: clear content before adding fetched parts
                                                            // (should already be empty, but this prevents duplication)
                                                            message.content.clear();

                                                            // Translate parts and add to message
                                                            for oc_part in &msg_with_parts.parts {
                                                                match dirigent_protocol::adapters::opencode::OpenCodeAdapter::translate_part_for_list(oc_part.clone()) {
                                                                    Ok(part) => {
                                                                        message.content.push(part);
                                                                    }
                                                                    Err(e) => {
                                                                        warn!(
                                                                            connector_id = %id,
                                                                            message_id = %message.id,
                                                                            error = %e,
                                                                            "Failed to translate part"
                                                                        );
                                                                    }
                                                                }
                                                            }

                                                            debug!(
                                                                connector_id = %id,
                                                                message_id = %message.id,
                                                                parts_count = message.content.len(),
                                                                "Fetched user message content from API"
                                                            );
                                                        } else {
                                                            warn!(
                                                                connector_id = %id,
                                                                message_id = %message.id,
                                                                session_id = %message.session_id,
                                                                "Message not found in list_messages response"
                                                            );
                                                        }
                                                    }
                                                    Err(e) => {
                                                        warn!(
                                                            connector_id = %id,
                                                            message_id = %message.id,
                                                            session_id = %message.session_id,
                                                            error = %e,
                                                            "Failed to fetch user message content from API"
                                                        );
                                                        // Continue with empty content - better than failing
                                                    }
                                                }
                                            }
                                        }

                                        debug!(connector_id = %id, event = ?dirigent_event, "Translated to Dirigent event");

                                        // Track if this is a MessageCompleted event for TurnComplete and SessionIdle emission
                                        let should_emit_turn_complete = if let Event::MessageCompleted { message, .. } = &dirigent_event {
                                            // Emit TurnComplete and SessionIdle after messages complete
                                            // User messages also get MessageCompleted, but we emit for all for consistency
                                            Some((message.session_id.clone(), message.id.clone()))
                                        } else {
                                            None
                                        };

                                        // Broadcast the event (via bus + per-connector channel)
                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            dirigent_event,
                                        )
                                        .await;

                                        // Emit TurnComplete then SessionIdle after MessageCompleted to signal completion
                                        // Order: MessageCompleted → TurnComplete → SessionIdle
                                        // This ensures archivist receives complete turn signal before idle flush
                                        if let Some((session_id, message_id)) = should_emit_turn_complete {
                                            // First emit TurnComplete with ExplicitSignal trigger
                                            // (OpenCode session.idle events are explicit completion signals)
                                            emit_event(
                                                &sharing_bus,
                                                &events_tx,
                                                &id,
                                                connector_uid,
                                                Event::TurnComplete {
                                                    connector_id: id.clone(),
                                                    session_id: session_id.clone(),
                                                    message_id: message_id.clone(),
                                                    trigger: TurnCompleteTrigger::ExplicitSignal,
                                                },
                                            )
                                            .await;
                                            debug!(
                                                connector_id = %id,
                                                session_id = %session_id,
                                                message_id = %message_id,
                                                "Emitted TurnComplete with ExplicitSignal trigger"
                                            );

                                            // Then emit SessionIdle
                                            emit_event(
                                                &sharing_bus,
                                                &events_tx,
                                                &id,
                                                connector_uid,
                                                Event::SessionIdle {
                                                    connector_id: id.to_string(),
                                                    session_id: session_id.clone(),
                                                },
                                            )
                                            .await;
                                            debug!(connector_id = %id, session_id = %session_id, "Emitted SessionIdle after TurnComplete");
                                        }
                                    }
                                    Err(dirigent_protocol::adapters::opencode::TranslationError::Duplicate) => {
                                        // Skip duplicate events (this is normal)
                                        debug!(connector_id = %id, "Skipped duplicate event");
                                    }
                                    Err(e) => {
                                        warn!(connector_id = %id, error = %e, "Failed to translate event");
                                    }
                                }
                            }
                            Some(Err(e)) => {
                                error!(connector_id = %id, error = %e, "SSE stream error");

                                // Update state to Error
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Error(format!("SSE stream error: {}", e));
                                }

                                // Emit state change + error events
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::ConnectorStateChanged {
                                        connector_id: id.clone(),
                                        state: format!("Error(SSE stream error: {})", e),
                                        error_kind: None,
                                    },
                                )
                                .await;
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::Error {
                                        message: format!("SSE stream error: {}", e),
                                    },
                                )
                                .await;

                                // Attempt to reconnect
                                break 'event_loop;
                            }
                            None => {
                                warn!(connector_id = %id, "SSE stream closed");

                                // Update state to Connecting (will reconnect)
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Connecting;
                                }

                                // Emit state change + disconnected events
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::ConnectorStateChanged {
                                        connector_id: id.clone(),
                                        state: "Connecting".to_string(),
                                        error_kind: None,
                                    },
                                )
                                .await;
                                emit_event(
                                    &sharing_bus,
                                    &events_tx,
                                    &id,
                                    connector_uid,
                                    Event::Disconnected,
                                )
                                .await;

                                // Attempt to reconnect
                                break 'event_loop;
                            }
                        }
                    }

                    // Handle commands
                    cmd = cmd_rx.recv() => {
                        match cmd {
                            Some(ConnectorCommand::ListSessions) => {
                                debug!(connector_id = %id, "Processing ListSessions command");

                                match client.list_sessions().await {
                                    Ok(oc_sessions) => {
                                        info!(connector_id = %id, count = oc_sessions.len(), "Listed sessions");

                                        // Translate all sessions to Dirigent protocol sessions
                                        let mut sessions = Vec::new();
                                        for oc_session in oc_sessions {
                                            match adapter.translate_event(opencode_client::types::Event::SessionCreated {
                                                properties: opencode_client::types::SessionEventInfo {
                                                    info: oc_session,
                                                },
                                            }) {
                                                Ok(Event::SessionCreated { session, .. }) => {
                                                    sessions.push(session);
                                                }
                                                Err(e) => {
                                                    warn!(connector_id = %id, error = %e, "Failed to translate session");
                                                }
                                                _ => {
                                                    // Unexpected event type from translation
                                                    warn!(connector_id = %id, "Unexpected event type from session translation");
                                                }
                                            }
                                        }

                                        // Emit a single SessionsListed event with all sessions
                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::SessionsListed {
                                                connector_id: id.clone(),
                                                sessions,
                                            },
                                        )
                                        .await;
                                    }
                                    Err(e) => {
                                        error!(connector_id = %id, error = %e, "Failed to list sessions");

                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::Error {
                                                message: format!("Failed to list sessions: {}", e),
                                            },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::ListMessages { session_id }) => {
                                debug!(connector_id = %id, session_id = %session_id, "Processing ListMessages command");

                                match client.list_messages(&session_id).await {
                                    Ok(messages) => {
                                        info!(connector_id = %id, session_id = %session_id, count = messages.len(), "Listed messages");

                                        // Collect all translated messages with their parts
                                        let mut dirigent_messages = Vec::new();

                                        // Use a fresh adapter for batch translation to avoid deduplication issues
                                        // The stateful adapter is for SSE streams, not for batch list operations
                                        let batch_adapter = dirigent_protocol::adapters::opencode::OpenCodeAdapter::new();

                                        for msg_with_parts in messages {
                                            // Translate the message using fresh adapter
                                            match batch_adapter.translate_event(opencode_client::types::Event::MessageUpdated {
                                                properties: opencode_client::types::MessageEventInfo {
                                                    info: msg_with_parts.info.clone(),
                                                },
                                            }) {
                                                Ok(Event::MessageCompleted { message: mut msg, .. }) | Ok(Event::MessageStarted { message: mut msg, .. }) => {
                                                    // Translate all parts and add them to the message
                                                    for oc_part in msg_with_parts.parts {
                                                        // Directly translate part to MessagePart using translate_part
                                                        // This avoids the roundtrip through SessionUpdate
                                                        match dirigent_protocol::adapters::opencode::OpenCodeAdapter::translate_part_for_list(oc_part.clone()) {
                                                            Ok(part) => {
                                                                msg.content.push(part);
                                                            }
                                                            Err(e) => {
                                                                warn!(connector_id = %id, error = %e, "Failed to translate part");
                                                            }
                                                        }
                                                    }
                                                    dirigent_messages.push(msg);
                                                }
                                                Ok(Event::SessionSystemMessageSet { .. }) => {
                                                    // Skip system message events when listing messages
                                                    // (they're already part of session metadata)
                                                }
                                                Err(dirigent_protocol::adapters::opencode::TranslationError::Duplicate) => {
                                                    // Skip duplicates
                                                }
                                                Err(e) => {
                                                    warn!(connector_id = %id, error = %e, "Failed to translate message");
                                                }
                                                _ => {
                                                    // Unexpected event type from translation
                                                    warn!(connector_id = %id, "Unexpected event type from message translation");
                                                }
                                            }
                                        }

                                        // Calculate message count for metadata update
                                        let message_count = dirigent_messages.len() as u32;

                                        // Emit a single MessagesListed event with all messages
                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::MessagesListed { messages: dirigent_messages },
                                        )
                                        .await;

                                        // Emit SessionMetadataUpdated with the accurate count
                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::SessionMetadataUpdated {
                                                connector_id: id.clone(),
                                                session_id: session_id.clone(),
                                                title: None, // OpenCode doesn't support title updates yet
                                                total_messages: Some(message_count),
                                                model: None,
                                            },
                                        )
                                        .await;
                                    }
                                    Err(e) => {
                                        error!(connector_id = %id, session_id = %session_id, error = %e, "Failed to list messages");

                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::Error {
                                                message: format!("Failed to list messages: {}", e),
                                            },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::CreateSession { .. }) => {
                                warn!(connector_id = %id, "CreateSession command not implemented for OpenCode connector");
                                // OpenCode doesn't have a create session API endpoint
                            }

                            Some(ConnectorCommand::LoadSession { session_id, .. }) => {
                                warn!(connector_id = %id, session_id = %session_id, "LoadSession command not implemented for OpenCode connector");
                                // OpenCode doesn't have a session/load API endpoint
                            }

                            Some(ConnectorCommand::SendMessage { session_id, text }) => {
                                debug!(connector_id = %id, session_id = %session_id, text_len = text.len(), "Processing SendMessage command");

                                match client.send_message(&session_id, text).await {
                                    Ok(response) => {
                                        info!(connector_id = %id, session_id = %session_id, "Message sent successfully");

                                        // The response streaming is handled by SSE events
                                        // We just log success here
                                        debug!(connector_id = %id, response = ?response, "Send message response");
                                    }
                                    Err(e) => {
                                        error!(connector_id = %id, session_id = %session_id, error = %e, "Failed to send message");

                                        emit_event(
                                            &sharing_bus,
                                            &events_tx,
                                            &id,
                                            connector_uid,
                                            Event::Error {
                                                message: format!("Failed to send message: {}", e),
                                            },
                                        )
                                        .await;
                                    }
                                }
                            }

                            Some(ConnectorCommand::CancelGeneration { .. }) => {
                                warn!(connector_id = %id, "CancelGeneration command not implemented for OpenCode connector");
                                // OpenCode doesn't have a cancel generation API endpoint
                            }

                            Some(ConnectorCommand::Reconnect) => {
                                info!(connector_id = %id, "Received Reconnect command, restarting SSE stream");

                                // Reset all retry counters and break to reconnect loop
                                retry_count = 0;
                                rapid_disconnect_count = 0;
                                connected_at = None; // Clear so stability check doesn't misclassify this
                                break 'event_loop;
                            }

                            Some(ConnectorCommand::AgentResponse { .. }) => {
                                // OpenCode connector does not support agent-initiated requests
                                warn!(
                                    connector_id = %id,
                                    "Received AgentResponse command but OpenCode connector does not support agent-initiated requests"
                                );
                            }

                            Some(ConnectorCommand::SetSessionMode { .. }) => {
                                // OpenCode connector does not support mode switching
                                warn!(
                                    connector_id = %id,
                                    "Received SetSessionMode command but OpenCode connector does not support mode switching"
                                );
                            }

                            Some(ConnectorCommand::SetSessionModel { .. }) => {
                                // OpenCode connector does not support model switching
                                warn!(
                                    connector_id = %id,
                                    "Received SetSessionModel command but OpenCode connector does not support model switching"
                                );
                            }

                            Some(ConnectorCommand::CloseSession { session_id }) => {
                                warn!(connector_id = %id, session_id = %session_id, "CloseSession not supported by OpenCode connector");
                            }

                            Some(ConnectorCommand::SetConfigOption { .. }) => {
                                warn!(
                                    connector_id = %id,
                                    "Received SetConfigOption command but OpenCode connector does not support config options"
                                );
                            }

                            Some(ConnectorCommand::Shutdown) => {
                                info!(connector_id = %id, "Received Shutdown command, stopping connector");

                                // Update state to Stopped
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Stopped;
                                }

                                // Exit both loops
                                break 'reconnect_loop;
                            }

                            None => {
                                error!(connector_id = %id, "Command channel closed, stopping connector");

                                // Update state to Stopped
                                {
                                    let mut state_guard = state.write().await;
                                    *state_guard = ConnectorState::Stopped;
                                }

                                // Exit both loops
                                break 'reconnect_loop;
                            }
                        }
                    }
                }
            }

            // If we exited the event loop (not due to Shutdown), check connection stability
            if let Some(connect_time) = connected_at.take() {
                let uptime = connect_time.elapsed();
                if uptime >= stability_threshold {
                    // Established connection broke - reset rapid counter, retry immediately
                    info!(
                        connector_id = %id,
                        uptime_secs = uptime.as_secs(),
                        "Established connection lost, reconnecting immediately"
                    );
                    rapid_disconnect_count = 0;
                    ever_stable = true;
                } else {
                    // Rapid disconnect - connection died too quickly
                    let offline_threshold = if ever_stable { rapid_disconnect_max } else { 2 };
                    if rapid_disconnect_count < offline_threshold {
                        let delay = rapid_disconnect_delays
                            [rapid_disconnect_count.min(rapid_disconnect_delays.len() - 1)];
                        rapid_disconnect_count += 1;
                        warn!(
                            connector_id = %id,
                            rapid_disconnect_count,
                            rapid_disconnect_max,
                            uptime_secs = uptime.as_secs(),
                            delay_secs = delay.as_secs(),
                            "Rapid disconnect, backing off"
                        );

                        // Update state to Error while backing off
                        {
                            let mut state_guard = state.write().await;
                            *state_guard = ConnectorState::Error(format!(
                                "Rapid disconnect #{}/{} (connected for {}s), retrying in {}s",
                                rapid_disconnect_count,
                                rapid_disconnect_max,
                                uptime.as_secs(),
                                delay.as_secs()
                            ));
                        }
                        {
                            let mut ek = error_kind.write().await;
                            *ek = Some(ConnectorErrorKind::Unstable);
                        }

                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::ConnectorStateChanged {
                                connector_id: id.clone(),
                                state: format!("Error(Rapid disconnect #{}/{})", rapid_disconnect_count, rapid_disconnect_max),
                                error_kind: Some("unstable".to_string()),
                            },
                        )
                        .await;
                        emit_event(&sharing_bus, &events_tx, &id, connector_uid, Event::Disconnected).await;

                        tokio::time::sleep(delay).await;
                        continue 'reconnect_loop;
                    } else {
                        error!(
                            connector_id = %id,
                            "Max rapid disconnects exceeded, pausing until manual Reconnect"
                        );

                        // Update state to Error
                        {
                            let mut state_guard = state.write().await;
                            *state_guard = ConnectorState::Error(
                                "Connection unstable: too many rapid disconnects. Use Reconnect to retry.".to_string()
                            );
                        }
                        let final_error_kind = if ever_stable {
                            ConnectorErrorKind::Unstable
                        } else {
                            ConnectorErrorKind::Offline
                        };
                        let ek_str = match &final_error_kind {
                            ConnectorErrorKind::Offline => "offline",
                            ConnectorErrorKind::Unstable => "unstable",
                            ConnectorErrorKind::ConnectionFailed => "connection_failed",
                        };
                        {
                            let mut ek = error_kind.write().await;
                            *ek = Some(final_error_kind);
                        }

                        emit_event(
                            &sharing_bus,
                            &events_tx,
                            &id,
                            connector_uid,
                            Event::ConnectorStateChanged {
                                connector_id: id.clone(),
                                state: "Error(Connection unstable: too many rapid disconnects)".to_string(),
                                error_kind: Some(ek_str.to_string()),
                            },
                        )
                        .await;
                        emit_event(&sharing_bus, &events_tx, &id, connector_uid, Event::Disconnected).await;

                        // Reuse the error-state command loop pattern
                        loop {
                            match cmd_rx.recv().await {
                                Some(ConnectorCommand::Reconnect) => {
                                    info!(connector_id = %id, "Received Reconnect command, resetting all retry counters");
                                    retry_count = 0;
                                    rapid_disconnect_count = 0;
                                    {
                                        let mut ek = error_kind.write().await;
                                        *ek = None;
                                    }
                                    continue 'reconnect_loop;
                                }
                                Some(ConnectorCommand::Shutdown) => {
                                    info!(connector_id = %id, "Received Shutdown command");
                                    break 'reconnect_loop;
                                }
                                Some(cmd) => {
                                    warn!(
                                        connector_id = %id,
                                        command = ?cmd,
                                        "Received command while in Error state, ignoring"
                                    );
                                }
                                None => {
                                    error!(connector_id = %id, "Command channel closed");
                                    break 'reconnect_loop;
                                }
                            }
                        }
                    }
                }
            } else {
                // No connected_at means we never fully connected - just retry
                info!(connector_id = %id, "Event loop exited before connection established, attempting reconnect");
            }
        }

        info!(connector_id = %id, "OpenCode connector task stopped");
    }
}

impl Connector for OpenCodeConnector {
    fn id(&self) -> &ConnectorId {
        &self.id
    }

    fn kind(&self) -> ConnectorKind {
        ConnectorKind::OpenCode
    }

    fn owner(&self) -> &UserId {
        &self.owner
    }

    fn title(&self) -> &str {
        &self.title
    }

    fn state(&self) -> ConnectorState {
        // Try to read the state without blocking
        match self.state.try_read() {
            Ok(state_guard) => state_guard.clone(),
            Err(_) => {
                // Lock is held, return Initializing as a safe default
                ConnectorState::Initializing
            }
        }
    }

    fn command_tx(&self) -> mpsc::Sender<ConnectorCommand> {
        self.cmd_tx.clone()
    }

    fn subscribe(&self) -> broadcast::Receiver<Event> {
        self.events_tx.subscribe()
    }

    fn stop(&self) {
        // Send shutdown command
        let cmd_tx = self.cmd_tx.clone();
        tokio::spawn(async move {
            let _ = cmd_tx.send(ConnectorCommand::Shutdown).await;
        });
    }
}

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

    #[test]
    fn test_opencode_config_serialization() {
        let config = OpenCodeConfig {
            base_url: "http://localhost:12225".to_string(),
            initial_session: Some("session-123".to_string()),
        };

        let json = serde_json::to_string(&config).unwrap();
        let deserialized: OpenCodeConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(deserialized.base_url, config.base_url);
        assert_eq!(deserialized.initial_session, config.initial_session);
    }

    #[tokio::test]
    async fn test_opencode_connector_creation() {
        let config = OpenCodeConfig {
            base_url: "http://localhost:12225".to_string(),
            initial_session: None,
        };

        let connector = OpenCodeConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test Connector".to_string(),
            config,
            SharingBus::new(),
        );

        assert_eq!(connector.id(), "test-conn");
        assert_eq!(*connector.owner(), uuid::Uuid::nil());
        assert_eq!(connector.title(), "Test Connector");
        assert_eq!(connector.kind(), ConnectorKind::OpenCode);
        assert_eq!(connector.state(), ConnectorState::Initializing);
    }

    #[tokio::test]
    async fn test_opencode_connector_clone_command_tx() {
        let config = OpenCodeConfig {
            base_url: "http://localhost:12225".to_string(),
            initial_session: None,
        };

        let connector = OpenCodeConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test".to_string(),
            config,
            SharingBus::new(),
        );

        // Should be able to clone the command sender
        let cmd_tx1 = connector.command_tx();
        let cmd_tx2 = connector.command_tx();

        // Both should be valid senders
        drop(cmd_tx1);
        drop(cmd_tx2);
    }

    #[tokio::test]
    async fn test_opencode_connector_subscribe() {
        let config = OpenCodeConfig {
            base_url: "http://localhost:12225".to_string(),
            initial_session: None,
        };

        let connector = OpenCodeConnector::new(
            "test-conn".to_string(),
            uuid::Uuid::nil(),
            "Test".to_string(),
            config,
            SharingBus::new(),
        );

        // Should be able to create multiple subscriptions
        let _rx1 = connector.subscribe();
        let _rx2 = connector.subscribe();
    }
}
