//! Tests for remote session support: SessionActorRef, SessionRegistry,
//! RemoteNodeManager, EventRelayActor, and EventForwarder.
//!
//! These tests exercise the local-variant code paths end-to-end without
//! requiring a live kameo mesh / libp2p swarm.  Feature-gated tests that
//! exercise the `Remote` variant live in a separate `#[cfg(feature = "remote")]`
//! submodule at the bottom.

use crate::agent::agent_config::AgentConfig;
use crate::agent::agent_config_builder::AgentConfigBuilder;
use crate::agent::core::{AgentMode, SessionRuntime};
use crate::agent::remote::SessionActorRef;
use crate::agent::session_actor::SessionActor;
use crate::agent::session_registry::SessionRegistry;
use crate::events::{AgentEvent, AgentEventKind, EventOrigin};
use crate::session::backend::StorageBackend;
use crate::session::sqlite_storage::SqliteStorage;
use kameo::actor::Spawn;
use querymt::LLMParams;
use querymt::plugin::host::PluginRegistry;
use std::collections::HashMap;
use std::sync::Arc;
use tempfile::TempDir;

// ═══════════════════════════════════════════════════════════════════════════
//  Test helpers
// ═══════════════════════════════════════════════════════════════════════════

/// Build a minimal `AgentConfig` suitable for unit tests that don't need LLM calls.
///
/// Returns `(Arc<AgentConfig>, TempDir)` — the `TempDir` must be kept alive for
/// the duration of the test.
async fn test_agent_config() -> (Arc<AgentConfig>, TempDir) {
    let temp_dir = TempDir::new().expect("create temp dir");
    let config_path = temp_dir.path().join("providers.toml");
    std::fs::write(&config_path, "providers = []\n").expect("write providers.toml");
    let registry = PluginRegistry::from_path(&config_path).expect("create plugin registry");
    let plugin_registry = Arc::new(registry);

    let storage = Arc::new(
        SqliteStorage::connect(":memory:".into())
            .await
            .expect("create sqlite storage"),
    );
    let llm = LLMParams::new().provider("mock").model("mock");

    let mut builder = AgentConfigBuilder::new(
        plugin_registry,
        storage.session_store(),
        storage.event_journal(),
        llm,
    );
    if let Some(repo) = storage.schedule_repository() {
        builder = builder.with_schedule_repository(repo);
    }
    let config = Arc::new(builder.build());
    (config, temp_dir)
}

/// Spawn a `SessionActor` with default runtime and return a `SessionActorRef::Local`.
fn spawn_test_session(config: Arc<AgentConfig>, session_id: &str) -> SessionActorRef {
    let runtime = SessionRuntime::new(
        None,
        HashMap::new(),
        crate::agent::core::McpToolState::empty(),
    );
    let actor = SessionActor::new(config, session_id.to_string(), runtime);
    let actor_ref = SessionActor::spawn(actor);
    SessionActorRef::Local(actor_ref)
}

// ═══════════════════════════════════════════════════════════════════════════
//  SessionActorRef — Local variant
// ═══════════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn test_local_ref_is_not_remote() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-local-1");
    assert!(!session_ref.is_remote());
    assert_eq!(session_ref.node_label(), "local");
}

#[tokio::test]
async fn test_local_ref_get_mode_default_is_build() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-mode-1");
    let mode = session_ref
        .get_mode()
        .await
        .expect("get_mode should succeed");
    assert_eq!(mode, AgentMode::Build);
}

#[tokio::test]
async fn test_local_ref_set_mode_roundtrip() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-mode-2");

    session_ref
        .set_mode(AgentMode::Plan)
        .await
        .expect("set_mode Plan");
    let mode = session_ref.get_mode().await.expect("get_mode");
    assert_eq!(mode, AgentMode::Plan);

    session_ref
        .set_mode(AgentMode::Build)
        .await
        .expect("set_mode Build");
    let mode = session_ref.get_mode().await.expect("get_mode");
    assert_eq!(mode, AgentMode::Build);
}

#[tokio::test]
async fn test_local_ref_cancel_idle_session() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-cancel-1");
    // Cancel on idle session should succeed (no-op)
    session_ref.cancel().await.expect("cancel on idle session");
}

#[tokio::test]
async fn test_local_ref_shutdown() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-shutdown-1");
    session_ref.shutdown().await.expect("shutdown");
}

#[tokio::test]
async fn test_local_ref_get_session_limits_none() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-limits-1");
    let limits = session_ref
        .get_session_limits()
        .await
        .expect("get_session_limits");
    assert!(limits.is_none(), "no limits configured → None");
}

#[tokio::test]
async fn test_local_ref_get_llm_config() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-llm-1");
    // Session doesn't exist in the DB → handler returns an error.
    // We just verify the message round-trips without panic.
    let result = session_ref.get_llm_config().await;
    assert!(result.is_err(), "no session in DB → should return error");
}

