#![cfg(all(feature = "integration-real-tests", not(target_arch = "wasm32")))]
#![allow(clippy::all)]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

//! Kitchen Sink E2E: Two-Mob Joke Collaboration with Chaos
//!
//! Capstone stress test that exercises everything at once: two mobs with
//! cross-mob comms, identity-first lifecycle, mid-conversation shutdown/restore,
//! member failure/recovery, topology changes, fencing races, and reset vs
//! respawn semantics.
//!
//! Run:
//! ```bash
//! ANTHROPIC_API_KEY=... ./scripts/repo-cargo test -p meerkat-mobkit \
//!     --test identity_first_kitchen_sink --features integration-real-tests \
//!     -- --ignored --test-threads=1 --nocapture
//! ```

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use tokio::sync::RwLock;

use meerkat_mob::definition::WiringRules;
use meerkat_mob::{MobDefinition, MobId, MobRuntimeMode, Profile, ProfileName, ToolConfig};

use meerkat_mobkit::UnifiedRuntimeBuilder;
use meerkat_mobkit::contact_directory::ContactDirectory;
use meerkat_mobkit::identity_first::contracts::{AgentCustomizer, TopologyProvider};
use meerkat_mobkit::identity_first::orchestrator::RestoreOutcome;
use meerkat_mobkit::identity_first::{
    AgentAddressability, AgentBuildContext, AgentBuildDraft, AgentIdentity, CheckpointVersion,
    ContinuityGeneration, ContinuityRecord, ContinuityStore, ContinuityStoreError, CustomizerError,
    DurabilityPolicy, DurableAgentSpec, FencingToken, IdentityLifecycleState, IdentityRuntime,
    IdentityRuntimeConfig, IdentityRuntimeError, LocalContinuityStore, LocalLeaseProvider,
    ManagedPeerEdge, SessionBridge, SessionSnapshot, TopologyContext, TopologyError,
    wire_cross_mob_by_identity,
};
use meerkat_mobkit::mob_handle_runtime::SessionCreatedContext;

// ---------------------------------------------------------------------------
// Env helpers
// ---------------------------------------------------------------------------

fn first_env(vars: &[&str]) -> Option<String> {
    for name in vars {
        if let Ok(value) = std::env::var(name) {
            if !value.is_empty() {
                return Some(value);
            }
        }
    }
    None
}

fn has_api_key() -> bool {
    first_env(&["RKAT_ANTHROPIC_API_KEY", "ANTHROPIC_API_KEY"]).is_some()
}

fn smoke_model() -> String {
    std::env::var("SMOKE_MODEL").unwrap_or_else(|_| "claude-sonnet-4-5".to_string())
}

// ---------------------------------------------------------------------------
// Identity helpers
// ---------------------------------------------------------------------------

fn id(name: &str) -> AgentIdentity {
    AgentIdentity::parse(name).unwrap()
}

/// Default timeout for wait operations in the kitchen sink test.
const WAIT_TIMEOUT: Duration = Duration::from_secs(90);

fn spec(name: &str, addr: AgentAddressability, profile: &str) -> DurableAgentSpec {
    DurableAgentSpec {
        identity: id(name),
        profile: ProfileName::from(profile),
        addressability: addr,
        display_name: None,
        labels: BTreeMap::new(),
        context: None,
        additional_instructions: Vec::new(),
    }
}

// ---------------------------------------------------------------------------
// Mob definitions
// ---------------------------------------------------------------------------

/// Author mob: coordinator, lead, writer profiles — all TurnDriven + comms.
fn author_definition(model: &str) -> MobDefinition {
    let mut profiles = BTreeMap::new();

    for (name, desc, ext_addr) in [
        ("coordinator", "Task coordinator", true),
        ("lead", "Lead author orchestrator", true),
        ("writer", "Joke writer", true),
    ] {
        profiles.insert(
            ProfileName::from(name),
            Profile {
                model: model.to_string(),
                skills: vec![],
                tools: ToolConfig {
                    comms: true,
                    ..Default::default()
                },
                peer_description: desc.to_string(),
                external_addressable: ext_addr,
                backend: None,
                runtime_mode: MobRuntimeMode::TurnDriven,
                max_inline_peer_notifications: None,
                output_schema: None,
                provider_params: None,
            },
        );
    }

    MobDefinition {
        id: MobId::from("authors"),
        orchestrator: None,
        profiles,
        mcp_servers: BTreeMap::new(),
        wiring: WiringRules {
            auto_wire_orchestrator: false,
            role_wiring: vec![],
        },
        skills: BTreeMap::new(),
        backend: Default::default(),
        flows: Default::default(),
        topology: None,
        supervisor: None,
        limits: None,
        spawn_policy: None,
        event_router: None,
    }
}

/// Critic mob: lead, judge profiles — all TurnDriven + comms.
fn critic_definition(model: &str) -> MobDefinition {
    let mut profiles = BTreeMap::new();

    for (name, desc) in [
        ("lead", "Lead critic orchestrator"),
        ("judge", "Joke judge/scorer"),
    ] {
        profiles.insert(
            ProfileName::from(name),
            Profile {
                model: model.to_string(),
                skills: vec![],
                tools: ToolConfig {
                    comms: true,
                    ..Default::default()
                },
                peer_description: desc.to_string(),
                external_addressable: true,
                backend: None,
                runtime_mode: MobRuntimeMode::TurnDriven,
                max_inline_peer_notifications: None,
                output_schema: None,
                provider_params: None,
            },
        );
    }

    MobDefinition {
        id: MobId::from("critics"),
        orchestrator: None,
        profiles,
        mcp_servers: BTreeMap::new(),
        wiring: WiringRules {
            auto_wire_orchestrator: false,
            role_wiring: vec![],
        },
        skills: BTreeMap::new(),
        backend: Default::default(),
        flows: Default::default(),
        topology: None,
        supervisor: None,
        limits: None,
        spawn_policy: None,
        event_router: None,
    }
}

