use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use futures::stream;
use roder_api::catalog::PROVIDER_MOCK;
use roder_api::events::RoderEvent;
use roder_api::extension::{ExtensionRegistryBuilder, InferenceEngineId, ToolProviderId};
use roder_api::inference::*;
use roder_api::remote_runner::{
    RemoteRunnerProvider, RemoteRunnerProviderId, RemoteRunnerSession, RunnerCapabilities,
    RunnerCommandId, RunnerCommandRequest, RunnerCommandResult, RunnerDestination,
    RunnerFileReadRequest, RunnerFileReadResult, RunnerFileWriteRequest, RunnerManifest,
    RunnerPortRequest, RunnerPortResult, RunnerSessionState, RunnerSnapshotRef,
};
use roder_api::thread::ThreadStore;
use roder_api::tools::{
    ToolCall, ToolContributor, ToolExecutionContext, ToolExecutor, ToolRegistry, ToolResult,
    ToolSpec,
};
use roder_core::{Runtime, RuntimeConfig, StartTurnRequest, default_instructions};
use roder_ext_jsonl_thread_store::store::{JsonlThreadStore, JsonlThreadStoreFactory};
use roder_ext_runner_sprites::{
    LIVE_ENV as SPRITES_LIVE_ENV, PROVIDER_ID as SPRITES_PROVIDER_ID, SpritesRunnerProvider,
};
use roder_ext_runner_unix_local::UnixLocalRunnerProvider;
use time::OffsetDateTime;

struct FinalEngine {
    requests: Mutex<usize>,
}

struct ToolScriptEngine {
    requests: Mutex<usize>,
}

#[derive(Default)]
struct MockRunnerState {
    files: Mutex<HashMap<String, Vec<u8>>>,
    created: Mutex<usize>,
    resumed: Mutex<usize>,
    commands: Mutex<Vec<RunnerCommandRequest>>,
    default_workspace: Mutex<Option<String>>,
}

#[derive(Clone, Default)]
struct MockRunnerProvider {
    state: Arc<MockRunnerState>,
}

struct MockRunnerToolContributor {
    state: Arc<MockRunnerState>,
}

struct MockRunnerWriteTool {
    state: Arc<MockRunnerState>,
}

struct MockRunnerReadTool {
    state: Arc<MockRunnerState>,
}

#[async_trait::async_trait]
impl InferenceEngine for FinalEngine {
    fn id(&self) -> InferenceEngineId {
        PROVIDER_MOCK.to_string()
    }

    fn capabilities(&self) -> InferenceCapabilities {
        InferenceCapabilities::coding_agent_default()
    }

    async fn list_models(
        &self,
        _ctx: InferenceProviderContext<'_>,
    ) -> anyhow::Result<Vec<ModelDescriptor>> {
        Ok(Vec::new())
    }

    async fn stream_turn(
        &self,
        _ctx: InferenceTurnContext<'_>,
        _request: AgentInferenceRequest,
    ) -> anyhow::Result<InferenceEventStream> {
        *self.requests.lock().unwrap() += 1;
        Ok(Box::pin(stream::iter(vec![
            Ok(InferenceEvent::MessageDelta(MessageDelta {
                text: "done".to_string(),
                phase: None,
            })),
            Ok(InferenceEvent::Completed(CompletionMetadata {
                stop_reason: Some("stop".to_string()),
                provider_response_id: Some("resp".to_string()),
            })),
        ])))
    }
}

#[async_trait::async_trait]
impl InferenceEngine for ToolScriptEngine {
    fn id(&self) -> InferenceEngineId {
        PROVIDER_MOCK.to_string()
    }

    fn capabilities(&self) -> InferenceCapabilities {
        InferenceCapabilities::coding_agent_default()
    }

    async fn list_models(
        &self,
        _ctx: InferenceProviderContext<'_>,
    ) -> anyhow::Result<Vec<ModelDescriptor>> {
        Ok(Vec::new())
    }

    async fn stream_turn(
        &self,
        _ctx: InferenceTurnContext<'_>,
        _request: AgentInferenceRequest,
    ) -> anyhow::Result<InferenceEventStream> {
        let mut requests = self.requests.lock().unwrap();
        *requests += 1;
        let index = *requests;
        drop(requests);
        let events = match index {
            1 => vec![
                Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
                    id: "write".to_string(),
                    name: "write_file".to_string(),
                    arguments: r#"{"path":"out.txt","content":"hello from runner"}"#.to_string(),
                })),
                Ok(InferenceEvent::Completed(CompletionMetadata {
                    stop_reason: Some("tool_calls".to_string()),
                    provider_response_id: Some("resp-write".to_string()),
                })),
            ],
            2 => vec![
                Ok(InferenceEvent::ToolCallCompleted(ToolCallCompleted {
                    id: "read".to_string(),
                    name: "read_file".to_string(),
                    arguments: r#"{"path":"out.txt"}"#.to_string(),
                })),
                Ok(InferenceEvent::Completed(CompletionMetadata {
                    stop_reason: Some("tool_calls".to_string()),
                    provider_response_id: Some("resp-read".to_string()),
                })),
            ],
            _ => vec![
                Ok(InferenceEvent::MessageDelta(MessageDelta {
                    text: "runner workflow complete".to_string(),
                    phase: None,
                })),
                Ok(InferenceEvent::Completed(CompletionMetadata {
                    stop_reason: Some("stop".to_string()),
                    provider_response_id: Some("resp-final".to_string()),
                })),
            ],
        };
        Ok(Box::pin(stream::iter(events)))
    }
}

