//! Integration tests for dirigent_archivist
//!
//! These tests verify the end-to-end functionality of the archivist,
//! including storage, retrieval, and event streaming.

#[cfg(test)]
mod tests {
    use chrono::Utc;
    use dirigent_archivist::{
        Archivist, MessageRecord, RegisterConnectorRequest,
        RegisterSessionRequest, RegisterStatus, Result, SessionKind, SessionListQuery,
        SessionMetadata,
    };
    use dirigent_archivist::storage::ndjson::append_ndjson;
    use dirigent_archivist::storage::json::write_json;
    use uuid::Uuid;

    #[tokio::test]
    async fn test_archivist_creation() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Verify coordinator construction succeeded (smoke test).
        let _ = &archivist;

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_register_connector_acceptance() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let response = archivist.register_connector(req, None).await?;
        assert_eq!(response.status, RegisterStatus::Accepted);
        assert!(response.alias_of.is_none());

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_register_connector_aliasing() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let req1 = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let response1 = archivist.register_connector(req1, None).await?;
        assert_eq!(response1.status, RegisterStatus::Accepted);

        // Register again with same client_native_id
        let req2 = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector 2".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let response2 = archivist.register_connector(req2, None).await?;
        assert_eq!(response2.status, RegisterStatus::Aliased);
        assert_eq!(response2.connector_uid, response1.connector_uid);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_register_session_acceptance() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector first
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        // Register session
        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;
        assert_eq!(session_response.status, RegisterStatus::Accepted);
        assert!(session_response.alias_of.is_none());

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_register_session_aliasing() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        // Register session
        let session_req1 = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response1 = archivist.register_session(session_req1, None).await?;

