use super::super::*;
use super::contracts::{
    GraphContract, assert_all_processes_terminal, assert_completed_process_graph,
    assert_labeled_node, assert_labeled_resource_operation,
    assert_min_completed_child_session_exec_graphs, assert_min_completed_process_graphs,
    assert_no_duplicate_label_step, assert_session_turn_child_graph, assert_successful_turn_case,
};
use std::collections::VecDeque;

#[derive(Default)]
pub(super) struct ExpectedContracts {
    pub(super) labeled_resource_titles: Vec<&'static str>,
    pub(super) labeled_node_titles: Vec<&'static str>,
    pub(super) completed_process_entries: Vec<&'static str>,
    pub(super) min_completed_child_session_exec_graphs: usize,
    pub(super) min_completed_process_graphs: usize,
}

pub(super) struct LashE2eCase {
    pub(super) name: &'static str,
    pub(super) session_id: &'static str,
    pub(super) scripted_provider_responses: Vec<&'static str>,
    pub(super) root_prompt: &'static str,
    pub(super) expected_submitted_value: Option<serde_json::Value>,
    pub(super) tool_provider: Option<Arc<dyn ToolProvider>>,
    pub(super) install_subagents: bool,
    pub(super) max_turns: Option<usize>,
    pub(super) expected_contracts: ExpectedContracts,
}

pub(super) struct LashE2eRun {
    pub(super) turn_output: Option<TurnResult>,
    pub(super) streamed_events: Vec<TurnActivity>,
    pub(super) graph_snapshots: Vec<crate::tracing::TraceLashlangGraph>,
    pub(super) prompt_captures: Vec<LlmRequest>,
    pub(super) final_process_list: Vec<lash_core::ProcessHandleSummary>,
}

pub(super) async fn run_turn_case(case: LashE2eCase) -> Result<LashE2eRun> {
    let run = run_turn_case_without_success_assertions(case).await?;
    assert_successful_turn_case(&run);
    Ok(run)
}

pub(super) async fn run_turn_case_without_success_assertions(
    case: LashE2eCase,
) -> Result<LashE2eRun> {
    let graph_store = Arc::new(crate::tracing::TraceLashlangGraphStore::default());
    let process_registry = Arc::new(TestLocalProcessRegistry::default());
    let prompt_captures = Arc::new(StdMutex::new(Vec::new()));
    let provider = scripted_provider(
        case.scripted_provider_responses.clone(),
        Arc::clone(&prompt_captures),
    );
    let mut builder = explicit_ephemeral_facets(LashCore::rlm())
        .provider(provider)
        .model(mock_model_spec())
        .store_factory(Arc::new(lash_core::InMemorySessionStoreFactory::new()))
        .process_registry(Arc::clone(&process_registry) as Arc<dyn ProcessRegistry>)
        .lashlang_execution_sink(Arc::clone(&graph_store) as Arc<dyn crate::tracing::TraceSink>);
    if let Some(tools) = case.tool_provider.clone() {
        builder = builder.tools(tools);
    }
    if case.install_subagents {
        builder = builder.plugin(subagents_plugin());
    }
    if let Some(max_turns) = case.max_turns {
        builder = builder.max_turns(max_turns);
    }
    let core = builder.build()?;
    let session = core.session(case.session_id).open().await?;
    let events = Arc::new(RecordingEvents::default());

    let turn_output = session
        .turn(TurnInput::text(case.root_prompt))
        .stream_to(events.as_ref())
        .await?;
    session.processes().await_all().await?;
    let prompt_captures_snapshot = prompt_captures.lock().expect("prompt captures").clone();
    let final_process_list = all_host_process_summaries(&core).await?;
    let run = LashE2eRun {
        turn_output: Some(turn_output),
        streamed_events: events.snapshot().await,
        graph_snapshots: graph_store.graphs(),
        prompt_captures: prompt_captures_snapshot,
        final_process_list,
    };

    if let Some(expected) = &case.expected_submitted_value {
        let Some(output) = run.turn_output.as_ref() else {
            panic!("{} did not run a turn", case.name);
        };
        assert_eq!(
            output.submitted_value(),
            Some(expected),
            "{} submitted value mismatch",
            case.name
        );
    }

    let contract = GraphContract::from_graphs(&run.graph_snapshots);
    for title in case.expected_contracts.labeled_resource_titles {
        assert_labeled_resource_operation(
            &contract,
            title,
            crate::tracing::TraceLashlangNodeStatus::Completed,
        );
        assert_no_duplicate_label_step(&contract, title);
    }
    for title in case.expected_contracts.labeled_node_titles {
        assert_labeled_node(
            &contract,
            title,
            crate::tracing::TraceLashlangNodeStatus::Completed,
        );
        assert_no_duplicate_label_step(&contract, title);
    }
    for entry_name in case.expected_contracts.completed_process_entries {
        assert_completed_process_graph(&contract, entry_name);
    }
    assert_min_completed_process_graphs(
        &contract,
        case.expected_contracts.min_completed_process_graphs,
    );
    assert_min_completed_child_session_exec_graphs(
        &run,
        case.session_id,
        case.expected_contracts
            .min_completed_child_session_exec_graphs,
    );
    Ok(run)
}

async fn all_host_process_summaries(
    core: &LashCore,
) -> Result<Vec<lash_core::ProcessHandleSummary>> {
    let processes = core
        .processes()
        .list(&lash_core::ProcessListFilter {
            definition: None,
            status: lash_core::ProcessStatusFilter::Any,
            waiting: None,
        })
        .await?;
    Ok(processes
        .into_iter()
        .map(observed_process_summary)
        .collect())
}