#[tokio::test]
async fn test_local_ref_subscribe_unsubscribe_events() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-events-1");

    // Subscribe with a dummy relay_actor_id
    session_ref
        .subscribe_events(42, "event_relay::test-events-1::local".to_string())
        .await
        .expect("subscribe_events");
    // Unsubscribe
    session_ref
        .unsubscribe_events(42, "event_relay::test-events-1::local".to_string())
        .await
        .expect("unsubscribe_events");
}

#[tokio::test]
async fn test_local_ref_get_history() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "test-history-1");
    // Session doesn't exist in the DB → handler returns an error.
    // We just verify the message round-trips without panic.
    let result = session_ref.get_history().await;
    assert!(result.is_err(), "no session in DB → should return error");
}

// ═══════════════════════════════════════════════════════════════════════════
//  SessionRegistry — CRUD operations
// ═══════════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn test_registry_initially_empty() {
    let (config, _td) = test_agent_config().await;
    let registry = SessionRegistry::new(config);
    assert!(registry.is_empty());
    assert_eq!(registry.len(), 0);
    assert!(registry.session_ids().is_empty());
}

#[tokio::test]
async fn test_registry_insert_and_get() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config.clone(), "reg-1");

    let mut registry = SessionRegistry::new(config);
    registry.insert("reg-1".to_string(), session_ref);

    assert_eq!(registry.len(), 1);
    assert!(!registry.is_empty());
    assert!(registry.get("reg-1").is_some());
    assert!(registry.get("reg-nonexistent").is_none());
}

#[tokio::test]
async fn test_registry_insert_and_remove() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config.clone(), "reg-2");

    let mut registry = SessionRegistry::new(config);
    registry.insert("reg-2".to_string(), session_ref);
    assert_eq!(registry.len(), 1);

    let removed = registry.remove("reg-2");
    assert!(removed.is_some());
    assert_eq!(registry.len(), 0);
    assert!(registry.get("reg-2").is_none());
}

#[tokio::test]
async fn test_registry_remove_nonexistent() {
    let (config, _td) = test_agent_config().await;
    let mut registry = SessionRegistry::new(config);
    let removed = registry.remove("does-not-exist");
    assert!(removed.is_none());
}

#[tokio::test]
async fn test_registry_session_ids() {
    let (config, _td) = test_agent_config().await;
    let ref1 = spawn_test_session(config.clone(), "id-a");
    let ref2 = spawn_test_session(config.clone(), "id-b");
    let ref3 = spawn_test_session(config.clone(), "id-c");

    let mut registry = SessionRegistry::new(config);
    registry.insert("id-a".to_string(), ref1);
    registry.insert("id-b".to_string(), ref2);
    registry.insert("id-c".to_string(), ref3);

    let mut ids = registry.session_ids();
    ids.sort();
    assert_eq!(ids, vec!["id-a", "id-b", "id-c"]);
}

#[tokio::test]
async fn test_registry_overwrite() {
    let (config, _td) = test_agent_config().await;
    let ref1 = spawn_test_session(config.clone(), "ow-1");
    let ref2 = spawn_test_session(config.clone(), "ow-1-replacement");

    let mut registry = SessionRegistry::new(config);
    registry.insert("ow-1".to_string(), ref1);
    assert_eq!(registry.len(), 1);

    // Insert again with same key — should overwrite
    registry.insert("ow-1".to_string(), ref2);
    assert_eq!(registry.len(), 1);

    // The stored ref should still be gettable
    assert!(registry.get("ow-1").is_some());
}

// ═══════════════════════════════════════════════════════════════════════════
//  EventRelayActor — extended tests
// ═══════════════════════════════════════════════════════════════════════════

#[cfg(feature = "remote")]
mod event_relay_extended {
    use super::*;
    use crate::agent::remote::event_relay::{EventRelayActor, RelayedEvent};
    use crate::event_fanout::EventFanout;
    use crate::event_sink::EventSink;

    /// Create an EventSink backed by an in-memory journal.
    async fn make_sink() -> Arc<EventSink> {
        let storage = Arc::new(
            crate::session::sqlite_storage::SqliteStorage::connect(":memory:".into())
                .await
                .unwrap(),
        );
        let journal = storage.event_journal();
        let fanout = Arc::new(EventFanout::new());
        Arc::new(EventSink::new(journal, fanout))
    }

