use chrono::TimeZone;
use chrono::Utc;
use codex_config::types::DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION;
use codex_memories_write::clear_memory_roots_contents;
use codex_memories_write::ensure_layout;
use codex_memories_write::memory_root;
use codex_memories_write::raw_memories_file;
use codex_memories_write::rebuild_raw_memories_file_from_memories;
use codex_memories_write::rollout_summaries_dir;
use codex_memories_write::sync_rollout_summaries_from_memories;
use codex_protocol::ThreadId;
use codex_state::Stage1Output;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::path::Path;
use std::path::PathBuf;
use tempfile::tempdir;

#[test]
fn memory_root_uses_shared_global_path() {
    let codex_home = AbsolutePathBuf::current_dir().expect("cwd").join("codex");
    assert_eq!(memory_root(&codex_home), codex_home.join("memories"));
}

#[test]
fn stage_one_output_schema_requires_rollout_slug_and_keeps_it_nullable() {
    let schema = crate::memories::phase1::output_schema();
    let properties = schema
        .get("properties")
        .and_then(Value::as_object)
        .expect("properties object");
    let required = schema
        .get("required")
        .and_then(Value::as_array)
        .expect("required array");

    let mut required_keys = required
        .iter()
        .map(|key| key.as_str().expect("required key string"))
        .collect::<Vec<_>>();
    required_keys.sort_unstable();

    assert!(
        properties.contains_key("rollout_slug"),
        "schema should declare rollout_slug"
    );

    let rollout_slug_type = properties
        .get("rollout_slug")
        .and_then(Value::as_object)
        .and_then(|schema| schema.get("type"))
        .and_then(Value::as_array)
        .expect("rollout_slug type array");
    let mut rollout_slug_types = rollout_slug_type
        .iter()
        .map(|entry| entry.as_str().expect("type entry string"))
        .collect::<Vec<_>>();
    rollout_slug_types.sort_unstable();

    assert_eq!(
        required_keys,
        vec!["raw_memory", "rollout_slug", "rollout_summary"]
    );
    assert_eq!(rollout_slug_types, vec!["null", "string"]);
}

#[tokio::test]
async fn clear_memory_root_contents_preserves_root_directory() {
    let dir = tempdir().expect("tempdir");
    let root = dir.path().join("memories");
    let nested_dir = root.join("rollout_summaries");
    tokio::fs::create_dir_all(&nested_dir)
        .await
        .expect("create rollout summaries dir");
    tokio::fs::write(root.join("MEMORY.md"), "stale memory index\n")
        .await
        .expect("write memory index");
    tokio::fs::write(nested_dir.join("rollout.md"), "stale rollout\n")
        .await
        .expect("write rollout summary");

    clear_memory_roots_contents(dir.path())
        .await
        .expect("clear memory root contents");

    assert!(
        tokio::fs::try_exists(&root)
            .await
            .expect("check memory root existence"),
        "memory root should still exist after clearing contents"
    );
    let mut entries = tokio::fs::read_dir(&root)
        .await
        .expect("read memory root after clear");
    assert!(
        entries
            .next_entry()
            .await
            .expect("read next entry")
            .is_none(),
        "memory root should be empty after clearing contents"
    );
}

#[cfg(unix)]
#[tokio::test]
async fn clear_memory_root_contents_rejects_symlinked_root() {
    let dir = tempdir().expect("tempdir");
    let target = dir.path().join("outside");
    tokio::fs::create_dir_all(&target)
        .await
        .expect("create symlink target dir");
    let target_file = target.join("keep.txt");
    tokio::fs::write(&target_file, "keep\n")
        .await
        .expect("write target file");

    let root = dir.path().join("memories");
    std::os::unix::fs::symlink(&target, &root).expect("create memory root symlink");

    let err = clear_memory_roots_contents(dir.path())
        .await
        .expect_err("symlinked memory root should be rejected");
    assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
    assert!(
        tokio::fs::try_exists(&target_file)
            .await
            .expect("check target file existence"),
        "rejecting a symlinked memory root should not delete the symlink target"
    );
}

struct ConsolidatedOutputPaths {
    memory_index: PathBuf,
    memory_summary: PathBuf,
    skill: PathBuf,
}

async fn write_consolidated_outputs(root: &Path) -> ConsolidatedOutputPaths {
    let paths = ConsolidatedOutputPaths {
        memory_index: root.join("MEMORY.md"),
        memory_summary: root.join("memory_summary.md"),
        skill: root.join("skills/demo/SKILL.md"),
    };

    tokio::fs::write(&paths.memory_index, "consolidated memory index\n")
        .await
        .expect("write memory index");
    tokio::fs::write(&paths.memory_summary, "consolidated memory summary\n")
        .await
        .expect("write memory summary");
    tokio::fs::create_dir_all(paths.skill.parent().expect("skill parent"))
        .await
        .expect("create skill dir");
    tokio::fs::write(&paths.skill, "consolidated skill\n")
        .await
        .expect("write skill");

    paths
}

async fn assert_consolidated_outputs_exist(paths: &ConsolidatedOutputPaths, context: &str) {
    assert!(
        tokio::fs::try_exists(&paths.memory_index)
            .await
            .expect("check memory index existence"),
        "{context} should leave MEMORY.md untouched"
    );
    assert!(
        tokio::fs::try_exists(&paths.memory_summary)
            .await
            .expect("check memory summary existence"),
        "{context} should leave memory_summary.md untouched"
    );
    assert!(
        tokio::fs::try_exists(&paths.skill)
            .await
            .expect("check skill existence"),
        "{context} should leave skills untouched"
    );
}