#[async_trait::async_trait]
impl RemoteRunnerProvider for MockRunnerProvider {
    fn id(&self) -> RemoteRunnerProviderId {
        "mock-hosted".to_string()
    }

    fn capabilities(&self) -> RunnerCapabilities {
        RunnerCapabilities {
            command_exec: true,
            file_read: true,
            file_write: true,
            port_preview: true,
            snapshots: true,
            cancellation: true,
            artifact_export: true,
            mounts: Default::default(),
            pausable: true,
            detachable: true,
        }
    }

    fn default_workspace(&self) -> Option<String> {
        self.state.default_workspace.lock().unwrap().clone()
    }

    async fn create_session(
        &self,
        destination: RunnerDestination,
    ) -> anyhow::Result<Arc<dyn RemoteRunnerSession>> {
        *self.state.created.lock().unwrap() += 1;
        Ok(Arc::new(MockRunnerSession {
            state: self.state.clone(),
            destination_id: destination.id,
            session_id: "mock-session".to_string(),
            paused: std::sync::atomic::AtomicBool::new(false),
        }))
    }

    async fn resume_session(
        &self,
        state: RunnerSessionState,
    ) -> anyhow::Result<Arc<dyn RemoteRunnerSession>> {
        *self.state.resumed.lock().unwrap() += 1;
        Ok(Arc::new(MockRunnerSession {
            state: self.state.clone(),
            destination_id: state.destination_id,
            session_id: state.session_id,
            paused: std::sync::atomic::AtomicBool::new(false),
        }))
    }
}

struct MockRunnerSession {
    state: Arc<MockRunnerState>,
    destination_id: String,
    session_id: String,
    paused: std::sync::atomic::AtomicBool,
}

#[async_trait::async_trait]
impl RemoteRunnerSession for MockRunnerSession {
    fn state(&self) -> RunnerSessionState {
        RunnerSessionState {
            provider_id: "mock-hosted".to_string(),
            session_id: self.session_id.clone(),
            destination_id: self.destination_id.clone(),
            snapshot: None,
            metadata: serde_json::json!({
                "mock": true,
                "paused": self.paused.load(std::sync::atomic::Ordering::SeqCst),
            }),
        }
    }

    async fn pause(&self) -> anyhow::Result<RunnerSessionState> {
        self.paused.store(true, std::sync::atomic::Ordering::SeqCst);
        Ok(self.state())
    }

    async fn resume(&self) -> anyhow::Result<RunnerSessionState> {
        self.paused
            .store(false, std::sync::atomic::Ordering::SeqCst);
        Ok(self.state())
    }

    async fn detach(&self) -> anyhow::Result<RunnerSessionState> {
        Ok(self.state())
    }

    async fn run_command(
        &self,
        request: RunnerCommandRequest,
    ) -> anyhow::Result<RunnerCommandResult> {
        self.state.commands.lock().unwrap().push(request.clone());
        Ok(RunnerCommandResult {
            command_id: request.command_id,
            exit_code: Some(0),
            stdout: "command ok\n".to_string(),
            stderr: String::new(),
        })
    }

    async fn cancel_command(&self, _command_id: &RunnerCommandId) -> anyhow::Result<bool> {
        Ok(true)
    }

    async fn read_file(
        &self,
        request: RunnerFileReadRequest,
    ) -> anyhow::Result<RunnerFileReadResult> {
        let key = request.path.to_string_lossy().to_string();
        let contents = self
            .state
            .files
            .lock()
            .unwrap()
            .get(&key)
            .cloned()
            .unwrap_or_default();
        Ok(RunnerFileReadResult {
            path: request.path,
            contents,
        })
    }

    async fn write_file(&self, request: RunnerFileWriteRequest) -> anyhow::Result<()> {
        self.state
            .files
            .lock()
            .unwrap()
            .insert(request.path.to_string_lossy().to_string(), request.contents);
        Ok(())
    }

    async fn expose_port(&self, request: RunnerPortRequest) -> anyhow::Result<RunnerPortResult> {
        Ok(RunnerPortResult {
            port: request.port,
            url: Some(format!("https://mock.runner/{}", request.port)),
        })
    }

    async fn snapshot(&self) -> anyhow::Result<Option<RunnerSnapshotRef>> {
        Ok(Some(RunnerSnapshotRef {
            provider_id: "mock-hosted".to_string(),
            snapshot_id: "snap-1".to_string(),
            metadata: serde_json::json!({ "kind": "mock" }),
        }))
    }

    async fn close(&self) -> anyhow::Result<()> {
        Ok(())
    }
}

impl ToolContributor for MockRunnerToolContributor {
    fn id(&self) -> ToolProviderId {
        "mock-runner-tools".to_string()
    }

