roder-core 0.3.2

use super::*;

use futures::stream;
use roder_api::catalog::PROVIDER_MOCK;
use roder_api::extension::{ExtensionRegistryBuilder, InferenceEngineId};
use roder_api::inference::{
    AgentInferenceRequest, InferenceCapabilities, InferenceEngine, InferenceEvent,
    InferenceEventStream, InferenceFailure, InferenceProviderContext, InferenceTurnContext,
    MessageDelta, ModelDescriptor,
};
use roder_ext_jsonl_thread_store::store::JsonlThreadStoreFactory;

struct PendingInferenceEngine;

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

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

    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> {
        Ok(Box::pin(stream::pending()))
    }
}

struct PartialFailureInferenceEngine;

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

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

    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> {
        Ok(Box::pin(stream::iter(vec![
            Ok(InferenceEvent::MessageDelta(MessageDelta {
                text: "partial teammate result".to_string(),
                phase: None,
            })),
            Ok(InferenceEvent::Failed(InferenceFailure {
                message: "provider disconnected".to_string(),
            })),
        ])))
    }
}

#[tokio::test]
async fn interrupt_agent_is_a_noop_without_an_active_turn() {
    let (runtime, parent_thread_id, thread_root, team_root) =
        lifecycle_runtime("interrupt-idle", Arc::new(FakeInferenceEngine)).await;
    let team = spawn_idle_member(&runtime, &parent_thread_id, "parent-turn", "reviewer").await;
    let member = team.members.last().unwrap().clone();
    let mut events = runtime.subscribe_events();

    let idle = execute_control_tool(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "interrupt_agent",
        serde_json::json!({ "target": member.thread_id.clone() }),
    )
    .await;
    assert!(!idle.is_error, "{}", idle.text);
    assert_eq!(idle.data["previous_status"], "idle");
    assert!(idle.data["interrupted_turn_id"].is_null());
    assert_eq!(
        runtime.read_team(&team.id).await.unwrap().members[1].status,
        TeamMemberStatus::Idle
    );

    runtime
        .teams
        .update_member(&team.id, &member.id, |member| {
            member.status = TeamMemberStatus::Completed;
            member.final_message = Some("already finished".to_string());
        })
        .await
        .unwrap();
    let completed = execute_control_tool(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "interrupt_agent",
        serde_json::json!({ "target": member.thread_id.clone() }),
    )
    .await;
    assert!(!completed.is_error, "{}", completed.text);
    assert_eq!(completed.data["previous_status"], "completed");
    assert!(completed.data["interrupted_turn_id"].is_null());
    let unchanged = runtime.read_team(&team.id).await.unwrap();
    assert_eq!(unchanged.members[1].status, TeamMemberStatus::Completed);
    assert_eq!(
        unchanged.members[1].final_message.as_deref(),
        Some("already finished")
    );
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(50), events.recv())
            .await
            .is_err(),
        "an inactive interrupt must not emit another completion"
    );

    cleanup(thread_root, team_root);
}

#[tokio::test]
async fn wait_agent_returns_stored_terminal_result_details() {
    let (runtime, parent_thread_id, thread_root, team_root) =
        lifecycle_runtime("wait-terminal", Arc::new(FakeInferenceEngine)).await;
    let team = spawn_idle_member(&runtime, &parent_thread_id, "parent-turn", "reviewer").await;
    let member = team.members.last().unwrap().clone();
    runtime
        .teams
        .update_member(&team.id, &member.id, |member| {
            member.status = TeamMemberStatus::Failed;
            member.final_message = Some("partial review".to_string());
            member.terminal_error = Some("provider failed".to_string());
        })
        .await
        .unwrap();

    let result = execute_control_tool(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "wait_agent",
        serde_json::json!({ "target": member.thread_id.clone(), "timeout_ms": 10_000 }),
    )
    .await;

    assert!(!result.is_error, "{}", result.text);
    assert_eq!(result.data["timed_out"], false);
    assert_eq!(result.data["agents"][0]["status"], "failed");
    assert_eq!(result.data["agents"][0]["final_message"], "partial review");
    assert_eq!(
        result.data["agents"][0]["terminal_error"],
        "provider failed"
    );

    cleanup(thread_root, team_root);
}