#[tokio::test]
async fn sync_rollout_summaries_and_raw_memories_file_keeps_latest_memories_only() {
    let dir = tempdir().expect("tempdir");
    let root = dir.path().join("memory");
    ensure_layout(&root).await.expect("ensure layout");

    let keep_id = ThreadId::default().to_string();
    let drop_id = ThreadId::default().to_string();
    let keep_path = rollout_summaries_dir(&root).join(format!("{keep_id}.md"));
    let drop_path = rollout_summaries_dir(&root).join(format!("{drop_id}.md"));
    tokio::fs::write(&keep_path, "keep")
        .await
        .expect("write keep");
    tokio::fs::write(&drop_path, "drop")
        .await
        .expect("write drop");

    let memories = vec![Stage1Output {
        thread_id: ThreadId::try_from(keep_id.clone()).expect("thread id"),
        source_updated_at: Utc.timestamp_opt(100, 0).single().expect("timestamp"),
        raw_memory: "raw memory".to_string(),
        rollout_summary: "short summary".to_string(),
        rollout_slug: None,
        rollout_path: PathBuf::from("/tmp/rollout-100.jsonl"),
        cwd: PathBuf::from("/tmp/workspace"),
        git_branch: None,
        generated_at: Utc.timestamp_opt(101, 0).single().expect("timestamp"),
    }];

    sync_rollout_summaries_from_memories(
        &root,
        &memories,
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("sync rollout summaries");
    rebuild_raw_memories_file_from_memories(
        &root,
        &memories,
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("rebuild raw memories");

    assert!(
        !tokio::fs::try_exists(&keep_path)
            .await
            .expect("check stale keep path"),
        "sync should prune stale filename that used thread id only"
    );
    assert!(
        !tokio::fs::try_exists(&drop_path)
            .await
            .expect("check stale drop path"),
        "sync should prune stale filename for dropped thread"
    );

    let mut dir = tokio::fs::read_dir(rollout_summaries_dir(&root))
        .await
        .expect("open rollout summaries dir");
    let mut files = Vec::new();
    while let Some(entry) = dir.next_entry().await.expect("read dir entry") {
        files.push(entry.file_name().to_string_lossy().to_string());
    }
    files.sort_unstable();
    assert_eq!(files.len(), 1);
    let canonical_rollout_summary_file = &files[0];

    let raw_memories = tokio::fs::read_to_string(raw_memories_file(&root))
        .await
        .expect("read raw memories");
    assert!(raw_memories.contains("raw memory"));
    assert!(raw_memories.contains(&keep_id));
    assert!(raw_memories.contains("cwd: /tmp/workspace"));
    assert!(raw_memories.contains("rollout_path: /tmp/rollout-100.jsonl"));
    assert!(raw_memories.contains(&format!(
        "rollout_summary_file: {canonical_rollout_summary_file}"
    )));
    let thread_header = format!("## Thread `{keep_id}`");
    let thread_pos = raw_memories
        .find(&thread_header)
        .expect("thread header should exist");
    let updated_pos = raw_memories[thread_pos..]
        .find("updated_at: ")
        .map(|offset| thread_pos + offset)
        .expect("updated_at should exist after thread header");
    let cwd_pos = raw_memories[thread_pos..]
        .find("cwd: /tmp/workspace")
        .map(|offset| thread_pos + offset)
        .expect("cwd should exist after thread header");
    let rollout_path_pos = raw_memories[thread_pos..]
        .find("rollout_path: /tmp/rollout-100.jsonl")
        .map(|offset| thread_pos + offset)
        .expect("rollout_path should exist after thread header");
    let file_pos = raw_memories[thread_pos..]
        .find(&format!(
            "rollout_summary_file: {canonical_rollout_summary_file}"
        ))
        .map(|offset| thread_pos + offset)
        .expect("rollout_summary_file should exist after thread header");
    assert!(thread_pos < updated_pos);
    assert!(updated_pos < cwd_pos);
    assert!(cwd_pos < rollout_path_pos);
    assert!(rollout_path_pos < file_pos);
}

#[tokio::test]
async fn sync_empty_inputs_preserves_consolidated_outputs() {
    let dir = tempdir().expect("tempdir");
    let root = dir.path().join("memory");
    ensure_layout(&root).await.expect("ensure layout");

    let stale_rollout_summary_path = rollout_summaries_dir(&root).join("stale.md");
    tokio::fs::write(&stale_rollout_summary_path, "stale summary\n")
        .await
        .expect("write stale rollout summary");
    let outputs = write_consolidated_outputs(&root).await;

    sync_rollout_summaries_from_memories(
        &root,
        &[],
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("sync empty rollout summaries");
    rebuild_raw_memories_file_from_memories(
        &root,
        &[],
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("rebuild empty raw memories");

    assert!(
        !tokio::fs::try_exists(&stale_rollout_summary_path)
            .await
            .expect("check stale rollout summary existence"),
        "empty sync should prune stale rollout summaries"
    );
    let raw_memories = tokio::fs::read_to_string(raw_memories_file(&root))
        .await
        .expect("read raw memories");
    assert_eq!(raw_memories, "# Raw Memories\n\nNo raw memories yet.\n");
    assert_consolidated_outputs_exist(&outputs, "empty sync").await;
}

#[tokio::test]
async fn sync_rollout_summaries_uses_timestamp_hash_and_sanitized_slug_filename() {
    let dir = tempdir().expect("tempdir");
    let root = dir.path().join("memory");
    ensure_layout(&root).await.expect("ensure layout");

    let thread_id = ThreadId::new();
    let stale_unslugged_path = rollout_summaries_dir(&root).join(format!("{thread_id}.md"));
    let stale_old_slug_path =
        rollout_summaries_dir(&root).join(format!("{thread_id}--old-slug.md"));
    tokio::fs::write(&stale_unslugged_path, "stale")
        .await
        .expect("write stale unslugged file");
    tokio::fs::write(&stale_old_slug_path, "stale")
        .await
        .expect("write stale old-slug file");

    let memories = vec![Stage1Output {
        thread_id,
        source_updated_at: Utc.timestamp_opt(200, 0).single().expect("timestamp"),
        raw_memory: "raw memory".to_string(),
        rollout_summary: "short summary".to_string(),
        rollout_slug: Some("Unsafe Slug/With Spaces & Symbols + EXTRA_LONG_12345".to_string()),
        rollout_path: PathBuf::from("/tmp/rollout-200.jsonl"),
        cwd: PathBuf::from("/tmp/workspace"),
        git_branch: Some("feature/memory-branch".to_string()),
        generated_at: Utc.timestamp_opt(201, 0).single().expect("timestamp"),
    }];

    sync_rollout_summaries_from_memories(
        &root,
        &memories,
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("sync rollout summaries");

    let mut dir = tokio::fs::read_dir(rollout_summaries_dir(&root))
        .await
        .expect("open rollout summaries dir");
    let mut files = Vec::new();
    while let Some(entry) = dir.next_entry().await.expect("read dir entry") {
        files.push(entry.file_name().to_string_lossy().to_string());
    }
    files.sort_unstable();

    assert_eq!(files.len(), 1);
    let file_name = &files[0];
    let stem = file_name
        .strip_suffix(".md")
        .expect("rollout summary file should end with .md");
    let (prefix, slug) = stem
        .rsplit_once('-')
        .expect("rollout summary filename should include slug");
    let (timestamp, short_hash) = prefix
        .rsplit_once('-')
        .expect("rollout summary filename should include short hash");

    assert_eq!(timestamp.len(), 19, "timestamp should be second precision");
    let parsed_timestamp = chrono::NaiveDateTime::parse_from_str(timestamp, "%Y-%m-%dT%H-%M-%S");
    assert!(
        parsed_timestamp.is_ok(),
        "timestamp should use YYYY-MM-DDThh-mm-ss"
    );
    assert_eq!(short_hash.len(), 4, "short hash should be exactly 4 chars");
    assert!(
        short_hash.chars().all(|ch| ch.is_ascii_alphanumeric()),
        "short hash should use only alphanumeric chars"
    );
    assert!(slug.len() <= 60, "slug should be capped at 60 chars");
    assert!(
        slug.chars()
            .all(|ch| ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '_'),
        "slug should be file-safe lowercase ascii with underscores"
    );

    let summary = tokio::fs::read_to_string(rollout_summaries_dir(&root).join(file_name))
        .await
        .expect("read rollout summary");
    assert!(summary.contains(&format!("thread_id: {thread_id}")));
    assert!(summary.contains("rollout_path: /tmp/rollout-200.jsonl"));
    assert!(summary.contains("git_branch: feature/memory-branch"));
    assert!(
        !tokio::fs::try_exists(&stale_unslugged_path)
            .await
            .expect("check stale unslugged path"),
        "slugged sync should prune stale unslugged filename for same thread"
    );
    assert!(
        !tokio::fs::try_exists(&stale_old_slug_path)
            .await
            .expect("check stale old slug path"),
        "slugged sync should prune stale slugged filename for same thread"
    );
}

#[tokio::test]
async fn rebuild_raw_memories_file_adds_canonical_rollout_summary_file_header() {
    let dir = tempdir().expect("tempdir");
    let root = dir.path().join("memory");
    ensure_layout(&root).await.expect("ensure layout");

    let thread_id =
        ThreadId::try_from("0194f5a6-89ab-7cde-8123-456789abcdef").expect("valid thread id");
    let memories = vec![Stage1Output {
        thread_id,
        source_updated_at: Utc.timestamp_opt(200, 0).single().expect("timestamp"),
        raw_memory: "\
---
description: Added a migration test
keywords: codex-state, migrations
---
### Task 1: migration-test
task: add-migration-test
task_group: codex-state
task_outcome: success
- Added regression coverage for migration uniqueness.

### Task 2: validate-migration
task: validate-migration-ordering
task_group: codex-state
task_outcome: success
- Confirmed no ordering regressions."
            .to_string(),
        rollout_summary: "short summary".to_string(),
        rollout_slug: Some("Unsafe Slug/With Spaces & Symbols + EXTRA_LONG_12345".to_string()),
        rollout_path: PathBuf::from("/tmp/rollout-200.jsonl"),
        cwd: PathBuf::from("/tmp/workspace"),
        git_branch: None,
        generated_at: Utc.timestamp_opt(201, 0).single().expect("timestamp"),
    }];

    sync_rollout_summaries_from_memories(
        &root,
        &memories,
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("sync rollout summaries");
    rebuild_raw_memories_file_from_memories(
        &root,
        &memories,
        DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
    )
    .await
    .expect("rebuild raw memories");

    let mut dir = tokio::fs::read_dir(rollout_summaries_dir(&root))
        .await
        .expect("open rollout summaries dir");
    let mut files = Vec::new();
    while let Some(entry) = dir.next_entry().await.expect("read dir entry") {
        files.push(entry.file_name().to_string_lossy().to_string());
    }
    files.sort_unstable();
    assert_eq!(files.len(), 1);
    let canonical_rollout_summary_file = &files[0];

    let raw_memories = tokio::fs::read_to_string(raw_memories_file(&root))
        .await
        .expect("read raw memories");
    let summary = tokio::fs::read_to_string(
        rollout_summaries_dir(&root).join(canonical_rollout_summary_file),
    )
    .await
    .expect("read rollout summary");
    assert!(summary.contains("rollout_path: /tmp/rollout-200.jsonl"));
    assert!(raw_memories.contains(&format!(
        "rollout_summary_file: {canonical_rollout_summary_file}"
    )));
    assert!(raw_memories.contains("description: Added a migration test"));
    assert!(raw_memories.contains("### Task 1: migration-test"));
    assert!(raw_memories.contains("task: add-migration-test"));
    assert!(raw_memories.contains("task_group: codex-state"));
    assert!(raw_memories.contains("task_outcome: success"));
}

mod phase2 {
    use crate::ThreadManager;
    use crate::agent::AgentControl;
    use crate::config::Config;
    use crate::config::test_config;
    use crate::memories::phase2;
    use crate::session::session::Session;
    use crate::session::tests::make_session_and_context;
    use chrono::Duration as ChronoDuration;
    use chrono::Utc;
    use codex_config::Constrained;
    use codex_config::types::McpServerConfig;
    use codex_features::Feature;
    use codex_login::CodexAuth;
    use codex_memories_write::memory_root;
    use codex_memories_write::raw_memories_file;
    use codex_memories_write::rebuild_raw_memories_file_from_memories;
    use codex_memories_write::rollout_summaries_dir;
    use codex_memories_write::sync_rollout_summaries_from_memories;
    use codex_memories_write::workspace::prepare_memory_workspace;
    use codex_protocol::AgentPath;
    use codex_protocol::ThreadId;
    use codex_protocol::models::PermissionProfile;
    use codex_protocol::permissions::FileSystemSandboxPolicy;
    use codex_protocol::permissions::NetworkSandboxPolicy;
    use codex_protocol::protocol::AskForApproval;
    use codex_protocol::protocol::Op;
    use codex_protocol::protocol::SandboxPolicy;
    use codex_protocol::protocol::SessionSource;
    use codex_state::Phase2JobClaimOutcome;
    use codex_state::Stage1Output;
    use codex_state::ThreadMetadataBuilder;
    use std::collections::HashMap;
    use std::path::PathBuf;
    use std::sync::Arc;
    use tempfile::TempDir;

    fn stage1_output_with_source_updated_at(source_updated_at: i64) -> Stage1Output {
        Stage1Output {
            thread_id: ThreadId::new(),
            source_updated_at: chrono::DateTime::<Utc>::from_timestamp(source_updated_at, 0)
                .expect("valid source_updated_at timestamp"),
            raw_memory: "raw memory".to_string(),
            rollout_summary: "rollout summary".to_string(),
            rollout_slug: None,
            rollout_path: PathBuf::from("/tmp/rollout-summary.jsonl"),
            cwd: PathBuf::from("/tmp/workspace"),
            git_branch: None,
            generated_at: chrono::DateTime::<Utc>::from_timestamp(source_updated_at + 1, 0)
                .expect("valid generated_at timestamp"),
        }
    }

    struct DispatchHarness {
        _codex_home: TempDir,
        config: Arc<Config>,
        session: Arc<Session>,
        manager: ThreadManager,
        state_db: Arc<codex_state::StateRuntime>,
    }

    impl DispatchHarness {
        async fn new() -> Self {
            Self::new_with_config(|_| {}).await
        }

        async fn new_with_config(configure: impl FnOnce(&mut Config)) -> Self {
            let codex_home = tempfile::tempdir().expect("create temp codex home");
            let mut config = test_config().await;
            config.codex_home =
                codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path(codex_home.path())
                    .expect("codex home is absolute");
            config.cwd = config.codex_home.clone();
            let permission_profile = PermissionProfile::from_runtime_permissions(
                &FileSystemSandboxPolicy::unrestricted(),
                NetworkSandboxPolicy::Enabled,
            );
            config
                .permissions
                .set_permission_profile(permission_profile)
                .expect("permissions are configurable");
            configure(&mut config);
            let config = Arc::new(config);

            let state_db = codex_state::StateRuntime::init(
                config.codex_home.to_path_buf(),
                config.model_provider_id.clone(),
            )
            .await
            .expect("initialize state db");

            let manager = ThreadManager::with_models_provider_and_home_for_tests(
                CodexAuth::from_api_key("dummy"),
                config.model_provider.clone(),
                config.codex_home.to_path_buf(),
                std::sync::Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
            );
            let (mut session, _turn_context) = make_session_and_context().await;
            session.services.state_db = Some(Arc::clone(&state_db));
            session.services.agent_control = manager.agent_control();

            Self {
                _codex_home: codex_home,
                config,
                session: Arc::new(session),
                manager,
                state_db,
            }
        }

        async fn seed_stage1_output(&self, source_updated_at: i64) -> ThreadId {
            let thread_id = ThreadId::new();
            let mut metadata_builder = ThreadMetadataBuilder::new(
                thread_id,
                self.config
                    .codex_home
                    .join(format!("rollout-{thread_id}.jsonl"))
                    .to_path_buf(),
                Utc::now(),
                SessionSource::Cli,
            );
            metadata_builder.cwd = self.config.cwd.to_path_buf();
            metadata_builder.model_provider = Some(self.config.model_provider_id.clone());
            let metadata = metadata_builder.build(&self.config.model_provider_id);

            self.state_db
                .upsert_thread(&metadata)
                .await
                .expect("upsert thread metadata");

            let claim = self
                .state_db
                .try_claim_stage1_job(
                    thread_id,
                    self.session.conversation_id,
                    source_updated_at,
                    /*lease_seconds*/ 3_600,
                    /*max_running_jobs*/ 64,
                )
                .await
                .expect("claim stage-1 job");
            let ownership_token = match claim {
                codex_state::Stage1JobClaimOutcome::Claimed { ownership_token } => ownership_token,
                other => panic!("unexpected stage-1 claim outcome: {other:?}"),
            };
            assert!(
                self.state_db
                    .mark_stage1_job_succeeded(
                        thread_id,
                        &ownership_token,
                        source_updated_at,
                        "raw memory",
                        "rollout summary",
                        /*rollout_slug*/ None,
                    )
                    .await
                    .expect("mark stage-1 success"),
                "stage-1 success should enqueue global consolidation"
            );
            thread_id
        }

        async fn shutdown_threads(&self) {
            let report = self
                .manager
                .shutdown_all_threads_bounded(std::time::Duration::from_secs(10))
                .await;
            assert!(report.submit_failed.is_empty());
            assert!(report.timed_out.is_empty());
        }

        fn user_input_ops_count(&self) -> usize {
            self.manager
                .captured_ops()
                .into_iter()
                .filter(|(_, op)| matches!(op, Op::UserInput { .. }))
                .count()
        }
    }

    #[test]
    fn completion_watermark_never_regresses_below_claimed_input_watermark() {
        let stage1_output = stage1_output_with_source_updated_at(/*source_updated_at*/ 123);

        let completion = phase2::get_watermark(/*claimed_watermark*/ 1_000, &[stage1_output]);
        pretty_assertions::assert_eq!(completion, 1_000);
    }

    #[test]
    fn completion_watermark_uses_claimed_watermark_when_there_are_no_memories() {
        let completion = phase2::get_watermark(/*claimed_watermark*/ 777, &[]);
        pretty_assertions::assert_eq!(completion, 777);
    }

    #[test]
    fn completion_watermark_uses_latest_memory_timestamp_when_it_is_newer() {
        let older = stage1_output_with_source_updated_at(/*source_updated_at*/ 123);
        let newer = stage1_output_with_source_updated_at(/*source_updated_at*/ 456);

        let completion = phase2::get_watermark(/*claimed_watermark*/ 200, &[older, newer]);
        pretty_assertions::assert_eq!(completion, 456);
    }

    #[tokio::test]
    async fn dispatch_skips_when_memory_workspace_is_not_dirty() {
        let harness = DispatchHarness::new().await;
        let root = memory_root(&harness.config.codex_home);
        rebuild_raw_memories_file_from_memories(
            &root,
            &[],
            /*max_raw_memories_for_consolidation*/ 0,
        )
        .await
        .expect("write empty raw memories baseline");
        let outputs = super::write_consolidated_outputs(&root).await;
        prepare_memory_workspace(&root)
            .await
            .expect("commit empty memory workspace as baseline");

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 0);
        super::assert_consolidated_outputs_exist(&outputs, "clean no-input phase2").await;
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 0);
    }

    #[tokio::test]
    async fn dispatch_uses_git_dirty_state_without_db_dirty_watermark() {
        let harness = DispatchHarness::new().await;
        let root = memory_root(&harness.config.codex_home);
        rebuild_raw_memories_file_from_memories(
            &root,
            &[],
            /*max_raw_memories_for_consolidation*/ 0,
        )
        .await
        .expect("write empty raw memories baseline");
        prepare_memory_workspace(&root)
            .await
            .expect("commit empty memory workspace as baseline");
        let extension_resource = root
            .join("extensions")
            .join("chronicle")
            .join("resources")
            .join("2026-04-22T12-00-00-abcd-10min-memory.md");
        tokio::fs::create_dir_all(
            extension_resource
                .parent()
                .expect("extension resource parent"),
        )
        .await
        .expect("create extension resource dir");
        tokio::fs::write(
            root.join("extensions/chronicle/instructions.md"),
            "instructions\n",
        )
        .await
        .expect("write extension instructions");
        tokio::fs::write(&extension_resource, "extension memory\n")
            .await
            .expect("write extension resource");

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 1);
        let workspace_diff = tokio::fs::read_to_string(root.join("phase2_workspace_diff.md"))
            .await
            .expect("read workspace diff");
        assert!(
            workspace_diff.contains("- A extensions/chronicle/instructions.md"),
            "git-only extension instructions should dirty phase2: {workspace_diff}"
        );
        assert!(
            workspace_diff.contains("- A extensions/chronicle/resources/"),
            "git-only extension resource should dirty phase2: {workspace_diff}"
        );
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 1);

        harness.shutdown_threads().await;
    }

    #[tokio::test]
    async fn dispatch_skips_when_global_job_is_already_running() {
        let harness = DispatchHarness::new().await;
        harness
            .state_db
            .enqueue_global_consolidation(/*input_watermark*/ 123)
            .await
            .expect("enqueue global consolidation");
        let claimed = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim running global lock");
        assert!(
            matches!(claimed, Phase2JobClaimOutcome::Claimed { .. }),
            "precondition should claim the running lock"
        );

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        let running_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim while lock is still running");
        pretty_assertions::assert_eq!(running_claim, Phase2JobClaimOutcome::SkippedRunning);
        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 0);
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 0);
    }

    #[tokio::test]
    async fn dispatch_reclaims_stale_global_lock_and_starts_consolidation() {
        let harness = DispatchHarness::new_with_config(|config| {
            let server: McpServerConfig =
                toml::from_str("command = \"docs-server\"").expect("deserialize MCP server");
            config
                .mcp_servers
                .set(HashMap::from([("docs".to_string(), server)]))
                .expect("parent MCP servers are configurable");
            config
                .features
                .enable(Feature::Apps)
                .expect("apps feature is configurable");
            config
                .features
                .enable(Feature::Plugins)
                .expect("plugins feature is configurable");
            config.include_apps_instructions = true;
        })
        .await;
        harness.seed_stage1_output(Utc::now().timestamp()).await;

        let stale_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 0)
            .await
            .expect("claim stale global lock");
        assert!(
            matches!(stale_claim, Phase2JobClaimOutcome::Claimed { .. }),
            "stale lock precondition should be claimed"
        );

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        let post_dispatch_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim after stale lock dispatch");
        assert!(
            matches!(
                post_dispatch_claim,
                Phase2JobClaimOutcome::SkippedRunning
                    | Phase2JobClaimOutcome::SkippedRetryUnavailable
            ),
            "stale-lock dispatch should either keep the reclaimed job running or finish it before re-claim"
        );

        let user_input_ops = harness.user_input_ops_count();
        pretty_assertions::assert_eq!(user_input_ops, 1);
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 1);
        let thread_id = thread_ids[0];
        let subagent = harness
            .manager
            .get_thread(thread_id)
            .await
            .expect("get consolidation thread");
        let config_snapshot = subagent.config_snapshot().await;
        pretty_assertions::assert_eq!(config_snapshot.approval_policy, AskForApproval::Never);
        assert!(config_snapshot.ephemeral);
        pretty_assertions::assert_eq!(
            config_snapshot.cwd.as_path(),
            memory_root(&harness.config.codex_home).as_path()
        );
        match &config_snapshot.sandbox_policy {
            SandboxPolicy::WorkspaceWrite { network_access, .. } => {
                assert!(!*network_access);
                let effective_writable_roots: Vec<_> = config_snapshot
                    .sandbox_policy
                    .get_writable_roots_with_cwd(config_snapshot.cwd.as_path())
                    .into_iter()
                    .map(|root| root.root)
                    .collect();
                pretty_assertions::assert_eq!(
                    effective_writable_roots.as_slice(),
                    [memory_root(&harness.config.codex_home)],
                    "consolidation subagent should only be able to write the memory root"
                );
            }
            other => panic!("unexpected sandbox policy: {other:?}"),
        }
        pretty_assertions::assert_eq!(
            config_snapshot.session_source.get_agent_path(),
            Some(AgentPath::morpheus())
        );
        assert!(
            harness
                .session
                .services
                .agent_control
                .get_agent_metadata(thread_id)
                .is_none(),
            "memory consolidation should not be registered in the root collab agent registry"
        );
        let turn_context = subagent.codex.session.new_default_turn().await;
        let file_system_sandbox_policy = turn_context.file_system_sandbox_policy();
        let legacy_file_system_sandbox_policy =
            FileSystemSandboxPolicy::from_legacy_sandbox_policy_for_cwd(
                &config_snapshot.sandbox_policy,
                config_snapshot.cwd.as_path(),
            );
        assert!(
            file_system_sandbox_policy.is_semantically_equivalent_to(
                &legacy_file_system_sandbox_policy,
                config_snapshot.cwd.as_path(),
            ),
            "consolidation subagent split filesystem policy should match the memory-root legacy policy"
        );
        assert!(
            file_system_sandbox_policy.can_write_path_with_cwd(
                memory_root(&harness.config.codex_home).as_path(),
                config_snapshot.cwd.as_path(),
            ),
            "consolidation subagent should be able to write the memory root"
        );
        assert!(
            !file_system_sandbox_policy.can_write_path_with_cwd(
                harness.config.codex_home.join("config.toml").as_path(),
                config_snapshot.cwd.as_path(),
            ),
            "consolidation subagent should not inherit codex_home write access"
        );
        pretty_assertions::assert_eq!(
            turn_context.network_sandbox_policy(),
            NetworkSandboxPolicy::Restricted,
            "consolidation subagent split network policy should preserve no-network sandboxing"
        );
        assert!(
            !turn_context.features.enabled(Feature::MemoryTool),
            "consolidation subagent should have the memories feature disabled"
        );
        assert!(
            turn_context.config.mcp_servers.get().is_empty(),
            "consolidation subagent should not inherit configured MCP servers"
        );
        assert!(
            !subagent
                .codex
                .session
                .services
                .mcp_connection_manager
                .read()
                .await
                .has_servers(),
            "consolidation subagent should not initialize MCP servers"
        );
        assert!(
            !turn_context.features.enabled(Feature::Apps),
            "consolidation subagent should not expose app-backed MCP"
        );
        assert!(
            !turn_context.features.enabled(Feature::Plugins),
            "consolidation subagent should not expose plugin-backed MCP"
        );
        assert!(
            !turn_context.config.include_apps_instructions,
            "consolidation subagent should not include apps instructions"
        );
        assert!(
            !turn_context.config.memories.generate_memories,
            "consolidation subagent should not generate memories"
        );
        assert!(
            !turn_context.config.memories.use_memories,
            "consolidation subagent should not read memories"
        );
        assert!(
            subagent.rollout_path().is_none(),
            "ephemeral consolidation thread should not materialize a rollout"
        );
        let memory_mode = harness
            .state_db
            .get_thread_memory_mode(thread_id)
            .await
            .expect("read consolidation thread memory mode");
        pretty_assertions::assert_eq!(memory_mode, None);

        harness.shutdown_threads().await;
    }

    #[tokio::test]
    async fn dispatch_with_empty_stage1_outputs_spawns_for_workspace_changes() {
        let harness = DispatchHarness::new().await;
        let root = memory_root(&harness.config.codex_home);
        let summaries_dir = rollout_summaries_dir(&root);
        tokio::fs::create_dir_all(&summaries_dir)
            .await
            .expect("create rollout summaries dir");

        let stale_summary_path = summaries_dir.join(format!("{}.md", ThreadId::new()));
        tokio::fs::write(&stale_summary_path, "stale summary\n")
            .await
            .expect("write stale rollout summary");
        let raw_memories_path = raw_memories_file(&root);
        tokio::fs::write(&raw_memories_path, "stale raw memories\n")
            .await
            .expect("write stale raw memories");
        let outputs = super::write_consolidated_outputs(&root).await;

        harness
            .state_db
            .enqueue_global_consolidation(/*input_watermark*/ 999)
            .await
            .expect("enqueue global consolidation");

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        assert!(
            !tokio::fs::try_exists(&stale_summary_path)
                .await
                .expect("check stale summary existence"),
            "empty consolidation should prune stale rollout summary files"
        );
        let raw_memories = tokio::fs::read_to_string(&raw_memories_path)
            .await
            .expect("read rebuilt raw memories");
        pretty_assertions::assert_eq!(raw_memories, "# Raw Memories\n\nNo raw memories yet.\n");
        super::assert_consolidated_outputs_exist(&outputs, "empty consolidation").await;
        let next_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim global job after empty consolidation dispatch");
        pretty_assertions::assert_eq!(next_claim, Phase2JobClaimOutcome::SkippedRunning);
        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 1);
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 1);

        harness.shutdown_threads().await;
    }

    #[tokio::test]
    async fn dispatch_with_empty_selected_inputs_preserves_consolidated_outputs() {
        let harness = DispatchHarness::new().await;
        let source_updated_at = Utc::now().timestamp();
        let thread_id = harness.seed_stage1_output(source_updated_at).await;
        let root = memory_root(&harness.config.codex_home);
        let selected = harness
            .state_db
            .get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 30)
            .await
            .expect("load phase2 input selection");
        sync_rollout_summaries_from_memories(&root, &selected, selected.len())
            .await
            .expect("sync selected rollout summaries");
        rebuild_raw_memories_file_from_memories(&root, &selected, selected.len())
            .await
            .expect("sync selected raw memories");
        let outputs = super::write_consolidated_outputs(&root).await;
        prepare_memory_workspace(&root)
            .await
            .expect("commit current memory workspace as baseline");

        let claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim global phase2 job");
        let Phase2JobClaimOutcome::Claimed {
            ownership_token, ..
        } = claim
        else {
            panic!("unexpected phase2 claim outcome: {claim:?}");
        };
        assert!(
            harness
                .state_db
                .mark_global_phase2_job_succeeded(&ownership_token, source_updated_at, &selected)
                .await
                .expect("mark phase2 succeeded"),
            "phase2 success should update selected baseline"
        );
        assert!(
            harness
                .state_db
                .mark_thread_memory_mode_polluted(thread_id)
                .await
                .expect("mark thread polluted"),
            "polluted selected thread should enqueue phase2 forgetting"
        );

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 1);
        super::assert_consolidated_outputs_exist(&outputs, "empty selected phase2").await;
        let workspace_diff = tokio::fs::read_to_string(root.join("phase2_workspace_diff.md"))
            .await
            .expect("read workspace diff");
        assert!(
            workspace_diff.contains("- D rollout_summaries/"),
            "empty selected phase2 should surface deleted rollout summaries: {workspace_diff}"
        );
        assert!(
            !workspace_diff.contains("- D MEMORY.md"),
            "empty selected phase2 should not delete MEMORY.md directly: {workspace_diff}"
        );
        assert!(
            !workspace_diff.contains("- D memory_summary.md"),
            "empty selected phase2 should not delete memory_summary.md directly: {workspace_diff}"
        );
        assert!(
            !workspace_diff.contains("- D skills/demo/SKILL.md"),
            "empty selected phase2 should not delete skills directly: {workspace_diff}"
        );

        harness.shutdown_threads().await;
    }

    #[tokio::test]
    async fn dispatch_with_clean_workspace_rebuilds_selected_phase2_baseline() {
        let harness = DispatchHarness::new().await;
        let source_updated_at = (Utc::now() - ChronoDuration::days(1)).timestamp();
        let thread_id = harness.seed_stage1_output(source_updated_at).await;
        let root = memory_root(&harness.config.codex_home);
        let selected = harness
            .state_db
            .get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 30)
            .await
            .expect("load phase2 input selection");

        sync_rollout_summaries_from_memories(&root, &selected, selected.len())
            .await
            .expect("sync selected rollout summaries");
        rebuild_raw_memories_file_from_memories(&root, &selected, selected.len())
            .await
            .expect("sync selected raw memories");
        prepare_memory_workspace(&root)
            .await
            .expect("commit current memory workspace as baseline");

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 0);
        let pruned = harness
            .state_db
            .prune_stage1_outputs_for_retention(/*max_unused_days*/ 0, /*limit*/ 10)
            .await
            .expect("prune stage1 outputs after clean phase2");
        pretty_assertions::assert_eq!(pruned, 0);

        let selected = harness
            .state_db
            .get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 30)
            .await
            .expect("load phase2 input selection after clean workspace success");
        pretty_assertions::assert_eq!(selected.len(), 1);
        pretty_assertions::assert_eq!(selected[0].thread_id, thread_id);
    }

    #[tokio::test]
    async fn dispatch_marks_job_for_retry_when_sandbox_policy_cannot_be_overridden() {
        let harness = DispatchHarness::new().await;
        harness
            .state_db
            .enqueue_global_consolidation(/*input_watermark*/ 99)
            .await
            .expect("enqueue global consolidation");
        let mut constrained_config = harness.config.as_ref().clone();
        constrained_config.permissions.permission_profile = Constrained::allow_only(
            PermissionProfile::from_legacy_sandbox_policy(&SandboxPolicy::DangerFullAccess),
        );

        phase2::run(&harness.session, Arc::new(constrained_config)).await;

        let retry_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim global job after sandbox policy failure");
        pretty_assertions::assert_eq!(retry_claim, Phase2JobClaimOutcome::SkippedRetryUnavailable);
        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 0);
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 0);
    }

    #[tokio::test]
    async fn dispatch_marks_job_for_retry_when_syncing_artifacts_fails() {
        let harness = DispatchHarness::new().await;
        harness.seed_stage1_output(/*source_updated_at*/ 100).await;
        let root = memory_root(&harness.config.codex_home);
        tokio::fs::write(&root, "not a directory")
            .await
            .expect("create file at memory root");

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        let retry_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim global job after sync failure");
        pretty_assertions::assert_eq!(retry_claim, Phase2JobClaimOutcome::SkippedRetryUnavailable);
        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 0);
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 0);
    }

    #[tokio::test]
    async fn dispatch_marks_job_for_retry_when_rebuilding_raw_memories_fails() {
        let harness = DispatchHarness::new().await;
        harness.seed_stage1_output(/*source_updated_at*/ 100).await;
        let root = memory_root(&harness.config.codex_home);
        tokio::fs::create_dir_all(raw_memories_file(&root))
            .await
            .expect("create raw_memories.md as a directory");

        phase2::run(&harness.session, Arc::clone(&harness.config)).await;

        let retry_claim = harness
            .state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim global job after rebuild failure");
        pretty_assertions::assert_eq!(retry_claim, Phase2JobClaimOutcome::SkippedRetryUnavailable);
        pretty_assertions::assert_eq!(harness.user_input_ops_count(), 0);
        let thread_ids = harness.manager.list_thread_ids().await;
        pretty_assertions::assert_eq!(thread_ids.len(), 0);
    }

    #[tokio::test]
    async fn dispatch_marks_job_for_retry_when_spawn_agent_fails() {
        let codex_home = tempfile::tempdir().expect("create temp codex home");
        let mut config = test_config().await;
        config.codex_home =
            codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path(codex_home.path())
                .expect("codex home is absolute");
        config.cwd = config.codex_home.clone();
        let config = Arc::new(config);

        let state_db = codex_state::StateRuntime::init(
            config.codex_home.to_path_buf(),
            config.model_provider_id.clone(),
        )
        .await
        .expect("initialize state db");

        let (mut session, _turn_context) = make_session_and_context().await;
        session.services.state_db = Some(Arc::clone(&state_db));
        session.services.agent_control = AgentControl::default();
        let session = Arc::new(session);

        let thread_id = ThreadId::new();
        let mut metadata_builder = ThreadMetadataBuilder::new(
            thread_id,
            config
                .codex_home
                .join(format!("rollout-{thread_id}.jsonl"))
                .to_path_buf(),
            Utc::now(),
            SessionSource::Cli,
        );
        metadata_builder.cwd = config.cwd.to_path_buf();
        metadata_builder.model_provider = Some(config.model_provider_id.clone());
        let metadata = metadata_builder.build(&config.model_provider_id);
        state_db
            .upsert_thread(&metadata)
            .await
            .expect("upsert thread metadata");

        let claim = state_db
            .try_claim_stage1_job(
                thread_id,
                session.conversation_id,
                /*source_updated_at*/ 100,
                /*lease_seconds*/ 3_600,
                /*max_running_jobs*/ 64,
            )
            .await
            .expect("claim stage-1 job");
        let ownership_token = match claim {
            codex_state::Stage1JobClaimOutcome::Claimed { ownership_token } => ownership_token,
            other => panic!("unexpected stage-1 claim outcome: {other:?}"),
        };
        assert!(
            state_db
                .mark_stage1_job_succeeded(
                    thread_id,
                    &ownership_token,
                    /*source_updated_at*/ 100,
                    "raw memory",
                    "rollout summary",
                    /*rollout_slug*/ None,
                )
                .await
                .expect("mark stage-1 success"),
            "stage-1 success should enqueue global consolidation"
        );

        let chronicle_resources = config
            .codex_home
            .join("memories/extensions/chronicle/resources");
        tokio::fs::create_dir_all(&chronicle_resources)
            .await
            .expect("create chronicle resources");
        tokio::fs::write(
            config
                .codex_home
                .join("memories/extensions/chronicle/instructions.md"),
            "instructions",
        )
        .await
        .expect("write chronicle instructions");
        let old_file = chronicle_resources.join(format!(
            "{}-abcd-10min-old.md",
            (Utc::now() - ChronoDuration::days(8)).format("%Y-%m-%dT%H-%M-%S")
        ));
        tokio::fs::write(&old_file, "old resource")
            .await
            .expect("write old extension resource");

        phase2::run(&session, Arc::clone(&config)).await;

        let retry_claim = state_db
            .try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
            .await
            .expect("claim global job after spawn failure");
        pretty_assertions::assert_eq!(
            retry_claim,
            Phase2JobClaimOutcome::SkippedRetryUnavailable,
            "spawn failures should leave the job in retry backoff instead of running"
        );
        assert!(
            !tokio::fs::try_exists(&old_file)
                .await
                .expect("check old extension resource"),
            "old extension resources should still be pruned on failed phase2 attempts"
        );
        let workspace_diff =
            tokio::fs::read_to_string(config.codex_home.join("memories/phase2_workspace_diff.md"))
                .await
                .expect("read workspace diff");
        assert!(
            workspace_diff.contains("- D extensions/chronicle/resources/"),
            "spawn failures should keep a retryable workspace diff: {workspace_diff}"
        );
    }
}