//! dispatch_stream_demo —— 流式子代理调度示例。
//! Required features: cargo run --example dispatch_stream_demo --features "engine"
//!
//! 演示：
//! 1. 创建父 session
//! 2. dispatch_stream 单个子 agent
//! 3. 消费 SubTaskStreamEvent 序列
//! 4. 验证事件序列：ChildCreated → Stream(...) × N → Completed
//! 5. 验证：完成时 turn_index 已递增（finalize 副作用）
//!
//! 运行：`cargo run --example dispatch_stream_demo`

use std::sync::Arc;

use agcore::agent::{Agent, AgentBuilder};
use agcore::engine::{SessionManager, SubTaskStreamEvent};
use agcore::llm::hooks::HookExecutor;
use agcore::llm::mock::MockProvider;
use agcore::llm::types::Usage;
use agcore::llm::types::message::{ContentBlock, Message};
use agcore::llm::types::response_v2::{MessageResponse, StopReason};
use agcore::memory::store::InMemoryStore;
use agcore::tools::ToolRegistry;
use futures_util::StreamExt;

struct StreamWorkerAgent;

impl Agent for StreamWorkerAgent {
    fn name(&self) -> &str {
        "stream_worker"
    }
    fn system_prompt(&self) -> Option<&str> {
        Some("You are a streaming worker.")
    }
}

fn assistant_text(text: &str) -> MessageResponse {
    MessageResponse {
        id: String::new(),
        model: String::new(),
        message: Message::Assistant {
            content: vec![ContentBlock::Text { text: text.into() }],
        },
        usage: Usage::from_input_output(8, 4),
        stop_reason: StopReason::Stop,
        extra: Default::default(),
    }
}

#[tokio::main]
async fn main() {
    println!("=== Dispatch Stream Demo ===\n");

    let store: Arc<dyn agcore::memory::store::MemoryStore> = Arc::new(InMemoryStore::new());
    let provider = Arc::new(MockProvider::new(vec![assistant_text("streamed response")]));
    let bundle = Arc::new(
        AgentBuilder::new()
            .provider(provider)
            .tool_registry(Arc::new(ToolRegistry::new()))
            .hook_executor(Arc::new(HookExecutor::new()))
            .session_memory_backend(store.clone())
            .build()
            .expect("RuntimeBundle"),
    );

    let worker: Arc<dyn Agent> = Arc::new(StreamWorkerAgent);

    let sm = Arc::new(SessionManager::new(store));
    let parent_id = sm
        .create(worker.clone(), bundle.clone())
        .await
        .expect("create");
    println!("[1] parent session: {parent_id}");

    // dispatch_stream
    let mut stream = sm
        .dispatch_stream(&parent_id, worker, "do streaming work", Default::default())
        .await
        .expect("dispatch_stream");

    println!("[2] consuming SubTaskStreamEvent sequence...\n");
    let mut saw_child_created = false;
    let mut saw_stream_count = 0;
    let mut completed = None;

    while let Some(event) = stream.next().await {
        match event {
            SubTaskStreamEvent::ChildCreated { child_id } => {
                println!("  → ChildCreated({})", &child_id[..20]);
                saw_child_created = true;
            }
            SubTaskStreamEvent::Stream(_) => {
                saw_stream_count += 1;
            }
            SubTaskStreamEvent::Completed(r) => {
                println!(
                    "  → Completed(child_id={}, {} tokens)",
                    &r.child_id[..20],
                    r.usage.total().total_tokens
                );
                completed = Some(r);
                break;
            }
            SubTaskStreamEvent::Error { child_id, error } => {
                panic!("unexpected error: child_id={child_id}, error={error}");
            }
        }
    }

    let result = completed.expect("Completed should arrive");

    // 验证事件序列
    assert!(saw_child_created, "ChildCreated should be received");
    assert!(saw_stream_count > 0, "at least one Stream event");
    println!("\n[3] received {} stream events", saw_stream_count);

    // 验证 finalize 已发生（turn_index 递增）
    let child_session = sm.get(&result.child_id).await.unwrap();
    let child_guard = child_session.lock().await;
    let child_turn_index = child_guard.turn_index();
    drop(child_guard);
    assert_eq!(
        child_turn_index, 1,
        "turn_index should increment after finalize"
    );
    println!("[4] child session turn_index = {child_turn_index} (finalize works)");

    println!("\n✓ dispatch_stream completed successfully");
}