    fn contribute(&self, registry: &mut ToolRegistry) -> anyhow::Result<()> {
        registry.register(Arc::new(MockRunnerWriteTool {
            state: self.state.clone(),
        }))?;
        registry.register(Arc::new(MockRunnerReadTool {
            state: self.state.clone(),
        }))
    }
}

#[async_trait::async_trait]
impl ToolExecutor for MockRunnerWriteTool {
    fn spec(&self) -> ToolSpec {
        ToolSpec {
            name: "write_file".to_string(),
            description: "Write a file through the mock runner".to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "path": { "type": "string" },
                    "content": { "type": "string" }
                },
                "required": ["path", "content"]
            }),
        }
    }

    async fn execute(
        &self,
        _ctx: ToolExecutionContext,
        call: ToolCall,
    ) -> anyhow::Result<ToolResult> {
        let path = call
            .arguments
            .get("path")
            .and_then(serde_json::Value::as_str)
            .unwrap_or_default()
            .to_string();
        let content = call
            .arguments
            .get("content")
            .and_then(serde_json::Value::as_str)
            .unwrap_or_default()
            .as_bytes()
            .to_vec();
        self.state
            .files
            .lock()
            .unwrap()
            .insert(path.clone(), content);
        Ok(ToolResult {
            id: call.id,
            name: call.name,
            text: format!("wrote {path}"),
            data: serde_json::json!({ "path": path }),
            is_error: false,
        })
    }
}

#[async_trait::async_trait]
impl ToolExecutor for MockRunnerReadTool {
    fn spec(&self) -> ToolSpec {
        ToolSpec {
            name: "read_file".to_string(),
            description: "Read a file through the mock runner".to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": { "path": { "type": "string" } },
                "required": ["path"]
            }),
        }
    }

    async fn execute(
        &self,
        _ctx: ToolExecutionContext,
        call: ToolCall,
    ) -> anyhow::Result<ToolResult> {
        let path = call
            .arguments
            .get("path")
            .and_then(serde_json::Value::as_str)
            .unwrap_or_default()
            .to_string();
        let contents = self
            .state
            .files
            .lock()
            .unwrap()
            .get(&path)
            .cloned()
            .unwrap_or_default();
        let text = String::from_utf8(contents)?;
        Ok(ToolResult {
            id: call.id,
            name: call.name,
            text: text.clone(),
            data: serde_json::json!({ "path": path, "contents": text }),
            is_error: false,
        })
    }
}

#[tokio::test]
async fn remote_runner_state_persists_and_resumes_between_turns() {
    let session_dir = temp_dir("remote-runner-sessions");
    let workspace = temp_dir("remote-runner-workspace");
    let runtime = runtime(session_dir.clone(), workspace.clone()).await;
    let mut events = runtime.subscribe_events();
    let metadata = runtime.create_thread(None).await.unwrap();
    assert!(metadata.runner_destination.is_some());
    assert!(metadata.runner_state.is_none());

    start_and_wait(&runtime, &mut events, &metadata.thread_id, "first").await;
    let first_state = runtime
        .load_thread(&metadata.thread_id)
        .await
        .unwrap()
        .unwrap()
        .metadata
        .unwrap()
        .runner_state
        .unwrap();

    start_and_wait(&runtime, &mut events, &metadata.thread_id, "second").await;
    let second_state = runtime
        .load_thread(&metadata.thread_id)
        .await
        .unwrap()
        .unwrap()
        .metadata
        .unwrap()
        .runner_state
        .unwrap();

    assert_eq!(second_state.provider_id, "unix-local");
    assert_eq!(second_state.destination_id, "unix-local");
    assert_eq!(second_state.session_id, first_state.session_id);

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(workspace);
}

#[tokio::test]
async fn stale_remote_runner_state_falls_back_to_fresh_session() {
    let session_dir = temp_dir("remote-runner-stale-sessions");
    let workspace = temp_dir("remote-runner-stale-workspace");
    let runtime = runtime(session_dir.clone(), workspace.clone()).await;
    let mut events = runtime.subscribe_events();
    let metadata = runtime.create_thread(None).await.unwrap();
    let store = JsonlThreadStore {
        base_path: session_dir.clone(),
    };
    let mut stale = metadata;
    stale.runner_state = Some(RunnerSessionState {
        provider_id: "unix-local".to_string(),
        session_id: "stale-session".to_string(),
        destination_id: "unix-local".to_string(),
        snapshot: None,
        metadata: serde_json::json!({ "root": "/definitely/missing/runner/root" }),
    });
    stale.updated_at = OffsetDateTime::now_utc();
    store.update_thread_metadata(stale.clone()).await.unwrap();

    start_and_wait(&runtime, &mut events, &stale.thread_id, "recover").await;
    let recovered = runtime
        .load_thread(&stale.thread_id)
        .await
        .unwrap()
        .unwrap()
        .metadata
        .unwrap()
        .runner_state
        .unwrap();

    assert_ne!(recovered.session_id, "stale-session");
    assert_eq!(recovered.destination_id, "unix-local");

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(workspace);
}