#[tokio::test]
async fn targetless_wait_prioritizes_live_agents_over_stale_terminal_results() {
    let (runtime, parent_thread_id, thread_root, team_root) =
        lifecycle_runtime("wait-live-over-stale", Arc::new(FakeInferenceEngine)).await;
    let first = spawn_idle_member(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "finished_reviewer",
    )
    .await;
    let finished = first.members.last().unwrap().clone();
    let second = spawn_idle_member(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "active_reviewer",
    )
    .await;
    let active = second.members.last().unwrap().clone();
    runtime
        .teams
        .update_member(&second.id, &finished.id, |member| {
            member.status = TeamMemberStatus::Completed;
            member.final_message = Some("old result".to_string());
        })
        .await
        .unwrap();

    let mut wait = Box::pin(execute_control_tool(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "wait_agent",
        serde_json::json!({ "timeout_ms": 10_000 }),
    ));
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(50), wait.as_mut())
            .await
            .is_err(),
        "targetless wait must not return a stale terminal result while another agent is live"
    );

    runtime
        .emit(RoderEvent::TeamMemberCompleted(TeamMemberCompleted {
            team_id: second.id.clone(),
            member_id: active.id.clone(),
            member_thread_id: active.thread_id.clone(),
            turn_id: Some("active-turn".to_string()),
            status: TeamMemberStatus::Completed,
            final_message: Some("new result".to_string()),
            error: None,
            timestamp: OffsetDateTime::now_utc(),
        }))
        .await;
    let result = tokio::time::timeout(std::time::Duration::from_secs(1), wait)
        .await
        .expect("live agent completion should wake targetless wait");
    assert!(!result.is_error, "{}", result.text);
    assert_eq!(result.data["agents"][0]["member_id"], active.id);
    assert_eq!(result.data["agents"][0]["final_message"], "new result");

    let explicit = execute_control_tool(
        &runtime,
        &parent_thread_id,
        "parent-turn",
        "wait_agent",
        serde_json::json!({
            "target": finished.thread_id,
            "timeout_ms": 10_000
        }),
    )
    .await;
    assert!(!explicit.is_error, "{}", explicit.text);
    assert_eq!(explicit.data["timed_out"], false);
    assert_eq!(explicit.data["agents"][0]["member_id"], finished.id);
    assert_eq!(explicit.data["agents"][0]["final_message"], "old result");

    cleanup(thread_root, team_root);
}

#[tokio::test]
async fn session_start_hook_is_claimed_once_per_thread() {
    let (runtime, thread_id, thread_root, team_root) =
        lifecycle_runtime("session-start-once", Arc::new(PendingInferenceEngine)).await;

    // Creating the thread already emitted ThreadCreated, which opened the
    // session, so the claim for it is spent.
    assert!(
        !runtime.claim_session_hook_start(&thread_id).await,
        "ThreadCreated should already have claimed this session"
    );

    // A fresh session claims once; the repeat ThreadLoaded events that fire on
    // every read of a thread from its store must not re-run SessionStart hooks.
    let fresh = "session-start-dedup".to_string();
    assert!(runtime.claim_session_hook_start(&fresh).await);
    assert!(!runtime.claim_session_hook_start(&fresh).await);
    assert!(!runtime.claim_session_hook_start(&fresh).await);

    cleanup(thread_root, team_root);
}

#[tokio::test]
async fn wait_agent_observes_mailbox_activity_queued_before_subscription() {
    let (runtime, parent_thread_id, thread_root, team_root) =
        lifecycle_runtime("wait-pending-steer", Arc::new(PendingInferenceEngine)).await;
    let parent_turn_id = runtime
        .start_turn(StartTurnRequest {
            thread_id: parent_thread_id.clone(),
            message: "coordinate the review".to_string(),
            images: Vec::new(),
            provider_override: None,
            model_override: None,
            reasoning_override: None,
            workspace: std::env::current_dir().unwrap().display().to_string(),
            instructions: crate::default_instructions(),
            developer_context: None,
            task_ledger_required: false,
            service_tier_override: None,
        })
        .await
        .unwrap();
    let team = spawn_idle_member(&runtime, &parent_thread_id, &parent_turn_id, "reviewer").await;
    let member = team.members.last().unwrap().clone();
    runtime
        .queue_team_member_message(
            &member.thread_id,
            &team.id,
            &team.members[0].id,
            "agent update".to_string(),
        )
        .await
        .unwrap();

    let result = tokio::time::timeout(
        std::time::Duration::from_secs(1),
        execute_control_tool(
            &runtime,
            &parent_thread_id,
            &parent_turn_id,
            "wait_agent",
            serde_json::json!({ "target": member.thread_id.clone(), "timeout_ms": 10_000 }),
        ),
    )
    .await
    .expect("pending mailbox activity should wake wait_agent without waiting for its timeout");

    assert!(!result.is_error, "{}", result.text);
    assert_eq!(result.data["timed_out"], false);
    assert_eq!(result.data["activity"], "mailbox_or_steer");
    runtime
        .interrupt_turn(parent_thread_id, parent_turn_id)
        .await
        .unwrap();
    cleanup(thread_root, team_root);
}