    #[tokio::test]
    async fn test_relay_multiple_events() {
        let sink = make_sink().await;
        let relay = EventRelayActor::new(
            sink.clone(),
            "s1".to_string(),
            "multi-test".to_string(),
            None,
            None,
        );
        let relay_ref = <EventRelayActor as Spawn>::spawn(relay);

        let mut rx = sink.fanout().subscribe();

        let events = vec![
            AgentEvent {
                seq: 1,
                timestamp: 100,
                session_id: "s1".to_string(),
                origin: EventOrigin::Local,
                source_node: None,
                kind: AgentEventKind::SessionCreated,
            },
            AgentEvent {
                seq: 2,
                timestamp: 200,
                session_id: "s1".to_string(),
                origin: EventOrigin::Local,
                source_node: None,
                kind: AgentEventKind::PromptReceived {
                    content: "hello".to_string(),
                    message_id: None,
                },
            },
            AgentEvent {
                seq: 3,
                timestamp: 300,
                session_id: "s1".to_string(),
                origin: EventOrigin::Local,
                source_node: None,
                kind: AgentEventKind::LlmRequestStart { message_count: 5 },
            },
        ];

        for event in &events {
            relay_ref
                .tell(RelayedEvent {
                    event: event.clone(),
                })
                .await
                .expect("tell should succeed");
        }

        // Receive all 3 events. The relay routes through EventSink which
        // assigns new journal stream_seq values (not the original seq).
        // We verify: session_id preserved, events arrive in order, seq monotonic.
        let mut received_seqs = Vec::new();
        for expected_event in &events {
            let received = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
                .await
                .expect("should receive within timeout")
                .expect("recv should succeed");

            assert_eq!(received.session_id(), expected_event.session_id);
            if let crate::events::EventEnvelope::Durable(de) = &received {
                assert!(de.stream_seq >= 1, "journal-assigned seq must be >= 1");
                received_seqs.push(de.stream_seq);
            } else {
                panic!("expected durable event envelope");
            }
        }

        // Journal assigns monotonically increasing seqs.
        for window in received_seqs.windows(2) {
            assert!(
                window[0] < window[1],
                "journal stream_seq must be monotonically increasing: {} < {}",
                window[0],
                window[1]
            );
        }
    }

    #[tokio::test]
    async fn test_relay_preserves_session_id() {
        let sink = make_sink().await;
        let relay = EventRelayActor::new(
            sink.clone(),
            "unique-session-id-xyz".to_string(),
            "preserve-test".to_string(),
            None,
            None,
        );
        let relay_ref = <EventRelayActor as Spawn>::spawn(relay);

        let mut rx = sink.fanout().subscribe();

        let event = AgentEvent {
            seq: 99,
            timestamp: 555,
            session_id: "unique-session-id-xyz".to_string(),
            origin: EventOrigin::Local,
            source_node: None,
            kind: AgentEventKind::SessionCreated,
        };

        relay_ref
            .tell(RelayedEvent {
                event: event.clone(),
            })
            .await
            .expect("tell");

        let received = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
            .await
            .expect("timeout")
            .expect("recv");

        assert_eq!(received.session_id(), "unique-session-id-xyz");
    }