#[tokio::test]
async fn mock_runner_e2e_tools_command_port_snapshot_resume_and_continue() {
    let session_dir = temp_dir("remote-runner-e2e-sessions");
    let workspace = temp_dir("remote-runner-e2e-workspace");
    let provider = MockRunnerProvider::default();
    let engine = Arc::new(ToolScriptEngine {
        requests: Mutex::new(0),
    });
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(engine);
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: session_dir.clone(),
    }));
    builder.remote_runner_provider(Arc::new(provider.clone()));
    builder.tool_contributor(Arc::new(MockRunnerToolContributor {
        state: provider.state.clone(),
    }));
    let runtime = Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                default_provider: PROVIDER_MOCK.to_string(),
                default_model: "mock".to_string(),
                reasoning: None,
                auto_compact_token_limit: None,
                file_backed_dynamic_context: true,
                hosted_web_search: HostedWebSearchConfig::disabled(),
                model_edit_tools: std::collections::HashMap::new(),
                model_parallel_tool_calls: std::collections::HashMap::new(),
                model_profiles: std::collections::HashMap::new(),
                tool_allowlist: Vec::new(),
                command_shell: roder_api::command_shell::default_command_shell(),
                workspace: Some(workspace.display().to_string()),
                policy_mode: roder_api::policy_mode::PolicyMode::AcceptAll,
                runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
                speed_policy: Default::default(),
                dynamic_workflows: Default::default(),
                reliability: Default::default(),
                turn_deadline_seconds: None,
                remote_runner_destination: Some(RunnerDestination {
                    id: "mock-hosted".to_string(),
                    provider_id: "mock-hosted".to_string(),
                    config: serde_json::Value::Null,
                    default_manifest: RunnerManifest::default(),
                }),
                team_data_dir: None,
                roadmap_data_dir: None,
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    );
    let mut events = runtime.subscribe_events();
    let metadata = runtime.create_thread(None).await.unwrap();

    start_and_wait(&runtime, &mut events, &metadata.thread_id, "write and read").await;
    assert_eq!(
        provider
            .state
            .files
            .lock()
            .unwrap()
            .get("out.txt")
            .cloned()
            .unwrap(),
        b"hello from runner".to_vec()
    );

    let runner_state = runtime
        .load_thread(&metadata.thread_id)
        .await
        .unwrap()
        .unwrap()
        .metadata
        .unwrap()
        .runner_state
        .unwrap();
    let resumed = provider.resume_session(runner_state).await.unwrap();
    let command = resumed
        .run_command(RunnerCommandRequest {
            command_id: "cmd-1".to_string(),
            program: "echo".to_string(),
            args: vec!["ok".to_string()],
            cwd: None,
            env: Vec::new(),
        })
        .await
        .unwrap();
    assert_eq!(command.stdout, "command ok\n");
    assert_eq!(
        resumed
            .expose_port(RunnerPortRequest {
                port: 3000,
                label: Some("web".to_string())
            })
            .await
            .unwrap()
            .url
            .as_deref(),
        Some("https://mock.runner/3000")
    );
    assert!(resumed.snapshot().await.unwrap().is_some());

    start_and_wait(&runtime, &mut events, &metadata.thread_id, "continue").await;
    assert_eq!(*provider.state.created.lock().unwrap(), 1);
    assert!(
        *provider.state.resumed.lock().unwrap() >= 2,
        "runtime and explicit resume should reuse the mock runner session"
    );

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(workspace);
}

#[tokio::test]
async fn runtime_destination_auto_binds_new_thread_when_provider_has_default_workspace() {
    let session_dir = temp_dir("remote-runner-autobind-sessions");
    let workspace = temp_dir("remote-runner-autobind-workspace");
    let provider = MockRunnerProvider::default();
    // Provider advertises a default workspace -> a runtime-level destination
    // (as set by the TUI runner picker / config default) should auto-bind.
    *provider.state.default_workspace.lock().unwrap() = Some("/runner/workspace".to_string());
    let engine = Arc::new(ToolScriptEngine {
        requests: Mutex::new(0),
    });
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(engine);
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: session_dir.clone(),
    }));
    builder.remote_runner_provider(Arc::new(provider.clone()));
    let runtime = Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                default_provider: PROVIDER_MOCK.to_string(),
                default_model: "mock".to_string(),
                reasoning: None,
                auto_compact_token_limit: None,
                file_backed_dynamic_context: true,
                hosted_web_search: HostedWebSearchConfig::disabled(),
                model_edit_tools: std::collections::HashMap::new(),
                model_parallel_tool_calls: std::collections::HashMap::new(),
                model_profiles: std::collections::HashMap::new(),
                tool_allowlist: Vec::new(),
                command_shell: roder_api::command_shell::default_command_shell(),
                workspace: Some(workspace.display().to_string()),
                policy_mode: roder_api::policy_mode::PolicyMode::AcceptAll,
                runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
                speed_policy: Default::default(),
                dynamic_workflows: Default::default(),
                reliability: Default::default(),
                turn_deadline_seconds: None,
                remote_runner_destination: Some(RunnerDestination {
                    id: "mock-hosted".to_string(),
                    provider_id: "mock-hosted".to_string(),
                    config: serde_json::Value::Null,
                    default_manifest: RunnerManifest::default(),
                }),
                team_data_dir: None,
                roadmap_data_dir: None,
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    );

    let metadata = runtime.create_thread(None).await.unwrap();
    let binding = metadata
        .runner_binding
        .expect("runtime-level destination should auto-bind a new thread");
    assert_eq!(binding.destination.provider_id, "mock-hosted");
    assert_eq!(
        binding.workspace,
        std::path::PathBuf::from("/runner/workspace")
    );

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(workspace);
}