#[tokio::test]
async fn failed_agent_preserves_partial_final_message_and_error() {
    let (runtime, parent_thread_id, thread_root, team_root) =
        lifecycle_runtime("partial-failure", Arc::new(PartialFailureInferenceEngine)).await;
    let team = spawn_idle_member(&runtime, &parent_thread_id, "parent-turn", "reviewer").await;
    let member = team.members.last().unwrap().clone();
    let mut events = runtime.subscribe_events();

    runtime
        .followup_team_member(
            &parent_thread_id,
            &team.id,
            &member.id,
            "review the patch".to_string(),
        )
        .await
        .unwrap();
    let completed = tokio::time::timeout(std::time::Duration::from_secs(2), async {
        loop {
            let envelope = events.recv().await.unwrap();
            if let RoderEvent::TeamMemberCompleted(completed) = envelope.event
                && completed.member_id == member.id
            {
                break completed;
            }
        }
    })
    .await
    .expect("failed teammate should publish a terminal result");

    assert_eq!(completed.status, TeamMemberStatus::Failed);
    assert_eq!(
        completed.final_message.as_deref(),
        Some("partial teammate result")
    );
    assert_eq!(completed.error.as_deref(), Some("provider disconnected"));
    let stored = runtime.read_team(&team.id).await.unwrap();
    assert_eq!(
        stored.members[1].final_message.as_deref(),
        Some("partial teammate result")
    );
    assert_eq!(
        stored.members[1].terminal_error.as_deref(),
        Some("provider disconnected")
    );
    let terminal = stored
        .mailbox
        .iter()
        .find(|message| message.kind == TeamMailboxMessageKind::FinalAnswer)
        .expect("terminal result mailbox message");
    assert_eq!(terminal.from_member_id.as_deref(), Some(member.id.as_str()));
    assert_eq!(terminal.to_member_id, stored.members[0].id);
    assert!(terminal.text.contains("partial teammate result"));
    assert!(terminal.text.contains("provider disconnected"));

    cleanup(thread_root, team_root);
}

async fn execute_control_tool(
    runtime: &Arc<Runtime>,
    parent_thread_id: &ThreadId,
    parent_turn_id: &str,
    name: &str,
    arguments: serde_json::Value,
) -> roder_api::tools::ToolResult {
    runtime
        .execute_agent_control_tool(
            parent_thread_id,
            &parent_turn_id.to_string(),
            &ToolCallCompleted {
                id: format!("call-{name}"),
                name: name.to_string(),
                arguments: arguments.to_string(),
            },
            arguments,
        )
        .await
}

async fn spawn_idle_member(
    runtime: &Arc<Runtime>,
    parent_thread_id: &ThreadId,
    parent_turn_id: &str,
    task_name: &str,
) -> TeamState {
    runtime
        .spawn_team_member_for_caller(
            parent_thread_id,
            &parent_turn_id.to_string(),
            TeamMemberStartRequest {
                name: task_name.to_string(),
                model_provider: None,
                model: None,
            },
            None,
            "none".to_string(),
            None,
        )
        .await
        .unwrap()
}

async fn lifecycle_runtime(
    test_name: &str,
    engine: Arc<dyn InferenceEngine>,
) -> (Arc<Runtime>, ThreadId, PathBuf, PathBuf) {
    let suffix = uuid::Uuid::new_v4();
    let thread_root =
        std::env::temp_dir().join(format!("roder-codex-v2-{test_name}-threads-{suffix}"));
    let team_root = std::env::temp_dir().join(format!("roder-codex-v2-{test_name}-teams-{suffix}"));
    let mut builder = ExtensionRegistryBuilder::new();
    builder.inference_engine(engine);
    builder.thread_store_factory(Arc::new(JsonlThreadStoreFactory {
        base_path: thread_root.clone(),
    }));
    let runtime = Arc::new(
        Runtime::new(
            builder.build().unwrap(),
            RuntimeConfig {
                default_provider: PROVIDER_MOCK.to_string(),
                default_model: "mock".to_string(),
                reasoning: None,
                team_data_dir: Some(team_root.clone()),
                ..RuntimeConfig::default()
            },
        )
        .unwrap(),
    );
    let parent = runtime
        .create_thread_with(CreateThreadRequest {
            title: Some("Codex V2 lifecycle parent".to_string()),
            workspace: std::env::current_dir().unwrap().display().to_string(),
            workspace_id: Some("workspace-test".to_string()),
            root_id: Some("root-test".to_string()),
            provider: Some(PROVIDER_MOCK.to_string()),
            model: Some("mock".to_string()),
            selection_mode: None,
            tool_allowlist: Vec::new(),
            developer_instructions: None,
            external_tools: Vec::new(),
            runner: None,
        })
        .await
        .unwrap();
    (runtime, parent.thread_id, thread_root, team_root)
}

fn cleanup(thread_root: PathBuf, team_root: PathBuf) {
    let _ = std::fs::remove_dir_all(thread_root);
    let _ = std::fs::remove_dir_all(team_root);
}