    #[tokio::test]
    async fn test_relay_preserves_event_kind() {
        let sink = make_sink().await;
        let relay = EventRelayActor::new(
            sink.clone(),
            "s1".to_string(),
            "kind-test".to_string(),
            None,
            None,
        );
        let relay_ref = <EventRelayActor as Spawn>::spawn(relay);

        let mut rx = sink.fanout().subscribe();

        let event = AgentEvent {
            seq: 7,
            timestamp: 777,
            session_id: "s1".to_string(),
            origin: EventOrigin::Local,
            source_node: None,
            kind: AgentEventKind::Error {
                message: "test error".to_string(),
            },
        };

        relay_ref
            .tell(RelayedEvent {
                event: event.clone(),
            })
            .await
            .expect("tell");

        let received = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
            .await
            .expect("timeout")
            .expect("recv");

        match received.kind() {
            AgentEventKind::Error { message } => {
                assert_eq!(message, "test error");
            }
            other => panic!("expected Error event, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_relay_ignores_mismatched_session() {
        let sink = make_sink().await;
        let relay = EventRelayActor::new(
            sink.clone(),
            "session-a".to_string(),
            "multi-session".to_string(),
            None,
            None,
        );
        let relay_ref = <EventRelayActor as Spawn>::spawn(relay);

        let mut rx = sink.fanout().subscribe();

        let event_a = AgentEvent {
            seq: 1,
            timestamp: 100,
            session_id: "session-a".to_string(),
            origin: EventOrigin::Local,
            source_node: None,
            kind: AgentEventKind::SessionCreated,
        };
        let event_b = AgentEvent {
            seq: 2,
            timestamp: 200,
            session_id: "session-b".to_string(),
            origin: EventOrigin::Local,
            source_node: None,
            kind: AgentEventKind::SessionCreated,
        };

        relay_ref
            .tell(RelayedEvent {
                event: event_a.clone(),
            })
            .await
            .expect("tell a");
        relay_ref
            .tell(RelayedEvent {
                event: event_b.clone(),
            })
            .await
            .expect("tell b");

        let received = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
            .await
            .expect("timeout")
            .expect("recv");
        assert_eq!(received.session_id(), "session-a");

        let extra = tokio::time::timeout(std::time::Duration::from_millis(50), rx.recv()).await;
        assert!(extra.is_err(), "mismatched session event should be ignored");
    }
}

// ═══════════════════════════════════════════════════════════════════════════
//  EventForwarder — stub tests (no remote feature)
// ═══════════════════════════════════════════════════════════════════════════

#[cfg(not(feature = "remote"))]
mod event_forwarder_stub {
    use crate::agent::remote::EventForwarder;

    #[test]
    #[should_panic(expected = "requires the 'remote' feature")]
    fn test_stub_panics_on_start() {
        let fanout = std::sync::Arc::new(crate::event_fanout::EventFanout::new());
        let _ = EventForwarder::start(fanout, (), "test".to_string(), "sess".to_string());
    }
}

// ═══════════════════════════════════════════════════════════════════════════
//  RemoteNodeManager — lifecycle tests (requires `remote` feature)
// ═══════════════════════════════════════════════════════════════════════════

#[cfg(feature = "remote")]
mod node_manager_tests {
    use super::*;
    use crate::agent::messages::GetMode;
    use crate::agent::remote::node_manager::{
        AdmissionRequest, AdmissionResponse, CreateRemoteSchedule, CreateRemoteSession,
        DeleteRemoteSchedule, ForkRemoteSession, GetNodeInfo, ListRemoteSchedules,
        ListRemoteSessions, PauseRemoteSchedule, RemoteNodeManager, ResumeRemoteSchedule,
        ResumeRemoteSession, SessionHandoff, StopRemoteSessionRuntime,
    };
    use crate::agent::remote::test_helpers::fixtures::get_test_mesh;
    use crate::model::{AgentMessage, MessagePart};
    use crate::session::domain::IntentSnapshot;
    use kameo::actor::{ActorRef, Spawn};
    use kameo::error::{RemoteSendError, SendError};
    use querymt::chat::ChatRole;
    use std::time::Duration;
    use tokio::sync::Mutex;
    use uuid::Uuid;

    /// Spawn a `RemoteNodeManager` actor without mesh and return its `ActorRef`.
    async fn spawn_test_node_manager_no_mesh()
    -> (ActorRef<RemoteNodeManager>, Arc<AgentConfig>, TempDir) {
        let (config, td) = test_agent_config().await;
        let registry = Arc::new(Mutex::new(SessionRegistry::new(config.clone())));
        let nm = RemoteNodeManager::new(
            config.clone(),
            registry,
            None,
            Arc::new(parking_lot::Mutex::new(None)),
        );
        let actor_ref = RemoteNodeManager::spawn(nm);
        (actor_ref, config, td)
    }

    /// Spawn a `RemoteNodeManager` actor with mesh enabled and return its `ActorRef`.
    async fn spawn_test_node_manager_with_mesh()
    -> (ActorRef<RemoteNodeManager>, Arc<AgentConfig>, TempDir) {
        let mesh = get_test_mesh().await;
        let (config, td) = test_agent_config().await;
        let registry = Arc::new(Mutex::new(SessionRegistry::new(config.clone())));
        let nm = RemoteNodeManager::new(
            config.clone(),
            registry,
            Some(mesh.clone()),
            Arc::new(parking_lot::Mutex::new(None)),
        );
        let actor_ref = RemoteNodeManager::spawn(nm);
        (actor_ref, config, td)
    }

    #[tokio::test]
    async fn test_admission_request_invite_preserves_mesh_name() {
        let mesh = get_test_mesh().await;
        let (config, _td) = test_agent_config().await;
        let registry = Arc::new(Mutex::new(SessionRegistry::new(config.clone())));
        let nm_ref = RemoteNodeManager::spawn(RemoteNodeManager::new(
            config,
            registry,
            Some(mesh.clone()),
            Arc::new(parking_lot::Mutex::new(None)),
        ));

        let invite = mesh
            .create_invite(Some("named-mesh".to_string()), None, Some(1), false)
            .unwrap();
        let joiner_peer_id = libp2p::identity::Keypair::generate_ed25519()
            .public()
            .to_peer_id()
            .to_string();

        let response = nm_ref
            .ask(AdmissionRequest::Invite {
                invite_id: invite.grant.invite_id.clone(),
                mesh_name: invite.grant.mesh_name.clone(),
                peer_id: joiner_peer_id,
            })
            .await
            .expect("admission request should succeed");

        match response {
            AdmissionResponse::Admitted {
                membership_token, ..
            } => {
                assert_eq!(
                    membership_token.mesh_id,
                    crate::agent::remote::invite::mesh_id_for(
                        &mesh.peer_id().to_string(),
                        Some("named-mesh"),
                    )
                );
            }
            other => panic!("expected admitted response, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn test_create_session_no_cwd() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create session should succeed");

        assert!(
            !resp.session_id.is_empty(),
            "session_id should not be empty"
        );
        assert!(resp.created_at >= 0, "created_at should be populated");
        assert!(matches!(resp.handoff, SessionHandoff::DirectRemote { .. }));
    }

    #[tokio::test]
    async fn test_create_session_no_mesh_never_claims_lookup_fallback() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_no_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create session without mesh should succeed");

        assert!(
            matches!(
                resp.handoff,
                SessionHandoff::DirectRemote { .. } | SessionHandoff::NoAttachPath
            ),
            "no-mesh environments must not advertise lookup-only handoff"
        );

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        assert!(sessions.iter().any(|s| s.session_id == resp.session_id));
    }

    #[tokio::test]
    async fn test_create_session_valid_cwd() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;
        let cwd_dir = TempDir::new().unwrap();

        let resp = nm_ref
            .ask(CreateRemoteSession {
                cwd: Some(cwd_dir.path().to_string_lossy().to_string()),
            })
            .await
            .expect("create session with valid cwd should succeed");

        assert!(!resp.session_id.is_empty());
    }

    #[tokio::test]
    async fn test_create_session_nonexistent_cwd() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession {
                cwd: Some("/this/path/does/not/exist/anywhere".to_string()),
            })
            .await
            .expect("create session with nonexistent cwd should succeed (workspace index skipped)");