// ---------------------------------------------------------------------------
// Mutable topology provider
// ---------------------------------------------------------------------------

/// Topology provider with dynamically updatable edges.
struct MutableTopology {
    edges: Arc<RwLock<Vec<(String, String)>>>,
}

impl MutableTopology {
    fn new(edges: Vec<(String, String)>) -> Self {
        Self {
            edges: Arc::new(RwLock::new(edges)),
        }
    }

    async fn set_edges(&self, edges: Vec<(String, String)>) {
        *self.edges.write().await = edges;
    }
}

#[async_trait]
impl TopologyProvider for MutableTopology {
    async fn compute_edges(
        &self,
        _target_identities: &[AgentIdentity],
        _context: &TopologyContext,
    ) -> Result<Vec<ManagedPeerEdge>, TopologyError> {
        let edges = self.edges.read().await;
        let mut result = Vec::new();
        for (a, b) in edges.iter() {
            result.push(
                ManagedPeerEdge::new(id(a), id(b))
                    .map_err(|e| TopologyError::InvalidEdge(format!("{e}")))?,
            );
        }
        Ok(result)
    }
}

// ---------------------------------------------------------------------------
// Topology-aware customizer (injects subscriber names into prompts)
// ---------------------------------------------------------------------------

struct JokeCollabCustomizer;

#[async_trait]
impl AgentCustomizer for JokeCollabCustomizer {
    async fn customize_build(
        &self,
        context: &AgentBuildContext,
        spec: &DurableAgentSpec,
        draft: &mut AgentBuildDraft,
    ) -> Result<(), CustomizerError> {
        let my_id = &context.identity;
        let mut peers = Vec::new();
        for edge in &context.managed_edges {
            if edge.a() == my_id {
                peers.push(edge.b().to_string());
            } else if edge.b() == my_id {
                peers.push(edge.a().to_string());
            }
        }
        let peers_line = if peers.is_empty() {
            String::new()
        } else {
            peers.sort();
            format!("\nYour connected peers: {}", peers.join(", "))
        };

        let profile = spec.profile.as_str();
        let role_prompt = match profile {
            "coordinator" => format!(
                "You are the Joke Coordinator for a comedy writing room focused on AI humor. \
                 Your job is to present the final joke clearly and celebrate the team's work. \
                 When you receive a joke with scores, present it as the team's masterpiece.{peers_line}"
            ),
            "lead" if my_id.as_str().contains("author") => format!(
                "You are the Lead Author of an AI comedy writing room. You manage joke writers \
                 and refine jokes based on critic feedback. When picking between drafts, choose \
                 the one with sharper comedic timing. When revising, focus on the specific \
                 feedback — tighten punchlines, sharpen wordplay, raise the stakes. \
                 Your goal: craft the world's greatest joke about large language models. \
                 Always output the current best version of the joke clearly.{peers_line}"
            ),
            "lead" if my_id.as_str().contains("critic") => format!(
                "You are the Lead Critic. You evaluate jokes about AI and LLMs. \
                 Score 1-10 and give specific, actionable feedback. Be tough but fair — \
                 a 10 means instant classic, a 7 means good but needs work.{peers_line}"
            ),
            "writer" => format!(
                "You are a Comedy Writer specializing in AI and LLM humor. Write original, \
                 clever jokes that land with both technical and general audiences. \
                 Aim for jokes that reveal something true about LLMs in a surprising way. \
                 Keep jokes to 2-4 sentences. No explaining the joke.{peers_line}"
            ),
            "judge" => format!(
                "You are a Joke Judge specializing in tech humor. Score jokes 1-10. \
                 Be specific about what works and what doesn't. A great LLM joke should \
                 be surprising, insightful, and funny to both AI insiders and normal people. \
                 Start your response with 'SCORE: N/10' on the first line.{peers_line}"
            ),
            _ => format!("You are a helpful assistant.{peers_line}"),
        };
        draft.system_prompt = Some(role_prompt);
        Ok(())
    }

    async fn after_create(
        &self,
        _identity: &AgentIdentity,
        _session_id: &meerkat_core::types::SessionId,
        _context: &SessionCreatedContext,
    ) -> Result<(), CustomizerError> {
        Ok(())
    }
}

// ===========================================================================
// THE KITCHEN SINK
// ===========================================================================

