//! Modules D (extended) + G — Session lifecycle integration tests.
//!
//! Two logical nodes ("alpha" and "beta") share a single in-process mesh.
//! Alpha creates, inspects, and stops session runtimes on Beta.
//!
//! Module D: additional `RemoteNodeManager` tests (event emission, HOSTNAME
//! env override, stop-then-create, create-stop loop).
//!
//! Module G: full two-node lifecycle (create/list/stop/attach/set-mode).

#[cfg(all(test, feature = "remote"))]
mod node_manager_extended_tests {
    use crate::agent::remote::node_manager::{
        CreateRemoteSession, GetNodeInfo, ListRemoteSessions, StopRemoteSessionRuntime,
    };
    use crate::agent::remote::test_helpers::fixtures::NodeManagerFixture;
    use crate::events::AgentEventKind;

    // ── D.3 — HOSTNAME env override ───────────────────────────────────────────

    #[tokio::test]
    async fn test_get_node_info_hostname_env_override() {
        // Set HOSTNAME before spawning the node manager so the env is visible.
        // SAFETY: test is single-threaded at this point; no concurrent env reads.
        unsafe { std::env::set_var("HOSTNAME", "testhost-d3") };

        let f = NodeManagerFixture::new_with_mesh().await;
        let info = f
            .actor_ref
            .ask(GetNodeInfo)
            .await
            .expect("get_node_info should succeed");

        // Restore env (tests run in the same process).
        // SAFETY: same single-threaded context as set_var above.
        unsafe { std::env::remove_var("HOSTNAME") };

        assert_eq!(
            info.hostname, "testhost-d3",
            "hostname should reflect HOSTNAME env var override"
        );
    }

    // ── D.4 — Stop kills the session actor (runtime only, history persists) ───

    #[tokio::test]
    async fn test_stop_session_runtime_shuts_down_actor() {
        let f = NodeManagerFixture::new_with_mesh().await;

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

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

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

        // The session row persists in SQLite after stopping the runtime.
        // Verify the session is still listed but no longer has a live actor.
        let stopped = sessions
            .iter()
            .find(|s| s.session_id == resp.session_id)
            .expect("stopped session should still be listed from SQLite");
        assert_eq!(
            stopped.actor_id, 0,
            "stopped session should have no live actor (actor_id == 0)"
        );
        assert_eq!(
            stopped.runtime_state,
            Some("persisted".to_string()),
            "stopped session should have runtime_state 'persisted'"
        );
    }

    // ── D.5 — No credentials → empty model list ───────────────────────────────

    #[tokio::test]
    async fn test_node_manager_get_node_info_still_works_without_provider_listing() {
        // Keep one no-mesh regression path to ensure the node manager still
        // responds on core control-plane APIs after provider discovery moved out.
        let f = NodeManagerFixture::new_with_mesh().await;
        let info = f
            .actor_ref
            .ask(GetNodeInfo)
            .await
            .expect("get_node_info should succeed");

        assert!(
            !info.node_id.to_string().is_empty(),
            "node manager should still expose a stable node id"
        );
    }

    // ── D.6 — 10 create+stop cycles leave no live actors ─────────────────────

    #[tokio::test]
    async fn test_create_stop_create_sequence_no_runtime_leak() {
        let f = NodeManagerFixture::new_with_mesh().await;

        let timeout = std::time::Duration::from_secs(3);

        for i in 0..10 {
            let resp = f
                .actor_ref
                .ask(CreateRemoteSession { cwd: None })
                .mailbox_timeout(timeout)
                .reply_timeout(timeout)
                .send()
                .await
                .unwrap_or_else(|_| panic!("create timed out at iteration {i}"));

            f.actor_ref
                .ask(StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                })
                .mailbox_timeout(timeout)
                .reply_timeout(timeout)
                .send()
                .await
                .unwrap_or_else(|_| panic!("stop timed out at iteration {i}"));
        }

        let sessions = f
            .actor_ref
            .ask(ListRemoteSessions {
                offset: None,
                limit: None,
            })
            .mailbox_timeout(timeout)
            .reply_timeout(timeout)
            .send()
            .await
            .expect("list")
            .sessions;