        assert!(!resp.session_id.is_empty());
    }

    #[tokio::test]
    async fn test_create_session_relative_cwd_rejected() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let result = nm_ref
            .ask(CreateRemoteSession {
                cwd: Some("relative/path".to_string()),
            })
            .await;

        match result {
            Err(SendError::HandlerError(err)) => {
                assert!(
                    err.to_string().contains("absolute"),
                    "error should mention absolute path: {:?}",
                    err
                );
            }
            Ok(_) => panic!("relative cwd should be rejected, but got Ok"),
            Err(e) => panic!("unexpected error variant: {:?}", e),
        }
    }

    #[tokio::test]
    async fn test_list_sessions_empty() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list should succeed")
            .sessions;

        assert!(sessions.is_empty());
    }

    #[tokio::test]
    async fn test_list_sessions_after_create() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create");

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;

        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].session_id, resp.session_id);
        assert!(!sessions[0].peer_label.is_empty());
    }

    #[tokio::test]
    async fn test_remote_schedule_create_rejects_missing_persisted_session() {
        let (nm_ref, config, _td) = spawn_test_node_manager_with_mesh().await;

        let session = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create session");

        config
            .provider
            .history_store()
            .delete_session(&session.session_id)
            .await
            .expect("delete persisted session row");

        let result = nm_ref
            .ask(CreateRemoteSchedule {
                session_id: session.session_id.clone(),
                prompt: "run every minute".to_string(),
                trigger: crate::session::domain_schedule::ScheduleTrigger::Interval { seconds: 60 },
                max_steps: Some(5),
                max_cost_usd: Some(1.5),
                max_runs: Some(3),
            })
            .await;

        match result {
            Err(SendError::HandlerError(err)) => {
                let message = err.to_string();
                assert!(
                    message.contains(&session.session_id),
                    "missing-session error should include session id: {message}"
                );
            }
            Ok(_) => panic!("expected missing persisted session to reject schedule creation"),
            Err(e) => panic!("unexpected error variant: {e:?}"),
        }
    }

    #[tokio::test]
    async fn test_remote_schedule_lifecycle() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let session = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create session");

        let created = nm_ref
            .ask(CreateRemoteSchedule {
                session_id: session.session_id.clone(),
                prompt: "run every minute".to_string(),
                trigger: crate::session::domain_schedule::ScheduleTrigger::Interval { seconds: 60 },
                max_steps: Some(5),
                max_cost_usd: Some(1.5),
                max_runs: Some(3),
            })
            .await
            .expect("create schedule");

        let list = nm_ref
            .ask(ListRemoteSchedules {
                session_id: Some(session.session_id.clone()),
            })
            .await
            .expect("list schedules");
        assert_eq!(list.schedules.len(), 1);
        assert_eq!(list.schedules[0].public_id, created.schedule_public_id);
        assert_eq!(list.schedules[0].config.max_runs, Some(3));
        assert_eq!(
            list.schedules[0]
                .config
                .execution_limits
                .as_ref()
                .and_then(|limits| limits.max_steps),
            Some(5)
        );

        nm_ref
            .ask(PauseRemoteSchedule {
                schedule_public_id: created.schedule_public_id.clone(),
            })
            .await
            .expect("pause schedule");
        let paused = nm_ref
            .ask(ListRemoteSchedules {
                session_id: Some(session.session_id.clone()),
            })
            .await
            .expect("list paused schedules");
        assert_eq!(paused.schedules[0].state.to_string(), "paused");

        nm_ref
            .ask(ResumeRemoteSchedule {
                schedule_public_id: created.schedule_public_id.clone(),
            })
            .await
            .expect("resume schedule");
        nm_ref
            .ask(DeleteRemoteSchedule {
                schedule_public_id: created.schedule_public_id.clone(),
            })
            .await
            .expect("delete schedule");
        let after_delete = nm_ref
            .ask(ListRemoteSchedules {
                session_id: Some(session.session_id),
            })
            .await
            .expect("list after delete");
        assert!(after_delete.schedules.is_empty());
    }

    #[tokio::test]
    async fn test_list_sessions_uses_initial_intent_snapshot_as_title() {
        let (nm_ref, config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create");

        let session = config
            .provider
            .history_store()
            .get_session(&resp.session_id)
            .await
            .expect("session lookup")
            .expect("created session should exist");

        config
            .provider
            .history_store()
            .create_intent_snapshot(IntentSnapshot {
                id: 0,
                session_id: session.id,
                task_id: None,
                summary: "Remote title from snapshot".to_string(),
                constraints: None,
                next_step_hint: None,
                created_at: time::OffsetDateTime::now_utc(),
            })
            .await
            .expect("create intent snapshot");

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        let listed = sessions
            .iter()
            .find(|s| s.session_id == resp.session_id)
            .expect("created session should be listed");

        assert_eq!(listed.title.as_deref(), Some("Remote title from snapshot"));
    }

    #[tokio::test]
    async fn test_create_multiple_sessions() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let mut session_ids = Vec::new();
        for _ in 0..3 {
            let resp = nm_ref
                .ask(CreateRemoteSession { cwd: None })
                .await
                .expect("create");

            session_ids.push(resp.session_id);
        }

        // All IDs should be unique
        let mut sorted = session_ids.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(sorted.len(), 3, "all session IDs should be unique");

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        assert_eq!(sessions.len(), 3);
    }

    #[tokio::test]
    async fn test_stop_session_runtime() {
        let (nm_ref, config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create");

        nm_ref
            .ask(StopRemoteSessionRuntime {
                session_id: resp.session_id.clone(),
            })
            .await
            .expect("stop runtime should succeed");

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        // Stop removes the runtime actor but the session row persists in
        // SQLite, so ListRemoteSessions (which now queries the session store)
        // still returns it.
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].session_id, resp.session_id);
        assert!(
            config
                .provider
                .history_store()
                .get_session(&resp.session_id)
                .await
                .expect("session lookup")
                .is_some(),
            "stop should remove only the runtime, not persisted history"
        );
    }

    #[tokio::test]
    async fn test_stop_nonexistent_session_runtime() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let result = nm_ref
            .ask(StopRemoteSessionRuntime {
                session_id: "does-not-exist".to_string(),
            })
            .await;

        assert!(
            matches!(result, Err(SendError::HandlerError(_))),
            "stopping nonexistent session should fail with HandlerError"
        );
    }

    #[tokio::test]
    async fn test_get_node_info() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let info = nm_ref
            .ask(GetNodeInfo)
            .await
            .expect("get_node_info should succeed");

        assert!(!info.hostname.is_empty(), "hostname should not be empty");
        assert_eq!(info.active_sessions, 0);
        assert!(!info.capabilities.is_empty());
    }

    #[tokio::test]
    async fn test_get_node_info_reflects_session_count() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        for _ in 0..2 {
            nm_ref
                .ask(CreateRemoteSession { cwd: None })
                .await
                .expect("create");
        }

        let info = nm_ref.ask(GetNodeInfo).await.expect("info");
        assert_eq!(info.active_sessions, 2);
    }

    #[tokio::test]
    async fn test_create_session_emits_events() {
        let (nm_ref, config, _td) = spawn_test_node_manager_no_mesh().await;

        let mut rx = config.subscribe_events();

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create");

        let event = tokio::time::timeout(std::time::Duration::from_millis(500), async {
            loop {
                let event = rx.recv().await.expect("recv");
                if event.session_id() == resp.session_id
                    && matches!(event.kind(), AgentEventKind::SessionCreated)
                {
                    break event;
                }
            }
        })
        .await
        .expect("should receive SessionCreated event within timeout");

        assert_eq!(event.session_id(), resp.session_id);
        assert!(matches!(event.kind(), AgentEventKind::SessionCreated));
    }

    #[tokio::test]
    async fn test_create_session_cwd_tracked_in_list() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;
        let cwd_dir = TempDir::new().unwrap();
        let cwd_str = cwd_dir.path().to_string_lossy().to_string();

        let resp = nm_ref
            .ask(CreateRemoteSession {
                cwd: Some(cwd_str.clone()),
            })
            .await
            .expect("create");

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;

        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].session_id, resp.session_id);
        assert_eq!(sessions[0].cwd, Some(cwd_str));
    }

    #[tokio::test]
    async fn test_stop_then_create_reuses_slot() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        // Create + destroy
        let resp1 = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create 1");

        nm_ref
            .ask(StopRemoteSessionRuntime {
                session_id: resp1.session_id.clone(),
            })
            .await
            .expect("stop");

        // Create again — should succeed and produce a different session ID
        let resp2 = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create 2");

        assert_ne!(resp1.session_id, resp2.session_id);

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        // Both sessions appear: resp1 is stopped but still persisted in
        // SQLite, resp2 is the newly created live session.
        assert_eq!(sessions.len(), 2);
        let ids: Vec<&str> = sessions.iter().map(|s| s.session_id.as_str()).collect();
        assert!(
            ids.contains(&resp1.session_id.as_str()),
            "stopped session should still be listed from SQLite"
        );
        assert!(
            ids.contains(&resp2.session_id.as_str()),
            "new session should be listed"
        );
    }

    #[tokio::test]
    async fn test_resume_stopped_session() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create");

        nm_ref
            .ask(StopRemoteSessionRuntime {
                session_id: resp.session_id.clone(),
            })
            .await
            .expect("stop");

        let resumed = nm_ref
            .ask(ResumeRemoteSession {
                session_id: resp.session_id.clone(),
            })
            .await
            .expect("resume");

        assert_eq!(resumed.session_id, resp.session_id);

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].session_id, resp.session_id);
    }

    /// Regression test: after stopping and resuming a session, the
    /// materialized actor's event stream should include previously persisted
    /// events.  This validates that `ResumeRemoteSession` →
    /// `materialize_remote_session` → `GetEventStream` reads from the
    /// durable journal, not just from ephemeral actor state.
    #[tokio::test]
    async fn test_resume_session_exposes_persisted_event_stream() {
        let (nm_ref, config, _td) = spawn_test_node_manager_with_mesh().await;

        let control_timeout = Duration::from_secs(5);
        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .mailbox_timeout(control_timeout)
            .reply_timeout(control_timeout)
            .send()
            .await
            .unwrap_or_else(|err| panic!("create timed out after {:?}: {err}", control_timeout));

        // Persist a durable event through the agent's event sink.
        config
            .emit_event_persisted(
                &resp.session_id,
                crate::events::AgentEventKind::SessionCreated,
            )
            .await
            .expect("persist SessionCreated event");

        // Stop the live actor so the session is no longer in-memory.
        nm_ref
            .ask(StopRemoteSessionRuntime {
                session_id: resp.session_id.clone(),
            })
            .mailbox_timeout(control_timeout)
            .reply_timeout(control_timeout)
            .send()
            .await
            .unwrap_or_else(|err| panic!("stop timed out after {:?}: {err}", control_timeout));

        // Resume — materializes a new actor from SQLite.
        let resumed = nm_ref
            .ask(ResumeRemoteSession {
                session_id: resp.session_id.clone(),
            })
            .mailbox_timeout(control_timeout)
            .reply_timeout(control_timeout)
            .send()
            .await
            .unwrap_or_else(|err| panic!("resume timed out after {:?}: {err}", control_timeout));

        // Resolve the handoff to get a live actor reference.
        let session_ref = match resumed.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref,
            other => panic!("expected DirectRemote handoff, got {other:?}"),
        };

        // Ask the resumed actor for its full event stream with bounded remote
        // mailbox/reply timeouts so suite-level mesh issues fail fast.
        let timeout = Duration::from_secs(5);
        let events: Vec<crate::events::AgentEvent> = querymt_remote::ask_remote_with_timeout(
            &session_ref,
            &crate::agent::messages::GetEventStream,
            timeout,
        )
        .await
        .unwrap_or_else(|err| match err {
            RemoteSendError::ReplyTimeout => {
                panic!("GetEventStream remote reply timed out after {:?}", timeout)
            }
            other => panic!("GetEventStream RPC failed after {:?}: {other}", timeout),
        });

        assert!(
            !events.is_empty(),
            "resumed session should expose at least one persisted event"
        );
        assert!(
            events.iter().any(|e| e.session_id == resp.session_id),
            "events should belong to the resumed session"
        );
    }

    /// Sequential resume of the same session must reuse the existing actor
    /// instead of spawning a duplicate.
    #[tokio::test]
    async fn test_resume_same_session_reuses_actor() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let resp = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create");

        let first_handoff = nm_ref
            .ask(ResumeRemoteSession {
                session_id: resp.session_id.clone(),
            })
            .await
            .expect("first resume");

        let first_id = match first_handoff.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref.id().sequence_id(),
            other => panic!("expected DirectRemote, got {other:?}"),
        };

        let second_handoff = nm_ref
            .ask(ResumeRemoteSession {
                session_id: resp.session_id.clone(),
            })
            .await
            .expect("second resume");

        let second_id = match second_handoff.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref.id().sequence_id(),
            other => panic!("expected DirectRemote, got {other:?}"),
        };

        assert_eq!(
            first_id, second_id,
            "second resume must return the same actor as the first"
        );

        // Only one entry in the session list.
        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        let count = sessions
            .iter()
            .filter(|s| s.session_id == resp.session_id)
            .count();
        assert_eq!(count, 1, "session should appear exactly once in the list");
    }

    /// Resuming two different sessions concurrently should succeed for both.
    #[tokio::test]
    async fn test_resume_different_sessions_both_succeed() {
        let (nm_ref, _config, _td) = spawn_test_node_manager_with_mesh().await;

        let s1 = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create s1");

        let s2 = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create s2");

        // Resume both concurrently — neither should block the other.
        let (r1, r2) = tokio::join!(
            nm_ref.ask(ResumeRemoteSession {
                session_id: s1.session_id.clone()
            }),
            nm_ref.ask(ResumeRemoteSession {
                session_id: s2.session_id.clone()
            }),
        );

        let h1 = r1.expect("resume s1");
        let h2 = r2.expect("resume s2");

        let id1 = match h1.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref.id().sequence_id(),
            other => panic!("s1: expected DirectRemote, got {other:?}"),
        };
        let id2 = match h2.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref.id().sequence_id(),
            other => panic!("s2: expected DirectRemote, got {other:?}"),
        };

        assert_ne!(id1, id2, "different sessions must have different actors");
    }

    #[tokio::test]
    async fn test_fork_remote_session_returns_live_child_ref() {
        let (nm_ref, config, _td) = spawn_test_node_manager_with_mesh().await;

        let parent = nm_ref
            .ask(CreateRemoteSession { cwd: None })
            .await
            .expect("create parent");

        let message_id = Uuid::new_v4().to_string();
        config
            .provider
            .history_store()
            .add_message(
                &parent.session_id,
                AgentMessage {
                    id: message_id.clone(),
                    session_id: parent.session_id.clone(),
                    role: ChatRole::User,
                    parts: vec![MessagePart::Text {
                        content: "fork here".to_string(),
                    }],
                    created_at: time::OffsetDateTime::now_utc().unix_timestamp(),
                    parent_message_id: None,
                    source_provider: None,
                    source_model: None,
                },
            )
            .await
            .expect("insert parent message");

        let resp = nm_ref
            .ask(ForkRemoteSession {
                source_session_id: parent.session_id.clone(),
                message_id,
            })
            .await
            .expect("fork child");

        assert_ne!(resp.session_id, parent.session_id);
        let session_ref = match resp.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref,
            SessionHandoff::LookupOnly => panic!("expected direct handoff in mesh test"),
            SessionHandoff::NoAttachPath => panic!("expected attachable handoff in mesh test"),
        };
        assert!(session_ref.ask(&GetMode).await.is_ok());

        let sessions = nm_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .await
            .expect("list")
            .sessions;
        assert!(sessions.iter().any(|s| s.session_id == resp.session_id));
    }
}