        // Register again with same native_session_id
        let session_req2 = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session 2".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response2 = archivist.register_session(session_req2, None).await?;
        assert_eq!(session_response2.status, RegisterStatus::Aliased);
        assert_eq!(session_response2.scroll_id, session_response1.scroll_id);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_append_and_get_messages() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Create and append messages
        let message1 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: None,
            ts: Utc::now(),
            role: "user".to_string(),
            author: Some("test".to_string()),
            content_md: "Hello, world!".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        let message2 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(message1.message_id),
            ts: Utc::now(),
            role: "assistant".to_string(),
            author: Some("assistant".to_string()),
            content_md: "Hi there!".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![message1.clone(), message2.clone()],
                None,
            )
            .await?;

        // Retrieve messages
        let messages = archivist.get_messages(session_response.scroll_id, None).await?;
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].message_id, message1.message_id);
        assert_eq!(messages[1].message_id, message2.message_id);
        assert_eq!(messages[0].content_md, "Hello, world!");
        assert_eq!(messages[1].content_md, "Hi there!");

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_list_sessions() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        // Register multiple sessions
        let session_req1 = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-1".to_string(),
            title: Some("Session 1".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response1 = archivist.register_session(session_req1, None).await?;

        // Wait a moment to ensure different timestamps
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        let session_req2 = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-2".to_string(),
            title: Some("Session 2".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response2 = archivist.register_session(session_req2, None).await?;

        // List sessions
        let sessions = archivist
            .list_sessions_paged(
                SessionListQuery::default()
                    .with_connector(connector_response.connector_uid)
                    .with_limit(100),
            )
            .await?
            .items;
        assert_eq!(sessions.len(), 2);

        // Verify sessions are sorted by updated_at descending (newest first)
        // Session 2 should be first because it was created later
        assert_eq!(sessions[0].scroll_id, session_response2.scroll_id);
        assert_eq!(sessions[1].scroll_id, session_response1.scroll_id);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_get_session_metadata() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Get session metadata
        let metadata = archivist
            .get_session_metadata(session_response.scroll_id, None)
            .await?;

        assert_eq!(metadata.scroll_id, session_response.scroll_id);
        assert_eq!(metadata.title, Some("Test Session".to_string()));
        assert_eq!(metadata.connector_uid, connector_response.connector_uid);
        assert_eq!(metadata.native_session_id, Some("native-123".to_string()));

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_resolve_session() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Resolve native session ID to scroll ID
        let scroll_id = archivist
            .resolve_session(connector_response.connector_uid, "native-123", None)
            .await?;

        assert_eq!(scroll_id, session_response.scroll_id);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_register_connector_custom_uid_collision() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let custom_uid = Uuid::now_v7();

        // Register connector with custom UID
        let req1 = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector 1".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: Some(custom_uid),
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let response1 = archivist.register_connector(req1, None).await?;
        assert_eq!(response1.status, RegisterStatus::Accepted);
        assert_eq!(response1.connector_uid, custom_uid);

        // Try to register another connector with same custom UID
        let req2 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Test Connector 2".to_string(),
            client_native_id: "acp@localhost:3000".to_string(),
            custom_uid: Some(custom_uid),
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let result2 = archivist.register_connector(req2, None).await;
        assert!(result2.is_err(), "Expected error for custom_uid collision");

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_register_session_with_unknown_connector() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Try to register session with unknown connector
        let unknown_connector = Uuid::now_v7();
        let session_req = RegisterSessionRequest {
            connector_uid: unknown_connector,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let result = archivist.register_session(session_req, None).await;
        assert!(result.is_err());

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_get_messages_empty_session() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Get messages from empty session
        let messages = archivist.get_messages(session_response.scroll_id, None).await?;
        assert_eq!(messages.len(), 0);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_get_messages_unknown_scroll_id() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Try to get messages from unknown session
        let unknown_scroll_id = Uuid::now_v7();
        let messages = archivist.get_messages(unknown_scroll_id, None).await?;

        // Should return empty vector for unknown session
        assert_eq!(messages.len(), 0);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_multiple_message_appends() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Append messages in multiple batches
        let message1 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: None,
            ts: Utc::now(),
            role: "user".to_string(),
            author: Some("test".to_string()),
            content_md: "First message".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        archivist
            .append_messages(session_response.scroll_id, vec![message1.clone()], None)
            .await?;

        let message2 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(message1.message_id),
            ts: Utc::now(),
            role: "assistant".to_string(),
            author: Some("assistant".to_string()),
            content_md: "Second message".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        archivist
            .append_messages(session_response.scroll_id, vec![message2.clone()], None)
            .await?;

        let message3 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(message2.message_id),
            ts: Utc::now(),
            role: "user".to_string(),
            author: Some("test".to_string()),
            content_md: "Third message".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        archivist
            .append_messages(session_response.scroll_id, vec![message3.clone()], None)
            .await?;

        // Retrieve all messages
        let messages = archivist.get_messages(session_response.scroll_id, None).await?;
        assert_eq!(messages.len(), 3);
        assert_eq!(messages[0].content_md, "First message");
        assert_eq!(messages[1].content_md, "Second message");
        assert_eq!(messages[2].content_md, "Third message");

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_messages_sorted_chronologically() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Create messages with specific timestamps in chronological order
        use chrono::TimeZone;
        let base_time = Utc.with_ymd_and_hms(2025, 11, 18, 18, 23, 36).unwrap();

        let msg_snake_user = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: None,
            ts: base_time + chrono::Duration::milliseconds(947),
            role: "user".to_string(),
            author: Some("user".to_string()),
            content_md: "hello please tell me a joke about snakes".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        let msg_snake_assistant = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(msg_snake_user.message_id),
            ts: base_time + chrono::Duration::milliseconds(969),
            role: "assistant".to_string(),
            author: Some("claude".to_string()),
            content_md: "Why don't snakes need cutlery? They have forked tongues!".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        let msg_tiger_user = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(msg_snake_assistant.message_id),
            ts: base_time + chrono::Duration::milliseconds(13429),
            role: "user".to_string(),
            author: Some("user".to_string()),
            content_md: "now one about tigers".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        let msg_tiger_assistant = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(msg_tiger_user.message_id),
            ts: base_time + chrono::Duration::milliseconds(13448),
            role: "assistant".to_string(),
            author: Some("claude".to_string()),
            content_md: "What do tigers wear to bed? Striped pajamas!".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        let msg_hyena_user = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(msg_tiger_assistant.message_id),
            ts: base_time + chrono::Duration::milliseconds(32623),
            role: "user".to_string(),
            author: Some("user".to_string()),
            content_md: "and a third one about hyenas".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        // Append messages OUT OF ORDER to simulate real-world event arrival
        // (assistant replies often arrive after subsequent user messages)
        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg_snake_user.clone()],
                None,
            )
            .await?;

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg_tiger_user.clone()],
                None,
            )
            .await?;

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg_snake_assistant.clone()],
                None,
            )
            .await?;

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg_hyena_user.clone()],
                None,
            )
            .await?;

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg_tiger_assistant.clone()],
                None,
            )
            .await?;

        // Retrieve messages - should be sorted chronologically despite out-of-order appends
        let messages = archivist.get_messages(session_response.scroll_id, None).await?;

        assert_eq!(messages.len(), 5);

        // Verify chronological order by timestamp
        assert_eq!(messages[0].message_id, msg_snake_user.message_id);
        assert_eq!(messages[0].content_md, "hello please tell me a joke about snakes");

        assert_eq!(messages[1].message_id, msg_snake_assistant.message_id);
        assert_eq!(messages[1].content_md, "Why don't snakes need cutlery? They have forked tongues!");

        assert_eq!(messages[2].message_id, msg_tiger_user.message_id);
        assert_eq!(messages[2].content_md, "now one about tigers");

        assert_eq!(messages[3].message_id, msg_tiger_assistant.message_id);
        assert_eq!(messages[3].content_md, "What do tigers wear to bed? Striped pajamas!");

        assert_eq!(messages[4].message_id, msg_hyena_user.message_id);
        assert_eq!(messages[4].content_md, "and a third one about hyenas");

        // Verify timestamps are strictly increasing
        for i in 1..messages.len() {
            assert!(
                messages[i].ts >= messages[i - 1].ts,
                "Messages not in chronological order: message {} has ts {} which is before message {} with ts {}",
                i,
                messages[i].ts,
                i - 1,
                messages[i - 1].ts
            );
        }

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_messages_with_identical_timestamps() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Create multiple messages with the exact same timestamp
        // This tests the secondary sorting by message_id
        let same_timestamp = Utc::now();

        // Create messages with explicitly ordered UUIDs (v7 includes timestamp)
        // Sleep briefly between creations to ensure UUIDv7 ordering
        let msg1 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: None,
            ts: same_timestamp,
            role: "user".to_string(),
            author: Some("user".to_string()),
            content_md: "First message".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        tokio::time::sleep(tokio::time::Duration::from_micros(1)).await;

        let msg2 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(msg1.message_id),
            ts: same_timestamp,
            role: "assistant".to_string(),
            author: Some("assistant".to_string()),
            content_md: "Second message".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        tokio::time::sleep(tokio::time::Duration::from_micros(1)).await;

        let msg3 = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: Some(msg2.message_id),
            ts: same_timestamp,
            role: "user".to_string(),
            author: Some("user".to_string()),
            content_md: "Third message".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        // Append in reverse order to ensure sorting is working
        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg3.clone()],
                None,
            )
            .await?;

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg1.clone()],
                None,
            )
            .await?;

        archivist
            .append_messages(
                session_response.scroll_id,
                vec![msg2.clone()],
                None,
            )
            .await?;

        // Retrieve messages - should be sorted by message_id since timestamps are identical
        let messages = archivist.get_messages(session_response.scroll_id, None).await?;

        assert_eq!(messages.len(), 3);

        // All timestamps should be the same
        assert_eq!(messages[0].ts, same_timestamp);
        assert_eq!(messages[1].ts, same_timestamp);
        assert_eq!(messages[2].ts, same_timestamp);

        // Messages should be ordered by message_id (UUIDv7 preserves creation order)
        assert_eq!(messages[0].message_id, msg1.message_id);
        assert_eq!(messages[1].message_id, msg2.message_id);
        assert_eq!(messages[2].message_id, msg3.message_id);

        assert_eq!(messages[0].content_md, "First message");
        assert_eq!(messages[1].content_md, "Second message");
        assert_eq!(messages[2].content_md, "Third message");

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_append_messages_updates_timestamp() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "native-123".to_string(),
            title: Some("Test Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Get initial metadata
        let metadata_before = archivist
            .get_session_metadata(session_response.scroll_id, None)
            .await?;

        // Wait a moment
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

        // Append a message
        let message = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: session_response.scroll_id,
            parent_id: None,
            ts: Utc::now(),
            role: "user".to_string(),
            author: Some("test".to_string()),
            content_md: "Hello!".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };

        archivist
            .append_messages(session_response.scroll_id, vec![message], None)
            .await?;

        // Get updated metadata
        let metadata_after = archivist
            .get_session_metadata(session_response.scroll_id, None)
            .await?;

        // Verify updated_at changed
        assert!(metadata_after.updated_at > metadata_before.updated_at);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    // ========================================================================
    // Performance Benchmarks
    // These tests are marked with #[ignore] to avoid slowing down regular test runs
    // Run with: cargo test --package dirigent_archivist -- --ignored
    // ========================================================================

    #[tokio::test]
    #[ignore]
    async fn bench_append_1000_messages() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_bench_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Bench Connector".to_string(),
            client_native_id: "bench@localhost".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "bench-session".to_string(),
            title: Some("Bench Session".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Create 1000 messages
        let messages: Vec<MessageRecord> = (0..1000)
            .map(|i| MessageRecord {
                version: 1,
                message_id: Uuid::now_v7(),
                session: session_response.scroll_id,
                parent_id: None,
                ts: Utc::now(),
                role: if i % 2 == 0 { "user" } else { "assistant" }.to_string(),
                author: Some("bench".to_string()),
                content_md: format!("Message number {} with some realistic content that might appear in a conversation. This helps simulate real-world usage patterns.", i),
                content_parts: None,
                attachments: vec![],
                metadata: serde_json::json!({"index": i}),
            })
            .collect();

        // Benchmark appending messages
        let start = std::time::Instant::now();
        archivist
            .append_messages(session_response.scroll_id, messages, None)
            .await?;
        let elapsed = start.elapsed();

        let messages_per_sec = 1000.0 / elapsed.as_secs_f64();
        println!("\nBenchmark: Append 1000 messages");
        println!("  Total time: {:?}", elapsed);
        println!("  Messages/sec: {:.2}", messages_per_sec);
        println!("  Avg time per message: {:?}", elapsed / 1000);

        // Target: >100 msg/s
        assert!(
            messages_per_sec > 100.0,
            "Performance degraded: {:.2} msg/s < 100 msg/s",
            messages_per_sec
        );

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    #[ignore]
    async fn bench_read_100_messages() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_bench_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Bench Connector".to_string(),
            client_native_id: "bench@localhost".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "bench-session-100".to_string(),
            title: Some("Bench Session 100".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Create and append 100 messages
        let messages: Vec<MessageRecord> = (0..100)
            .map(|i| MessageRecord {
                version: 1,
                message_id: Uuid::now_v7(),
                session: session_response.scroll_id,
                parent_id: None,
                ts: Utc::now(),
                role: if i % 2 == 0 { "user" } else { "assistant" }.to_string(),
                author: Some("bench".to_string()),
                content_md: format!("Message {}", i),
                content_parts: None,
                attachments: vec![],
                metadata: serde_json::json!({}),
            })
            .collect();

        archivist
            .append_messages(session_response.scroll_id, messages, None)
            .await?;

        // Benchmark reading messages
        let start = std::time::Instant::now();
        let retrieved = archivist.get_messages(session_response.scroll_id, None).await?;
        let elapsed = start.elapsed();

        println!("\nBenchmark: Read 100 messages");
        println!("  Total time: {:?}", elapsed);
        println!("  Messages retrieved: {}", retrieved.len());

        // Target: sub-100ms for typical sessions
        assert!(
            elapsed.as_millis() < 100,
            "Read performance degraded: {:?} > 100ms",
            elapsed
        );

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    #[ignore]
    async fn bench_read_1000_messages() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_bench_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector and session
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Bench Connector".to_string(),
            client_native_id: "bench@localhost".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        let session_req = RegisterSessionRequest {
            connector_uid: connector_response.connector_uid,
            native_session_id: "bench-session-1000".to_string(),
            title: Some("Bench Session 1000".to_string()),
            custom_scroll_id: None,
            metadata: serde_json::json!({}),
            completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let session_response = archivist.register_session(session_req, None).await?;

        // Create and append 1000 messages
        let messages: Vec<MessageRecord> = (0..1000)
            .map(|i| MessageRecord {
                version: 1,
                message_id: Uuid::now_v7(),
                session: session_response.scroll_id,
                parent_id: None,
                ts: Utc::now(),
                role: if i % 2 == 0 { "user" } else { "assistant" }.to_string(),
                author: Some("bench".to_string()),
                content_md: format!("Message {}", i),
                content_parts: None,
                attachments: vec![],
                metadata: serde_json::json!({}),
            })
            .collect();

        archivist
            .append_messages(session_response.scroll_id, messages, None)
            .await?;

        // Benchmark reading messages
        let start = std::time::Instant::now();
        let retrieved = archivist.get_messages(session_response.scroll_id, None).await?;
        let elapsed = start.elapsed();

        println!("\nBenchmark: Read 1000 messages");
        println!("  Total time: {:?}", elapsed);
        println!("  Messages retrieved: {}", retrieved.len());

        // Log for tracking (no strict requirement for large sessions)
        println!("  Note: Performance acceptable for large session");

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    #[ignore]
    async fn bench_list_100_sessions() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_bench_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector
        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "Bench Connector".to_string(),
            client_native_id: "bench@localhost".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let connector_response = archivist.register_connector(connector_req, None).await?;

        // Register 100 sessions
        for i in 0..100 {
            let session_req = RegisterSessionRequest {
                connector_uid: connector_response.connector_uid,
                native_session_id: format!("bench-session-{}", i),
                title: Some(format!("Session {}", i)),
                custom_scroll_id: None,
                metadata: serde_json::json!({"index": i}),
                completeness: Default::default(),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
            };

            archivist.register_session(session_req, None).await?;
        }

        // Benchmark listing sessions
        let start = std::time::Instant::now();
        let sessions = archivist
            .list_sessions_paged(
                SessionListQuery::default()
                    .with_connector(connector_response.connector_uid)
                    .with_limit(dirigent_archivist::MAX_PAGE_LIMIT),
            )
            .await?
            .items;
        let elapsed = start.elapsed();

        println!("\nBenchmark: List 100 sessions");
        println!("  Total time: {:?}", elapsed);
        println!("  Sessions listed: {}", sessions.len());

        // Target: sub-100ms for typical connector
        assert!(
            elapsed.as_millis() < 100,
            "List performance degraded: {:?} > 100ms",
            elapsed
        );

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_mixed_format_compatibility() {
        // Create archivist with temp directory
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await.unwrap()
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await.unwrap();

        // Register connector
        let connector_req = RegisterConnectorRequest {
            r#type: "Test".to_string(),
            title: "Test Connector".to_string(),
            client_native_id: "test-connector".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };
        let connector_resp = archivist.register_connector(connector_req, None).await.unwrap();
        let connector_uid = connector_resp.connector_uid;

        // Manually create a session with .jsonl format for messages
        let scroll_id = Uuid::now_v7();
        let session_metadata = SessionMetadata {
            version: 1,
            scroll_id,
            created_at: Utc::now(),
            updated_at: Utc::now(),
            title: Some("Test Session".to_string()),
            connector_uid,
            native_session_id: Some("test-123".to_string()),
            agent_id: None,
            parent_scroll_id: None,
            continuation: None,
            tags: vec![],
            metadata: serde_json::json!({}),
            no_update: false,
            kind: SessionKind::Chat,
            acp_client_id: None,
            is_connected: None,
            current_session_id: None,
            models: None,
            modes: None,
            config_options: None,
            completeness: Default::default(),
            matrix_room_id: None,
            matrix_sharing_active: false,
            matrix_shared_at: None,
            is_subagent: false,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        backend.paths().ensure_dirs(scroll_id).await.unwrap();
        write_json(&backend.paths().session_json(scroll_id), &session_metadata).await.unwrap();

        // Create messages.jsonl (not .ndjson)
        let jsonl_path = backend.paths().session_dir(scroll_id).join("messages.jsonl");
        let message = MessageRecord {
            version: 1,
            message_id: Uuid::now_v7(),
            session: scroll_id,
            parent_id: None,
            ts: Utc::now(),
            role: "user".to_string(),
            author: None,
            content_md: "Hello from .jsonl file".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };
        append_ndjson(&jsonl_path, &message).await.unwrap();

        // Read messages using archivist API
        let messages = archivist.get_messages(scroll_id, None).await.unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].content_md, "Hello from .jsonl file");

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
    }

    #[tokio::test]
    async fn test_fingerprint_registration_and_matching() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register first connector with a fingerprint
        let fingerprint = "acp/stdio:/usr/bin/claude".to_string();
        let req1 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Claude CLI".to_string(),
            client_native_id: "acp-session-abc123".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some(fingerprint.clone()),
        };

        let response1 = archivist.register_connector(req1, None).await?;
        assert_eq!(response1.status, RegisterStatus::Accepted);
        let original_uid = response1.connector_uid;

        // Register a second connector with a DIFFERENT client_native_id
        // but the SAME fingerprint. Should be ALIASED to the original.
        let req2 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Claude CLI (re-added)".to_string(),
            client_native_id: "acp-session-xyz789".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({"version": 2}),
            fingerprint: Some(fingerprint.clone()),
        };

        let response2 = archivist.register_connector(req2, None).await?;
        assert_eq!(
            response2.status,
            RegisterStatus::Aliased,
            "Same fingerprint should cause ALIASED status"
        );
        assert_eq!(
            response2.connector_uid, original_uid,
            "Aliased connector should return the original UID"
        );
        assert_eq!(response2.alias_of, Some(original_uid));
        assert!(
            response2.note.as_deref().unwrap_or("").contains("fingerprint"),
            "Note should mention fingerprint matching"
        );

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_fingerprint_no_match_different_fingerprints() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register first connector with fingerprint A
        let req1 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Claude CLI".to_string(),
            client_native_id: "acp-claude-1".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some("acp/stdio:/usr/bin/claude".to_string()),
        };

        let response1 = archivist.register_connector(req1, None).await?;
        assert_eq!(response1.status, RegisterStatus::Accepted);

        // Register second connector with fingerprint B (different)
        let req2 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Codex Agent".to_string(),
            client_native_id: "acp-codex-1".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some("acp/stdio:/usr/bin/codex".to_string()),
        };

        let response2 = archivist.register_connector(req2, None).await?;
        assert_eq!(
            response2.status,
            RegisterStatus::Accepted,
            "Different fingerprints should both be ACCEPTED"
        );
        assert_ne!(
            response2.connector_uid, response1.connector_uid,
            "Different fingerprints should get different UIDs"
        );

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_fingerprint_none_skips_matching() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register first connector WITH a fingerprint
        let req1 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Claude CLI".to_string(),
            client_native_id: "acp-claude-1".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some("acp/stdio:/usr/bin/claude".to_string()),
        };

        let response1 = archivist.register_connector(req1, None).await?;
        assert_eq!(response1.status, RegisterStatus::Accepted);

        // Register second connector WITHOUT a fingerprint (different native ID)
        // Should NOT match the first connector even though one exists with a fingerprint
        let req2 = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Unknown ACP Agent".to_string(),
            client_native_id: "acp-unknown-1".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let response2 = archivist.register_connector(req2, None).await?;
        assert_eq!(
            response2.status,
            RegisterStatus::Accepted,
            "Connector with fingerprint=None should not match existing fingerprints"
        );
        assert_ne!(
            response2.connector_uid, response1.connector_uid,
            "Should get a new UID when no fingerprint is provided"
        );

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_connector_fingerprint_persistence() -> Result<()> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_test_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // Register connector with a fingerprint
        let fingerprint_value = "acp/stdio:/usr/bin/claude".to_string();
        let connector_req = RegisterConnectorRequest {
            r#type: "ACP".to_string(),
            title: "Claude CLI".to_string(),
            client_native_id: "acp-claude-1".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some(fingerprint_value.clone()),
        };

        let response = archivist.register_connector(connector_req, None).await?;
        assert_eq!(response.status, RegisterStatus::Accepted);
        let connector_uid = response.connector_uid;

        // Verify fingerprint is persisted in connector.json
        let connector_json_path = backend.paths()
            .connector_dir(connector_uid)
            .join("connector.json");
        let raw_json = tokio::fs::read_to_string(&connector_json_path).await.unwrap();
        let connector_record: serde_json::Value = serde_json::from_str(&raw_json).unwrap();
        assert_eq!(
            connector_record["fingerprint"].as_str(),
            Some(fingerprint_value.as_str()),
            "Fingerprint should be persisted in connector.json"
        );

        // Verify fingerprint is persisted in TSV index
        let index_path = backend.paths().connector_index_tsv();
        let tsv_content = tokio::fs::read_to_string(&index_path).await.unwrap();
        assert!(
            tsv_content.contains(&fingerprint_value),
            "Fingerprint should appear in TSV index"
        );

        // Verify fingerprint can be read back via TSV reader
        let rows = dirigent_archivist::storage::tsv::read_connector_index(&index_path).await.unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].fingerprint, Some(fingerprint_value.clone()));

        // Register a second connector WITHOUT a fingerprint (ensure None is handled)
        let connector_req2 = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "OpenCode Local".to_string(),
            client_native_id: "opencode@localhost:12225".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };

        let response2 = archivist.register_connector(connector_req2, None).await?;
        assert_eq!(response2.status, RegisterStatus::Accepted);

        // Re-read TSV and verify both connectors
        let rows = dirigent_archivist::storage::tsv::read_connector_index(&index_path).await.unwrap();
        assert_eq!(rows.len(), 2);

        // First connector should have fingerprint
        let row_with_fp = rows.iter().find(|r| r.connector_uid == connector_uid).unwrap();
        assert_eq!(row_with_fp.fingerprint, Some(fingerprint_value));

        // Second connector should have no fingerprint
        let row_without_fp = rows.iter().find(|r| r.connector_uid == response2.connector_uid).unwrap();
        assert_eq!(row_without_fp.fingerprint, None);

        // Clean up
        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_list_connectors() -> std::result::Result<(), Box<dyn std::error::Error>> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_lc_{}", uuid::Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let req1 = RegisterConnectorRequest {
            r#type: "Acp".to_string(),
            title: "Claude".to_string(),
            client_native_id: "c1".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some("acp/stdio:/usr/bin/claude".to_string()),
        };
        archivist.register_connector(req1, None).await?;

        let req2 = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "OC".to_string(),
            client_native_id: "c2".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };
        archivist.register_connector(req2, None).await?;

        let connectors = archivist.list_connectors(None).await?;
        assert_eq!(connectors.len(), 2);

        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_move_session() -> std::result::Result<(), Box<dyn std::error::Error>> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_mv_{}", uuid::Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let c1 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "Acp".to_string(),
                    title: "Source".to_string(),
                    client_native_id: "src".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        let c2 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "Acp".to_string(),
                    title: "Target".to_string(),
                    client_native_id: "tgt".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        let session = archivist
            .register_session(
                RegisterSessionRequest {
                    connector_uid: c1,
                    native_session_id: "s1".to_string(),
                    title: Some("Test Session".to_string()),
                    custom_scroll_id: None,
                    metadata: serde_json::json!({}),
                    completeness: Default::default(),
                parent_scroll_id: None,
                is_subagent: false,
                continuation: None,
                agent_id: None,
                subagent_type: None,
                spawning_tool_use_id: None,
                },
                None,
            )
            .await?;

        // Verify under c1
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default().with_connector(c1).with_limit(100),
                )
                .await?
                .items
                .len(),
            1
        );

        // Move to c2
        archivist
            .move_session_to_connector(session.scroll_id, c2, None)
            .await?;

        // c1 should be empty, c2 should have the session
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default().with_connector(c1).with_limit(100),
                )
                .await?
                .items
                .len(),
            0
        );
        let c2_sessions = archivist
            .list_sessions_paged(
                SessionListQuery::default().with_connector(c2).with_limit(100),
            )
            .await?
            .items;
        assert_eq!(c2_sessions.len(), 1);
        assert_eq!(c2_sessions[0].connector_uid, c2);

        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_copy_session() -> std::result::Result<(), Box<dyn std::error::Error>> {
        let temp_dir = std::env::temp_dir().join(format!("archivist_cp_{}", uuid::Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let c1 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "Acp".to_string(),
                    title: "Source".to_string(),
                    client_native_id: "src".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        let c2 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "Acp".to_string(),
                    title: "Target".to_string(),
                    client_native_id: "tgt".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        let session = archivist
            .register_session(
                RegisterSessionRequest {
                    connector_uid: c1,
                    native_session_id: "s1".to_string(),
                    title: Some("Original".to_string()),
                    custom_scroll_id: None,
                    metadata: serde_json::json!({}),
                    completeness: Default::default(),
                parent_scroll_id: None,
                is_subagent: false,
                continuation: None,
                agent_id: None,
                subagent_type: None,
                spawning_tool_use_id: None,
                },
                None,
            )
            .await?;

        // Add a message
        let msg = MessageRecord {
            version: 1,
            message_id: uuid::Uuid::now_v7(),
            session: session.scroll_id,
            parent_id: None,
            ts: chrono::Utc::now(),
            role: "user".to_string(),
            author: Some("test".to_string()),
            content_md: "Hello".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };
        archivist
            .append_messages(session.scroll_id, vec![msg], None)
            .await?;

        // Copy
        let new_scroll_id = archivist
            .copy_session_to_connector(session.scroll_id, c2, None)
            .await?;
        assert_ne!(new_scroll_id, session.scroll_id);

        // Original still under c1
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default().with_connector(c1).with_limit(100),
                )
                .await?
                .items
                .len(),
            1
        );
        // Copy under c2
        let c2_sessions = archivist
            .list_sessions_paged(
                SessionListQuery::default().with_connector(c2).with_limit(100),
            )
            .await?
            .items;
        assert_eq!(c2_sessions.len(), 1);
        assert_eq!(c2_sessions[0].connector_uid, c2);
        // Messages copied
        let msgs = archivist.get_messages(new_scroll_id, None).await?;
        assert_eq!(msgs.len(), 1);
        assert_eq!(msgs[0].content_md, "Hello");

        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_move_sessions_bulk() -> std::result::Result<(), Box<dyn std::error::Error>> {
        let temp_dir =
            std::env::temp_dir().join(format!("archivist_mvb_{}", uuid::Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let c1 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "Acp".to_string(),
                    title: "Source".to_string(),
                    client_native_id: "src".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        let c2 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "Acp".to_string(),
                    title: "Target".to_string(),
                    client_native_id: "tgt".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        let mut scroll_ids = Vec::new();
        for i in 0..3 {
            let s = archivist
                .register_session(
                    RegisterSessionRequest {
                        connector_uid: c1,
                        native_session_id: format!("s{}", i),
                        title: Some(format!("Session {}", i)),
                        custom_scroll_id: None,
                        metadata: serde_json::json!({}),
                        completeness: Default::default(),
                    parent_scroll_id: None,
                    is_subagent: false,
                    continuation: None,
                    agent_id: None,
                    subagent_type: None,
                    spawning_tool_use_id: None,
                    },
                    None,
                )
                .await?;
            scroll_ids.push(s.scroll_id);
        }

        let report = archivist
            .move_sessions_to_connector(scroll_ids, c2, None)
            .await?;
        assert_eq!(report.moved, 3);
        assert_eq!(report.failed, 0);
        assert!(report.errors.is_empty());
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default().with_connector(c1).with_limit(100),
                )
                .await?
                .items
                .len(),
            0
        );
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default().with_connector(c2).with_limit(100),
                )
                .await?
                .items
                .len(),
            3
        );

        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_connector_identity_persistence_e2e() -> std::result::Result<(), Box<dyn std::error::Error>>
    {
        let temp_dir =
            std::env::temp_dir().join(format!("archivist_e2e_{}", uuid::Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        // 1. Register connector with fingerprint
        let req1 = RegisterConnectorRequest {
            r#type: "Acp".to_string(),
            title: "Claude v1".to_string(),
            client_native_id: "first-run-id".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some("acp/stdio:/usr/bin/claude".to_string()),
        };
        let resp1 = archivist.register_connector(req1, None).await?;
        let original_uid = resp1.connector_uid;
        assert_eq!(resp1.status, RegisterStatus::Accepted);

        // 2. Create sessions under this connector
        let s1 = archivist
            .register_session(
                RegisterSessionRequest {
                    connector_uid: original_uid,
                    native_session_id: "session-1".to_string(),
                    title: Some("Important Session".to_string()),
                    custom_scroll_id: None,
                    metadata: serde_json::json!({}),
                    completeness: Default::default(),
                parent_scroll_id: None,
                is_subagent: false,
                continuation: None,
                agent_id: None,
                subagent_type: None,
                spawning_tool_use_id: None,
                },
                None,
            )
            .await?;

        // 3. Add messages
        let msg = MessageRecord {
            version: 1,
            message_id: uuid::Uuid::now_v7(),
            session: s1.scroll_id,
            parent_id: None,
            ts: chrono::Utc::now(),
            role: "user".to_string(),
            author: Some("test".to_string()),
            content_md: "Don't lose me!".to_string(),
            content_parts: None,
            attachments: vec![],
            metadata: serde_json::json!({}),
        };
        archivist
            .append_messages(s1.scroll_id, vec![msg], None)
            .await?;

        // 4. Simulate "remove and re-add" -- new connector_id, same fingerprint
        let req2 = RegisterConnectorRequest {
            r#type: "Acp".to_string(),
            title: "Claude v2 (reinstalled)".to_string(),
            client_native_id: "second-run-id".to_string(),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: Some("acp/stdio:/usr/bin/claude".to_string()),
        };
        let resp2 = archivist.register_connector(req2, None).await?;

        // 5. Verify: same UID, ALIASED status
        assert_eq!(resp2.status, RegisterStatus::Aliased);
        assert_eq!(resp2.connector_uid, original_uid);

        // 6. Verify: sessions still accessible under the same UID
        let sessions = archivist
            .list_sessions_paged(
                SessionListQuery::default()
                    .with_connector(original_uid)
                    .with_limit(100),
            )
            .await?
            .items;
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].title, Some("Important Session".to_string()));

        // 7. Verify: messages intact
        let messages = archivist.get_messages(s1.scroll_id, None).await?;
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].content_md, "Don't lose me!");

        // 8. Verify: connector record is preserved under the original UID.
        //
        // NOTE: Pre-Phase-2 `FileBasedArchivist` ALSO refreshed the matched
        // connector's `title`/`metadata` on fingerprint-based ALIASED
        // registration. The `Archivist` deliberately drops that
        // refresh (see `coordinator/connectors.rs` for the rationale — the
        // `ConnectorRegistryBackend` trait has no "update metadata" method
        // yet, and `put_connector` would append rather than mutate). The
        // identity (UID) is stable; the title stays the original.
        let connectors = archivist.list_connectors(None).await?;
        let connector = connectors
            .iter()
            .find(|c| c.connector_uid == original_uid)
            .unwrap();
        assert_eq!(connector.title, "Claude v1");

        // 9. Test move_session works after fingerprint re-association
        let c2 = archivist
            .register_connector(
                RegisterConnectorRequest {
                    r#type: "OpenCode".to_string(),
                    title: "Secondary".to_string(),
                    client_native_id: "secondary".to_string(),
                    custom_uid: None,
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await?
            .connector_uid;

        archivist
            .move_session_to_connector(s1.scroll_id, c2, None)
            .await?;
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default()
                        .with_connector(original_uid)
                        .with_limit(100),
                )
                .await?
                .items
                .len(),
            0
        );
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default().with_connector(c2).with_limit(100),
                )
                .await?
                .items
                .len(),
            1
        );

        // Move it back
        archivist
            .move_session_to_connector(s1.scroll_id, original_uid, None)
            .await?;
        assert_eq!(
            archivist
                .list_sessions_paged(
                    SessionListQuery::default()
                        .with_connector(original_uid)
                        .with_limit(100),
                )
                .await?
                .items
                .len(),
            1
        );

        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }

    #[tokio::test]
    async fn test_paged_walk_fifty_sessions() -> Result<()> {
        use chrono::{Duration, Utc};
        use dirigent_archivist::SessionListQuery;

        let temp_dir = std::env::temp_dir().join(format!("paged_walk_{}", Uuid::now_v7()));
        let backend = std::sync::Arc::new(
            dirigent_archivist::backends::JsonlBackend::new(temp_dir.clone()).await?
        );
        let archivist = Archivist::from_single_backend(
            "main".into(), backend.clone()
        ).await?;

        let connector_req = RegisterConnectorRequest {
            r#type: "OpenCode".to_string(),
            title: "paged-walk".to_string(),
            client_native_id: format!("paged-walk@{}", Uuid::now_v7()),
            custom_uid: None,
            metadata: serde_json::json!({}),
            fingerprint: None,
        };
        let cresp = archivist.register_connector(connector_req, None).await?;
        let uid = cresp.connector_uid;

        let base = Utc::now();
        for i in 0..50 {
            let tag = if i % 2 == 0 { "even" } else { "odd" };

            let req = RegisterSessionRequest {
                connector_uid: uid,
                native_session_id: format!("walk-{i}"),
                title: Some(format!("title-{i}")),
                custom_scroll_id: None,
                metadata: serde_json::json!({"model": "claude-3-5-sonnet"}),
                completeness: Default::default(),
                parent_scroll_id: None,
                is_subagent: false,
                continuation: None,
                agent_id: None,
                subagent_type: None,
                spawning_tool_use_id: None,
            };
            let r = archivist.register_session(req, None).await?;

            let mut meta = archivist.get_session_metadata(r.scroll_id, None).await?;
            meta.updated_at = base - Duration::seconds(i);
            meta.tags = vec![tag.to_string()];
            let path = backend.paths().session_json(r.scroll_id);
            dirigent_archivist::storage::json::write_json(&path, &meta)
                .await
                .map_err(dirigent_archivist::ArchivistError::Io)?;
        }

        // Walk in chunks of 10 — 5 pages, 50 items, no dupes.
        let mut seen: std::collections::HashSet<Uuid> = std::collections::HashSet::new();
        let mut cursor = None;
        let mut page_count = 0;
        loop {
            let page = archivist
                .list_sessions_paged(
                    SessionListQuery::default()
                        .with_connector(uid)
                        .with_limit(10)
                        .with_cursor(cursor.clone()),
                )
                .await?;
            page_count += 1;
            for s in &page.items {
                assert!(seen.insert(s.scroll_id), "duplicate scroll_id across pages");
            }
            if page.next_cursor.is_none() {
                break;
            }
            cursor = page.next_cursor;
            assert!(page_count <= 10, "runaway pagination");
        }
        assert_eq!(seen.len(), 50);
        assert_eq!(page_count, 5);

        // Compose filter: tag=even AND title contains "1" → titles 10, 12, 14, 16, 18.
        let mut q = SessionListQuery::default().with_connector(uid).with_limit(50);
        q.tags = vec!["even".into()];
        q.title_query = Some("1".into());
        let page = archivist.list_sessions_paged(q).await?;
        assert_eq!(
            page.items.len(),
            5,
            "got titles {:?}",
            page.items.iter().map(|s| s.title.clone()).collect::<Vec<_>>()
        );

        tokio::fs::remove_dir_all(temp_dir).await.ok();
        Ok(())
    }
}