        // Sessions persist in SQLite but no live actors should remain.
        assert_eq!(
            sessions.len(),
            10,
            "all 10 sessions should be listed from SQLite"
        );
        assert!(
            sessions.iter().all(|s| s.actor_id == 0),
            "all stopped sessions should have actor_id == 0, got actors: {:?}",
            sessions
                .iter()
                .filter(|s| s.actor_id != 0)
                .collect::<Vec<_>>()
        );
    }

    // ── D.1 — CreateRemoteSession emits ProviderChanged event ─────────────────

    #[tokio::test]
    async fn test_create_session_emits_provider_changed_event() {
        let f = NodeManagerFixture::new_with_mesh().await;

        let mut rx = f.config.subscribe_events();

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

        // Collect events until we see SessionCreated and optionally ProviderChanged.
        let mut saw_session_created = false;
        let mut saw_provider_changed = false;

        let deadline = tokio::time::Instant::now() + std::time::Duration::from_millis(500);

        loop {
            match tokio::time::timeout_at(deadline, rx.recv()).await {
                Ok(Ok(event)) if event.session_id() == resp.session_id => {
                    match event.kind() {
                        AgentEventKind::SessionCreated => saw_session_created = true,
                        AgentEventKind::ProviderChanged { .. } => saw_provider_changed = true,
                        _ => {}
                    }
                    if saw_session_created {
                        break;
                    }
                }
                Ok(Ok(_)) => continue, // event for another session
                _ => break,
            }
        }

        assert!(
            saw_session_created,
            "SessionCreated event should be emitted on CreateRemoteSession"
        );
        // ProviderChanged is emitted after model configuration; it may or may
        // not appear depending on the config.  We document the expectation.
        let _ = saw_provider_changed; // optionally assert when config emits it
    }
}

// ═════════════════════════════════════════════════════════════════════════════
//  Module G — Full two-node session lifecycle integration tests
// ═════════════════════════════════════════════════════════════════════════════

#[cfg(all(test, feature = "remote"))]
mod remote_session_lifecycle_integration_tests {
    use crate::agent::core::AgentMode;
    use crate::agent::messages::GetMode;
    use crate::agent::remote::SessionActorRef;
    use crate::agent::remote::node_manager::SessionHandoff;
    use crate::agent::remote::node_manager::{
        CreateRemoteSession, ForkRemoteSession, GetNodeInfo, ListRemoteSessions,
        StopRemoteSessionRuntime,
    };
    use crate::agent::remote::test_helpers::fixtures::{TwoNodeFixture, get_test_mesh};
    use crate::agent::session_actor::SessionActor;
    use crate::model::{AgentMessage, MessagePart};
    use querymt::chat::ChatRole;
    use std::future::Future;
    use std::time::Duration;
    use uuid::Uuid;

    const STEP_TIMEOUT: Duration = Duration::from_secs(15);

    async fn within_timeout<T>(label: &str, fut: impl Future<Output = T>) -> T {
        tokio::time::timeout(STEP_TIMEOUT, fut)
            .await
            .unwrap_or_else(|_| panic!("{} timed out after {:?}", label, STEP_TIMEOUT))
    }