// ═══════════════════════════════════════════════════════════════════════════
//  SessionActorRef identification edge cases
// ═══════════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn test_from_actor_ref_into_local() {
    let (config, _td) = test_agent_config().await;
    let runtime = SessionRuntime::new(
        None,
        HashMap::new(),
        crate::agent::core::McpToolState::empty(),
    );
    let actor = SessionActor::new(config, "from-test".to_string(), runtime);
    let actor_ref = SessionActor::spawn(actor);

    // Test the From<ActorRef<SessionActor>> conversion
    let session_ref: SessionActorRef = actor_ref.into();
    assert!(!session_ref.is_remote());
    assert_eq!(session_ref.node_label(), "local");
}

#[tokio::test]
async fn test_session_actor_ref_clone() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config, "clone-test");
    let cloned = session_ref.clone();

    // Both should work independently
    let m1 = session_ref.get_mode().await.expect("get_mode original");
    let m2 = cloned.get_mode().await.expect("get_mode clone");
    assert_eq!(m1, m2);
}

// ═══════════════════════════════════════════════════════════════════════════
//  SessionRegistry — interaction with SessionActorRef
// ═══════════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn test_registry_get_returns_working_ref() {
    let (config, _td) = test_agent_config().await;
    let session_ref = spawn_test_session(config.clone(), "reg-work-1");

    let mut registry = SessionRegistry::new(config);
    registry.insert("reg-work-1".to_string(), session_ref);

    // Get the ref from registry and use it
    let stored_ref = registry.get("reg-work-1").expect("should exist");
    let mode = stored_ref
        .get_mode()
        .await
        .expect("get_mode via registry ref");
    assert_eq!(mode, AgentMode::Build);
}

#[tokio::test]
async fn test_registry_insert_via_actor_ref() {
    let (config, _td) = test_agent_config().await;
    let runtime = SessionRuntime::new(
        None,
        HashMap::new(),
        crate::agent::core::McpToolState::empty(),
    );
    let actor = SessionActor::new(config.clone(), "raw-actor".to_string(), runtime);
    let actor_ref = SessionActor::spawn(actor);

    let mut registry = SessionRegistry::new(config);
    // insert() accepts `impl Into<SessionActorRef>`, including bare ActorRef
    registry.insert("raw-actor".to_string(), actor_ref);

    let stored = registry.get("raw-actor").expect("should exist");
    assert!(!stored.is_remote());
}