#[tokio::test]
async fn runner_pause_resume_detach_and_rejoin_reuse_one_session() {
    let session_dir = temp_dir("remote-runner-lifecycle-sessions");
    let workspace = temp_dir("remote-runner-lifecycle-workspace");
    let provider = MockRunnerProvider::default();
    let engine = Arc::new(ToolScriptEngine {
        requests: Mutex::new(0),
    });
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(engine);
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: session_dir.clone(),
    }));
    builder.remote_runner_provider(Arc::new(provider.clone()));
    let runtime = Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                default_provider: PROVIDER_MOCK.to_string(),
                default_model: "mock".to_string(),
                reasoning: None,
                auto_compact_token_limit: None,
                file_backed_dynamic_context: true,
                hosted_web_search: HostedWebSearchConfig::disabled(),
                model_edit_tools: std::collections::HashMap::new(),
                model_parallel_tool_calls: std::collections::HashMap::new(),
                model_profiles: std::collections::HashMap::new(),
                tool_allowlist: Vec::new(),
                command_shell: roder_api::command_shell::default_command_shell(),
                workspace: Some(workspace.display().to_string()),
                policy_mode: roder_api::policy_mode::PolicyMode::AcceptAll,
                runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
                speed_policy: Default::default(),
                dynamic_workflows: Default::default(),
                reliability: Default::default(),
                turn_deadline_seconds: None,
                remote_runner_destination: Some(RunnerDestination {
                    id: "mock-hosted".to_string(),
                    provider_id: "mock-hosted".to_string(),
                    config: serde_json::Value::Null,
                    default_manifest: RunnerManifest::default(),
                }),
                team_data_dir: None,
                roadmap_data_dir: None,
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    );
    let metadata = runtime.create_thread(None).await.unwrap();
    let thread_id = metadata.thread_id.clone();

    // Pause establishes (creates) the session, then marks standby intent.
    let paused = runtime.pause_thread_runner(&thread_id).await.unwrap();
    assert_eq!(paused.metadata["paused"].as_bool(), Some(true));
    assert_eq!(*provider.state.created.lock().unwrap(), 1);

    // Resume wakes the (resumed) session.
    let resumed = runtime.resume_thread_runner(&thread_id).await.unwrap();
    assert_eq!(resumed.metadata["paused"].as_bool(), Some(false));

    // Detach persists durable, rejoinable state without deleting the sandbox.
    let detached = runtime.detach_thread_runner(&thread_id).await.unwrap();
    assert_eq!(detached.provider_id, "mock-hosted");
    let persisted = runtime
        .load_thread(&thread_id)
        .await
        .unwrap()
        .unwrap()
        .metadata
        .unwrap()
        .runner_state
        .unwrap();
    assert_eq!(persisted.session_id, detached.session_id);

    // Rejoin (simulating a fresh process) reuses the same sandbox via the
    // persisted state and never provisions a new one.
    let rejoined = runtime.rejoin_thread_runner(&thread_id, None).await.unwrap();
    assert_eq!(rejoined.session_id, detached.session_id);
    assert_eq!(
        *provider.state.created.lock().unwrap(),
        1,
        "rejoin must not create a new sandbox"
    );
    assert!(*provider.state.resumed.lock().unwrap() >= 1);

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(workspace);
}