    async fn assert_node_manager_lookup(
        mesh: &crate::agent::remote::mesh::MeshHandle,
        dht_name: &str,
        expected_actor_id: kameo::actor::ActorId,
    ) {
        let looked_up = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(dht_name),
        )
        .await
        .expect("lookup failed")
        .expect("lookup returned none");
        assert_eq!(
            looked_up.id(),
            expected_actor_id,
            "lookup beta node manager should resolve to fixture actor"
        );
    }

    // ── G.1 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_creates_session_on_beta() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;

        // Alpha looks up Beta's node manager via DHT.
        let beta_ref = within_timeout(
            "lookup beta node manager",
            f.mesh
                .lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("DHT lookup")
        .expect("beta not in DHT");

        let resp = within_timeout("CreateRemoteSession on beta", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("CreateRemoteSession on beta");

        assert!(
            !resp.session_id.is_empty(),
            "session_id should not be empty"
        );
        assert!(resp.created_at >= 0, "created_at should be populated");

        // Verify via Beta's local list.
        let sessions = within_timeout("ListRemoteSessions on beta", async {
            f.beta
                .actor_ref
                .ask(ListRemoteSessions {
                    offset: None,
                    limit: None,
                })
                .await
        })
        .await
        .map(|r| r.sessions)
        .expect("ListRemoteSessions on beta");
        assert!(
            sessions.iter().any(|s| s.session_id == resp.session_id),
            "created session should appear in Beta's list"
        );

        let _ = within_timeout("destroy created session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.2 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_creates_session_on_beta_appears_in_dht() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup beta")
        .expect("beta not found");

        let resp = within_timeout("create session on beta", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create");

        // Give DHT a moment to propagate the session registration.
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        let dht_name = crate::agent::remote::scope::scoped_session(
            &crate::agent::remote::scope::MeshScopeId::lan_default(),
            &resp.session_id,
        );
        let session_actor_ref = within_timeout(
            "lookup session in DHT",
            mesh.lookup_actor::<SessionActor>(&dht_name),
        )
        .await
        .expect("DHT lookup for session");

        assert!(
            session_actor_ref.is_some(),
            "session '{}' should be findable in DHT after creation",
            resp.session_id
        );

        let _ = within_timeout("destroy created session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.3 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_lists_betas_sessions() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        // Create two sessions on beta.
        let r1 = within_timeout("create 1", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create 1");
        let r2 = within_timeout("create 2", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create 2");

        // Alpha lists beta's sessions.
        let sessions: Vec<_> = within_timeout("list", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &ListRemoteSessions {
                    offset: None,
                    limit: None,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("list")
        .sessions;

        assert_eq!(sessions.len(), 2, "beta should have 2 sessions");
        let ids: Vec<&str> = sessions.iter().map(|s| s.session_id.as_str()).collect();
        assert!(ids.contains(&r1.session_id.as_str()));
        assert!(ids.contains(&r2.session_id.as_str()));

        let _ = within_timeout("stop 1", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: r1.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        let _ = within_timeout("stop 2", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: r2.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    #[tokio::test]
    async fn test_fork_remote_session_returns_attachable_child_without_lookup() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let parent = within_timeout("create parent", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create parent");

        let message_id = Uuid::new_v4().to_string();
        f.beta
            .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 this point".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 child = within_timeout("fork child", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &ForkRemoteSession {
                    source_session_id: parent.session_id.clone(),
                    message_id: message_id.clone(),
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("fork child");

        assert_ne!(child.session_id, parent.session_id);
        let session_ref = match child.handoff {
            SessionHandoff::DirectRemote { session_ref } => session_ref,
            SessionHandoff::LookupOnly => {
                panic!("expected direct handoff in integration mesh test")
            }
            SessionHandoff::NoAttachPath => {
                panic!("expected attachable handoff in integration mesh test")
            }
        };
        let mode = within_timeout(
            "forked child get_mode",
            querymt_remote::ask_remote_with_timeout(&session_ref, &GetMode, STEP_TIMEOUT),
        )
        .await
        .expect("forked child should be directly usable");
        assert_eq!(mode, AgentMode::Build);

        let sessions: Vec<_> = within_timeout("list", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &ListRemoteSessions {
                    offset: None,
                    limit: None,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("list")
        .sessions;
        assert!(sessions.iter().any(|s| s.session_id == child.session_id));

        let _ = within_timeout("stop parent", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: parent.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        let _ = within_timeout("stop child", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: child.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.4 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_stops_betas_session_runtime() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let resp = within_timeout("create", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create");

        within_timeout("stop", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id.clone(),
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("stop");

        let sessions = within_timeout("list", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &ListRemoteSessions {
                    offset: None,
                    limit: None,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("list")
        .sessions;
        // The session row persists in SQLite after stopping the runtime.
        let stopped = sessions
            .iter()
            .find(|s| s.session_id == resp.session_id)
            .expect("stopped session should still be listed from SQLite");
        assert_eq!(
            stopped.actor_id, 0,
            "stopped session should have no live actor (actor_id == 0)"
        );
        f.cleanup().await;
    }

    // ── G.5 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_lookup_survives_two_stop_cycles() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let first = within_timeout("create first", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create first");

        let second = within_timeout("create second", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create second");

        within_timeout("stop first", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: first.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("stop first");

        assert_node_manager_lookup(mesh, &f.beta.dht_name, f.beta.actor_ref.id()).await;

        within_timeout("stop second", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: second.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("stop second");

        assert_node_manager_lookup(mesh, &f.beta.dht_name, f.beta.actor_ref.id()).await;

        f.cleanup().await;
    }

    #[tokio::test]
    async fn test_alpha_creates_multiple_sessions_on_beta() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");
        assert_eq!(
            beta_ref.id(),
            f.beta.actor_ref.id(),
            "lookup beta node manager should resolve to fixture actor"
        );

        let mut ids = Vec::new();
        for i in 0..3 {
            let resp = within_timeout(&format!("create {i}"), async {
                querymt_remote::ask_remote_with_timeout(
                    &beta_ref,
                    &CreateRemoteSession { cwd: None },
                    STEP_TIMEOUT,
                )
                .await
            })
            .await
            .expect("create");
            ids.push(resp.session_id);
        }

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

        let sessions = within_timeout("list", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &ListRemoteSessions {
                    offset: None,
                    limit: None,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("list")
        .sessions;
        assert_eq!(sessions.len(), 3);

        for id in ids {
            let refreshed_beta_ref = within_timeout(
                "refresh beta node manager",
                mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
            )
            .await
            .expect("refresh lookup")
            .expect("refreshed beta node manager not found");
            assert_eq!(
                refreshed_beta_ref.id(),
                f.beta.actor_ref.id(),
                "refreshed lookup beta node manager should resolve to fixture actor"
            );
            let stop_id = id.clone();
            let _ = within_timeout("stop created session", async {
                querymt_remote::ask_remote_with_timeout(
                    &refreshed_beta_ref,
                    &StopRemoteSessionRuntime {
                        session_id: stop_id,
                    },
                    STEP_TIMEOUT,
                )
                .await
            })
            .await;
        }
        f.cleanup().await;
    }

    // ── G.6 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_attaches_remote_session_to_local_registry() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let resp = within_timeout("create remote session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create");

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Attach by resolving via DHT.
        let dht_name = crate::agent::remote::scope::scoped_session(
            &crate::agent::remote::scope::MeshScopeId::lan_default(),
            &resp.session_id,
        );
        let remote_ref = within_timeout(
            "lookup session actor",
            mesh.lookup_actor::<SessionActor>(&dht_name),
        )
        .await
        .expect("session lookup")
        .expect("session not in DHT");

        let session_ref = SessionActorRef::Remote {
            actor_ref: remote_ref,
            peer_label: "beta".to_string(),
            remote_node_id: None,
        };

        assert!(session_ref.is_remote());
        assert_eq!(session_ref.node_label(), "beta");

        // Keep the shared test mesh clean to reduce cross-test interference.
        let _ = within_timeout("stop remote session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.7 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_sets_mode_on_betas_session() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let resp = within_timeout("create remote session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            )
            .await
        })
        .await
        .expect("create");

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let dht_name = crate::agent::remote::scope::scoped_session(
            &crate::agent::remote::scope::MeshScopeId::lan_default(),
            &resp.session_id,
        );
        let remote_ref = within_timeout(
            "lookup session actor",
            mesh.lookup_actor::<SessionActor>(&dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let session_ref = SessionActorRef::Remote {
            actor_ref: remote_ref,
            peer_label: "beta".to_string(),
            remote_node_id: None,
        };

        within_timeout("set_mode", session_ref.set_mode(AgentMode::Plan))
            .await
            .expect("set_mode");
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

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

        let _ = within_timeout("destroy created session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.8 — Bug #5 (documented) ────────────────────────────────────────────

    /// Documents the DHT propagation race in `attach_remote_session`.
    ///
    /// `SubscribeEvents` is sent before the `EventRelayActor` registration
    /// propagates in the DHT, silently dropping event forwarding.
    ///
    /// When the race is fixed, this test should assert that events from Beta's
    /// session appear on Alpha's local event bus.
    #[tokio::test]
    async fn test_alpha_events_arrive_on_local_bus() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        use crate::agent::remote::event_relay::EventRelayActor;
        use crate::event_fanout::EventFanout;
        use crate::event_sink::EventSink;
        use crate::session::backend::StorageBackend;
        use kameo::actor::Spawn;
        use std::sync::Arc;

        let beta_ref = mesh
            .lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name)
            .await
            .expect("lookup")
            .expect("not found");

        let resp = querymt_remote::ask_remote_with_timeout(
            &beta_ref,
            &CreateRemoteSession { cwd: None },
            STEP_TIMEOUT,
        )
        .await
        .expect("create");

        // Alpha sets up a relay actor and registers it in the DHT under the
        // name that SubscribeEvents will look for on Beta's session.
        let alpha_storage = Arc::new(
            crate::session::sqlite_storage::SqliteStorage::connect(":memory:".into())
                .await
                .unwrap(),
        );
        let alpha_journal = alpha_storage.event_journal();
        let alpha_fanout = Arc::new(EventFanout::new());
        let alpha_sink = Arc::new(EventSink::new(alpha_journal, alpha_fanout.clone()));
        let relay = EventRelayActor::new(
            alpha_sink.clone(),
            resp.session_id.clone(),
            "alpha-relay-g8".to_string(),
            None,
            None,
        );
        let relay_ref = EventRelayActor::spawn(relay);

        let relay_dht_name = crate::agent::remote::scope::scoped_event_relay(
            &crate::agent::remote::scope::MeshScopeId::lan_default(),
            &resp.session_id,
            mesh.peer_id(),
        );
        mesh.register_actor(relay_ref.clone(), relay_dht_name.clone())
            .await;
        let _ = relay_ref;

        // Allow DHT propagation before subscribing (mitigates Bug #5 for this test).
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Look up and subscribe.
        let dht_name = crate::agent::remote::scope::scoped_session(
            &crate::agent::remote::scope::MeshScopeId::lan_default(),
            &resp.session_id,
        );
        let remote_ref = mesh
            .lookup_actor::<SessionActor>(&dht_name)
            .await
            .expect("lookup")
            .expect("not found");

        let session_ref = SessionActorRef::Remote {
            actor_ref: remote_ref,
            peer_label: "beta".to_string(),
            remote_node_id: None,
        };

        session_ref
            .subscribe_events(1, relay_dht_name)
            .await
            .expect("subscribe");
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Trigger an event on Beta's session by changing mode.
        session_ref
            .set_mode(AgentMode::Plan)
            .await
            .expect("set_mode");
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Bug #5: if the DHT race fires, no events may arrive on alpha fanout.
        // We document expected behaviour without asserting strictly.
        // When fixed: assert alpha fanout has received at least one event.
        let _ = alpha_fanout.subscriber_count(); // just ensure no panic

        let _ = within_timeout("destroy created session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.9 ──────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_cancels_betas_idle_session() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = mesh
            .lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name)
            .await
            .expect("lookup")
            .expect("not found");

        let resp = querymt_remote::ask_remote_with_timeout(
            &beta_ref,
            &CreateRemoteSession { cwd: None },
            STEP_TIMEOUT,
        )
        .await
        .expect("create");

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        let dht_name = crate::agent::remote::scope::scoped_session(
            &crate::agent::remote::scope::MeshScopeId::lan_default(),
            &resp.session_id,
        );
        let remote_ref = mesh
            .lookup_actor::<SessionActor>(&dht_name)
            .await
            .expect("lookup")
            .expect("not found");

        let session_ref = SessionActorRef::Remote {
            actor_ref: remote_ref,
            peer_label: "beta".to_string(),
            remote_node_id: None,
        };

        let result = session_ref.cancel().await;
        assert!(
            result.is_ok(),
            "cancel on idle remote session should return Ok"
        );

        let _ = within_timeout("destroy created session", async {
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &StopRemoteSessionRuntime {
                    session_id: resp.session_id,
                },
                STEP_TIMEOUT,
            )
            .await
        })
        .await;
        f.cleanup().await;
    }

    // ── G.10 ─────────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_alpha_gets_node_info_from_beta() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = mesh
            .lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name)
            .await
            .expect("lookup")
            .expect("not found");

        let info = querymt_remote::ask_remote_with_timeout(&beta_ref, &GetNodeInfo, STEP_TIMEOUT)
            .await
            .expect("GetNodeInfo");
        assert!(!info.hostname.is_empty(), "hostname should not be empty");
        assert!(
            !info.capabilities.is_empty(),
            "capabilities should not be empty"
        );
        f.cleanup().await;
    }

    // ── G.11 — Concurrent session creation ───────────────────────────────────

    #[tokio::test]
    async fn test_concurrent_sessions_on_beta_from_alpha() {
        let test_id = Uuid::now_v7().to_string();
        let f = TwoNodeFixture::new(&test_id).await;
        let mesh = get_test_mesh().await;

        let beta_ref = within_timeout(
            "lookup beta node manager",
            mesh.lookup_actor::<crate::agent::remote::RemoteNodeManager>(&f.beta.dht_name),
        )
        .await
        .expect("lookup")
        .expect("not found");

        let (r1, r2, r3, r4, r5) = tokio::join!(
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            ),
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            ),
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            ),
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            ),
            querymt_remote::ask_remote_with_timeout(
                &beta_ref,
                &CreateRemoteSession { cwd: None },
                STEP_TIMEOUT,
            ),
        );

        let mut ids = vec![
            r1.expect("1").session_id,
            r2.expect("2").session_id,
            r3.expect("3").session_id,
            r4.expect("4").session_id,
            r5.expect("5").session_id,
        ];
        ids.sort();
        ids.dedup();
        assert_eq!(
            ids.len(),
            5,
            "all 5 concurrent sessions should have unique IDs"
        );

        for id in ids {
            let _ = within_timeout("stop created session", async {
                querymt_remote::ask_remote_with_timeout(
                    &beta_ref,
                    &StopRemoteSessionRuntime { session_id: id },
                    STEP_TIMEOUT,
                )
                .await
            })
            .await;
        }
        f.cleanup().await;
    }
}