#[tokio::test]
#[ignore = "integration-real: live API, ~120s"]
async fn e2e_kitchen_sink_two_mob_chaos() {
    if !has_api_key() {
        eprintln!("Skipping kitchen-sink: no ANTHROPIC_API_KEY");
        return;
    }

    let model = smoke_model();
    let temp = tempfile::TempDir::new().expect("temp dir");
    let author_state = temp.path().join("authors");
    let critic_state = temp.path().join("critics");
    let author_continuity_path = temp.path().join("author_continuity.db");
    let critic_continuity_path = temp.path().join("critic_continuity.db");

    // Shared session stores — survive across v1→v2 rebuild (like a persistent volume).
    // Using mob_storage_in_memory() avoids the redb exclusive file lock while still
    // getting persistent sessions via the shared SQLite store.
    let author_session_store: Arc<dyn meerkat::SessionStore> = Arc::new(
        meerkat_store::SqliteSessionStore::open(temp.path().join("author_sessions.db"))
            .expect("author session store"),
    );
    let critic_session_store: Arc<dyn meerkat::SessionStore> = Arc::new(
        meerkat_store::SqliteSessionStore::open(temp.path().join("critic_sessions.db"))
            .expect("critic session store"),
    );

    let start = Instant::now();

    eprintln!("╔══════════════════════════════════════════════════════════╗");
    eprintln!("║  KITCHEN SINK: Two-Mob Collaboration with Chaos        ║");
    eprintln!("║  model={model}");
    eprintln!("╚══════════════════════════════════════════════════════════╝");

    // ===================================================================
    // PHASE 1: Bootstrap both mobs (7 identities)
    // ===================================================================
    eprintln!("\n--- Phase 1: Bootstrap ---");

    let dir = ContactDirectory::from_toml("[mobs]\nauthors = \"inproc\"\ncritics = \"inproc\"")
        .expect("contact directory");

    // Build both UnifiedRuntimes
    let rt_authors = UnifiedRuntimeBuilder::default()
        .definition(author_definition(&model))
        .persistent_state(&author_state)
        .session_store(author_session_store.clone())
        .mob_storage_in_memory()
        .comms(true)
        .contact_directory(dir.clone())
        .build()
        .await
        .expect("build author runtime");

    let rt_critics = UnifiedRuntimeBuilder::default()
        .definition(critic_definition(&model))
        .persistent_state(&critic_state)
        .session_store(critic_session_store.clone())
        .mob_storage_in_memory()
        .comms(true)
        .contact_directory(dir.clone())
        .build()
        .await
        .expect("build critic runtime");

    // Register peer handles (bidirectional)
    rt_authors
        .register_peer_mob("critics", rt_critics.mob_handle())
        .await;
    rt_critics
        .register_peer_mob("authors", rt_authors.mob_handle())
        .await;

    // Get bridges
    let author_bridge: Arc<dyn SessionBridge> =
        rt_authors.session_bridge().expect("author bridge").clone();
    let critic_bridge: Arc<dyn SessionBridge> =
        rt_critics.session_bridge().expect("critic bridge").clone();

    // Continuity stores (SQLite — survive across drops)
    let author_store = Arc::new(
        LocalContinuityStore::open(&author_continuity_path).expect("author continuity store"),
    );
    let critic_store = Arc::new(
        LocalContinuityStore::open(&critic_continuity_path).expect("critic continuity store"),
    );

    // Lease providers (in-memory — fresh per "process")
    let author_leases = Arc::new(LocalLeaseProvider::new());
    let critic_leases = Arc::new(LocalLeaseProvider::new());

    // Identity runtimes
    let author_irt = IdentityRuntime::new(IdentityRuntimeConfig {
        continuity_store: author_store.clone() as Arc<dyn ContinuityStore>,
        lease_provider: author_leases.clone(),
        runtime_instance_id: "authors-v1".to_string(),
        has_runtime_store: true,
        durability_policy: DurabilityPolicy::SyncWriteThrough,
        bridge: Some(author_bridge),
        default_timeout: None,
    });
    let critic_irt = IdentityRuntime::new(IdentityRuntimeConfig {
        continuity_store: critic_store.clone() as Arc<dyn ContinuityStore>,
        lease_provider: critic_leases.clone(),
        runtime_instance_id: "critics-v1".to_string(),
        has_runtime_store: true,
        durability_policy: DurabilityPolicy::SyncWriteThrough,
        bridge: Some(critic_bridge),
        default_timeout: None,
    });

    // Rosters
    let author_roster = vec![
        spec(
            "coordinator:mc",
            AgentAddressability::Addressable,
            "coordinator",
        ),
        spec("lead:author", AgentAddressability::InternalOnly, "lead"),
        spec("writer:a", AgentAddressability::InternalOnly, "writer"),
        spec("writer:b", AgentAddressability::InternalOnly, "writer"),
    ];
    let critic_roster = vec![
        spec("lead:critic", AgentAddressability::InternalOnly, "lead"),
        spec("judge:x", AgentAddressability::InternalOnly, "judge"),
        spec("judge:y", AgentAddressability::InternalOnly, "judge"),
    ];

    // Topology providers (mutable for later phases)
    let author_topo = MutableTopology::new(vec![
        ("coordinator:mc".into(), "lead:author".into()),
        ("lead:author".into(), "writer:a".into()),
        ("lead:author".into(), "writer:b".into()),
    ]);
    let critic_topo = MutableTopology::new(vec![
        ("lead:critic".into(), "judge:x".into()),
        ("lead:critic".into(), "judge:y".into()),
    ]);

    let customizer = JokeCollabCustomizer;

    // restore_flow for both mobs
    eprintln!("[Phase 1] restore_flow for author mob (4 identities)...");
    let author_result = author_irt
        .restore_flow(
            &author_roster,
            Some(&author_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("author restore_flow");

    eprintln!("[Phase 1] restore_flow for critic mob (3 identities)...");
    let critic_result = critic_irt
        .restore_flow(
            &critic_roster,
            Some(&critic_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("critic restore_flow");

    // Collect records for all 7 identities
    let mut records: BTreeMap<String, ContinuityRecord> = BTreeMap::new();
    for (name, result) in [("author", &author_result), ("critic", &critic_result)] {
        for (identity, outcome) in &result.outcomes {
            match outcome {
                RestoreOutcome::Created { record, .. } => {
                    eprintln!(
                        "[Phase 1] {identity} created: rt={}, session={}",
                        record.agent_runtime_id, record.session_id
                    );
                    records.insert(identity.to_string(), record.clone());
                }
                other => {
                    panic!("[Phase 1] expected Created for {identity} in {name}, got {other:?}")
                }
            }
        }
    }
    assert_eq!(records.len(), 7, "should have 7 identities total");

    // Verify topology-aware prompts
    let lead_author_draft = match author_result.outcomes.get(&id("lead:author")).unwrap() {
        RestoreOutcome::Created { draft, .. } => draft,
        _ => unreachable!(),
    };
    let lead_prompt = lead_author_draft
        .system_prompt
        .as_ref()
        .expect("lead:author should have system prompt");
    assert!(
        lead_prompt.contains("writer:a") && lead_prompt.contains("writer:b"),
        "lead:author prompt should mention both writers: {lead_prompt}"
    );
    assert!(
        lead_prompt.contains("Lead Author"),
        "lead:author prompt should describe the Lead Author role: {lead_prompt}"
    );
    eprintln!("[Phase 1] lead:author prompt OK (mentions writers + role)");

    // Wire cross-mob by identity — no runtime ID extraction needed
    eprintln!("[Phase 1] Wiring cross-mob: lead:author ↔ lead:critic...");
    wire_cross_mob_by_identity(
        &author_irt,
        &id("lead:author"),
        &critic_irt,
        &id("lead:critic"),
        &rt_authors,
        "critics",
    )
    .await
    .expect("wire lead:author ↔ lead:critic");

    eprintln!("[Phase 1] Wiring cross-mob: coordinator:mc ↔ lead:critic...");
    wire_cross_mob_by_identity(
        &author_irt,
        &id("coordinator:mc"),
        &critic_irt,
        &id("lead:critic"),
        &rt_authors,
        "critics",
    )
    .await
    .expect("wire coordinator:mc ↔ lead:critic");

    eprintln!(
        "[Phase 1] PASSED ({:.1}s) — 7 identities bootstrapped, cross-mob wired",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 2: Write the world's greatest LLM joke (iterative refinement)
    // ===================================================================
    eprintln!("\n--- Phase 2: The World's Greatest LLM Joke ---");

    /// Extract a numeric score from judge output. Looks for "SCORE: N/10" or "N/10".
    fn extract_score(text: &str) -> Option<u32> {
        // Try "SCORE: N/10" first
        if let Some(pos) = text.find("SCORE:") {
            let after = &text[pos + 6..];
            let digits: String = after
                .trim_start()
                .chars()
                .take_while(|c| c.is_ascii_digit())
                .collect();
            if let Ok(n) = digits.parse::<u32>() {
                if n <= 10 {
                    return Some(n);
                }
            }
        }
        // Fallback: find any N/10 pattern
        for word in text.split_whitespace() {
            if word.ends_with("/10") || word.ends_with("/10.") || word.ends_with("/10,") {
                let num_part = word.trim_end_matches(|c: char| !c.is_ascii_digit());
                let digits: String = num_part
                    .chars()
                    .rev()
                    .take_while(|c| c.is_ascii_digit())
                    .collect::<String>()
                    .chars()
                    .rev()
                    .collect();
                if let Ok(n) = digits.parse::<u32>() {
                    if n <= 10 {
                        return Some(n);
                    }
                }
            }
        }
        None
    }

    // --- Round 0: Writers submit initial drafts ---
    eprintln!("[Phase 2] Round 0: Writers drafting...");
    author_irt
        .dispatch_text(
            &id("writer:a"),
            "Write the world's greatest joke about large language models. \
             It should be funny to both AI researchers and normal people. \
             2-4 sentences max. No explanation.",
        )
        .await
        .expect("dispatch to writer:a");

    author_irt
        .dispatch_text(
            &id("writer:b"),
            "Write the world's greatest joke about large language models. \
             Go for something that reveals a deep truth about LLMs in a surprising, \
             hilarious way. 2-4 sentences max. No explanation.",
        )
        .await
        .expect("dispatch to writer:b");

    let joke_a = author_irt
        .wait_for_output(&id("writer:a"), WAIT_TIMEOUT)
        .await
        .expect("writer:a output");
    let joke_b = author_irt
        .wait_for_output(&id("writer:b"), WAIT_TIMEOUT)
        .await
        .expect("writer:b output");
    eprintln!("[Phase 2] Writer A: {joke_a}");
    eprintln!("[Phase 2] Writer B: {joke_b}");

    // --- Lead author picks the best draft ---
    author_irt
        .dispatch_text(
            &id("lead:author"),
            format!(
                "Two writers submitted LLM jokes. Pick the funnier one and output ONLY \
                 the joke text (2-4 sentences, no commentary):\n\n\
                 Writer A: {joke_a}\n\nWriter B: {joke_b}"
            ),
        )
        .await
        .expect("dispatch to lead:author");

    let mut current_joke = author_irt
        .wait_for_output(&id("lead:author"), WAIT_TIMEOUT)
        .await
        .expect("lead:author pick output");
    eprintln!("[Phase 2] Lead's pick: {current_joke}");

    // --- Refinement loop: authors ↔ critics up to 10 rounds ---
    const MAX_ROUNDS: u32 = 10;
    const TARGET_SCORE: u32 = 9;
    let mut best_score: u32 = 0;
    let mut round = 1;

    loop {
        eprintln!("\n[Phase 2] === Round {round}/{MAX_ROUNDS} ===");

        // Send to both judges for scoring
        critic_irt
            .dispatch_text(
                &id("judge:x"),
                format!("Score this LLM joke. Start with 'SCORE: N/10' then give feedback.\n\n{current_joke}"),
            )
            .await
            .expect("dispatch to judge:x");

        critic_irt
            .dispatch_text(
                &id("judge:y"),
                format!("Score this LLM joke. Start with 'SCORE: N/10' then give feedback.\n\n{current_joke}"),
            )
            .await
            .expect("dispatch to judge:y");

        let review_x = critic_irt
            .wait_for_output(&id("judge:x"), WAIT_TIMEOUT)
            .await
            .expect("judge:x output");
        let review_y = critic_irt
            .wait_for_output(&id("judge:y"), WAIT_TIMEOUT)
            .await
            .expect("judge:y output");

        let score_x = extract_score(&review_x).unwrap_or(5);
        let score_y = extract_score(&review_y).unwrap_or(5);
        let avg_score = (score_x + score_y) / 2;
        best_score = best_score.max(avg_score);

        eprintln!("[Phase 2] Judge X: {score_x}/10 — Judge Y: {score_y}/10 — avg: {avg_score}/10");
        eprintln!("[Phase 2] Judge X feedback: {review_x}");
        eprintln!("[Phase 2] Judge Y feedback: {review_y}");

        if avg_score >= TARGET_SCORE || round >= MAX_ROUNDS {
            eprintln!(
                "[Phase 2] {} after {round} round(s) (avg score: {avg_score}/10)",
                if avg_score >= TARGET_SCORE {
                    "Target reached"
                } else {
                    "Max rounds reached"
                }
            );
            break;
        }

        // Send feedback to lead author for revision
        author_irt
            .dispatch_text(
                &id("lead:author"),
                format!(
                    "The critics scored your joke {score_x}/10 and {score_y}/10. \
                     Here's their feedback:\n\n\
                     Judge X: {review_x}\n\n\
                     Judge Y: {review_y}\n\n\
                     Revise the joke based on this feedback. Output ONLY the improved \
                     joke text (2-4 sentences, no commentary). Aim for {TARGET_SCORE}/10."
                ),
            )
            .await
            .expect("dispatch revision to lead:author");

        current_joke = author_irt
            .wait_for_output(&id("lead:author"), WAIT_TIMEOUT)
            .await
            .expect("lead:author revision output");
        eprintln!("[Phase 2] Revised joke: {current_joke}");

        round += 1;
    }

    // --- Deliver final joke to coordinator ---
    author_irt
        .send_text(
            &id("coordinator:mc"),
            format!(
                "After {round} round(s) of refinement, here is the team's final joke \
                 (best score: {best_score}/10):\n\n{current_joke}\n\n\
                 Present this as the team's masterpiece."
            ),
        )
        .await
        .expect("send final joke to coordinator");

    let final_delivery = author_irt
        .wait_for_output(&id("coordinator:mc"), WAIT_TIMEOUT)
        .await
        .expect("coordinator final output");
    eprintln!("[Phase 2] FINAL DELIVERY:\n{final_delivery}");

    assert!(
        best_score >= 5,
        "judges should score the joke at least 5/10 after refinement, got {best_score}/10"
    );

    eprintln!(
        "[Phase 2] PASSED ({:.1}s) — {round} rounds, best score {best_score}/10",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 3: Verify NotAddressable enforcement
    // ===================================================================
    eprintln!("\n--- Phase 3: Addressability enforcement ---");

    // send() to InternalOnly should fail
    let reject = author_irt.send_text(&id("lead:author"), "hi").await;
    assert!(
        matches!(reject, Err(IdentityRuntimeError::NotAddressable(_))),
        "send to InternalOnly lead:author should be NotAddressable: {reject:?}"
    );

    // dispatch() to InternalOnly should succeed (already proven in phase 2)
    // send() to Addressable should succeed
    author_irt
        .send_text(&id("coordinator:mc"), "Say ADDR_OK and nothing else.")
        .await
        .expect("send to addressable coordinator");

    author_irt
        .wait_for_output_containing(&id("coordinator:mc"), "ADDR_OK", WAIT_TIMEOUT)
        .await
        .expect("Phase 3 addressability output");

    eprintln!(
        "[Phase 3] PASSED ({:.1}s) — NotAddressable enforced, Addressable delivers",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 4: FULL SYSTEM DROP + RESTORE (chaos event 1)
    // ===================================================================
    eprintln!("\n--- Phase 4: System drop + restore ---");

    // Record pre-crash state for all 7 identities
    let pre_crash_sessions: BTreeMap<String, _> = records
        .iter()
        .map(|(k, r)| {
            (
                k.clone(),
                (r.session_id.clone(), r.agent_runtime_id.clone()),
            )
        })
        .collect();

    // Get a fencing token from writer:a for stale checkpoint test
    let wa_status = author_irt
        .status(&id("writer:a"))
        .await
        .expect("writer:a status");
    let wa_old_fencing = wa_status
        .lease
        .as_ref()
        .expect("writer:a should have lease")
        .fencing_token;

    eprintln!("[Phase 4] Shutting down both runtimes (simulating crash)...");
    // Drop identity runtimes first (they hold bridge Arcs → MobHandle clones).
    drop(author_irt);
    drop(critic_irt);
    drop(author_leases);
    drop(critic_leases);
    // Graceful shutdown stops the mob actor and drains background tasks.
    rt_authors.shutdown().await;
    rt_critics.shutdown().await;
    drop(rt_authors);
    drop(rt_critics);
    // Clear the InprocRegistry (process-global peer discovery).
    // Found bug: mob shutdown doesn't deregister comms identities from the
    // process-global InprocRegistry or SESSION_IDENTITY_CLAIMS. In production
    // the process dies and these are gone; in-process restart must clear manually.
    // The SESSION_IDENTITY_CLAIMS cannot be cleared externally (private static),
    // so resume_session will fall back to fresh spawn when the old comms identity
    // is still claimed. This means in-process restart loses conversation history.
    meerkat_comms::InprocRegistry::global().clear();

    // -- Fencing race: inject stale checkpoint --
    eprintln!("[Phase 4] Stale checkpoint injection test...");
    let stale_result = author_store
        .save_session_snapshot(
            &id("writer:a"),
            &pre_crash_sessions["writer:a"].0,
            ContinuityGeneration::new(0),
            CheckpointVersion::new(999), // version doesn't matter — token is stale
            FencingToken::new(wa_old_fencing.get().saturating_sub(1)), // stale token
            &SessionSnapshot {
                data: b"stale-data".to_vec(),
            },
        )
        .await;
    assert!(
        matches!(
            stale_result,
            Err(ContinuityStoreError::StaleFencingToken { .. })
        ),
        "stale checkpoint should be rejected: {stale_result:?}"
    );
    eprintln!("[Phase 4] Stale checkpoint correctly REJECTED");

    // -- Rebuild both runtimes (ephemeral + shared session store) --
    eprintln!("[Phase 4] Rebuilding runtimes...");
    let dir2 = ContactDirectory::from_toml("[mobs]\nauthors = \"inproc\"\ncritics = \"inproc\"")
        .expect("contact directory v2");

    let rt_authors = UnifiedRuntimeBuilder::default()
        .definition(author_definition(&model))
        .persistent_state(&author_state)
        .session_store(author_session_store.clone())
        .mob_storage_in_memory()
        .comms(true)
        .contact_directory(dir2.clone())
        .build()
        .await
        .expect("rebuild author runtime");

    let rt_critics = UnifiedRuntimeBuilder::default()
        .definition(critic_definition(&model))
        .persistent_state(&critic_state)
        .session_store(critic_session_store.clone())
        .mob_storage_in_memory()
        .comms(true)
        .contact_directory(dir2)
        .build()
        .await
        .expect("rebuild critic runtime");

    rt_authors
        .register_peer_mob("critics", rt_critics.mob_handle())
        .await;
    rt_critics
        .register_peer_mob("authors", rt_authors.mob_handle())
        .await;

    let author_bridge2: Arc<dyn SessionBridge> = rt_authors
        .session_bridge()
        .expect("author bridge v2")
        .clone();
    let critic_bridge2: Arc<dyn SessionBridge> = rt_critics
        .session_bridge()
        .expect("critic bridge v2")
        .clone();

    // Fresh lease providers (old leases died with the "crashed" process)
    let author_leases2 = Arc::new(LocalLeaseProvider::new());
    let critic_leases2 = Arc::new(LocalLeaseProvider::new());

    // Build author IdentityRuntime but DO NOT call restore_flow yet
    let author_irt2 = IdentityRuntime::new(IdentityRuntimeConfig {
        continuity_store: author_store.clone() as Arc<dyn ContinuityStore>,
        lease_provider: author_leases2.clone(),
        runtime_instance_id: "authors-v2".to_string(),
        has_runtime_store: true,
        durability_policy: DurabilityPolicy::SyncWriteThrough,
        bridge: Some(author_bridge2),
        default_timeout: None,
    });

    // -- Pre-restore dispatch test --
    eprintln!("[Phase 4] Pre-restore dispatch test (should fail with structured error)...");
    let pre_restore = author_irt2
        .dispatch_text(&id("lead:author"), "hello?")
        .await;
    assert!(
        matches!(
            pre_restore,
            Err(IdentityRuntimeError::UnknownIdentity(_))
                | Err(IdentityRuntimeError::NoActiveLease(_))
        ),
        "pre-restore dispatch should fail with structured error, got: {pre_restore:?}"
    );
    eprintln!("[Phase 4] Pre-restore dispatch correctly returned: {pre_restore:?}");

    // -- Now restore both mobs --
    eprintln!("[Phase 4] Running restore_flow for both mobs...");
    let author_result2 = author_irt2
        .restore_flow(
            &author_roster,
            Some(&author_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("author restore_flow v2");

    let critic_irt2 = IdentityRuntime::new(IdentityRuntimeConfig {
        continuity_store: critic_store.clone() as Arc<dyn ContinuityStore>,
        lease_provider: critic_leases2.clone(),
        runtime_instance_id: "critics-v2".to_string(),
        has_runtime_store: true,
        durability_policy: DurabilityPolicy::SyncWriteThrough,
        bridge: Some(critic_bridge2),
        default_timeout: None,
    });

    let critic_result2 = critic_irt2
        .restore_flow(
            &critic_roster,
            Some(&critic_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("critic restore_flow v2");

    // Verify all 7 identities resumed (or re-created) with same session IDs
    for (identity_str, (old_session, old_rt_id)) in &pre_crash_sessions {
        let identity = id(identity_str);
        let outcome = author_result2
            .outcomes
            .get(&identity)
            .or_else(|| critic_result2.outcomes.get(&identity))
            .unwrap_or_else(|| panic!("missing outcome for {identity_str}"));

        let (new_session, new_rt_id) = match outcome {
            RestoreOutcome::Created { record, .. } | RestoreOutcome::Resumed { record, .. } => {
                (&record.session_id, &record.agent_runtime_id)
            }
            RestoreOutcome::Broken(f) => {
                panic!("{identity_str} restore failed: {f:?}")
            }
        };

        assert_eq!(
            new_rt_id, old_rt_id,
            "{identity_str} should keep same AgentRuntimeId: old={old_rt_id}, new={new_rt_id}"
        );
        assert_eq!(
            new_session, old_session,
            "{identity_str} should keep same SessionId: old={old_session}, new={new_session}"
        );
    }
    eprintln!("[Phase 4] All 7 identities resumed with stable IDs");

    // Re-wire cross-mob after rebuild — identity-first, no runtime ID extraction
    wire_cross_mob_by_identity(
        &author_irt2,
        &id("lead:author"),
        &critic_irt2,
        &id("lead:critic"),
        &rt_authors,
        "critics",
    )
    .await
    .expect("re-wire lead:author ↔ lead:critic");
    wire_cross_mob_by_identity(
        &author_irt2,
        &id("coordinator:mc"),
        &critic_irt2,
        &id("lead:critic"),
        &rt_authors,
        "critics",
    )
    .await
    .expect("re-wire coordinator:mc ↔ lead:critic");

    // Post-restore dispatch: verify agents can receive messages after rebuild.
    // NOTE: In-process restart loses conversation history because the old mob
    // actor's SESSION_IDENTITY_CLAIMS are not released (they're held by the
    // detached tokio task). The bridge falls back to fresh spawn. In a real
    // crash/restart (different process), MemberLaunchMode::Resume would work
    // because the claims die with the process.
    eprintln!("[Phase 4] Post-restore dispatch test...");
    author_irt2
        .dispatch_text(&id("writer:b"), "Say POST_RESTORE_OK and nothing else.")
        .await
        .expect("dispatch post-restore");

    author_irt2
        .wait_for_output_containing(&id("writer:b"), "POST_RESTORE_OK", WAIT_TIMEOUT)
        .await
        .expect("Phase 4 post-restore output");
    eprintln!("[Phase 4] writer:b responds after restore");

    eprintln!(
        "[Phase 4] PASSED ({:.1}s) — stale checkpoint rejected, pre-restore error OK, all resumed",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 5: writer:b drops out (chaos event 2)
    // ===================================================================
    eprintln!("\n--- Phase 5: writer:b dropout ---");

    // Retire writer:b
    author_irt2
        .retire(&id("writer:b"))
        .await
        .expect("retire writer:b");
    eprintln!("[Phase 5] writer:b retired");

    // Update topology: remove writer:b edge
    author_topo
        .set_edges(vec![
            ("coordinator:mc".into(), "lead:author".into()),
            ("lead:author".into(), "writer:a".into()),
            // writer:b edge removed
        ])
        .await;

    // Re-run restore_flow with updated roster (without writer:b)
    let author_roster_v2 = vec![
        spec(
            "coordinator:mc",
            AgentAddressability::Addressable,
            "coordinator",
        ),
        spec("lead:author", AgentAddressability::InternalOnly, "lead"),
        spec("writer:a", AgentAddressability::InternalOnly, "writer"),
        // writer:b removed from roster
    ];

    let author_result3 = author_irt2
        .restore_flow(
            &author_roster_v2,
            Some(&author_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("author restore_flow v3");

    // Verify lead:author prompt now only mentions writer:a
    let lead_v3_draft = match author_result3.outcomes.get(&id("lead:author")).unwrap() {
        RestoreOutcome::Created { draft, .. } | RestoreOutcome::Resumed { draft, .. } => draft,
        other => panic!("unexpected lead:author outcome: {other:?}"),
    };
    let lead_v3_prompt = lead_v3_draft
        .system_prompt
        .as_ref()
        .expect("lead:author v3 should have prompt");
    assert!(
        lead_v3_prompt.contains("writer:a"),
        "lead:author prompt should still mention writer:a: {lead_v3_prompt}"
    );
    assert!(
        !lead_v3_prompt.contains("writer:b"),
        "lead:author prompt should NOT mention writer:b: {lead_v3_prompt}"
    );
    eprintln!("[Phase 5] lead:author prompt updated: {lead_v3_prompt}");

    // Verify writer:a is unaffected
    let wa_status = author_irt2
        .status(&id("writer:a"))
        .await
        .expect("writer:a status");
    assert_eq!(wa_status.state, IdentityLifecycleState::Active);

    eprintln!(
        "[Phase 5] PASSED ({:.1}s) — writer:b retired, topology updated, writer:a unaffected",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 6: judge:y drops out (chaos event 3)
    // ===================================================================
    eprintln!("\n--- Phase 6: judge:y dropout ---");

    critic_irt2
        .retire(&id("judge:y"))
        .await
        .expect("retire judge:y");

    // Update critic topology
    critic_topo
        .set_edges(vec![("lead:critic".into(), "judge:x".into())])
        .await;

    eprintln!(
        "[Phase 6] PASSED ({:.1}s) — judge:y retired",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 7: Critic mob functions after member loss
    // ===================================================================
    eprintln!("\n--- Phase 7: Critic mob post-dropout ---");

    // Verify remaining critic agents still function after judge:y dropout
    critic_irt2
        .dispatch_text(
            &id("judge:x"),
            "Score this joke 1-10: 'A programmer walks into a bar and orders 1.0 beers. \
             The bartender says: you sure you don\\'t want 1.00000000001?'",
        )
        .await
        .expect("dispatch to judge:x post-dropout");

    let jx_output = critic_irt2
        .wait_for_output(&id("judge:x"), WAIT_TIMEOUT)
        .await
        .expect("Phase 7 judge:x output");
    eprintln!("[Phase 7] judge:x scored post-dropout: {jx_output}");

    eprintln!(
        "[Phase 7] PASSED ({:.1}s) — critic mob functional after judge:y dropout",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 8: writer:b RETURNS via respawn (chaos event 4)
    // ===================================================================
    eprintln!("\n--- Phase 8: writer:b returns (respawn) ---");

    // Re-add writer:b to roster and topology
    author_topo
        .set_edges(vec![
            ("coordinator:mc".into(), "lead:author".into()),
            ("lead:author".into(), "writer:a".into()),
            ("lead:author".into(), "writer:b".into()),
        ])
        .await;

    // Restore with full roster again (writer:b back)
    let author_result4 = author_irt2
        .restore_flow(
            &author_roster, // original full roster
            Some(&author_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("author restore_flow v4");

    // writer:b should be re-created (was retired, continuity still in store)
    let wb_v4 = match author_result4.outcomes.get(&id("writer:b")).unwrap() {
        RestoreOutcome::Created { record, .. } | RestoreOutcome::Resumed { record, .. } => record,
        other => panic!("unexpected writer:b outcome: {other:?}"),
    };
    eprintln!(
        "[Phase 8] writer:b back: rt={}, gen={}, session={}",
        wb_v4.agent_runtime_id, wb_v4.generation, wb_v4.session_id
    );

    // Verify lead:author prompt is back to both writers
    let lead_v4_draft = match author_result4.outcomes.get(&id("lead:author")).unwrap() {
        RestoreOutcome::Created { draft, .. } | RestoreOutcome::Resumed { draft, .. } => draft,
        other => panic!("unexpected lead:author outcome: {other:?}"),
    };
    let lead_v4_prompt = lead_v4_draft
        .system_prompt
        .as_ref()
        .expect("lead:author v4 should have prompt");
    assert!(
        lead_v4_prompt.contains("writer:a") && lead_v4_prompt.contains("writer:b"),
        "lead:author prompt should mention both writers again: {lead_v4_prompt}"
    );

    // Verify writer:b can receive messages after return
    author_irt2
        .dispatch_text(&id("writer:b"), "Say WRITER_B_RETURNED_OK.")
        .await
        .expect("dispatch to writer:b after return");

    author_irt2
        .wait_for_output_containing(&id("writer:b"), "WRITER_B_RETURNED_OK", WAIT_TIMEOUT)
        .await
        .expect("Phase 8 writer:b returned output");
    eprintln!("[Phase 8] writer:b functional after return");

    eprintln!(
        "[Phase 8] PASSED ({:.1}s) — writer:b returned with history intact",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 9: judge:y RESET — clean slate (chaos event 5)
    // ===================================================================
    eprintln!("\n--- Phase 9: judge:y reset (destructive) ---");

    // First re-add judge:y to the critic roster
    critic_topo
        .set_edges(vec![
            ("lead:critic".into(), "judge:x".into()),
            ("lead:critic".into(), "judge:y".into()),
        ])
        .await;

    // Restore to bring judge:y back first
    let critic_result3 = critic_irt2
        .restore_flow(
            &critic_roster, // original full roster
            Some(&critic_topo as &dyn TopologyProvider),
            Some(&customizer as &dyn AgentCustomizer),
        )
        .await
        .expect("critic restore_flow v3");

    let jy_pre_reset = match critic_result3.outcomes.get(&id("judge:y")).unwrap() {
        RestoreOutcome::Created { record, .. } | RestoreOutcome::Resumed { record, .. } => {
            record.clone()
        }
        other => panic!("unexpected judge:y outcome: {other:?}"),
    };
    let jy_old_gen = jy_pre_reset.generation;
    let jy_old_session = jy_pre_reset.session_id.clone();
    eprintln!(
        "[Phase 9] judge:y pre-reset: gen={}, session={}",
        jy_old_gen, jy_old_session
    );

    // Now reset judge:y — destructive
    let reset_record = critic_irt2
        .reset(&id("judge:y"))
        .await
        .expect("reset judge:y");

    assert_eq!(
        reset_record.generation,
        ContinuityGeneration::new(jy_old_gen.get() + 1),
        "reset should advance generation"
    );
    assert_ne!(
        reset_record.session_id, jy_old_session,
        "reset should mint new SessionId"
    );
    assert_eq!(
        reset_record.checkpoint_version,
        CheckpointVersion::new(0),
        "reset should reset checkpoint version to 0"
    );
    eprintln!(
        "[Phase 9] judge:y reset: new gen={}, new session={}",
        reset_record.generation, reset_record.session_id
    );

    // judge:y should have NO memory of DOLPHIN
    eprintln!("[Phase 9] Memory test: judge:y should NOT remember DOLPHIN...");
    critic_irt2
        .dispatch_text(
            &id("judge:y"),
            "Do you remember any secret code from a previous conversation? If yes say the code, if no say NO_MEMORY. Then say RESET_CHECK_OK.",
        )
        .await
        .expect("dispatch to reset judge:y");

    let jy_reset_output = critic_irt2
        .wait_for_output_containing(&id("judge:y"), "RESET_CHECK_OK", WAIT_TIMEOUT)
        .await
        .expect("Phase 9 judge:y reset check output");
    assert!(
        !jy_reset_output.contains("DOLPHIN"),
        "judge:y should NOT remember DOLPHIN after reset: {jy_reset_output}"
    );
    eprintln!("[Phase 9] judge:y has no memory of DOLPHIN — reset is destructive");

    // judge:x should be unaffected by judge:y's reset — verify via status
    // (no extra LLM call needed; we already proved judge:x works in Phase 7)
    let jx_status = critic_irt2
        .status(&id("judge:x"))
        .await
        .expect("judge:x status after judge:y reset");
    assert_eq!(
        jx_status.state,
        IdentityLifecycleState::Active,
        "judge:x should still be Active after judge:y reset"
    );
    assert_eq!(
        jx_status.generation,
        Some(ContinuityGeneration::new(0)),
        "judge:x generation should be unchanged"
    );
    eprintln!("[Phase 9] judge:x still Active gen=0 — unaffected by judge:y reset");

    eprintln!(
        "[Phase 9] PASSED ({:.1}s) — reset vs respawn produce genuinely different outcomes",
        start.elapsed().as_secs_f64()
    );

    // ===================================================================
    // PHASE 10: Final delivery — system works after all chaos
    // ===================================================================
    eprintln!("\n--- Phase 10: Final delivery post-chaos ---");

    // Send to coordinator to prove the whole system still works after all chaos
    author_irt2
        .send_text(
            &id("coordinator:mc"),
            "The joke collaboration survived all the chaos. Summarize what happened \
             and declare the project complete.",
        )
        .await
        .expect("final delivery to coordinator");

    let final_output = author_irt2
        .wait_for_output(&id("coordinator:mc"), WAIT_TIMEOUT)
        .await
        .expect("Phase 10 final output");
    eprintln!("[Phase 10] coordinator:mc final: {final_output}");

    // Final status check on all remaining active identities
    for name in ["coordinator:mc", "lead:author", "writer:a", "writer:b"] {
        let s = author_irt2.status(&id(name)).await.expect("status");
        assert_eq!(
            s.state,
            IdentityLifecycleState::Active,
            "{name} should be Active"
        );
    }
    for name in ["lead:critic", "judge:x", "judge:y"] {
        let s = critic_irt2.status(&id(name)).await.expect("status");
        assert_eq!(
            s.state,
            IdentityLifecycleState::Active,
            "{name} should be Active"
        );
    }

    let elapsed = start.elapsed();
    eprintln!("\n╔══════════════════════════════════════════════════════════╗");
    eprintln!(
        "║  KITCHEN SINK PASSED in {:.1}s                           ║",
        elapsed.as_secs_f64()
    );
    eprintln!("║  7 identities, 2 mobs, system crash, fencing race,     ║");
    eprintln!("║  member dropout, respawn with memory, reset without,   ║");
    eprintln!("║  topology changes, cross-mob wiring — ALL verified.    ║");
    eprintln!("╚══════════════════════════════════════════════════════════╝");

    assert!(
        elapsed.as_secs() >= 5,
        "Completed in {elapsed:?} — too fast for real LLM calls across 7 agents"
    );
}