#[tokio::test]
#[ignore]
async fn live_sprites_runner_runtime_creates_session_and_offloads_operations() {
    if std::env::var(SPRITES_LIVE_ENV).ok().as_deref() != Some("1") {
        eprintln!("set {SPRITES_LIVE_ENV}=1 to run the live Sprites runtime smoke");
        return;
    }

    let session_dir = temp_dir("remote-runner-sprites-live-sessions");
    let workspace = temp_dir("remote-runner-sprites-live-workspace");
    let provider = Arc::new(SpritesRunnerProvider::default());
    let engine = Arc::new(FinalEngine {
        requests: Mutex::new(0),
    });
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(engine);
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: session_dir.clone(),
    }));
    builder.remote_runner_provider(provider.clone());
    let runtime = Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                default_provider: PROVIDER_MOCK.to_string(),
                default_model: "mock".to_string(),
                reasoning: None,
                auto_compact_token_limit: None,
                file_backed_dynamic_context: true,
                hosted_web_search: HostedWebSearchConfig::disabled(),
                model_edit_tools: std::collections::HashMap::new(),
                model_parallel_tool_calls: std::collections::HashMap::new(),
                model_profiles: std::collections::HashMap::new(),
                tool_allowlist: Vec::new(),
                command_shell: roder_api::command_shell::default_command_shell(),
                workspace: Some(workspace.display().to_string()),
                policy_mode: roder_api::policy_mode::PolicyMode::AcceptAll,
                runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
                speed_policy: Default::default(),
                dynamic_workflows: Default::default(),
                reliability: Default::default(),
                turn_deadline_seconds: None,
                remote_runner_destination: Some(RunnerDestination {
                    id: "sprites-live-runtime".to_string(),
                    provider_id: SPRITES_PROVIDER_ID.to_string(),
                    config: serde_json::json!({
                        "sprite_name_prefix": "roder-core-live",
                        "cleanup": "delete-on-close",
                        "working_dir": "/home/sprite/roder-core-live"
                    }),
                    default_manifest: RunnerManifest::default(),
                }),
                team_data_dir: None,
                roadmap_data_dir: None,
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    );
    let mut events = runtime.subscribe_events();
    let metadata = runtime.create_thread(None).await.unwrap();

    start_and_wait(
        &runtime,
        &mut events,
        &metadata.thread_id,
        "create live sprites runner session",
    )
    .await;
    let runner_state = runtime
        .load_thread(&metadata.thread_id)
        .await
        .unwrap()
        .unwrap()
        .metadata
        .unwrap()
        .runner_state
        .unwrap();
    assert_eq!(runner_state.provider_id, SPRITES_PROVIDER_ID);
    assert_eq!(runner_state.destination_id, "sprites-live-runtime");

    let session = provider.resume_session(runner_state).await.unwrap();
    let command = session
        .run_command(RunnerCommandRequest {
            command_id: "live-runtime-python".to_string(),
            program: "python3".to_string(),
            args: vec!["-c".to_string(), "print(2+2)".to_string()],
            cwd: None,
            env: Vec::new(),
        })
        .await
        .unwrap();
    assert_eq!(command.exit_code, Some(0));
    assert_eq!(command.stdout.trim(), "4");
    session
        .write_file(RunnerFileWriteRequest {
            path: "runtime-proof.txt".into(),
            contents: b"hello from roder runtime\n".to_vec(),
        })
        .await
        .unwrap();
    let read = session
        .read_file(RunnerFileReadRequest {
            path: "runtime-proof.txt".into(),
        })
        .await
        .unwrap();
    assert_eq!(read.contents, b"hello from roder runtime\n");
    assert!(session.snapshot().await.unwrap().is_some());
    session.close().await.unwrap();

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(workspace);
}

/// Serves one scripted event batch per inference round, then a final message.
struct ScriptedEngine {
    rounds: Mutex<Vec<Vec<InferenceEvent>>>,
}

#[async_trait::async_trait]
impl InferenceEngine for ScriptedEngine {
    fn id(&self) -> InferenceEngineId {
        PROVIDER_MOCK.to_string()
    }

    fn capabilities(&self) -> InferenceCapabilities {
        InferenceCapabilities::coding_agent_default()
    }

    async fn list_models(
        &self,
        _ctx: InferenceProviderContext<'_>,
    ) -> anyhow::Result<Vec<ModelDescriptor>> {
        Ok(Vec::new())
    }

    async fn stream_turn(
        &self,
        _ctx: InferenceTurnContext<'_>,
        _request: AgentInferenceRequest,
    ) -> anyhow::Result<InferenceEventStream> {
        let mut rounds = self.rounds.lock().unwrap();
        let events = if rounds.is_empty() {
            vec![
                InferenceEvent::MessageDelta(MessageDelta {
                    text: "done".to_string(),
                    phase: None,
                }),
                InferenceEvent::Completed(CompletionMetadata {
                    stop_reason: Some("stop".to_string()),
                    provider_response_id: Some("resp-final".to_string()),
                }),
            ]
        } else {
            rounds.remove(0)
        };
        Ok(Box::pin(stream::iter(events.into_iter().map(Ok))))
    }
}

fn tool_call_round(calls: &[(&str, &str, &str)]) -> Vec<InferenceEvent> {
    let mut events = calls
        .iter()
        .map(|(id, name, arguments)| {
            InferenceEvent::ToolCallCompleted(ToolCallCompleted {
                id: id.to_string(),
                name: name.to_string(),
                arguments: arguments.to_string(),
            })
        })
        .collect::<Vec<_>>();
    events.push(InferenceEvent::Completed(CompletionMetadata {
        stop_reason: Some("tool_calls".to_string()),
        provider_response_id: Some("resp-tools".to_string()),
    }));
    events
}

fn coding_tools_runtime(
    session_dir: PathBuf,
    scratch: PathBuf,
    provider: MockRunnerProvider,
    rounds: Vec<Vec<InferenceEvent>>,
) -> Arc<Runtime> {
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(Arc::new(ScriptedEngine {
        rounds: Mutex::new(rounds),
    }));
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: session_dir,
    }));
    builder.remote_runner_provider(Arc::new(provider));
    builder
        .tool_contributor(roder_tools::builtin_coding_tools_contributor(scratch.clone()).unwrap());
    Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                workspace: Some(scratch.display().to_string()),
                policy_mode: roder_api::policy_mode::PolicyMode::Bypass,
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    )
}

#[derive(Debug, Clone)]
struct CompletedToolCall {
    tool_name: Option<String>,
    output: Option<String>,
    is_error: bool,
}