fn observed_process_summary(
    process: lash_core::ObservedProcess,
) -> lash_core::ProcessHandleSummary {
    lash_core::ProcessHandleSummary::new(
        process.process_id,
        lash_core::ProcessHandleDescriptor::new(Some(process.kind), Some(process.label)),
        process.lifecycle,
    )
    .with_definition(lash_core::ProcessDefinitionSummary::from_input(
        &process.input,
    ))
}

pub(super) async fn run_session_turn_process_case() -> Result<()> {
    let session_id = "lash-e2e-session-turn-root";
    let child_session_id = "lash-e2e-session-turn-child";
    let process_id = "lash-e2e-session-turn-process";
    let graph_store = Arc::new(crate::tracing::TraceLashlangGraphStore::default());
    let process_registry = Arc::new(TestLocalProcessRegistry::default());
    let prompt_captures = Arc::new(StdMutex::new(Vec::new()));
    let provider = scripted_provider(
        vec!["```lashlang\nsubmit { child: \"done\", scoped: true }\n```"],
        Arc::clone(&prompt_captures),
    );
    let core = explicit_ephemeral_facets(LashCore::rlm())
        .provider(provider)
        .model(mock_model_spec())
        .plugin(subagents_plugin())
        .store_factory(Arc::new(lash_core::InMemorySessionStoreFactory::new()))
        .process_registry(Arc::clone(&process_registry) as Arc<dyn ProcessRegistry>)
        .lashlang_execution_sink(Arc::clone(&graph_store) as Arc<dyn crate::tracing::TraceSink>)
        .build()?;
    let session = core.session(session_id).open().await?;
    let child_policy = lash_core::SessionPolicy {
        model: mock_model_spec(),
        max_turns: Some(2),
        ..lash_core::SessionPolicy::default()
    };
    let create_request = lash_core::SessionCreateRequest::child(
        session_id,
        lash_core::SessionStartPoint::Empty,
        child_policy,
        lash_core::PluginOptions::default(),
        "e2e-session-turn",
    )
    .with_session_id(child_session_id);

    let handle = session
        .processes()
        .start(
            lash_core::ProcessStartRequest::new(
                process_id,
                lash_core::ProcessInput::SessionTurn {
                    create_request: Box::new(create_request),
                    turn_input: Box::new(TurnInput::text("run child session turn")),
                    output_contract: lash_core::ToolOutputContract::Static,
                },
                lash_core::ProcessOriginator::host(),
            )
            .with_grant(Some(lash_core::ProcessStartGrant {
                session_scope: lash_core::SessionScope::new("request-descriptor"),
                descriptor: lash_core::ProcessHandleDescriptor::new(
                    Some("session_turn"),
                    Some("child turn"),
                ),
            })),
            inline_scope(lash_core::ExecutionScope::process(process_id)),
        )
        .await?;
    assert_eq!(handle.process_id, process_id);
    session.processes().await_all().await?;

    let await_output = process_registry.await_process(process_id).await?;
    let lash_core::ProcessAwaitOutput::Success { value, .. } = await_output else {
        panic!("session-turn process did not succeed: {await_output:#?}");
    };
    assert_eq!(
        value.get("child_session_id"),
        Some(&serde_json::json!(child_session_id))
    );
    let turn: lash_core::AssembledTurn = value
        .get("turn")
        .cloned()
        .map(serde_json::from_value)
        .transpose()
        .expect("session-turn output should decode")
        .expect("session-turn output should contain a turn");
    assert_eq!(
        turn.outcome,
        TurnOutcome::Finished(lash_core::TurnFinish::SubmittedValue {
            value: serde_json::json!({ "child": "done", "scoped": true })
        })
    );

    let prompt_captures_snapshot = prompt_captures.lock().expect("prompt captures").clone();
    let final_process_list = all_host_process_summaries(&core).await?;
    let run = LashE2eRun {
        turn_output: None,
        streamed_events: Vec::new(),
        graph_snapshots: graph_store.graphs(),
        prompt_captures: prompt_captures_snapshot,
        final_process_list,
    };
    assert_eq!(run.prompt_captures.len(), 1);
    assert_all_processes_terminal(&run.final_process_list);
    assert_session_turn_child_graph(&run, child_session_id, process_id);
    Ok(())
}

fn scripted_provider(
    responses: Vec<&'static str>,
    prompt_captures: Arc<StdMutex<Vec<LlmRequest>>>,
) -> ProviderHandle {
    let responses = Arc::new(TokioMutex::new(VecDeque::from(
        responses
            .into_iter()
            .map(ToOwned::to_owned)
            .collect::<Vec<_>>(),
    )));
    crate::testing::TestProvider::builder()
        .kind("lash-e2e")
        .complete(move |request| {
            let responses = Arc::clone(&responses);
            let prompt_captures = Arc::clone(&prompt_captures);
            async move {
                prompt_captures
                    .lock()
                    .expect("prompt captures")
                    .push(request.clone());
                let text = responses
                    .lock()
                    .await
                    .pop_front()
                    .unwrap_or_else(|| panic!("no scripted e2e provider response left"));
                Ok(LlmResponse {
                    full_text: text.clone(),
                    parts: vec![LlmOutputPart::Text {
                        text,
                        response_meta: None,
                    }],
                    ..LlmResponse::default()
                })
            }
        })
        .build()
        .into_handle()
}

fn subagents_plugin() -> Arc<dyn PluginFactory> {
    Arc::new(lash_subagents::SubagentsPluginFactory::new(Arc::new(
        lash_subagents::CapabilityRegistry::new().with(Arc::new(
            lash_subagents::StaticCapability::new("default", SessionSpec::inherit()),
        )),
    )))
}