async fn run_turn_collecting_tool_calls(
    runtime: &Arc<Runtime>,
    thread_id: &str,
    workspace: &std::path::Path,
) -> Vec<CompletedToolCall> {
    let mut events = runtime.subscribe_events();
    runtime
        .start_turn(StartTurnRequest {
            thread_id: thread_id.to_string(),
            message: "run the scripted tools".to_string(),
            images: Vec::new(),
            provider_override: None,
            model_override: None,
            reasoning_override: None,
            workspace: workspace.display().to_string(),
            instructions: default_instructions(),
            developer_context: None,
            task_ledger_required: false,
        })
        .await
        .unwrap();
    let mut completed = Vec::new();
    loop {
        let event = tokio::time::timeout(Duration::from_secs(5), events.recv())
            .await
            .unwrap()
            .unwrap();
        match event.event {
            RoderEvent::ToolCallCompleted(call) => completed.push(CompletedToolCall {
                tool_name: call.tool_name,
                output: call.output,
                is_error: call.is_error,
            }),
            RoderEvent::TurnCompleted(_) => break,
            RoderEvent::TurnFailed(failed) => panic!("turn failed: {}", failed.error),
            _ => {}
        }
    }
    completed
}

#[tokio::test]
async fn runner_bound_thread_routes_coding_tools_through_remote_runner() {
    let session_dir = temp_dir("runner-bound-sessions");
    let scratch = temp_dir("runner-bound-scratch");
    let provider = MockRunnerProvider::default();
    let runtime = coding_tools_runtime(
        session_dir.clone(),
        scratch.clone(),
        provider.clone(),
        vec![
            tool_call_round(&[
                (
                    "write",
                    "write_file",
                    r#"{"path":"remote-out.txt","content":"hello from runner"}"#,
                ),
                ("shell", "shell", r#"{"command":"echo hi"}"#),
                ("exec", "exec_command", r#"{"cmd":"echo nope"}"#),
            ]),
            tool_call_round(&[("read", "read_file", r#"{"path":"remote-out.txt"}"#)]),
        ],
    );

    let metadata = runtime
        .create_thread_with(roder_core::CreateThreadRequest {
            title: None,
            workspace: scratch.display().to_string(),
            workspace_id: None,
            root_id: None,
            provider: None,
            model: None,
            selection_mode: None,
            tool_allowlist: Vec::new(),
            developer_instructions: None,
            external_tools: Vec::new(),
            runner: Some(roder_core::ThreadRunnerSelection {
                provider_id: "mock-hosted".to_string(),
                config: serde_json::json!({ "space_id": "space-1" }),
                workspace: "/sandbox/workspace".to_string(),
                read_roots: Vec::new(),
            }),
        })
        .await
        .unwrap();
    let binding = metadata.runner_binding.clone().expect("runner binding");
    assert_eq!(binding.destination.provider_id, "mock-hosted");
    assert_eq!(
        binding.destination.id,
        format!("thread-{}", metadata.thread_id)
    );
    assert_eq!(binding.destination.config["space_id"], "space-1");
    assert_eq!(binding.workspace, PathBuf::from("/sandbox/workspace"));

    let completed = run_turn_collecting_tool_calls(&runtime, &metadata.thread_id, &scratch).await;

    // write_file went to the runner, not local disk.
    assert_eq!(
        provider
            .state
            .files
            .lock()
            .unwrap()
            .get("remote-out.txt")
            .cloned(),
        Some(b"hello from runner".to_vec())
    );
    assert!(!scratch.join("remote-out.txt").exists());

    // shell executed through the runner session, scoped to the runner workspace root.
    let shell_command = provider
        .state
        .commands
        .lock()
        .unwrap()
        .iter()
        .find(|command| command.args.contains(&"echo hi".to_string()))
        .cloned()
        .expect("shell command routed to runner");
    assert_eq!(shell_command.program, "sh");
    assert_eq!(
        shell_command.cwd.as_deref(),
        Some(std::path::Path::new("/sandbox/workspace"))
    );

    // read_file returned the runner contents.
    let read = completed
        .iter()
        .find(|call| call.tool_name.as_deref() == Some("read_file"))
        .expect("read_file completed");
    assert!(!read.is_error);
    assert!(
        read.output
            .as_deref()
            .unwrap_or_default()
            .contains("hello from runner")
    );

    // Tools without a remote transport fail clearly instead of touching local disk.
    let exec = completed
        .iter()
        .find(|call| call.tool_name.as_deref() == Some("exec_command"))
        .expect("exec_command completed");
    assert!(exec.is_error);
    assert!(
        exec.output
            .as_deref()
            .unwrap_or_default()
            .contains("not supported on a remote runner workspace"),
        "unexpected exec output: {:?}",
        exec.output
    );

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(scratch);
}

#[tokio::test]
async fn unbound_thread_keeps_local_coding_tools_on_a_runner_capable_server() {
    let session_dir = temp_dir("runner-unbound-sessions");
    let scratch = temp_dir("runner-unbound-scratch");
    let provider = MockRunnerProvider::default();
    let runtime = coding_tools_runtime(
        session_dir.clone(),
        scratch.clone(),
        provider.clone(),
        vec![tool_call_round(&[(
            "write",
            "write_file",
            r#"{"path":"local-out.txt","content":"hello locally"}"#,
        )])],
    );

    let metadata = runtime.create_thread(None).await.unwrap();
    assert!(metadata.runner_binding.is_none());

    let completed = run_turn_collecting_tool_calls(&runtime, &metadata.thread_id, &scratch).await;
    let write = completed
        .iter()
        .find(|call| call.tool_name.as_deref() == Some("write_file"))
        .expect("write_file completed");
    assert!(!write.is_error);

    assert_eq!(
        std::fs::read_to_string(scratch.join("local-out.txt")).unwrap(),
        "hello locally"
    );
    assert!(provider.state.files.lock().unwrap().is_empty());
    assert_eq!(*provider.state.created.lock().unwrap(), 0);

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(scratch);
}

#[tokio::test]
async fn thread_runner_binding_rejects_unknown_providers_and_relative_workspaces() {
    let session_dir = temp_dir("runner-validate-sessions");
    let scratch = temp_dir("runner-validate-scratch");
    let runtime = coding_tools_runtime(
        session_dir.clone(),
        scratch.clone(),
        MockRunnerProvider::default(),
        Vec::new(),
    );

    let request = |provider_id: &str, workspace: &str| roder_core::CreateThreadRequest {
        title: None,
        workspace: scratch.display().to_string(),
        workspace_id: None,
        root_id: None,
        provider: None,
        model: None,
        selection_mode: None,
        tool_allowlist: Vec::new(),
        developer_instructions: None,
        external_tools: Vec::new(),
        runner: Some(roder_core::ThreadRunnerSelection {
            provider_id: provider_id.to_string(),
            config: serde_json::json!({}),
            workspace: workspace.to_string(),
            read_roots: Vec::new(),
        }),
    };

    let unknown = runtime
        .create_thread_with(request("missing-provider", "/sandbox/workspace"))
        .await
        .unwrap_err();
    assert!(unknown.to_string().contains("is not installed"));

    let relative = runtime
        .create_thread_with(request("mock-hosted", "sandbox/workspace"))
        .await
        .unwrap_err();
    assert!(relative.to_string().contains("absolute"));

    let _ = std::fs::remove_dir_all(session_dir);
    let _ = std::fs::remove_dir_all(scratch);
}

async fn runtime(session_dir: PathBuf, workspace: PathBuf) -> Arc<Runtime> {
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(Arc::new(FinalEngine {
        requests: Mutex::new(0),
    }));
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: session_dir,
    }));
    builder.remote_runner_provider(Arc::new(UnixLocalRunnerProvider::default()));
    Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                default_provider: PROVIDER_MOCK.to_string(),
                default_model: "mock".to_string(),
                reasoning: None,
                auto_compact_token_limit: None,
                file_backed_dynamic_context: true,
                hosted_web_search: HostedWebSearchConfig::disabled(),
                model_edit_tools: std::collections::HashMap::new(),
                model_parallel_tool_calls: std::collections::HashMap::new(),
                model_profiles: std::collections::HashMap::new(),
                tool_allowlist: Vec::new(),
                command_shell: roder_api::command_shell::default_command_shell(),
                workspace: Some(workspace.display().to_string()),
                policy_mode: roder_api::policy_mode::PolicyMode::Default,
                runtime_profile: roder_api::inference::RuntimeProfile::Interactive,
                speed_policy: Default::default(),
                dynamic_workflows: Default::default(),
                reliability: Default::default(),
                turn_deadline_seconds: None,
                remote_runner_destination: Some(RunnerDestination {
                    id: "unix-local".to_string(),
                    provider_id: "unix-local".to_string(),
                    config: serde_json::json!({ "root": workspace.display().to_string() }),
                    default_manifest: RunnerManifest::default(),
                }),
                team_data_dir: None,
                roadmap_data_dir: None,
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    )
}

async fn start_and_wait(
    runtime: &Arc<Runtime>,
    events: &mut tokio::sync::broadcast::Receiver<roder_api::events::EventEnvelope>,
    thread_id: &str,
    message: &str,
) {
    runtime
        .start_turn(StartTurnRequest {
            thread_id: thread_id.to_string(),
            message: message.to_string(),
            images: Vec::new(),
            provider_override: None,
            model_override: None,
            reasoning_override: None,
            workspace: std::env::current_dir().unwrap().display().to_string(),

            instructions: default_instructions(),
            developer_context: None,
            task_ledger_required: false,
        })
        .await
        .unwrap();
    loop {
        let event = tokio::time::timeout(Duration::from_secs(2), events.recv())
            .await
            .unwrap()
            .unwrap();
        match event.event {
            RoderEvent::TurnCompleted(_) => break,
            RoderEvent::TurnFailed(failed) => panic!("turn failed: {}", failed.error),
            _ => {}
        }
    }
    tokio::time::sleep(Duration::from_millis(25)).await;
}

fn temp_dir(name: &str) -> PathBuf {
    let path = std::env::temp_dir().join(format!("{name}-{}", uuid::Uuid::new_v4()));
    std::fs::create_dir_all(&path).unwrap();
    path
}
