//! Session store implementation
//!
//! File-based storage for sessions and chat history. Sessions are stored
//! as JSON metadata files, and chat history is stored as JSONL for
//! append-friendly operations.

use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap};
use std::fs::{self, File, OpenOptions};
use std::io::{BufReader, BufWriter, ErrorKind, Read, Write};
use std::path::{Path, PathBuf};
#[cfg(test)]
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use tracing::warn;
use uuid::Uuid;
use zeroize::Zeroizing;

use super::file_lock::FileLock;

/// Default message count threshold for auto-compaction
const DEFAULT_COMPACT_THRESHOLD: usize = 100;

/// Maximum message count before forcing compaction
const MAX_MESSAGES_BEFORE_COMPACT: usize = 500;

/// Cap on the session-meta JSON file at read. The meta payload is
/// a single session record (id, key, channel info, last-message
/// envelope); realistic size is a few KB. 4 MiB is generously
/// above any extension and refuses a planted-multi-GB DoS at the
/// session-load site.
const SESSION_META_FILE_MAX_BYTES: u64 = 4 * 1024 * 1024;

/// Cap on the session-history JSONL file at read. History grows
/// for long-running agents but is bounded by application-level
/// retention. 256 MiB is generous enough for legitimate long-
/// history sessions, refuses the planted-multi-GB attack.
pub(crate) const SESSION_HISTORY_FILE_MAX_BYTES: u64 = 256 * 1024 * 1024;

const SESSION_METADATA_PURPOSE: &str = "metadata";
const SESSION_HISTORY_PURPOSE: &str = "history";
const SESSION_ARCHIVE_PURPOSE: &str = "archive";
const PROTECTED_INBOUND_DEDUPE_CACHE_MAX_ENTRIES: usize = 4096;

#[cfg(test)]
static TEST_FORCE_SESSION_META_PARENT_SYNC_FAIL: AtomicBool = AtomicBool::new(false);
#[cfg(test)]
static TEST_FORCE_HISTORY_PARENT_SYNC_AFTER_RENAME_FAIL: AtomicBool = AtomicBool::new(false);
#[cfg(test)]
static TEST_PROTECTED_INBOUND_DEDUPE_HISTORY_SIGNATURE_READS: AtomicUsize = AtomicUsize::new(0);

/// Error types for session store operations
#[derive(Debug, Clone, thiserror::Error)]
pub enum SessionStoreError {
    #[error("Session not found: {0}")]
    NotFound(String),
    #[error("Session already exists: {0}")]
    AlreadyExists(String),
    #[error("IO error: {0}")]
    Io(String),
    #[error("Serialization error: {0}")]
    Serialization(String),
    #[error("Invalid session key: {0}")]
    InvalidSessionKey(String),
    #[error("Compaction in progress for session: {0}")]
    CompactionInProgress(String),
    #[error("Session is already archived: {0}")]
    AlreadyArchived(String),
    #[error("Session is not archived: {0}")]
    NotArchived(String),
    #[error("Archive not found: {0}")]
    ArchiveNotFound(String),
    #[error("Invalid user ID: {0}")]
    InvalidUserId(String),
    #[error("Invalid message batch: {0}")]
    InvalidMessageBatch(String),
    #[error("Session store is locked: {0}")]
    Locked(String),
    #[error("Session decryption failed: {0}")]
    DecryptionFailed(String),
    #[error("Session manifest integrity failed: {0}")]
    ManifestIntegrityFailed(String),
    #[error("Session integrity rejected: {0}")]
    IntegrityRejected(String),
    #[error("Session crypto error: {0}")]
    Crypto(String),
    /// A plaintext history line failed to parse during a dedupe scan.
    /// The caller (Matrix inbound replay) must NOT continue scanning
    /// because a missing prior `inbound_event_id` would let a
    /// redelivery re-dispatch the agent on a message already seen.
    /// Operator must inspect the on-disk file before the channel
    /// resumes.
    #[error("Session history is corrupt: {0}")]
    HistoryCorrupt(String),
    /// The history file is at or above `SESSION_HISTORY_FILE_MAX_BYTES`
    /// and the append (single or batch) would push it past the
    /// read-side cap. Classified as a separate variant from
    /// `Serialization` because matrix-channel DLQ retry classification
    /// treats `Serialization` as transient (the payload's encoding
    /// could change on the next attempt) while `HistoryFileFull` is
    /// permanent until an operator force-compacts the session —
    /// otherwise the DLQ retries the same oversize append forever
    /// and the same dead-letter slot accumulates indefinitely.
    #[error("Session history file is full: {0}")]
    HistoryFileFull(String),
}

impl SessionStoreError {
    /// Returns true for failures that can hide earlier protected inbound
    /// delivery markers from a replay scan.
    ///
    /// Future crypto/authenticity variants that can make prior
    /// `inbound_event_id` entries unreadable must be classified here so the
    /// Matrix channel fails closed until operator repair instead of retrying as
    /// a transient dispatch failure.
    pub fn is_permanent_history_corruption(&self) -> bool {
        matches!(
            self,
            SessionStoreError::HistoryCorrupt(_)
                | SessionStoreError::ManifestIntegrityFailed(_)
                | SessionStoreError::DecryptionFailed(_)
                | SessionStoreError::IntegrityRejected(_)
        )
    }
}

impl From<std::io::Error> for SessionStoreError {
    fn from(err: std::io::Error) -> Self {
        SessionStoreError::Io(err.to_string())
    }
}

impl From<serde_json::Error> for SessionStoreError {
    fn from(err: serde_json::Error) -> Self {
        SessionStoreError::Serialization(err.to_string())
    }
}

impl From<super::crypto::SessionCryptoError> for SessionStoreError {
    fn from(err: super::crypto::SessionCryptoError) -> Self {
        match err {
            super::crypto::SessionCryptoError::DecryptionFailed => {
                SessionStoreError::DecryptionFailed(
                    "wrong password or corrupted encrypted session data".to_string(),
                )
            }
            super::crypto::SessionCryptoError::ManifestIntegrityFailed => {
                SessionStoreError::ManifestIntegrityFailed(
                    "wrong password or tampered encrypted-session manifest".to_string(),
                )
            }
            super::crypto::SessionCryptoError::Io(message)
            | super::crypto::SessionCryptoError::Manifest(message)
            | super::crypto::SessionCryptoError::KeyDerivation(message)
            | super::crypto::SessionCryptoError::RandomFailure(message)
            | super::crypto::SessionCryptoError::BadFormat(message)
            | super::crypto::SessionCryptoError::EncryptionFailed(message) => {
                SessionStoreError::Crypto(message)
            }
            super::crypto::SessionCryptoError::Base64Decode { field, message } => {
                SessionStoreError::Crypto(format!("{}: {}", field, message))
            }
        }
    }
}

fn session_store_error_kind(err: &SessionStoreError) -> &'static str {
    match err {
        SessionStoreError::NotFound(_) => "not_found",
        SessionStoreError::AlreadyExists(_) => "already_exists",
        SessionStoreError::Io(_) => "io",
        SessionStoreError::Serialization(_) => "serialization",
        SessionStoreError::InvalidSessionKey(_) => "invalid_session_key",
        SessionStoreError::CompactionInProgress(_) => "compaction_in_progress",
        SessionStoreError::AlreadyArchived(_) => "already_archived",
        SessionStoreError::NotArchived(_) => "not_archived",
        SessionStoreError::ArchiveNotFound(_) => "archive_not_found",
        SessionStoreError::InvalidUserId(_) => "invalid_user_id",
        SessionStoreError::InvalidMessageBatch(_) => "invalid_message_batch",
        SessionStoreError::Locked(_) => "locked",
        SessionStoreError::DecryptionFailed(_) => "decryption_failed",
        SessionStoreError::ManifestIntegrityFailed(_) => "manifest_integrity_failed",
        SessionStoreError::IntegrityRejected(_) => "integrity_rejected",
        SessionStoreError::Crypto(_) => "crypto",
        SessionStoreError::HistoryCorrupt(_) => "history_corrupt",
        SessionStoreError::HistoryFileFull(_) => "history_file_full",
    }
}

fn session_store_error_export_warning(err: &SessionStoreError) -> Cow<'static, str> {
    match err {
        SessionStoreError::NotFound(_) => Cow::Borrowed("session data was not found on disk"),
        SessionStoreError::AlreadyExists(_) => Cow::Borrowed("session data already exists on disk"),
        SessionStoreError::Io(_) => Cow::Borrowed("session data could not be read from disk"),
        SessionStoreError::Serialization(_) => Cow::Borrowed("session data is malformed"),
        SessionStoreError::InvalidSessionKey(_) => {
            Cow::Borrowed("session data uses an invalid session key")
        }
        SessionStoreError::CompactionInProgress(_) => {
            Cow::Borrowed("session data is being compacted and is temporarily unavailable")
        }
        SessionStoreError::AlreadyArchived(_) => Cow::Borrowed("session data is already archived"),
        SessionStoreError::NotArchived(_) => Cow::Borrowed("session data is not archived"),
        SessionStoreError::ArchiveNotFound(_) => {
            Cow::Borrowed("session archive was not found on disk")
        }
        SessionStoreError::InvalidUserId(_) => {
            Cow::Borrowed("session data is associated with an invalid user ID")
        }
        SessionStoreError::InvalidMessageBatch(_) => {
            Cow::Borrowed("session message batch contains messages for multiple sessions")
        }
        SessionStoreError::Locked(message) => Cow::Owned(message.clone()),
        SessionStoreError::DecryptionFailed(_) => {
            Cow::Borrowed("encrypted session data could not be decrypted")
        }
        SessionStoreError::ManifestIntegrityFailed(_) => {
            Cow::Borrowed("encrypted session manifest integrity verification failed")
        }
        SessionStoreError::IntegrityRejected(_) => {
            Cow::Borrowed("session integrity verification failed")
        }
        SessionStoreError::Crypto(_) => {
            Cow::Borrowed("encrypted session data is unreadable or corrupted")
        }
        SessionStoreError::HistoryCorrupt(_) => {
            Cow::Borrowed("session history is corrupt or failed integrity verification")
        }
        SessionStoreError::HistoryFileFull(_) => {
            Cow::Borrowed("session history file is full and must be compacted")
        }
    }
}

fn integrity_error_kind(err: &super::integrity::IntegrityError) -> &'static str {
    match err {
        super::integrity::IntegrityError::Io(_) => "io",
        super::integrity::IntegrityError::VerificationFailed { .. } => "verification_failed",
        super::integrity::IntegrityError::Rejected { .. } => "rejected",
    }
}

/// Status of a session
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionStatus {
    /// Session is active and can receive messages
    #[default]
    Active,
    /// Session is paused (no new messages processed)
    Paused,
    /// Session is archived (read-only)
    Archived,
    /// Session is being compacted
    Compacting,
    /// Forward-compat sentinel for unknown status wire names written
    /// by a newer daemon. `load_session_with_locked_tracking` parses
    /// the per-session JSON; a parse failure today escalates to
    /// `SessionStoreError::Serialization` → silently drops the session
    /// from the listing (per-file isolated, but the operator can no
    /// longer find/manage it). Unknown maps to this sentinel; the
    /// session still loads and is visible. `Unknown` is not `Archived`
    /// or `Compacting`, so non-archived predicates continue to fire.
    Unknown,
}

impl std::fmt::Display for SessionStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Active => write!(f, "active"),
            Self::Paused => write!(f, "paused"),
            Self::Archived => write!(f, "archived"),
            Self::Compacting => write!(f, "compacting"),
            Self::Unknown => write!(f, "unknown"),
        }
    }
}

fn deserialize_session_status_forward_compat<'de, D>(
    deserializer: D,
) -> Result<SessionStatus, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value = String::deserialize(deserializer)?;
    let status = match value.as_str() {
        "active" => SessionStatus::Active,
        "paused" => SessionStatus::Paused,
        "archived" => SessionStatus::Archived,
        "compacting" => SessionStatus::Compacting,
        "unknown" => SessionStatus::Unknown,
        _ => {
            tracing::warn!(
                session_status = %value,
                "sessions: unrecognized session status wire name; treating as Unknown \
                 for forward-compat (downgrade-tolerant per-session load)"
            );
            SessionStatus::Unknown
        }
    };
    Ok(status)
}

/// Role of a chat message sender
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MessageRole {
    /// User-sent message
    #[default]
    User,
    /// Assistant response
    Assistant,
    /// System message (e.g., context, instructions)
    System,
    /// Tool call or result
    Tool,
    /// Forward-compat sentinel for unknown role wire names written by a
    /// newer daemon. Reads must NOT hard-error a session history line
    /// on an unknown role — `SessionStoreError::Serialization` from a
    /// historical replay escalates to `SessionStoreError::HistoryCorrupt`
    /// at `src/sessions/store.rs:1123+`, which `is_permanent_history_corruption`
    /// fails closed against, blocking Matrix inbound replay for the
    /// affected session until manual operator repair.
    Unknown,
}

impl std::fmt::Display for MessageRole {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::User => write!(f, "user"),
            Self::Assistant => write!(f, "assistant"),
            Self::System => write!(f, "system"),
            Self::Tool => write!(f, "tool"),
            Self::Unknown => write!(f, "unknown"),
        }
    }
}

fn deserialize_message_role_forward_compat<'de, D>(deserializer: D) -> Result<MessageRole, D::Error>
where
    D: serde::Deserializer<'de>,
{
    // COMPAT: see the `MessageRole::Unknown` doc comment above for the
    // failure-mode this defends against. Mirrors the pattern at
    // `deserialize_update_phase_forward_compat` / `deserialize_task_state_forward_compat`.
    let value = String::deserialize(deserializer)?;
    let role = match value.as_str() {
        "user" => MessageRole::User,
        "assistant" => MessageRole::Assistant,
        "system" => MessageRole::System,
        "tool" => MessageRole::Tool,
        "unknown" => MessageRole::Unknown,
        _ => {
            tracing::warn!(
                message_role = %value,
                "sessions: unrecognized message role wire name in history; \
                 treating as Unknown for forward-compat (downgrade-tolerant replay)"
            );
            MessageRole::Unknown
        }
    };
    Ok(role)
}

/// Metadata for compaction operations
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct CompactionMetadata {
    /// Number of messages compacted
    pub messages_compacted: usize,
    /// Timestamp of last compaction (Unix ms)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_compacted_at: Option<i64>,
    /// Number of compaction operations performed
    pub compaction_count: u32,
    /// Summary text from last compaction
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_summary: Option<String>,
    /// Original message count before compactions
    #[serde(default)]
    pub original_message_count: usize,
}

/// Result of an archive operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ArchiveResult {
    /// Session ID that was archived
    pub session_id: String,
    /// Path to the archive file
    pub archive_path: String,
    /// Number of messages archived
    pub message_count: usize,
    /// Size of the archive file in bytes
    pub archive_size: u64,
    /// Timestamp when the archive was created (Unix ms)
    pub archived_at: i64,
    /// Whether history was deleted after archiving
    pub history_deleted: bool,
}

/// Result of a restore operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RestoreResult {
    /// Session ID that was restored
    pub session_id: String,
    /// Number of messages restored
    pub message_count: usize,
    /// Timestamp when the restore happened (Unix ms)
    pub restored_at: i64,
}

/// Maximum `ArchivedSession.version` this binary understands. A file
/// with a higher version was written by a newer daemon and must be
/// refused at load time rather than silently downgraded — see
/// `Self::decode_archive` for the rationale.
pub const ARCHIVED_SESSION_VERSION: u32 = 1;

/// Archived session metadata stored in archive file
///
/// SECURITY: `#[serde(deny_unknown_fields)]` ensures that a
/// session archive written by a newer daemon with additional top-level
/// fields is REFUSED at parse time rather than silently downgraded.
/// Without this, the `version > ARCHIVED_SESSION_VERSION` guard
/// could be bypassed by a future v1.5 daemon that added fields without
/// bumping the version — this binary would parse them, drop the
/// unknown fields, and clobber the archive with v1 bytes on next
/// re-archive. With deny_unknown_fields, the parse fails loud and the
/// operator gets a clear "upgrade Carapace" surface.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ArchivedSession {
    /// Original session data
    pub session: Session,
    /// All messages at time of archiving
    pub messages: Vec<ChatMessage>,
    /// When the archive was created (Unix ms)
    pub archived_at: i64,
    /// Archive format version
    pub version: u32,
}

/// Metadata associated with a session
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct SessionMetadata {
    /// Human-readable session name
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// Session description
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// Agent ID this session belongs to
    #[serde(skip_serializing_if = "Option::is_none")]
    pub agent_id: Option<String>,
    /// Channel this session is associated with
    #[serde(skip_serializing_if = "Option::is_none")]
    pub channel: Option<String>,
    /// Chat/conversation ID within the channel
    #[serde(skip_serializing_if = "Option::is_none")]
    pub chat_id: Option<String>,
    /// User ID of the session owner
    #[serde(skip_serializing_if = "Option::is_none")]
    pub user_id: Option<String>,
    /// Named route for this session
    #[serde(skip_serializing_if = "Option::is_none")]
    pub route: Option<String>,
    /// Model being used for this session
    #[serde(skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    /// Thinking level for this session
    #[serde(skip_serializing_if = "Option::is_none")]
    pub thinking_level: Option<String>,
    /// Custom tags for organization
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tags: Vec<String>,
    /// Additional custom data
    #[serde(skip_serializing_if = "Option::is_none")]
    pub extra: Option<serde_json::Value>,
    /// Compaction metadata
    #[serde(default)]
    pub compaction: CompactionMetadata,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct InboundAppendOutcome {
    pub appended: bool,
    pub original_run_id: Option<String>,
    pub corrupt_index_lines: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct InboundEventIndexEntry {
    event_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    run_id: Option<String>,
}

#[derive(Debug, Default)]
struct InboundEventIndexLookup {
    entry: Option<InboundEventIndexEntry>,
    corrupt_lines: u64,
    matched_protected_history: bool,
    legacy_index_present: bool,
}

#[derive(Debug, Default)]
struct ProtectedInboundDedupeCache {
    complete: bool,
    history_signature: Option<ProtectedInboundDedupeHistorySignature>,
    history_hasher: Option<ProtectedInboundDedupeHistoryHasher>,
    entries: HashMap<String, InboundEventIndexEntry>,
}

#[derive(Clone)]
struct ProtectedInboundDedupeHistoryHasher(Sha256);

impl std::fmt::Debug for ProtectedInboundDedupeHistoryHasher {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("ProtectedInboundDedupeHistoryHasher(<redacted>)")
    }
}

struct ProtectedInboundDedupeHistoryCacheState {
    signature: ProtectedInboundDedupeHistorySignature,
    hasher: ProtectedInboundDedupeHistoryHasher,
}

#[derive(Clone, PartialEq, Eq)]
struct ProtectedInboundDedupeHistoryFileState {
    file_len: u64,
    #[cfg(unix)]
    device: u64,
    #[cfg(unix)]
    inode: u64,
    #[cfg(unix)]
    ctime_sec: i64,
    #[cfg(unix)]
    ctime_nsec: i64,
    #[cfg(not(unix))]
    modified_at_ms: Option<i64>,
}

impl std::fmt::Debug for ProtectedInboundDedupeHistoryFileState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut debug = f.debug_struct("ProtectedInboundDedupeHistoryFileState");
        debug.field("file_len", &self.file_len);
        #[cfg(unix)]
        {
            debug
                .field("device", &self.device)
                .field("inode", &self.inode)
                .field("ctime_sec", &self.ctime_sec)
                .field("ctime_nsec", &self.ctime_nsec);
        }
        #[cfg(not(unix))]
        {
            debug.field("modified_at_ms", &self.modified_at_ms);
        }
        debug.finish()
    }
}

#[derive(Clone, PartialEq, Eq)]
struct ProtectedInboundDedupeHistorySignature {
    file_len: u64,
    hmac_sidecar_sha256: Option<[u8; 32]>,
    content_sha256: [u8; 32],
}

impl std::fmt::Debug for ProtectedInboundDedupeHistorySignature {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ProtectedInboundDedupeHistorySignature")
            .field("file_len", &self.file_len)
            .field(
                "hmac_sidecar_sha256",
                &self.hmac_sidecar_sha256.as_ref().map(|_| "<redacted>"),
            )
            .field("content_sha256", &"<redacted>")
            .finish()
    }
}

/// A chat session
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Session {
    /// Unique session identifier (UUID)
    pub id: String,
    /// Session key (human-friendly, e.g., "telegram:123456:default")
    pub session_key: String,
    /// Current status
    #[serde(deserialize_with = "deserialize_session_status_forward_compat")]
    pub status: SessionStatus,
    /// Session metadata
    #[serde(default)]
    pub metadata: SessionMetadata,
    /// Number of messages in history
    pub message_count: usize,
    /// Timestamp when session was created (Unix ms)
    pub created_at: i64,
    /// Timestamp when session was last updated (Unix ms)
    pub updated_at: i64,
    /// Timestamp of last activity (Unix ms)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_activity_at: Option<i64>,
}

impl Session {
    /// Create a new session with the given agent ID and metadata
    pub fn new(agent_id: impl Into<String>, metadata: SessionMetadata) -> Self {
        let now = now_millis();
        let id = Uuid::new_v4().to_string();
        let agent_id_str = agent_id.into();

        // Generate session key from metadata
        let session_key = generate_session_key(&agent_id_str, &metadata);

        Self {
            id,
            session_key,
            status: SessionStatus::Active,
            metadata: SessionMetadata {
                agent_id: Some(agent_id_str),
                ..metadata
            },
            message_count: 0,
            created_at: now,
            updated_at: now,
            last_activity_at: Some(now),
        }
    }

    /// Create a session with a specific session key
    pub fn with_session_key(session_key: impl Into<String>, metadata: SessionMetadata) -> Self {
        let now = now_millis();
        Self {
            id: Uuid::new_v4().to_string(),
            session_key: session_key.into(),
            status: SessionStatus::Active,
            metadata,
            message_count: 0,
            created_at: now,
            updated_at: now,
            last_activity_at: Some(now),
        }
    }

    /// Check if the session needs compaction
    pub fn needs_compaction(&self) -> bool {
        self.message_count >= DEFAULT_COMPACT_THRESHOLD
    }

    /// Check if compaction should be forced
    pub fn force_compaction_needed(&self) -> bool {
        self.message_count >= MAX_MESSAGES_BEFORE_COMPACT
    }
}

/// Accessibility state for a session listing entry.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionAccessState {
    Available,
    Locked,
}

/// Session list entry that may represent either a full session or a locked stub.
#[derive(Debug, Clone)]
pub struct SessionListEntry {
    session_id: String,
    access: SessionAccessState,
    session: Option<Session>,
    updated_at: Option<i64>,
}

impl SessionListEntry {
    fn available(session: Session) -> Self {
        let updated_at = Some(session.updated_at);
        let session_id = session.id.clone();
        Self {
            session_id,
            access: SessionAccessState::Available,
            session: Some(session),
            updated_at,
        }
    }

    fn locked(session_id: String, updated_at: Option<i64>) -> Self {
        Self {
            session_id,
            access: SessionAccessState::Locked,
            session: None,
            updated_at,
        }
    }

    pub fn session_id(&self) -> &str {
        &self.session_id
    }

    pub fn access(&self) -> SessionAccessState {
        self.access
    }

    pub fn is_locked(&self) -> bool {
        self.access == SessionAccessState::Locked
    }

    pub fn session(&self) -> Option<&Session> {
        self.session.as_ref()
    }

    pub fn into_session(self) -> Option<Session> {
        self.session
    }

    pub fn updated_at(&self) -> Option<i64> {
        self.updated_at
    }
}

/// A chat message in session history
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChatMessage {
    /// Unique message identifier
    pub id: String,
    /// Session ID this message belongs to
    pub session_id: String,
    /// Role of the sender
    #[serde(deserialize_with = "deserialize_message_role_forward_compat")]
    pub role: MessageRole,
    /// Message content
    pub content: String,
    /// Tool call ID (for tool messages)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tool_call_id: Option<String>,
    /// Tool name (for tool messages)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tool_name: Option<String>,
    /// Token count for this message
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tokens: Option<u64>,
    /// Timestamp when message was created (Unix ms)
    pub created_at: i64,
    /// Additional metadata
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metadata: Option<serde_json::Value>,
    /// Forward-compat catch-all for top-level fields a newer daemon
    /// may add (e.g., `reasoning_blocks`, `tool_calls`, `cache_control`).
    /// Without this, an older binary reading a v2 history line would
    /// silently drop the unknown top-level fields, and the subsequent
    /// `compact_session` / encrypted-history rewrite would clobber the
    /// v2 bytes on disk with the v1 shape — same hazard the
    /// `ArchivedSession` deny_unknown_fields guard closes at the
    /// file level, but per-message inside the JSONL history. Mirror
    /// `ManagedPluginManifestEntry.extra` and
    /// `UpdateTransaction.extra` — capture unknowns into a map that
    /// round-trips on serialize.
    #[serde(flatten, default, skip_serializing_if = "BTreeMap::is_empty")]
    pub extra: BTreeMap<String, serde_json::Value>,
}

impl ChatMessage {
    /// Create a new chat message
    pub fn new(
        session_id: impl Into<String>,
        role: MessageRole,
        content: impl Into<String>,
    ) -> Self {
        Self {
            id: Uuid::new_v4().to_string(),
            session_id: session_id.into(),
            role,
            content: content.into(),
            tool_call_id: None,
            tool_name: None,
            tokens: None,
            created_at: now_millis(),
            metadata: None,
            extra: BTreeMap::new(),
        }
    }

    /// Create a user message
    pub fn user(session_id: impl Into<String>, content: impl Into<String>) -> Self {
        Self::new(session_id, MessageRole::User, content)
    }

    /// Create an assistant message
    pub fn assistant(session_id: impl Into<String>, content: impl Into<String>) -> Self {
        Self::new(session_id, MessageRole::Assistant, content)
    }

    /// Create a system message
    pub fn system(session_id: impl Into<String>, content: impl Into<String>) -> Self {
        Self::new(session_id, MessageRole::System, content)
    }

    /// Create a tool message
    pub fn tool(
        session_id: impl Into<String>,
        tool_name: impl Into<String>,
        tool_call_id: impl Into<String>,
        content: impl Into<String>,
    ) -> Self {
        Self {
            tool_name: Some(tool_name.into()),
            tool_call_id: Some(tool_call_id.into()),
            ..Self::new(session_id, MessageRole::Tool, content)
        }
    }

    /// Set token count
    pub fn with_tokens(mut self, tokens: u64) -> Self {
        self.tokens = Some(tokens);
        self
    }

    /// Set metadata
    pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self {
        self.metadata = Some(metadata);
        self
    }
}

/// Filter criteria for listing sessions
#[derive(Debug, Clone, Default)]
pub struct SessionFilter {
    /// Filter by agent ID
    pub agent_id: Option<String>,
    /// Filter by status
    pub status: Option<SessionStatus>,
    /// Filter by channel
    pub channel: Option<String>,
    /// Filter by user ID
    pub user_id: Option<String>,
    /// Filter sessions created after this timestamp (Unix ms)
    pub created_after: Option<i64>,
    /// Filter sessions created before this timestamp (Unix ms)
    pub created_before: Option<i64>,
    /// Filter sessions updated after this timestamp (Unix ms)
    pub updated_after: Option<i64>,
    /// Filter sessions updated before this timestamp (Unix ms)
    pub updated_before: Option<i64>,
    /// Maximum number of sessions to return
    pub limit: Option<usize>,
    /// Number of sessions to skip
    pub offset: Option<usize>,
}

impl SessionFilter {
    /// Create a new filter
    pub fn new() -> Self {
        Self::default()
    }

    /// Filter by agent ID
    pub fn with_agent_id(mut self, agent_id: impl Into<String>) -> Self {
        self.agent_id = Some(agent_id.into());
        self
    }

    /// Filter by status
    pub fn with_status(mut self, status: SessionStatus) -> Self {
        self.status = Some(status);
        self
    }

    /// Filter by channel
    pub fn with_channel(mut self, channel: impl Into<String>) -> Self {
        self.channel = Some(channel.into());
        self
    }

    /// Filter by user ID
    pub fn with_user_id(mut self, user_id: impl Into<String>) -> Self {
        self.user_id = Some(user_id.into());
        self
    }

    /// Set result limit
    pub fn with_limit(mut self, limit: usize) -> Self {
        self.limit = Some(limit);
        self
    }

    /// Set result offset
    pub fn with_offset(mut self, offset: usize) -> Self {
        self.offset = Some(offset);
        self
    }

    /// Check if a session matches this filter
    fn matches(&self, session: &Session) -> bool {
        if let Some(ref agent_id) = self.agent_id {
            if session.metadata.agent_id.as_ref() != Some(agent_id) {
                return false;
            }
        }
        if let Some(status) = self.status {
            if session.status != status {
                return false;
            }
        }
        if let Some(ref channel) = self.channel {
            if session.metadata.channel.as_ref() != Some(channel) {
                return false;
            }
        }
        if let Some(ref user_id) = self.user_id {
            if session.metadata.user_id.as_ref() != Some(user_id) {
                return false;
            }
        }
        if let Some(created_after) = self.created_after {
            if session.created_at < created_after {
                return false;
            }
        }
        if let Some(created_before) = self.created_before {
            if session.created_at > created_before {
                return false;
            }
        }
        if let Some(updated_after) = self.updated_after {
            if session.updated_at < updated_after {
                return false;
            }
        }
        if let Some(updated_before) = self.updated_before {
            if session.updated_at > updated_before {
                return false;
            }
        }
        true
    }

    fn requires_decrypted_metadata(&self) -> bool {
        self.agent_id.is_some()
            || self.status.is_some()
            || self.channel.is_some()
            || self.user_id.is_some()
            || self.created_after.is_some()
            || self.created_before.is_some()
            || self.updated_after.is_some()
            || self.updated_before.is_some()
    }
}

/// In-memory session cache entry
#[derive(Debug)]
struct CachedSession {
    session: Session,
    dirty: bool,
    history_file_current_confirmed: bool,
}

struct CachedHistoryHmac {
    state: super::integrity::AppendHmacState,
    file_len: u64,
    modified_at_ms: Option<i64>,
    hmac_sidecar_sha256: Option<[u8; 32]>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct CachedArtifactSignature {
    file_len: u64,
    modified_at_ms: Option<i64>,
    hmac_sidecar_sha256: Option<[u8; 32]>,
}

impl std::fmt::Debug for CachedHistoryHmac {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CachedHistoryHmac")
            .field("state", &"<redacted>")
            .field("file_len", &self.file_len)
            .field("modified_at_ms", &self.modified_at_ms)
            .finish()
    }
}

/// Thread-safe session store with file-based persistence
#[derive(Debug)]
pub struct SessionStore {
    /// Base path for session storage
    base_path: PathBuf,
    /// In-memory session cache
    sessions: RwLock<HashMap<String, CachedSession>>,
    /// Session key to ID mapping
    key_to_id: RwLock<HashMap<String, String>>,
    /// Compaction threshold
    compact_threshold: usize,
    /// Optional HMAC key for session integrity verification.
    hmac_key: Option<Zeroizing<[u8; 32]>>,
    /// Action to take when integrity verification fails.
    integrity_action: super::integrity::IntegrityAction,
    /// Session encryption mode.
    encryption_mode: super::crypto::EncryptionMode,
    /// Optional session crypto context for encrypted session artifacts.
    crypto: Option<Arc<super::crypto::SessionCryptoContext>>,
    /// Locked-session stubs discovered by the store-owned disk scan.
    locked_session_entries: RwLock<HashMap<String, SessionListEntry>>,
    /// Store-owned rolling HMAC state for history append fast paths.
    history_hmac_states: RwLock<HashMap<String, CachedHistoryHmac>>,
    /// Process-local protected-history Matrix inbound dedupe cache.
    protected_inbound_dedupe_cache: RwLock<HashMap<String, ProtectedInboundDedupeCache>>,
    /// Process-local archive currentness confirmations keyed by stable file signature.
    archive_current_signatures: RwLock<HashMap<String, CachedArtifactSignature>>,
    #[cfg(test)]
    history_current_file_read_count: Arc<AtomicUsize>,
    #[cfg(test)]
    archive_current_file_read_count: Arc<AtomicUsize>,
}

impl Default for SessionStore {
    fn default() -> Self {
        Self::new()
    }
}

impl SessionStore {
    /// Create a new session store with default path (~/.config/carapace/sessions/)
    pub fn new() -> Self {
        let base_path = dirs::config_dir()
            .map(|p| p.join("carapace").join("sessions"))
            .unwrap_or_else(|| PathBuf::from(".config/carapace/sessions"));
        Self::with_base_path(base_path)
    }

    /// Create a session store with a custom base path
    pub fn with_base_path(base_path: PathBuf) -> Self {
        Self {
            base_path,
            sessions: RwLock::new(HashMap::new()),
            key_to_id: RwLock::new(HashMap::new()),
            compact_threshold: DEFAULT_COMPACT_THRESHOLD,
            hmac_key: None,
            integrity_action: super::integrity::IntegrityAction::Warn,
            encryption_mode: super::crypto::EncryptionMode::Off,
            crypto: None,
            locked_session_entries: RwLock::new(HashMap::new()),
            history_hmac_states: RwLock::new(HashMap::new()),
            protected_inbound_dedupe_cache: RwLock::new(HashMap::new()),
            archive_current_signatures: RwLock::new(HashMap::new()),
            #[cfg(test)]
            history_current_file_read_count: Arc::new(AtomicUsize::new(0)),
            #[cfg(test)]
            archive_current_file_read_count: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// Set the compaction threshold
    pub fn with_compact_threshold(mut self, threshold: usize) -> Self {
        self.compact_threshold = threshold;
        self
    }

    /// Set the HMAC key for session integrity verification.
    pub fn with_hmac_key(mut self, key: Zeroizing<[u8; 32]>) -> Self {
        self.hmac_key = Some(key);
        self
    }

    /// Set the action to take when integrity verification fails.
    pub fn with_integrity_action(mut self, action: super::integrity::IntegrityAction) -> Self {
        self.integrity_action = action;
        self
    }

    /// Set the session encryption mode.
    pub fn with_encryption_mode(mut self, mode: super::crypto::EncryptionMode) -> Self {
        self.encryption_mode = mode;
        self
    }

    /// Set the session crypto context.
    pub fn with_crypto_context(mut self, crypto: Arc<super::crypto::SessionCryptoContext>) -> Self {
        self.crypto = Some(crypto);
        self
    }

    /// Ensure the base directory exists
    fn ensure_base_dir(&self) -> Result<(), SessionStoreError> {
        if !self.base_path.exists() {
            fs::create_dir_all(&self.base_path)?;
        }
        Ok(())
    }

    fn encryption_active(&self) -> bool {
        self.crypto.is_some() && self.encryption_mode.uses_encryption()
    }

    fn new_cached_session(&self, session: Session, dirty: bool) -> CachedSession {
        CachedSession {
            session,
            dirty,
            history_file_current_confirmed: false,
        }
    }

    fn history_file_current_confirmed(&self, session_id: &str) -> bool {
        self.encryption_active()
            && self
                .sessions
                .read()
                .get(session_id)
                .is_some_and(|cached| cached.history_file_current_confirmed)
    }

    fn mark_history_file_current(&self, session_id: &str) {
        let current = self.encryption_active();
        let mut sessions = self.sessions.write();
        if let Some(cached) = sessions.get_mut(session_id) {
            cached.history_file_current_confirmed = current;
        }
    }

    fn lock_message(reason: impl Into<String>) -> SessionStoreError {
        SessionStoreError::Locked(reason.into())
    }

    // SECURITY: not a credential — this returns a typed
    // `SessionStoreError::Locked` variant whose payload is a static
    // descriptive string telling the operator that the config
    // password (the unlock key supplied at daemon startup) is
    // required to decrypt the session store. The word "password"
    // refers to the OPERATOR INPUT needed to unlock, not to any
    // stored credential. CodeQL's `rust/cleartext-logging` rule
    // matches on the token "password" in this function's name +
    // return string; the alert is a false positive (the value is
    // never a credential and is not logged in cleartext anywhere —
    // it flows only through typed `Result` propagation and operator-
    // facing error responses). Dismiss the alert in the code-
    // scanning UI citing this comment.
    fn session_locked_without_password() -> SessionStoreError {
        Self::lock_message("session is encrypted and unavailable without the config password")
    }

    fn ensure_required_encryption_available(&self) -> Result<(), SessionStoreError> {
        if matches!(
            self.encryption_mode,
            super::crypto::EncryptionMode::Required
        ) && self.crypto.is_none()
        {
            return Err(Self::lock_message(
                "session encryption is required; provide the config password to access sessions",
            ));
        }
        Ok(())
    }

    fn encode_session_metadata(&self, session: &Session) -> Result<Vec<u8>, SessionStoreError> {
        if let Some(crypto) = self.crypto.as_ref().filter(|_| self.encryption_active()) {
            return crypto
                .encrypt_json(&session.id, SESSION_METADATA_PURPOSE, session)
                .map_err(Into::into);
        }
        serde_json::to_vec_pretty(session).map_err(Into::into)
    }

    fn decode_session_metadata(
        &self,
        session_id: &str,
        content: &[u8],
    ) -> Result<Session, SessionStoreError> {
        if super::crypto::has_encrypted_payload_prefix(content) {
            let Some(crypto) = self.crypto.as_ref() else {
                return Err(Self::session_locked_without_password());
            };
            return crypto
                .decrypt_json(session_id, SESSION_METADATA_PURPOSE, content)
                .map_err(Into::into);
        }
        if self.encryption_active() {
            return Err(Self::lock_message(
                "session metadata is not encrypted; current session encryption requires encrypted artifacts",
            ));
        }
        serde_json::from_slice(content).map_err(Into::into)
    }

    fn encode_history_message(&self, message: &ChatMessage) -> Result<Vec<u8>, SessionStoreError> {
        if let Some(crypto) = self.crypto.as_ref().filter(|_| self.encryption_active()) {
            return crypto
                .encrypt_json(&message.session_id, SESSION_HISTORY_PURPOSE, message)
                .map_err(Into::into);
        }
        serde_json::to_vec(message).map_err(Into::into)
    }

    fn decode_history_message(
        &self,
        session_id: &str,
        line: &[u8],
        encrypted_line: bool,
    ) -> Result<ChatMessage, SessionStoreError> {
        if encrypted_line {
            let Some(crypto) = self.crypto.as_ref() else {
                return Err(Self::session_locked_without_password());
            };
            return crypto
                .decrypt_json(session_id, SESSION_HISTORY_PURPOSE, line)
                .map_err(Into::into);
        }
        if self.encryption_active() {
            return Err(Self::lock_message(
                "session history is not encrypted; current session encryption requires encrypted artifacts",
            ));
        }
        serde_json::from_slice(line).map_err(Into::into)
    }

    fn encode_archive(
        &self,
        session_id: &str,
        archive: &ArchivedSession,
    ) -> Result<Vec<u8>, SessionStoreError> {
        if let Some(crypto) = self.crypto.as_ref().filter(|_| self.encryption_active()) {
            return crypto
                .encrypt_json(session_id, SESSION_ARCHIVE_PURPOSE, archive)
                .map_err(Into::into);
        }
        serde_json::to_vec_pretty(archive).map_err(Into::into)
    }

    fn decode_archive(
        &self,
        session_id: &str,
        content: &[u8],
    ) -> Result<ArchivedSession, SessionStoreError> {
        let archive: ArchivedSession = if super::crypto::has_encrypted_payload_prefix(content) {
            let Some(crypto) = self.crypto.as_ref() else {
                return Err(Self::session_locked_without_password());
            };
            crypto
                .decrypt_json(session_id, SESSION_ARCHIVE_PURPOSE, content)
                .map_err(SessionStoreError::from)?
        } else {
            if self.encryption_active() {
                return Err(Self::lock_message(
                    "session archive is not encrypted; current session encryption requires encrypted artifacts",
                ));
            }
            serde_json::from_slice(content).map_err(SessionStoreError::from)?
        };
        // SECURITY: refuse to load a session archive written by
        // a newer daemon. A bare downgrade would drop unknown fields,
        // the next re-archive would clobber the newer-version file
        // with v1 bytes, and operator-visible session history could
        // lose schema-required state silently. Mirror the same guard
        // applied to gateway/devices/nodes stores.
        if archive.version > ARCHIVED_SESSION_VERSION {
            return Err(SessionStoreError::Serialization(format!(
                "session archive {session_id} was written by a newer daemon (file version {}, this binary supports up to {ARCHIVED_SESSION_VERSION}); upgrade Carapace before continuing",
                archive.version,
            )));
        }
        Ok(archive)
    }

    fn history_integrity_error(
        session_id: &str,
        operation: &'static str,
        err: SessionStoreError,
    ) -> SessionStoreError {
        match err {
            SessionStoreError::IntegrityRejected(message) => SessionStoreError::HistoryCorrupt(
                format!(
                    "session history integrity verification failed during {operation} for session {session_id}: {message}"
                ),
            ),
            other => other,
        }
    }

    fn encrypted_history_decode_error(
        session_id: &str,
        operation: &'static str,
        encrypted_line: bool,
        err: SessionStoreError,
    ) -> SessionStoreError {
        match err {
            SessionStoreError::Locked(message) => SessionStoreError::Locked(message),
            SessionStoreError::DecryptionFailed(message) => SessionStoreError::HistoryCorrupt(
                format!(
                    "encrypted session history authentication failed during {operation} for session {session_id}: {message}"
                ),
            ),
            SessionStoreError::ManifestIntegrityFailed(message) => {
                SessionStoreError::HistoryCorrupt(format!(
                    "encrypted session manifest integrity failed during {operation} for session {session_id}: {message}"
                ))
            }
            other if encrypted_line => SessionStoreError::HistoryCorrupt(format!(
                "invalid encrypted session history line during {operation} for session {session_id}: {other}"
            )),
            other => other,
        }
    }

    fn archive_read_error(
        session_id: &str,
        operation: &'static str,
        err: SessionStoreError,
    ) -> SessionStoreError {
        match err {
            SessionStoreError::Locked(message) => SessionStoreError::Locked(message),
            SessionStoreError::DecryptionFailed(message) => SessionStoreError::HistoryCorrupt(
                format!(
                    "encrypted session archive authentication failed during {operation} for session {session_id}: {message}"
                ),
            ),
            SessionStoreError::ManifestIntegrityFailed(message) => {
                SessionStoreError::HistoryCorrupt(format!(
                    "encrypted session manifest integrity failed during {operation} for session {session_id}: {message}"
                ))
            }
            SessionStoreError::IntegrityRejected(message) => SessionStoreError::HistoryCorrupt(
                format!(
                    "session archive integrity verification failed during {operation} for session {session_id}: {message}"
                ),
            ),
            other => other,
        }
    }

    fn file_updated_at(path: &Path) -> Option<i64> {
        fs::metadata(path)
            .ok()
            .and_then(|meta| meta.modified().ok())
            .and_then(|ts| ts.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|dur| i64::try_from(dur.as_millis()).unwrap_or(i64::MAX))
    }

    fn record_locked_session_entry(&self, session_id: String, updated_at: Option<i64>) {
        self.locked_session_entries.write().insert(
            session_id.clone(),
            SessionListEntry::locked(session_id, updated_at),
        );
    }

    fn clear_locked_session_entry(&self, session_id: &str) {
        self.locked_session_entries.write().remove(session_id);
    }

    fn replace_locked_session_entries(&self, entries: HashMap<String, SessionListEntry>) {
        *self.locked_session_entries.write() = entries;
    }

    fn locked_session_count(&self) -> usize {
        self.locked_session_entries.read().len()
    }

    fn locked_session_entries_snapshot(&self) -> Vec<SessionListEntry> {
        self.locked_session_entries
            .read()
            .values()
            .cloned()
            .collect()
    }

    #[cfg_attr(not(test), allow(dead_code))]
    fn create_private_output_file(path: &Path) -> Result<File, SessionStoreError> {
        let mut options = OpenOptions::new();
        options.write(true).create(true).truncate(true);

        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;

            options.mode(0o600);
            Ok(options.open(path)?)
        }

        #[cfg(not(unix))]
        {
            Ok(options.open(path)?)
        }
    }

    /// Open a fresh tmp file with `create_new(true)` so a leftover tmp
    /// from a crashed write doesn't get silently overwritten. On
    /// `AlreadyExists`, sweep the stale tmp once and retry — matching
    /// the contract used by `persist_config_file_locked` for the
    /// config-write path.
    fn create_private_output_file_create_new(path: &Path) -> Result<File, SessionStoreError> {
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            // O_NOFOLLOW defense-in-depth: `O_EXCL` (create_new) already
            // refuses a planted symlink at the tmp path today, but
            // O_NOFOLLOW guards against a future refactor that weakens
            // create_new. Mirrors `paths::create_atomic_tmp_owner_only`.
            options.mode(0o600).custom_flags(libc::O_NOFOLLOW);
        }
        match options.open(path) {
            Ok(file) => Ok(file),
            Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
                // Best-effort sweep + one retry. A persistent
                // collision after sweep surfaces as AlreadyExists from
                // the second open and the caller sees that error.
                let _ = std::fs::remove_file(path);
                Ok(options.open(path)?)
            }
            Err(err) => Err(err.into()),
        }
    }

    fn open_private_append_file(path: &Path) -> Result<File, SessionStoreError> {
        let mut options = OpenOptions::new();
        options.create(true).append(true);

        #[cfg(unix)]
        {
            use std::os::unix::fs::{FileTypeExt, OpenOptionsExt, PermissionsExt};

            // SECURITY: O_NOFOLLOW + O_NONBLOCK + post-open file_type
            // revalidation. Session history files (history.jsonl)
            // carry plaintext message bytes — user prompts, assistant
            // content, tool arguments. Without O_NOFOLLOW a same-uid
            // attacker could pre-plant a symlink at the history path
            // and redirect append writes to an arbitrary daemon-uid-
            // writable file (information disclosure). The HMAC sidecar
            // catches the tampering on next READ, but the bytes have
            // already landed at the attacker-chosen target by then.
            // O_NONBLOCK additionally prevents `O_WRONLY | O_CREAT |
            // O_APPEND` from blocking when the dirent is a planted
            // FIFO with no reader — the post-open `is_fifo()` refusal
            // below only runs after open(2) returns. Companion to
            // the live-DLQ append hardening shipped in Batch 39.
            options
                .mode(0o600)
                .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
            let file = options.open(path)?;
            let opened_metadata = file.metadata()?;
            let file_type = opened_metadata.file_type();
            if !file_type.is_file()
                || file_type.is_symlink()
                || file_type.is_fifo()
                || file_type.is_socket()
                || file_type.is_block_device()
                || file_type.is_char_device()
            {
                return Err(SessionStoreError::Io(format!(
                    "session history path is not a regular file: {}",
                    path.display()
                )));
            }
            file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
            Ok(file)
        }

        #[cfg(not(unix))]
        {
            Ok(options.open(path)?)
        }
    }

    fn trim_ascii_whitespace(bytes: &[u8]) -> &[u8] {
        let start = bytes
            .iter()
            .position(|byte| !byte.is_ascii_whitespace())
            .unwrap_or(bytes.len());
        let end = bytes
            .iter()
            .rposition(|byte| !byte.is_ascii_whitespace())
            .map(|index| index + 1)
            .unwrap_or(start);
        &bytes[start..end]
    }

    fn read_verified_history_bytes(
        &self,
        history_path: &Path,
    ) -> Result<Vec<u8>, SessionStoreError> {
        // SECURITY: cap the history-file read. Conversation history
        // can grow large for long-running agents but is bounded by
        // application-level retention; a same-uid attacker planting
        // a multi-GB file at the history path otherwise OOMs the
        // daemon on the next history load. 256 MiB is the cap —
        // generous for legitimate long-history sessions, refuses
        // planted-multi-GB attack vector.
        let content = match crate::paths::read_to_vec_no_hang_no_follow_capped(
            history_path,
            SESSION_HISTORY_FILE_MAX_BYTES,
        ) {
            Ok(Some(content)) => content,
            Ok(None) => {
                return Err(SessionStoreError::Io(format!(
                    "history file not found at {}",
                    history_path.display()
                )))
            }
            Err(err) => return Err(SessionStoreError::Io(err.to_string())),
        };
        self.verify_integrity_bytes(&content, history_path)?;
        Ok(content)
    }

    fn verify_integrity_bytes(
        &self,
        content: &[u8],
        file_path: &Path,
    ) -> Result<(), SessionStoreError> {
        if self.encrypted_artifact_locked_without_crypto_bytes(content) {
            return Err(Self::session_locked_without_password());
        }
        let Some(ref key) = self.hmac_key else {
            return Ok(());
        };
        let integrity_config = super::integrity::IntegrityConfig {
            enabled: true,
            action: self.integrity_action,
        };
        self.handle_integrity_result(
            super::integrity::verify_hmac_file(key, content, file_path, &integrity_config),
            file_path,
        )
    }

    fn verify_integrity_path(&self, file_path: &Path) -> Result<(), SessionStoreError> {
        let Some(ref key) = self.hmac_key else {
            return Ok(());
        };
        let integrity_config = super::integrity::IntegrityConfig {
            enabled: true,
            action: self.integrity_action,
        };
        self.handle_integrity_result(
            super::integrity::verify_hmac_path(key, file_path, &integrity_config),
            file_path,
        )
    }

    fn handle_integrity_result(
        &self,
        result: Result<(), super::integrity::IntegrityError>,
        file_path: &Path,
    ) -> Result<(), SessionStoreError> {
        match result {
            Ok(()) => Ok(()),
            Err(super::integrity::IntegrityError::Rejected { .. }) => {
                Err(SessionStoreError::IntegrityRejected(format!(
                    "session integrity verification failed for {}",
                    file_path.display()
                )))
            }
            Err(err) => {
                if matches!(
                    self.integrity_action,
                    super::integrity::IntegrityAction::Reject
                ) {
                    tracing::warn!(
                        error_kind = integrity_error_kind(&err),
                        "session integrity verification issue"
                    );
                    return Err(SessionStoreError::IntegrityRejected(format!(
                        "session integrity verification failed for {}",
                        file_path.display()
                    )));
                }
                tracing::warn!(
                    error_kind = integrity_error_kind(&err),
                    "session integrity verification issue"
                );
                Ok(())
            }
        }
    }

    fn should_block_unknown_session_key_without_crypto(&self) -> bool {
        self.crypto.is_none() && self.locked_session_count() > 0
    }

    fn encrypted_artifact_locked_without_crypto_bytes(&self, content: &[u8]) -> bool {
        self.crypto.is_none() && super::crypto::is_encrypted_payload(content)
    }

    fn locked_filter_error(&self) -> SessionStoreError {
        if self.crypto.is_none() && self.locked_session_count() > 0 {
            return Self::lock_message(
                "encrypted sessions are present; provide the config password to apply metadata-based session filters",
            );
        }
        Self::lock_message(
            "unavailable sessions are present; repair or unlock them to apply metadata-based session filters",
        )
    }

    fn rewrite_history_file_from_messages(
        &self,
        history_path: &Path,
        session_id: &str,
        messages: &[ChatMessage],
    ) -> Result<(), SessionStoreError> {
        let temp_path = crate::paths::atomic_tmp_path(history_path, "jsonl");
        let write_result = (|| -> Result<(), SessionStoreError> {
            {
                let file = Self::create_private_output_file_create_new(&temp_path)?;
                let mut writer = BufWriter::new(file);
                for message in messages {
                    let encoded = self.encode_history_message(message)?;
                    writer.write_all(&encoded)?;
                    writeln!(writer)?;
                }
                writer.flush()?;
                writer
                    .into_inner()
                    .map_err(|e| std::io::Error::other(e.to_string()))?
                    .sync_all()?;
            }
            let hmac_state =
                self.prepare_history_hmac_for_path(&temp_path, history_path, session_id)?;
            fs::rename(&temp_path, history_path)?;
            #[cfg(test)]
            if TEST_FORCE_HISTORY_PARENT_SYNC_AFTER_RENAME_FAIL.swap(false, Ordering::SeqCst) {
                return Err(SessionStoreError::Io(
                    "forced history parent fsync failure after rename".to_string(),
                ));
            }
            crate::paths::sync_parent_dir_blocking(history_path)?;
            self.commit_history_hmac(history_path, session_id)?;
            if let Some(state) = hmac_state {
                self.store_history_hmac_state(session_id, history_path, state);
            } else {
                self.clear_history_hmac_state(session_id);
            }
            self.mark_history_file_current(session_id);
            self.invalidate_protected_inbound_dedupe_cache(session_id);
            Ok(())
        })();
        if write_result.is_err() {
            let _ = fs::remove_file(&temp_path);
        }
        write_result
    }

    fn read_raw_history_messages(
        &self,
        history_path: &Path,
        session_id: &str,
    ) -> Result<Vec<ChatMessage>, SessionStoreError> {
        if !history_path.exists() {
            return Ok(Vec::new());
        }
        let history_bytes = self
            .read_verified_history_bytes(history_path)
            .map_err(|err| {
                Self::history_integrity_error(session_id, "history rewrite read", err)
            })?;
        let mut messages = Vec::new();
        for raw_line in history_bytes.split(|byte| *byte == b'\n') {
            let line = Self::trim_ascii_whitespace(raw_line);
            if line.is_empty() {
                continue;
            }
            let encrypted_line = super::crypto::has_encrypted_payload_prefix(line);
            match self.decode_history_message(session_id, line, encrypted_line) {
                Ok(message) => messages.push(message),
                Err(SessionStoreError::Locked(message)) => {
                    return Err(SessionStoreError::Locked(message));
                }
                Err(err) => {
                    let err = Self::encrypted_history_decode_error(
                        session_id,
                        "history rewrite read",
                        encrypted_line,
                        err,
                    );
                    if err.is_permanent_history_corruption()
                        || matches!(err, SessionStoreError::Locked(_))
                    {
                        return Err(err);
                    }
                    tracing::warn!(
                        error_kind = "invalid_jsonl",
                        "skipping corrupt JSONL line in session history rewrite source"
                    );
                }
            }
        }
        Ok(messages)
    }

    fn inbound_dedupe_entry_from_message(
        message: &ChatMessage,
    ) -> Option<(String, InboundEventIndexEntry)> {
        let metadata = message.metadata.as_ref()?;
        let stored_event_id = metadata_string(
            metadata,
            crate::channels::inbound::INBOUND_EVENT_ID_META_KEY,
        )?;
        let canonical_event_id = canonical_inbound_event_id(&stored_event_id);
        let entry = InboundEventIndexEntry {
            event_id: stored_event_id,
            run_id: metadata_string(metadata, crate::channels::inbound::INBOUND_RUN_ID_META_KEY),
        };
        Some((canonical_event_id, entry))
    }

    fn collect_inbound_dedupe_entries<'a>(
        messages: impl IntoIterator<Item = &'a ChatMessage>,
    ) -> BTreeMap<String, InboundEventIndexEntry> {
        let mut entries = BTreeMap::new();
        for message in messages {
            if let Some((canonical_event_id, entry)) =
                Self::inbound_dedupe_entry_from_message(message)
            {
                entries.entry(canonical_event_id).or_insert(entry);
            }
        }
        entries
    }

    fn rewrite_history_file_preserving_inbound_dedupe(
        &self,
        history_path: &Path,
        session_id: &str,
        mut rewritten_messages: Vec<ChatMessage>,
        raw_history_messages: &[ChatMessage],
        discarded_visible_messages: &[ChatMessage],
    ) -> Result<(), SessionStoreError> {
        let existing_entries = Self::collect_inbound_dedupe_entries(rewritten_messages.iter());
        let mut preserved_entries = Self::collect_inbound_dedupe_entries(
            raw_history_messages
                .iter()
                .filter(|message| Self::is_protected_inbound_dedupe_marker(message))
                .chain(discarded_visible_messages.iter()),
        );
        for canonical_event_id in existing_entries.keys() {
            preserved_entries.remove(canonical_event_id);
        }
        // Compaction appends preserved/synthesized dedupe markers after the
        // visible rewritten history. Inbound-event lookup is keyed by
        // canonical event id, so marker order is intentionally
        // order-independent across compacted history rewrites.
        rewritten_messages.extend(
            preserved_entries
                .values()
                .map(|entry| Self::protected_inbound_dedupe_marker(session_id, entry)),
        );
        self.rewrite_history_file_from_messages(history_path, session_id, &rewritten_messages)
    }

    fn ensure_history_file_current(
        &self,
        history_path: &Path,
        session_id: &str,
    ) -> Result<(), SessionStoreError> {
        if !self.encryption_active() || self.history_file_current_confirmed(session_id) {
            return Ok(());
        }
        if !history_path.exists() {
            self.mark_history_file_current(session_id);
            return Ok(());
        }

        #[cfg(test)]
        self.history_current_file_read_count
            .fetch_add(1, Ordering::SeqCst);

        self.verify_integrity_path(history_path)?;
        // O_NOFOLLOW + O_NONBLOCK + fstat-validate: between
        // verify_integrity_path's open and this re-open of the same
        // path, a same-uid attacker who swaps the dirent for a
        // symlink-to-FIFO would otherwise hang here during open(2).
        // The post-open is_file() check on the held fd is the only
        // thing that catches a planted FIFO.
        let history_file = match crate::paths::open_regular_file_no_hang_no_follow(history_path)? {
            Some(file) => file,
            None => {
                return Err(SessionStoreError::Io(format!(
                    "history file disappeared between integrity verify and read at {}",
                    history_path.display()
                )));
            }
        };
        Self::ensure_history_reader_has_only_encrypted_payload_lines(history_file)?;
        self.mark_history_file_current(session_id);
        Ok(())
    }

    fn ensure_history_reader_has_only_encrypted_payload_lines<R: Read>(
        history_file: R,
    ) -> Result<(), SessionStoreError> {
        let prefix = super::crypto::encrypted_payload_prefix();
        let mut reader = BufReader::new(history_file);
        let mut buffer = [0_u8; 8192];
        let mut line_started = false;
        let mut prefix_index = 0_usize;
        let mut prefix_checked = false;

        loop {
            let read = reader.read(&mut buffer)?;
            if read == 0 {
                if line_started && !prefix_checked {
                    return Err(Self::lock_message(
                        "session history is not encrypted; current session encryption requires encrypted artifacts",
                    ));
                }
                return Ok(());
            }

            for byte in &buffer[..read] {
                if *byte == b'\n' {
                    if line_started && !prefix_checked {
                        return Err(Self::lock_message(
                            "session history is not encrypted; current session encryption requires encrypted artifacts",
                        ));
                    }
                    line_started = false;
                    prefix_index = 0;
                    prefix_checked = false;
                    continue;
                }

                if !line_started {
                    if byte.is_ascii_whitespace() {
                        continue;
                    }
                    line_started = true;
                }

                if !prefix_checked {
                    if prefix.get(prefix_index) != Some(byte) {
                        return Err(Self::lock_message(
                            "session history is not encrypted; current session encryption requires encrypted artifacts",
                        ));
                    }
                    prefix_index += 1;
                    if prefix_index == prefix.len() {
                        prefix_checked = true;
                    }
                }
            }
        }
    }

    fn ensure_archive_file_current(&self, session_id: &str) -> Result<(), SessionStoreError> {
        if !self.encryption_active() {
            return Ok(());
        }

        let archive_path = self.archive_path(session_id)?;
        if !archive_path.exists() {
            self.clear_archive_file_current(session_id);
            return Ok(());
        }

        let _lock =
            FileLock::acquire(&archive_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        let initial_signature = Self::artifact_file_signature(&archive_path)
            .ok_or_else(|| SessionStoreError::Io("failed to read archive metadata".to_string()))?;
        if self.archive_file_current_confirmed(session_id, initial_signature) {
            return Ok(());
        }

        #[cfg(test)]
        self.archive_current_file_read_count
            .fetch_add(1, Ordering::SeqCst);

        self.verify_integrity_path(&archive_path)?;
        if !Self::file_starts_with_encrypted_payload_prefix(&archive_path)? {
            return Err(Self::lock_message(
                "session archive is not encrypted; current session encryption requires encrypted artifacts",
            ));
        }
        let current_signature = Self::artifact_file_signature(&archive_path)
            .ok_or_else(|| SessionStoreError::Io("failed to read archive metadata".to_string()))?;
        self.mark_archive_file_current(session_id, current_signature);
        Ok(())
    }

    fn file_starts_with_encrypted_payload_prefix(path: &Path) -> Result<bool, SessionStoreError> {
        let prefix = super::crypto::encrypted_payload_prefix();
        // O_NOFOLLOW + O_NONBLOCK: archive-verify reads happen during
        // session-resume / archive-restore. A planted FIFO at the
        // archive path would otherwise hang open(2).
        let mut file = match crate::paths::open_regular_file_no_hang_no_follow(path)? {
            Some(file) => file,
            None => return Ok(false),
        };
        let mut buffer = vec![0_u8; prefix.len()];
        match file.read_exact(&mut buffer) {
            Ok(()) => Ok(buffer == prefix),
            Err(err) if err.kind() == ErrorKind::UnexpectedEof => Ok(false),
            Err(err) => Err(err.into()),
        }
    }

    fn ensure_session_artifacts_current(&self, session: &Session) -> Result<(), SessionStoreError> {
        if !self.encryption_active() {
            return Ok(());
        }

        let history_path = self.session_history_path(&session.id)?;
        if history_path.exists() {
            let _history_lock = FileLock::acquire(&history_path)
                .map_err(|e| SessionStoreError::Io(e.to_string()))?;
            self.ensure_history_file_current(&history_path, &session.id)?;
        }
        self.ensure_archive_file_current(&session.id)
    }

    /// Validate session_id to prevent path traversal attacks.
    /// Session IDs must be valid UUIDs or alphanumeric slugs (letters, numbers, hyphens, underscores).
    fn validate_session_id(session_id: &str) -> Result<(), SessionStoreError> {
        // Reject empty IDs
        if session_id.is_empty() {
            return Err(SessionStoreError::InvalidSessionKey(
                "session_id cannot be empty".to_string(),
            ));
        }

        // Reject path traversal attempts
        if session_id.contains("..") || session_id.contains('/') || session_id.contains('\\') {
            return Err(SessionStoreError::InvalidSessionKey(
                "invalid session_id (path traversal detected)".to_string(),
            ));
        }

        // Allow only safe characters: alphanumeric, hyphens, underscores
        // This covers UUIDs (with hyphens) and typical slug formats
        if !session_id
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
        {
            return Err(SessionStoreError::InvalidSessionKey(
                "invalid session_id (must be alphanumeric with hyphens/underscores)".to_string(),
            ));
        }

        Ok(())
    }

    /// Get the metadata file path for a session (validates session_id)
    fn session_meta_path(&self, session_id: &str) -> Result<PathBuf, SessionStoreError> {
        Self::validate_session_id(session_id)?;
        Ok(self.base_path.join(format!("{}.json", session_id)))
    }

    /// Get the history file path for a session (validates session_id)
    pub(crate) fn session_history_path(
        &self,
        session_id: &str,
    ) -> Result<PathBuf, SessionStoreError> {
        Self::validate_session_id(session_id)?;
        Ok(self.base_path.join(format!("{}.jsonl", session_id)))
    }

    fn session_inbound_event_index_path(
        &self,
        session_id: &str,
    ) -> Result<PathBuf, SessionStoreError> {
        Self::validate_session_id(session_id)?;
        Ok(self
            .base_path
            .join(format!("{}.inbound-events.jsonl", session_id)))
    }

    fn session_key_lock_path(&self, session_key: &str) -> PathBuf {
        let mut hasher = Sha256::new();
        hasher.update(session_key.as_bytes());
        let digest = hasher.finalize();
        self.base_path
            .join(format!("session-key-{}", hex::encode(digest)))
    }

    fn acquire_session_key_lock(&self, session_key: &str) -> Result<FileLock, SessionStoreError> {
        let lock_path = self.session_key_lock_path(session_key);
        FileLock::acquire(&lock_path).map_err(|e| SessionStoreError::Io(e.to_string()))
    }

    fn session_key_exists(&self, session_key: &str) -> Result<bool, SessionStoreError> {
        if self.key_to_id.read().contains_key(session_key) {
            return Ok(true);
        }

        self.load_sessions_from_disk()?;
        if self.key_to_id.read().contains_key(session_key) {
            return Ok(true);
        }

        if self.should_block_unknown_session_key_without_crypto() {
            return Err(Self::lock_message(
                "encrypted sessions are present; provide the config password before creating or resolving sessions by key",
            ));
        }

        Ok(false)
    }

    fn verify_archive_integrity(
        &self,
        archive_path: &Path,
        archive_bytes: &[u8],
    ) -> Result<(), SessionStoreError> {
        self.verify_integrity_bytes(archive_bytes, archive_path)
    }

    fn hmac_sidecar_path(path: &Path) -> PathBuf {
        let mut sidecar = path.as_os_str().to_owned();
        sidecar.push(".hmac");
        PathBuf::from(sidecar)
    }

    fn hmac_sidecar_sha256(path: &Path) -> Option<[u8; 32]> {
        let sidecar = Self::hmac_sidecar_path(path);
        // Hardened read: O_NOFOLLOW + regular-file fstat + byte cap.
        // Without this, a same-uid attacker who plants a symlink or
        // FIFO at `.hmac` redirects the dedupe-cache signature into
        // an attacker-chosen sidecar OR causes the read to block.
        // The HMAC sidecar is a fixed 32-byte tag-only artifact; cap
        // at 4 KiB as conservative headroom for legacy formats /
        // version-tagged shapes without enabling unbounded reads.
        const HMAC_SIDECAR_MAX_BYTES: u64 = 4 * 1024;
        let bytes =
            crate::paths::read_to_vec_no_hang_no_follow_capped(&sidecar, HMAC_SIDECAR_MAX_BYTES)
                .ok()??;
        Some(Sha256::digest(&bytes).into())
    }

    fn artifact_file_signature(path: &Path) -> Option<CachedArtifactSignature> {
        let metadata = fs::metadata(path).ok()?;
        let modified_at_ms = metadata
            .modified()
            .ok()
            .and_then(|ts| ts.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|dur| i64::try_from(dur.as_millis()).unwrap_or(i64::MAX));
        Some(CachedArtifactSignature {
            file_len: metadata.len(),
            modified_at_ms,
            hmac_sidecar_sha256: Self::hmac_sidecar_sha256(path),
        })
    }

    #[cfg(test)]
    fn protected_inbound_dedupe_history_signature(
        history_path: &Path,
    ) -> Option<ProtectedInboundDedupeHistorySignature> {
        Self::protected_inbound_dedupe_history_state(history_path).map(|state| state.signature)
    }

    fn protected_inbound_dedupe_history_state(
        history_path: &Path,
    ) -> Option<ProtectedInboundDedupeHistoryCacheState> {
        #[cfg(test)]
        TEST_PROTECTED_INBOUND_DEDUPE_HISTORY_SIGNATURE_READS.fetch_add(1, Ordering::SeqCst);
        // Hardened read: O_NOFOLLOW + regular-file fstat + 256 MiB
        // cap (matches `SESSION_HISTORY_FILE_MAX_BYTES`). Without
        // this, the dedupe-cache signature path reads history via
        // raw `fs::read`, which follows symlinks, blocks on FIFOs,
        // and has no upper bound — a same-uid attacker who plants
        // a multi-GiB regular file or a symlink at the history
        // path can OOM the dedupe-cache refresh or redirect it.
        // The cap matches the write-side guard so a legitimate
        // file always loads.
        let history_bytes = crate::paths::read_to_vec_no_hang_no_follow_capped(
            history_path,
            SESSION_HISTORY_FILE_MAX_BYTES,
        )
        .ok()??;
        Self::protected_inbound_dedupe_history_state_from_bytes(history_path, &history_bytes)
    }

    fn protected_inbound_dedupe_history_state_from_bytes(
        history_path: &Path,
        history_bytes: &[u8],
    ) -> Option<ProtectedInboundDedupeHistoryCacheState> {
        let file_state = Self::protected_inbound_dedupe_history_file_state(history_path)?;
        if file_state.file_len != u64::try_from(history_bytes.len()).ok()? {
            return None;
        }
        let mut hasher = Sha256::new();
        hasher.update(history_bytes);
        let content_sha256 = hasher.clone().finalize().into();
        Some(ProtectedInboundDedupeHistoryCacheState {
            signature: ProtectedInboundDedupeHistorySignature {
                file_len: file_state.file_len,
                hmac_sidecar_sha256: Self::hmac_sidecar_sha256(history_path),
                content_sha256,
            },
            hasher: ProtectedInboundDedupeHistoryHasher(hasher),
        })
    }

    fn protected_inbound_dedupe_history_file_state(
        history_path: &Path,
    ) -> Option<ProtectedInboundDedupeHistoryFileState> {
        let metadata = fs::metadata(history_path).ok()?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::MetadataExt;
            Some(ProtectedInboundDedupeHistoryFileState {
                file_len: metadata.len(),
                device: metadata.dev(),
                inode: metadata.ino(),
                ctime_sec: metadata.ctime(),
                ctime_nsec: metadata.ctime_nsec(),
            })
        }
        #[cfg(not(unix))]
        {
            let modified_at_ms = metadata
                .modified()
                .ok()
                .and_then(|ts| ts.duration_since(std::time::UNIX_EPOCH).ok())
                .map(|dur| i64::try_from(dur.as_millis()).unwrap_or(i64::MAX));
            Some(ProtectedInboundDedupeHistoryFileState {
                file_len: metadata.len(),
                modified_at_ms,
            })
        }
    }

    fn cached_history_hmac_state_for_append(
        &self,
        session_id: &str,
        history_path: &Path,
    ) -> Option<super::integrity::AppendHmacState> {
        let cached = {
            let states = self.history_hmac_states.read();
            states.get(session_id).map(|entry| {
                (
                    entry.state.clone(),
                    entry.file_len,
                    entry.modified_at_ms,
                    entry.hmac_sidecar_sha256,
                )
            })
        }?;

        if Self::artifact_file_signature(history_path)
            != Some(CachedArtifactSignature {
                file_len: cached.1,
                modified_at_ms: cached.2,
                hmac_sidecar_sha256: cached.3,
            })
        {
            self.clear_history_hmac_state(session_id);
            return None;
        }

        match super::integrity::sidecar_matches_state(history_path, &cached.0) {
            Ok(true) => Some(cached.0),
            Ok(false) | Err(_) => {
                self.clear_history_hmac_state(session_id);
                None
            }
        }
    }

    fn store_history_hmac_state(
        &self,
        session_id: &str,
        history_path: &Path,
        state: super::integrity::AppendHmacState,
    ) {
        let Some(signature) = Self::artifact_file_signature(history_path) else {
            self.clear_history_hmac_state(session_id);
            return;
        };
        let mut states = self.history_hmac_states.write();
        states.insert(
            session_id.to_string(),
            CachedHistoryHmac {
                state,
                file_len: signature.file_len,
                modified_at_ms: signature.modified_at_ms,
                hmac_sidecar_sha256: signature.hmac_sidecar_sha256,
            },
        );
    }

    fn clear_history_hmac_state(&self, session_id: &str) {
        let mut states = self.history_hmac_states.write();
        states.remove(session_id);
    }

    fn protected_inbound_dedupe_cache_lookup(
        &self,
        session_id: &str,
        history_path: &Path,
        canonical_event_id: &str,
    ) -> Option<Option<InboundEventIndexEntry>> {
        // This cache is an optimization owned by SessionStore's
        // per-history FileLock discipline: append_message_if_new_inbound
        // holds the history lock across lookup, legacy-index repair, and
        // append. Mutation paths that rewrite or replace history invalidate
        // the entry instead of trying to keep it current outside that lock.
        let (history_signature, entry) = {
            let cache = self.protected_inbound_dedupe_cache.read();
            let session_cache = cache.get(session_id)?;
            if !session_cache.complete {
                return None;
            }
            (
                session_cache.history_signature.clone(),
                session_cache.entries.get(canonical_event_id).cloned(),
            )
        };

        let Some(history_signature) = history_signature else {
            if !history_path.exists() {
                return Some(entry);
            }
            tracing::debug!(
                session_id = %session_id,
                path = %history_path.display(),
                "invalidating protected inbound dedupe cache because history appeared after a missing-history cache seed"
            );
            self.invalidate_protected_inbound_dedupe_cache(session_id);
            return None;
        };

        let Some(current_state) = Self::protected_inbound_dedupe_history_state(history_path) else {
            tracing::debug!(
                session_id = %session_id,
                path = %history_path.display(),
                "invalidating protected inbound dedupe cache because history identity could not be refreshed"
            );
            self.invalidate_protected_inbound_dedupe_cache(session_id);
            return None;
        };
        if current_state.signature != history_signature {
            self.invalidate_protected_inbound_dedupe_cache(session_id);
            return None;
        }

        {
            let mut cache = self.protected_inbound_dedupe_cache.write();
            if let Some(session_cache) = cache.get_mut(session_id) {
                if session_cache.history_signature.as_ref() == Some(&history_signature) {
                    session_cache.history_hasher = Some(current_state.hasher);
                }
            }
        }
        Some(entry)
    }

    fn replace_protected_inbound_dedupe_cache(
        &self,
        session_id: &str,
        history_state: Option<ProtectedInboundDedupeHistoryCacheState>,
        entries: HashMap<String, InboundEventIndexEntry>,
    ) {
        let mut cache = self.protected_inbound_dedupe_cache.write();
        let content_bound = history_state.is_some() || entries.is_empty();
        let complete = content_bound && entries.len() <= PROTECTED_INBOUND_DEDUPE_CACHE_MAX_ENTRIES;
        cache.insert(
            session_id.to_string(),
            ProtectedInboundDedupeCache {
                complete,
                history_signature: complete
                    .then(|| history_state.as_ref().map(|state| state.signature.clone()))
                    .flatten(),
                history_hasher: complete
                    .then(|| history_state.map(|state| state.hasher))
                    .flatten(),
                entries: if complete { entries } else { HashMap::new() },
            },
        );
    }

    fn cache_protected_inbound_event_from_message(
        &self,
        message: &ChatMessage,
        history_path: &Path,
        appended: &[u8],
    ) {
        if Self::is_protected_inbound_dedupe_marker(message) {
            return;
        }
        let Some(metadata) = message.metadata.as_ref() else {
            return;
        };
        let Some(stored_event_id) = metadata_string(
            metadata,
            crate::channels::inbound::INBOUND_EVENT_ID_META_KEY,
        ) else {
            return;
        };
        let entry = InboundEventIndexEntry {
            event_id: stored_event_id.clone(),
            run_id: metadata_string(metadata, crate::channels::inbound::INBOUND_RUN_ID_META_KEY),
        };
        let canonical_event_id = canonical_inbound_event_id(&stored_event_id);
        let mut cache = self.protected_inbound_dedupe_cache.write();
        if let Some(session_cache) = cache.get_mut(&message.session_id) {
            if session_cache.complete
                && session_cache.entries.len() < PROTECTED_INBOUND_DEDUPE_CACHE_MAX_ENTRIES
            {
                let Some(file_state) =
                    Self::protected_inbound_dedupe_history_file_state(history_path)
                else {
                    tracing::debug!(
                        session_id = %message.session_id,
                        path = %history_path.display(),
                        "invalidating protected inbound dedupe cache because appended history metadata could not be read"
                    );
                    session_cache.complete = false;
                    session_cache.history_signature = None;
                    session_cache.history_hasher = None;
                    session_cache.entries.clear();
                    return;
                };
                let previous_len = session_cache
                    .history_signature
                    .as_ref()
                    .map(|signature| signature.file_len)
                    .unwrap_or(0);
                let base_hasher = match (
                    session_cache.history_signature.as_ref(),
                    session_cache.history_hasher.as_ref(),
                ) {
                    (Some(_), Some(hasher)) => Some(hasher.0.clone()),
                    (None, None) if session_cache.entries.is_empty() => Some(Sha256::new()),
                    _ => None,
                };
                let Some(mut hasher) = base_hasher else {
                    tracing::debug!(
                        session_id = %message.session_id,
                        "invalidating protected inbound dedupe cache because append could not advance cached history hash state"
                    );
                    session_cache.complete = false;
                    session_cache.history_signature = None;
                    session_cache.history_hasher = None;
                    session_cache.entries.clear();
                    return;
                };
                if previous_len.saturating_add(appended.len() as u64) != file_state.file_len {
                    tracing::debug!(
                        session_id = %message.session_id,
                        path = %history_path.display(),
                        previous_len,
                        appended_len = appended.len(),
                        current_len = file_state.file_len,
                        "invalidating protected inbound dedupe cache because appended history length did not match cached state"
                    );
                    session_cache.complete = false;
                    session_cache.history_signature = None;
                    session_cache.history_hasher = None;
                    session_cache.entries.clear();
                    return;
                }
                hasher.update(appended);
                let content_sha256 = hasher.clone().finalize().into();
                session_cache.history_signature = Some(ProtectedInboundDedupeHistorySignature {
                    file_len: file_state.file_len,
                    hmac_sidecar_sha256: Self::hmac_sidecar_sha256(history_path),
                    content_sha256,
                });
                session_cache.history_hasher = Some(ProtectedInboundDedupeHistoryHasher(hasher));
                session_cache.entries.insert(canonical_event_id, entry);
            } else if session_cache.complete {
                session_cache.complete = false;
                session_cache.history_signature = None;
                session_cache.history_hasher = None;
                session_cache.entries.clear();
            }
        }
    }

    fn append_history_message_under_lock(
        &self,
        history_path: &Path,
        session_id: &str,
        message: &ChatMessage,
    ) -> Result<(), SessionStoreError> {
        self.ensure_history_file_current(history_path, session_id)?;
        let encoded = self.encode_history_message(message)?;
        let mut appended = encoded;
        appended.push(b'\n');

        // SECURITY: refuse to append if the result would push the
        // history file past `SESSION_HISTORY_FILE_MAX_BYTES`. The
        // read-side cap on `read_to_vec_no_hang_no_follow_capped`
        // makes the file unloadable once it exceeds 256 MiB; without
        // a write-side check, a channel pushing large multimodal /
        // tool-output payloads (Matrix accepts 16 MiB per event) can
        // grow the file past the cap before the message-count-based
        // auto-compaction (at 500 messages) fires, bricking the
        // session. Mode is `HistoryFileFull` → operator-visible
        // SessionStoreError that the caller can route through to
        // a force-compact or chunk-the-message remediation. Bounds
        // are computed against the current on-disk size + appended
        // length so the check is cheap (one fstat per append).
        const APPEND_HEADROOM: u64 = 16 * 1024;
        let current_len = match std::fs::metadata(history_path) {
            Ok(meta) => meta.len(),
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => 0,
            Err(err) => return Err(err.into()),
        };
        let projected = current_len.saturating_add(appended.len() as u64);
        // Compare against the cap minus headroom to avoid the
        // theoretical wrap when `projected == u64::MAX` from
        // saturating_add. `cap - headroom` is a stable u64 constant.
        if projected > SESSION_HISTORY_FILE_MAX_BYTES.saturating_sub(APPEND_HEADROOM) {
            return Err(SessionStoreError::HistoryFileFull(format!(
                "session {session_id} history file at {} would exceed the \
                 {SESSION_HISTORY_FILE_MAX_BYTES}-byte read cap after appending \
                 {} bytes (current {}). Force-compact the session or split the \
                 message before re-appending.",
                history_path.display(),
                appended.len(),
                current_len,
            )));
        }
        let hmac_state =
            self.prepare_history_hmac_for_appended_bytes(history_path, &appended, session_id)?;
        let file = Self::open_private_append_file(history_path)?;
        let mut writer = BufWriter::new(file);
        writer.write_all(&appended)?;
        writer.flush()?;
        writer
            .into_inner()
            .map_err(|e| std::io::Error::other(e.to_string()))?
            .sync_all()?;

        self.commit_history_hmac(history_path, session_id)?;
        if let Some(state) = hmac_state {
            self.store_history_hmac_state(session_id, history_path, state);
        }
        self.mark_history_file_current(session_id);
        self.cache_protected_inbound_event_from_message(message, history_path, &appended);
        Ok(())
    }

    fn protected_inbound_dedupe_marker(
        session_id: &str,
        entry: &InboundEventIndexEntry,
    ) -> ChatMessage {
        let mut metadata = serde_json::json!({
            crate::channels::inbound::INBOUND_EVENT_ID_META_KEY: entry.event_id.clone(),
            crate::channels::inbound::INBOUND_DEDUPE_MARKER_META_KEY: true,
        });
        if let Some(run_id) = entry.run_id.as_deref() {
            metadata[crate::channels::inbound::INBOUND_RUN_ID_META_KEY] =
                serde_json::Value::String(run_id.to_string());
        }
        ChatMessage::system(session_id.to_string(), "").with_metadata(metadata)
    }

    fn is_protected_inbound_dedupe_marker(message: &ChatMessage) -> bool {
        if message.role != MessageRole::System
            || !message.content.is_empty()
            || message.tool_call_id.is_some()
            || message.tool_name.is_some()
            || message.tokens.is_some()
        {
            return false;
        }
        let Some(metadata) = message
            .metadata
            .as_ref()
            .and_then(serde_json::Value::as_object)
        else {
            return false;
        };
        let marker_key = crate::channels::inbound::INBOUND_DEDUPE_MARKER_META_KEY;
        let event_key = crate::channels::inbound::INBOUND_EVENT_ID_META_KEY;
        let run_key = crate::channels::inbound::INBOUND_RUN_ID_META_KEY;
        if metadata
            .get(marker_key)
            .and_then(serde_json::Value::as_bool)
            != Some(true)
        {
            return false;
        }
        if metadata
            .get(event_key)
            .and_then(serde_json::Value::as_str)
            .is_none_or(str::is_empty)
        {
            return false;
        }
        if metadata
            .get(run_key)
            .is_some_and(|value| !value.is_string())
        {
            return false;
        }
        metadata
            .keys()
            .all(|key| key == marker_key || key == event_key || key == run_key)
    }

    fn invalidate_protected_inbound_dedupe_cache(&self, session_id: &str) {
        self.protected_inbound_dedupe_cache
            .write()
            .remove(session_id);
    }

    fn archive_file_current_confirmed(
        &self,
        session_id: &str,
        signature: CachedArtifactSignature,
    ) -> bool {
        self.encryption_active()
            && self
                .archive_current_signatures
                .read()
                .get(session_id)
                .is_some_and(|cached| *cached == signature)
    }

    fn mark_archive_file_current(&self, session_id: &str, signature: CachedArtifactSignature) {
        if self.encryption_active() {
            self.archive_current_signatures
                .write()
                .insert(session_id.to_string(), signature);
        }
    }

    fn mark_archive_file_current_from_path(&self, session_id: &str, archive_path: &Path) {
        if let Some(signature) = Self::artifact_file_signature(archive_path) {
            self.mark_archive_file_current(session_id, signature);
        }
    }

    fn clear_archive_file_current(&self, session_id: &str) {
        self.archive_current_signatures.write().remove(session_id);
    }

    fn prepare_history_hmac_for_path(
        &self,
        source_path: &Path,
        history_path: &Path,
        session_id: &str,
    ) -> Result<Option<super::integrity::AppendHmacState>, SessionStoreError> {
        if let Some(ref key) = self.hmac_key {
            let state =
                super::integrity::AppendHmacState::from_path(key, source_path).map_err(|e| {
                    SessionStoreError::Io(format!(
                        "failed to stage HMAC sidecar for history {}: {}",
                        session_id, e
                    ))
                })?;
            super::integrity::prepare_pending_hmac_file_for_state(history_path, &state).map_err(
                |e| {
                    SessionStoreError::Io(format!(
                        "failed to stage HMAC sidecar for history {}: {}",
                        session_id, e
                    ))
                },
            )?;
            return Ok(Some(state));
        }
        Ok(None)
    }

    fn prepare_history_hmac_for_appended_bytes(
        &self,
        history_path: &Path,
        appended: &[u8],
        session_id: &str,
    ) -> Result<Option<super::integrity::AppendHmacState>, SessionStoreError> {
        if let Some(ref key) = self.hmac_key {
            let next_state = if let Some(cached) =
                self.cached_history_hmac_state_for_append(session_id, history_path)
            {
                super::integrity::prepare_pending_hmac_file_for_appended_bytes_with_state(
                    history_path,
                    &cached,
                    appended,
                )
            } else {
                super::integrity::prepare_pending_hmac_file_for_appended_bytes(
                    key,
                    history_path,
                    appended,
                )
            }
            .map_err(|e| {
                SessionStoreError::Io(format!(
                    "failed to stage HMAC sidecar for history {}: {}",
                    session_id, e
                ))
            })?;
            return Ok(Some(next_state));
        }
        Ok(None)
    }

    fn commit_history_hmac(
        &self,
        history_path: &Path,
        session_id: &str,
    ) -> Result<(), SessionStoreError> {
        if self.hmac_key.is_some() {
            super::integrity::commit_pending_hmac_sidecar(history_path).map_err(|e| {
                SessionStoreError::Io(format!(
                    "failed to commit HMAC sidecar for history {}: {}",
                    session_id, e
                ))
            })?;
        }
        Ok(())
    }

    fn prepare_archive_hmac(
        &self,
        archive_path: &Path,
        archive_bytes: &[u8],
        session_id: &str,
    ) -> Result<(), SessionStoreError> {
        if let Some(ref key) = self.hmac_key {
            super::integrity::prepare_pending_hmac_file(key, archive_bytes, archive_path).map_err(
                |e| {
                    SessionStoreError::Io(format!(
                        "failed to stage HMAC sidecar for archive {}: {}",
                        session_id, e
                    ))
                },
            )?;
        }
        Ok(())
    }

    fn commit_archive_hmac(
        &self,
        archive_path: &Path,
        session_id: &str,
    ) -> Result<(), SessionStoreError> {
        if self.hmac_key.is_some() {
            super::integrity::commit_pending_hmac_sidecar(archive_path).map_err(|e| {
                SessionStoreError::Io(format!(
                    "failed to commit HMAC sidecar for archive {}: {}",
                    session_id, e
                ))
            })?;
        }
        Ok(())
    }

    fn delete_archive_hmac(
        &self,
        archive_path: &Path,
        session_id: &str,
    ) -> Result<(), SessionStoreError> {
        if let Err(e) = super::integrity::delete_hmac_sidecar(archive_path) {
            return Err(SessionStoreError::Io(format!(
                "failed to remove HMAC sidecar for archive {}: {}",
                session_id, e
            )));
        }
        Ok(())
    }

    fn write_archive_file(
        &self,
        archive_path: &Path,
        session_id: &str,
        archived: &ArchivedSession,
    ) -> Result<(), SessionStoreError> {
        let temp_path = crate::paths::atomic_tmp_path(archive_path, "archive");
        let encoded = self.encode_archive(session_id, archived)?;
        let write_result = (|| -> Result<(), SessionStoreError> {
            {
                let file = Self::create_private_output_file_create_new(&temp_path)?;
                let mut writer = BufWriter::new(file);
                writer.write_all(&encoded)?;
                writer.flush()?;
                writer
                    .into_inner()
                    .map_err(|e| std::io::Error::other(e.to_string()))?
                    .sync_all()?;
            }

            self.prepare_archive_hmac(archive_path, &encoded, session_id)?;
            fs::rename(&temp_path, archive_path)?;
            crate::paths::sync_parent_dir_blocking(archive_path)?;
            self.commit_archive_hmac(archive_path, session_id)?;
            self.mark_archive_file_current_from_path(session_id, archive_path);
            Ok(())
        })();
        if write_result.is_err() {
            let _ = fs::remove_file(&temp_path);
        }
        write_result
    }

    fn delete_history_hmac(
        &self,
        history_path: &Path,
        session_id: &str,
    ) -> Result<(), SessionStoreError> {
        if let Err(e) = super::integrity::delete_hmac_sidecar(history_path) {
            return Err(SessionStoreError::Io(format!(
                "failed to remove HMAC sidecar for history {}: {}",
                session_id, e
            )));
        }
        Ok(())
    }

    /// Create a new session
    pub fn create_session(
        &self,
        agent_id: impl Into<String>,
        metadata: SessionMetadata,
    ) -> Result<Session, SessionStoreError> {
        self.ensure_base_dir()?;

        let session = Session::new(agent_id, metadata);

        let _lock = self.acquire_session_key_lock(&session.session_key)?;

        // Check for existing session with same key (cache + disk)
        if self.session_key_exists(&session.session_key)? {
            return Err(SessionStoreError::AlreadyExists(session.session_key));
        }

        // Persist to disk
        self.write_session_meta(&session)?;

        // Update caches
        {
            let mut sessions = self.sessions.write();
            let mut key_map = self.key_to_id.write();
            key_map.insert(session.session_key.clone(), session.id.clone());
            sessions.insert(
                session.id.clone(),
                self.new_cached_session(session.clone(), false),
            );
        }
        self.mark_history_file_current(&session.id);

        Ok(session)
    }

    /// Get a session by ID
    pub fn get_session(&self, session_id: &str) -> Result<Session, SessionStoreError> {
        // Validate session_id upfront for defense in depth
        Self::validate_session_id(session_id)?;

        // Check cache first
        {
            let sessions = self.sessions.read();
            if let Some(cached) = sessions.get(session_id) {
                return Ok(cached.session.clone());
            }
        }

        // Load from disk
        self.load_session(session_id)
    }

    /// Get a session by session key
    pub fn get_session_by_key(&self, session_key: &str) -> Result<Session, SessionStoreError> {
        // SECURITY: snapshot the session-id out of `key_to_id` under
        // the read guard, then drop the guard BEFORE acquiring the
        // `sessions` lock (which `get_session` does). Holding
        // `key_to_id.read()` across `self.get_session(id)` re-enters
        // a lock pair that write paths in this file take in the
        // opposite order (`sessions.write()` then `key_to_id.write()`
        // — e.g. lines 2444, 2520, 2744, 3877). Under contention the
        // two orderings form a cycle and parking_lot's writer-queues
        // turn it into a real deadlock when a writer is queued
        // between the read-side `key_to_id.read()` and the inner
        // `sessions.read()`. Snapshot and release.
        let cached_id = {
            let key_map = self.key_to_id.read();
            key_map.get(session_key).cloned()
        };
        if let Some(id) = cached_id {
            return self.get_session(&id);
        }

        // Scan disk for matching session key
        self.load_sessions_from_disk()?;

        let cached_id = {
            let key_map = self.key_to_id.read();
            key_map.get(session_key).cloned()
        };
        if let Some(id) = cached_id {
            self.get_session(&id)
        } else if self.should_block_unknown_session_key_without_crypto() {
            Err(Self::lock_message(
                "encrypted sessions are present; provide the config password before resolving sessions by key",
            ))
        } else {
            Err(SessionStoreError::NotFound(session_key.to_string()))
        }
    }

    /// Get or create a session by session key
    pub fn get_or_create_session(
        &self,
        session_key: impl Into<String>,
        metadata: SessionMetadata,
    ) -> Result<Session, SessionStoreError> {
        let key = session_key.into();

        self.ensure_base_dir()?;
        let _lock = self.acquire_session_key_lock(&key)?;

        match self.get_session_by_key(&key) {
            Ok(session) => Ok(session),
            Err(SessionStoreError::NotFound(_)) => {
                let session = Session::with_session_key(key, metadata);
                self.write_session_meta(&session)?;

                {
                    let mut sessions = self.sessions.write();
                    let mut key_map = self.key_to_id.write();
                    key_map.insert(session.session_key.clone(), session.id.clone());
                    sessions.insert(
                        session.id.clone(),
                        self.new_cached_session(session.clone(), false),
                    );
                }
                // `mark_history_file_current` re-acquires `self.sessions`; keep
                // the cache insertion guards out of scope to avoid a non-reentrant
                // parking_lot RwLock deadlock.
                self.mark_history_file_current(&session.id);

                Ok(session)
            }
            Err(e) => Err(e),
        }
    }

    /// List sessions with optional filtering
    pub fn list_sessions(&self, filter: SessionFilter) -> Result<Vec<Session>, SessionStoreError> {
        let entries = self.list_session_entries(filter.clone())?;
        Ok(entries
            .into_iter()
            .filter_map(SessionListEntry::into_session)
            .collect())
    }

    /// List sessions, surfacing encrypted-but-locked sessions as stubs when allowed.
    pub fn list_session_entries(
        &self,
        filter: SessionFilter,
    ) -> Result<Vec<SessionListEntry>, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        self.load_sessions_from_disk()?;

        let mut available: Vec<SessionListEntry> = {
            let sessions = self.sessions.read();
            sessions
                .values()
                .map(|c| c.session.clone())
                .filter(|s| filter.matches(s))
                .map(SessionListEntry::available)
                .collect()
        };

        let mut locked = Vec::new();
        if self.locked_session_count() > 0 {
            if filter.requires_decrypted_metadata() {
                return Err(self.locked_filter_error());
            }
            locked.extend(self.locked_session_entries_snapshot());
        }

        available.extend(locked);
        available.sort_by(|a, b| b.updated_at.unwrap_or(0).cmp(&a.updated_at.unwrap_or(0)));

        let offset = filter.offset.unwrap_or(0);
        let limit = filter.limit.unwrap_or(usize::MAX);
        Ok(available.into_iter().skip(offset).take(limit).collect())
    }

    /// Update session metadata
    pub fn patch_session(
        &self,
        session_id: &str,
        updates: SessionMetadata,
    ) -> Result<Session, SessionStoreError> {
        let mut session = self.get_session(session_id)?;

        // Apply updates
        if updates.name.is_some() {
            session.metadata.name = updates.name;
        }
        if updates.description.is_some() {
            session.metadata.description = updates.description;
        }
        if updates.agent_id.is_some() {
            session.metadata.agent_id = updates.agent_id;
        }
        if updates.channel.is_some() {
            session.metadata.channel = updates.channel;
        }
        if updates.chat_id.is_some() {
            session.metadata.chat_id = updates.chat_id;
        }
        if updates.user_id.is_some() {
            session.metadata.user_id = updates.user_id;
        }
        if updates.route.is_some() {
            session.metadata.route = updates.route;
        }
        if updates.model.is_some() {
            session.metadata.model = updates.model;
        }
        if updates.thinking_level.is_some() {
            session.metadata.thinking_level = updates.thinking_level;
        }
        if !updates.tags.is_empty() {
            session.metadata.tags = updates.tags;
        }
        if updates.extra.is_some() {
            session.metadata.extra = updates.extra;
        }

        let now = now_millis();
        session.updated_at = now.max(session.updated_at.saturating_add(1));

        // Persist
        self.write_session_meta(&session)?;

        // Update cache
        {
            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session = session.clone();
                cached.dirty = false;
            }
        }

        Ok(session)
    }

    /// Reset a session (clear history but keep metadata)
    pub fn reset_session(&self, session_id: &str) -> Result<Session, SessionStoreError> {
        let mut session = self.get_session(session_id)?;

        // Delete history file
        let history_path = self.session_history_path(session_id)?;
        let inbound_index_path = self.session_inbound_event_index_path(session_id)?;
        let _history_lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;
        if history_path.exists() {
            fs::remove_file(&history_path)?;
        }
        self.delete_inbound_event_index(&inbound_index_path)?;
        self.delete_history_hmac(&history_path, session_id)?;
        self.clear_history_hmac_state(session_id);
        self.invalidate_protected_inbound_dedupe_cache(session_id);
        self.mark_history_file_current(session_id);

        // Reset message count and compaction metadata
        session.message_count = 0;
        session.metadata.compaction = CompactionMetadata::default();
        session.updated_at = now_millis();
        session.last_activity_at = Some(now_millis());

        // Persist
        self.write_session_meta(&session)?;

        // Update cache
        {
            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session = session.clone();
                cached.dirty = false;
            }
        }

        Ok(session)
    }

    /// Delete a session and its history
    pub fn delete_session(&self, session_id: &str) -> Result<(), SessionStoreError> {
        let session = self.get_session(session_id)?;

        // Delete files
        let meta_path = self.session_meta_path(session_id)?;
        let history_path = self.session_history_path(session_id)?;
        let inbound_index_path = self.session_inbound_event_index_path(session_id)?;
        // Acquire BOTH the history-path and meta-path advisory locks
        // before removing either file. `write_session_meta` only holds
        // the meta-path lock (line 3952), and `append_history_*` only
        // holds the history-path lock; without contending on the meta
        // lock here, a concurrent `flush_session` → `write_session_meta`
        // could land its `rename(temp, meta_path)` after we cleared
        // both files, resurrecting half of a deleted session on disk.
        //
        // Lock acquisition order is history → meta, MATCHING the
        // dominant order used by every other dual-locking path:
        // `compact_session`, `archive_session`, and `restore_session`
        // all acquire the history FileLock at the entry point and
        // then call `write_session_meta` internally, which acquires
        // the meta FileLock. Reversing the order here (meta → history)
        // would deadlock against a concurrent compact/archive/restore
        // on the same session id — they hold history awaiting meta,
        // we hold meta awaiting history. Both guards live to
        // scope-end via RAII; explicit naming avoids `_` shadowing
        // pitfalls.
        //
        // Declaration order also fixes drop order: Rust drops local
        // bindings in REVERSE declaration order, so `_meta_lock`
        // releases first then `_history_lock`. That's release in
        // meta→history — the inverse of acquire (history→meta) — and
        // is the only sound release pattern when other paths take the
        // same locks in history→meta order (releasing history first
        // here would briefly allow a queued compact/archive/restore
        // to grab history and then race us to meta-removal). Do NOT
        // reorder these two `let` lines without re-deriving the lock
        // graph against every other caller.
        let _history_lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;
        let _meta_lock =
            FileLock::acquire(&meta_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        if meta_path.exists() {
            fs::remove_file(&meta_path)?;
        }
        if history_path.exists() {
            fs::remove_file(&history_path)?;
        }
        self.delete_inbound_event_index(&inbound_index_path)?;
        if let Err(e) = super::integrity::delete_hmac_sidecar(&meta_path) {
            return Err(SessionStoreError::Io(format!(
                "failed to remove HMAC sidecar for session {}: {}",
                session_id, e
            )));
        }
        self.delete_history_hmac(&history_path, session_id)?;
        self.clear_history_hmac_state(session_id);
        self.invalidate_protected_inbound_dedupe_cache(session_id);

        // Remove from caches
        {
            let mut sessions = self.sessions.write();
            let mut key_map = self.key_to_id.write();
            sessions.remove(session_id);
            key_map.remove(&session.session_key);
        }

        Ok(())
    }

    /// Export all data for a user (GDPR Art. 20 — data portability).
    ///
    /// Returns all sessions and their chat histories for the given user_id
    /// as a portable JSON value.
    pub fn export_user_data(&self, user_id: &str) -> Result<serde_json::Value, SessionStoreError> {
        if user_id.trim().is_empty() {
            return Err(SessionStoreError::InvalidUserId(
                "userId must not be empty or whitespace-only".to_string(),
            ));
        }
        let filter = SessionFilter::new().with_user_id(user_id);
        let sessions = self.list_sessions(filter)?;

        let mut exported = Vec::new();
        let mut warnings: Vec<String> = Vec::new();
        for session in &sessions {
            match self.get_history(&session.id, None, None) {
                Ok(history) => {
                    exported.push(serde_json::json!({
                        "session": session,
                        "messages": history,
                    }));
                }
                Err(e) => {
                    warn!(
                        user_id = %user_id,
                        error_kind = session_store_error_kind(&e),
                        "failed to export session history during user data export"
                    );
                    warnings.push(format!(
                        "failed to export session {}: {}",
                        session.id,
                        session_store_error_export_warning(&e)
                    ));
                }
            }
        }

        Ok(serde_json::json!({
            "user_id": user_id,
            "exported_at": chrono::Utc::now().to_rfc3339(),
            "session_count": exported.len(),
            "sessions": exported,
            "warnings": warnings,
        }))
    }

    /// Delete all data for a user (GDPR Art. 17 — right to erasure).
    ///
    /// Deletes all sessions and their histories for the given user_id.
    /// Uses best-effort: logs individual failures and continues.
    /// Returns the number of sessions successfully deleted.
    pub fn purge_user_data(&self, user_id: &str) -> Result<(usize, usize), SessionStoreError> {
        if user_id.trim().is_empty() {
            return Err(SessionStoreError::InvalidUserId(
                "userId must not be empty or whitespace-only".to_string(),
            ));
        }
        let filter = SessionFilter::new().with_user_id(user_id);
        let sessions = self.list_sessions(filter)?;
        let total = sessions.len();
        let mut deleted = 0;

        for session in &sessions {
            if let Err(e) = self.delete_session(&session.id) {
                warn!(
                    user_id = %user_id,
                    error_kind = session_store_error_kind(&e),
                    "failed to delete session during user purge"
                );
            } else {
                deleted += 1;
            }
        }

        Ok((deleted, total))
    }

    /// Delete sessions that have not been updated within the given retention period.
    ///
    /// Returns the number of sessions deleted.
    pub fn cleanup_expired(&self, retention_days: u32) -> Result<usize, SessionStoreError> {
        let cutoff_ms = now_millis() - (retention_days as i64) * 24 * 60 * 60 * 1000;
        let filter = SessionFilter {
            updated_before: Some(cutoff_ms),
            ..SessionFilter::default()
        };
        let expired = self.list_sessions(filter)?;
        let count = expired.len();

        for session in &expired {
            if let Err(e) = self.delete_session(&session.id) {
                warn!(
                    error_kind = session_store_error_kind(&e),
                    "failed to delete expired session"
                );
            }
        }

        if count > 0 {
            tracing::info!(
                deleted = count,
                retention_days,
                "cleaned up expired sessions"
            );
        }

        Ok(count)
    }

    /// Append a message to session history
    ///
    /// Returns an error if the session is archived (read-only).
    pub fn append_message(&self, message: ChatMessage) -> Result<(), SessionStoreError> {
        self.ensure_base_dir()?;
        self.ensure_required_encryption_available()?;

        let session_id = message.session_id.clone();

        // Check session status - archived sessions are read-only.
        // Use get_session to check both cache and disk, ensuring the guard
        // works even if the session hasn't been loaded into memory yet.
        match self.get_session(&session_id) {
            Ok(session) if session.status == SessionStatus::Archived => {
                return Err(SessionStoreError::AlreadyArchived(session_id));
            }
            Ok(_) | Err(SessionStoreError::NotFound(_)) => {}
            Err(err) => return Err(err),
        }

        // Append to history file (JSONL format)
        let history_path = self.session_history_path(&session_id)?;
        let _lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;
        self.append_history_message_under_lock(&history_path, &session_id, &message)?;
        self.invalidate_protected_inbound_dedupe_cache(&session_id);

        // Update session message count
        self.increment_message_count(&session_id)?;

        Ok(())
    }

    /// Atomically check-and-append for inbound channel events.
    ///
    /// SECURITY: Holds the per-session `FileLock` across the dedupe
    /// scan AND the append, eliminating the read-then-write TOCTOU
    /// window in the previous two-step
    /// `session_contains_inbound_event` → `append_message` flow.
    /// Returns:
    ///
    /// - `Ok({ appended: true, .. })` if the message was appended.
    /// - `Ok({ appended: false, original_run_id })` if this session's
    ///   inbound-event index already carried this `inbound_event_id` — caller
    ///   should treat as a successful no-op dedupe.
    ///
    /// Dedupe is per-session: the caller's session-id selection is
    /// already channel-scoped (each channel maps inbound events to a
    /// distinct session), so a cross-channel collision on the same
    /// `inbound_event_id` would land in different sessions and never
    /// scan the same history. The match runs against
    /// `metadata.inbound_event_id` on each historical message — the
    /// same field `dispatch_inbound_text_with_options` writes.
    pub(crate) fn append_message_if_new_inbound(
        &self,
        message: ChatMessage,
        inbound_event_id: impl AsRef<str>,
    ) -> Result<InboundAppendOutcome, SessionStoreError> {
        let inbound_event_id = inbound_event_id.as_ref();
        self.ensure_base_dir()?;
        self.ensure_required_encryption_available()?;

        let session_id = message.session_id.clone();
        match self.get_session(&session_id) {
            Ok(session) if session.status == SessionStatus::Archived => {
                return Err(SessionStoreError::AlreadyArchived(session_id));
            }
            Ok(_) | Err(SessionStoreError::NotFound(_)) => {}
            Err(err) => return Err(err),
        }

        let history_path = self.session_history_path(&session_id)?;
        let inbound_index_path = self.session_inbound_event_index_path(&session_id)?;
        let _lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        let lookup = self.find_inbound_event_index_entry(
            &session_id,
            &history_path,
            &inbound_index_path,
            inbound_event_id,
        )?;
        if let Some(entry) = lookup.entry {
            if lookup.legacy_index_present && !lookup.matched_protected_history {
                let marker = Self::protected_inbound_dedupe_marker(&session_id, &entry);
                self.invalidate_protected_inbound_dedupe_cache(&session_id);
                self.append_history_message_under_lock(&history_path, &session_id, &marker)?;
            }
            if lookup.legacy_index_present {
                self.delete_inbound_event_index(&inbound_index_path)?;
            }
            return Ok(InboundAppendOutcome {
                appended: false,
                original_run_id: entry.run_id,
                corrupt_index_lines: lookup.corrupt_lines,
            });
        }

        self.append_history_message_under_lock(&history_path, &session_id, &message)?;
        if lookup.legacy_index_present {
            self.delete_inbound_event_index(&inbound_index_path)?;
        }
        self.increment_message_count(&session_id)?;
        Ok(InboundAppendOutcome {
            appended: true,
            original_run_id: None,
            corrupt_index_lines: lookup.corrupt_lines,
        })
    }

    fn find_inbound_event_index_entry(
        &self,
        session_id: &str,
        history_path: &Path,
        inbound_index_path: &Path,
        inbound_event_id: &str,
    ) -> Result<InboundEventIndexLookup, SessionStoreError> {
        let mut lookup =
            self.find_inbound_event_in_history(session_id, history_path, inbound_event_id)?;
        if inbound_index_path.exists() {
            lookup.legacy_index_present = true;
            let legacy_lookup = self.find_legacy_inbound_event_index_entry(
                session_id,
                inbound_index_path,
                inbound_event_id,
            )?;
            lookup.corrupt_lines = lookup
                .corrupt_lines
                .saturating_add(legacy_lookup.corrupt_lines);
            if lookup.entry.is_none() {
                lookup.entry = legacy_lookup.entry;
            }
        }
        Ok(lookup)
    }

    fn find_inbound_event_in_history(
        &self,
        session_id: &str,
        history_path: &Path,
        inbound_event_id: &str,
    ) -> Result<InboundEventIndexLookup, SessionStoreError> {
        let target = canonical_inbound_event_id(inbound_event_id);
        if let Some(entry) =
            self.protected_inbound_dedupe_cache_lookup(session_id, history_path, &target)
        {
            let matched_protected_history = entry.is_some();
            return Ok(InboundEventIndexLookup {
                entry,
                corrupt_lines: 0,
                matched_protected_history,
                legacy_index_present: false,
            });
        }
        if !history_path.exists() {
            self.replace_protected_inbound_dedupe_cache(session_id, None, HashMap::new());
            return Ok(InboundEventIndexLookup::default());
        }
        let history_bytes = self
            .read_verified_history_bytes(history_path)
            .map_err(|err| {
                let err =
                    Self::history_integrity_error(session_id, "inbound-event dedupe scan", err);
                if err.is_permanent_history_corruption() {
                    warn!(
                        session_id = %session_id,
                        path = %history_path.display(),
                        error = %err,
                        "session history integrity failure blocks inbound-event dedupe scan"
                    );
                }
                err
            })?;
        let history_state =
            Self::protected_inbound_dedupe_history_state_from_bytes(history_path, &history_bytes);
        let mut lookup = InboundEventIndexLookup::default();
        let mut cache_entries = HashMap::new();
        for raw_line in history_bytes.split(|byte| *byte == b'\n') {
            let line = Self::trim_ascii_whitespace(raw_line);
            if line.is_empty() {
                continue;
            }
            let encrypted_line = super::crypto::has_encrypted_payload_prefix(line);
            let msg = match self.decode_history_message(session_id, line, encrypted_line) {
                Ok(message) => message,
                Err(SessionStoreError::Locked(message)) => {
                    return Err(SessionStoreError::Locked(message));
                }
                Err(SessionStoreError::DecryptionFailed(message)) => {
                    warn!(
                        session_id = %session_id,
                        path = %history_path.display(),
                        error = %message,
                        "encrypted session history authentication failure blocks inbound-event dedupe scan"
                    );
                    return Err(SessionStoreError::HistoryCorrupt(format!(
                        "encrypted session history authentication failed during inbound-event dedupe scan for session {session_id}: {message}"
                    )));
                }
                Err(SessionStoreError::ManifestIntegrityFailed(message)) => {
                    warn!(
                        session_id = %session_id,
                        path = %history_path.display(),
                        error = %message,
                        "encrypted session manifest integrity failure blocks inbound-event dedupe scan"
                    );
                    return Err(SessionStoreError::HistoryCorrupt(format!(
                        "encrypted session manifest integrity failed during inbound-event dedupe scan for session {session_id}: {message}"
                    )));
                }
                Err(err) if encrypted_line => {
                    warn!(
                        session_id = %session_id,
                        path = %history_path.display(),
                        error = %err,
                        "invalid encrypted session history line blocks inbound-event dedupe scan"
                    );
                    return Err(SessionStoreError::HistoryCorrupt(format!(
                        "invalid encrypted session history line during inbound-event dedupe scan for session {session_id}: {err}"
                    )));
                }
                Err(err) => {
                    warn!(
                        session_id = %session_id,
                        path = %history_path.display(),
                        error = %err,
                        "corrupt session history line blocks inbound-event dedupe scan"
                    );
                    return Err(SessionStoreError::HistoryCorrupt(format!(
                        "corrupt session history line during inbound-event dedupe scan for session {session_id}: {err}"
                    )));
                }
            };
            let Some(metadata) = msg.metadata.as_ref() else {
                continue;
            };
            let Some(stored_event_id) = metadata_string(
                metadata,
                crate::channels::inbound::INBOUND_EVENT_ID_META_KEY,
            ) else {
                continue;
            };
            let canonical_stored_event_id = canonical_inbound_event_id(stored_event_id.as_str());
            let entry = InboundEventIndexEntry {
                event_id: stored_event_id,
                run_id: metadata_string(
                    metadata,
                    crate::channels::inbound::INBOUND_RUN_ID_META_KEY,
                ),
            };
            if canonical_stored_event_id == target {
                lookup.matched_protected_history = true;
                if lookup.entry.is_none() {
                    lookup.entry = Some(entry.clone());
                }
            }
            cache_entries
                .entry(canonical_stored_event_id)
                .or_insert(entry);
        }
        self.replace_protected_inbound_dedupe_cache(session_id, history_state, cache_entries);
        Ok(lookup)
    }

    fn find_legacy_inbound_event_index_entry(
        &self,
        session_id: &str,
        inbound_index_path: &Path,
        inbound_event_id: &str,
    ) -> Result<InboundEventIndexLookup, SessionStoreError> {
        // SECURITY: O_NOFOLLOW + O_NONBLOCK + 64 MiB cap. The
        // legacy inbound-event index is a JSONL file of per-event
        // dedupe entries; reachable on the dedupe-fallback path
        // for every inbound Matrix message. Plain `fs::read` here
        // (a) followed symlinks so a same-uid attacker could
        // redirect the read to an attacker-chosen file, (b) had no
        // cap so a runaway index or a /dev/zero plant would OOM
        // the daemon mid-dispatch, and (c) had no FIFO guard so a
        // planted FIFO with no writer would hang the dispatcher.
        // The B82 helper closes (a)+(c); the 64 MiB cap closes
        // (b) — legitimate indexes carry one short JSON line per
        // event so even a year of high-traffic sessions stays
        // well under the ceiling.
        const LEGACY_INBOUND_INDEX_MAX_BYTES: u64 = 64 * 1024 * 1024;
        let index_bytes = crate::paths::read_to_vec_no_hang_no_follow_capped(
            inbound_index_path,
            LEGACY_INBOUND_INDEX_MAX_BYTES,
        )?
        .ok_or_else(|| {
            std::io::Error::new(
                std::io::ErrorKind::NotFound,
                format!(
                    "legacy inbound event index disappeared between caller's existence check and read at {}",
                    inbound_index_path.display()
                ),
            )
        })?;
        let mut lookup = InboundEventIndexLookup::default();
        let target = canonical_inbound_event_id(inbound_event_id);
        for raw_line in index_bytes.split(|byte| *byte == b'\n') {
            let line = Self::trim_ascii_whitespace(raw_line);
            if line.is_empty() {
                continue;
            }
            match serde_json::from_slice::<InboundEventIndexEntry>(line) {
                Ok(entry) if canonical_inbound_event_id(entry.event_id.as_str()) == target => {
                    if lookup.entry.is_none() {
                        lookup.entry = Some(entry);
                    }
                }
                Ok(_) => {}
                Err(err) => {
                    lookup.corrupt_lines = lookup.corrupt_lines.saturating_add(1);
                    warn!(
                        session_id = %session_id,
                        path = %inbound_index_path.display(),
                        error = %err,
                        "corrupt inbound-event dedupe index line ignored"
                    );
                }
            }
        }
        Ok(lookup)
    }

    fn delete_inbound_event_index(
        &self,
        inbound_index_path: &Path,
    ) -> Result<(), SessionStoreError> {
        match fs::remove_file(inbound_index_path) {
            Ok(()) => Ok(()),
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(err) => Err(SessionStoreError::Io(format!(
                "failed to remove legacy inbound-event dedupe index '{}': {err}",
                inbound_index_path.display()
            ))),
        }
    }

    /// Append multiple messages in a single file open/write/close cycle.
    /// More efficient than calling `append_message` repeatedly for batch writes
    /// (e.g., assistant message + tool results in the executor).
    pub fn append_messages(&self, messages: &[ChatMessage]) -> Result<(), SessionStoreError> {
        if messages.is_empty() {
            return Ok(());
        }
        self.ensure_base_dir()?;
        self.ensure_required_encryption_available()?;

        let session_id = &messages[0].session_id;
        if let Some(message) = messages
            .iter()
            .find(|message| message.session_id != *session_id)
        {
            return Err(SessionStoreError::InvalidMessageBatch(format!(
                "cannot append a batch for session {} containing message for session {}",
                session_id, message.session_id
            )));
        }

        // Check session status
        match self.get_session(session_id) {
            Ok(session) if session.status == SessionStatus::Archived => {
                return Err(SessionStoreError::AlreadyArchived(session_id.to_string()));
            }
            Ok(_) | Err(SessionStoreError::NotFound(_)) => {}
            Err(err) => return Err(err),
        }

        let history_path = self.session_history_path(session_id)?;
        let _lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;
        self.ensure_history_file_current(&history_path, session_id)?;
        let mut appended = Vec::new();

        for msg in messages {
            let encoded = self.encode_history_message(msg)?;
            appended.extend_from_slice(&encoded);
            appended.push(b'\n');
        }

        // SECURITY: mirror the byte-cap pre-check from
        // `append_history_message_under_lock`. Without it, a batch of
        // assistant + tool-result messages from a single agent turn
        // can cross `SESSION_HISTORY_FILE_MAX_BYTES` while each
        // individual message would have been refused on its own
        // single-message path. The read-side cap then makes the file
        // unloadable and the session bricks until manual compaction.
        // Reject the whole batch atomically rather than half-writing.
        const APPEND_HEADROOM: u64 = 16 * 1024;
        let current_len = match std::fs::metadata(&history_path) {
            Ok(meta) => meta.len(),
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => 0,
            Err(err) => return Err(err.into()),
        };
        let projected = current_len.saturating_add(appended.len() as u64);
        // Compare against the cap minus headroom to avoid the
        // theoretical wrap when `projected == u64::MAX` from
        // saturating_add. `cap - headroom` is a stable u64 constant.
        if projected > SESSION_HISTORY_FILE_MAX_BYTES.saturating_sub(APPEND_HEADROOM) {
            return Err(SessionStoreError::HistoryFileFull(format!(
                "session {session_id} history file at {} would exceed the \
                 {SESSION_HISTORY_FILE_MAX_BYTES}-byte read cap after appending \
                 {} bytes for a {}-message batch (current {}). Force-compact the \
                 session or split the batch before re-appending.",
                history_path.display(),
                appended.len(),
                messages.len(),
                current_len,
            )));
        }

        let hmac_state =
            self.prepare_history_hmac_for_appended_bytes(&history_path, &appended, session_id)?;
        let file = Self::open_private_append_file(&history_path)?;
        let mut writer = BufWriter::new(file);
        writer.write_all(&appended)?;
        writer.flush()?;
        writer
            .into_inner()
            .map_err(|e| std::io::Error::other(e.to_string()))?
            .sync_all()?;

        self.commit_history_hmac(&history_path, session_id)?;
        if let Some(state) = hmac_state {
            self.store_history_hmac_state(session_id, &history_path, state);
        }
        self.mark_history_file_current(session_id);
        self.invalidate_protected_inbound_dedupe_cache(session_id);

        // Update message count
        for _ in messages {
            self.increment_message_count(session_id)?;
        }

        Ok(())
    }

    /// Get chat history for a session
    pub fn get_history(
        &self,
        session_id: &str,
        limit: Option<usize>,
        before_id: Option<&str>,
    ) -> Result<Vec<ChatMessage>, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        let history_path = self.session_history_path(session_id)?;

        if !history_path.exists() {
            return Ok(Vec::new());
        }

        let history_bytes = self
            .read_verified_history_bytes(&history_path)
            .map_err(|err| Self::history_integrity_error(session_id, "get history", err))?;

        let mut messages: Vec<ChatMessage> = Vec::new();
        let mut found_before = before_id.is_none();

        for raw_line in history_bytes.split(|byte| *byte == b'\n') {
            let line = Self::trim_ascii_whitespace(raw_line);
            if line.is_empty() {
                continue;
            }
            let encrypted_line = super::crypto::has_encrypted_payload_prefix(line);

            let msg = match self.decode_history_message(session_id, line, encrypted_line) {
                Ok(m) => m,
                Err(SessionStoreError::Locked(message)) => {
                    return Err(SessionStoreError::Locked(message));
                }
                Err(err) => {
                    let err = Self::encrypted_history_decode_error(
                        session_id,
                        "get history",
                        encrypted_line,
                        err,
                    );
                    if err.is_permanent_history_corruption()
                        || matches!(err, SessionStoreError::Locked(_))
                    {
                        return Err(err);
                    }
                    tracing::warn!(
                        error_kind = "invalid_jsonl",
                        "skipping corrupt JSONL line in session history"
                    );
                    continue;
                }
            };
            if Self::is_protected_inbound_dedupe_marker(&msg) {
                continue;
            }

            if !found_before {
                if Some(msg.id.as_str()) == before_id {
                    found_before = true;
                }
                continue;
            }

            messages.push(msg);
        }

        // Apply limit (from end)
        if let Some(limit) = limit {
            if messages.len() > limit {
                messages = messages.into_iter().rev().take(limit).rev().collect();
            }
        }

        Ok(messages)
    }

    /// Clear all history for a session
    pub fn clear_history(&self, session_id: &str) -> Result<(), SessionStoreError> {
        let _ = self.get_session(session_id)?;
        let history_path = self.session_history_path(session_id)?;
        let inbound_index_path = self.session_inbound_event_index_path(session_id)?;
        // Match the per-history FileLock discipline used by every
        // other history-mutating path (reset_session, delete_session,
        // append_message, append_message_if_new_inbound,
        // append_messages, compact_session, archive_session). Without
        // it, a concurrent inbound-append under FileLock can be
        // racing this unlink: the appender writes a stale-HMAC line
        // onto a recreated file, breaking the integrity-cache
        // invariant the cache comment declares.
        let _history_lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        if history_path.exists() {
            fs::remove_file(&history_path)?;
        }
        self.delete_inbound_event_index(&inbound_index_path)?;
        self.delete_history_hmac(&history_path, session_id)?;
        self.clear_history_hmac_state(session_id);
        self.invalidate_protected_inbound_dedupe_cache(session_id);
        self.mark_history_file_current(session_id);

        // Reset message count
        {
            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session.message_count = 0;
                cached.session.updated_at = now_millis();
                cached.dirty = true;
            }
        }

        self.flush_session(session_id)?;

        Ok(())
    }

    /// Compact a session by summarizing old messages
    ///
    /// This replaces old messages with a summary, keeping recent messages intact.
    /// The summary_fn is called with the messages to summarize and should return
    /// a summary text.
    pub fn compact_session<F>(
        &self,
        session_id: &str,
        keep_recent: usize,
        summary_fn: F,
    ) -> Result<CompactionMetadata, SessionStoreError>
    where
        F: FnOnce(&[ChatMessage]) -> String,
    {
        let mut session = self.get_session(session_id)?;

        if session.status == SessionStatus::Archived {
            return Err(SessionStoreError::AlreadyArchived(session_id.to_string()));
        }

        if session.status == SessionStatus::Compacting {
            return Err(SessionStoreError::CompactionInProgress(
                session_id.to_string(),
            ));
        }

        // Acquire lock on the history file for the duration of compaction
        let history_path = self.session_history_path(session_id)?;
        let _lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        // Mark as compacting
        session.status = SessionStatus::Compacting;
        self.write_session_meta(&session)?;

        let raw_messages = match self.read_raw_history_messages(&history_path, session_id) {
            Ok(messages) => messages,
            Err(err) => {
                return Err(self.restore_compaction_status_after_precommit_failure(
                    session,
                    err,
                    "read history",
                ));
            }
        };
        let messages: Vec<_> = raw_messages
            .iter()
            .filter(|message| !Self::is_protected_inbound_dedupe_marker(message))
            .cloned()
            .collect();

        if messages.len() <= keep_recent {
            // Not enough messages to compact
            session.status = SessionStatus::Active;
            self.write_session_meta(&session)?;
            return Ok(session.metadata.compaction);
        }

        // Split into messages to compact and messages to keep
        let compact_count = messages.len() - keep_recent;
        let to_compact: Vec<_> = messages.iter().take(compact_count).cloned().collect();
        let to_keep: Vec<_> = messages.into_iter().skip(compact_count).collect();

        // Generate summary
        let summary = summary_fn(&to_compact);

        let mut rewritten = Vec::with_capacity(to_keep.len() + 1);
        rewritten.push(ChatMessage::system(session_id, &summary));
        rewritten.extend(to_keep.iter().cloned());
        if let Err(err) = self.rewrite_history_file_preserving_inbound_dedupe(
            &history_path,
            session_id,
            rewritten,
            &raw_messages,
            &to_compact,
        ) {
            return Err(self.restore_compaction_status_after_precommit_failure(
                session,
                err,
                "rewrite history",
            ));
        }

        // Update session metadata
        session.status = SessionStatus::Active;
        session.message_count = to_keep.len() + 1; // +1 for summary
        session.metadata.compaction.messages_compacted += compact_count;
        session.metadata.compaction.last_compacted_at = Some(now_millis());
        session.metadata.compaction.compaction_count += 1;
        session.metadata.compaction.last_summary = Some(summary);
        session.metadata.compaction.original_message_count += compact_count;
        session.updated_at = now_millis();

        self.write_session_meta(&session)?;

        // Update cache
        {
            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session = session.clone();
                cached.dirty = false;
            }
        }

        Ok(session.metadata.compaction)
    }

    fn restore_compaction_status_after_precommit_failure(
        &self,
        mut session: Session,
        original_error: SessionStoreError,
        operation: &'static str,
    ) -> SessionStoreError {
        session.status = SessionStatus::Active;
        session.updated_at = now_millis();
        match self.write_session_meta(&session) {
            Ok(()) => {
                let mut sessions = self.sessions.write();
                if let Some(cached) = sessions.get_mut(&session.id) {
                    cached.session = session;
                    cached.dirty = false;
                }
                original_error
            }
            Err(restore_error) => SessionStoreError::Io(format!(
                "session compaction failed during {operation}: {original_error}; additionally failed to restore session status to active: {restore_error}"
            )),
        }
    }

    /// Auto-compact hook - triggers compaction if message count exceeds threshold
    pub fn auto_compact_if_needed<F>(
        &self,
        session_id: &str,
        keep_recent: usize,
        summary_fn: F,
    ) -> Result<Option<CompactionMetadata>, SessionStoreError>
    where
        F: FnOnce(&[ChatMessage]) -> String,
    {
        let session = self.get_session(session_id)?;

        if session.message_count >= self.compact_threshold {
            Ok(Some(self.compact_session(
                session_id,
                keep_recent,
                summary_fn,
            )?))
        } else {
            Ok(None)
        }
    }

    // ========================================================================
    // Archive Operations
    // ========================================================================

    /// Get the archive directory path
    fn archive_dir(&self) -> PathBuf {
        self.base_path.join("archives")
    }

    /// Get the archive file path for a session
    fn archive_path(&self, session_id: &str) -> Result<PathBuf, SessionStoreError> {
        Self::validate_session_id(session_id)?;
        Ok(self
            .archive_dir()
            .join(format!("{}.archive.json", session_id)))
    }

    /// Ensure the archive directory exists
    fn ensure_archive_dir(&self) -> Result<(), SessionStoreError> {
        let archive_dir = self.archive_dir();
        if !archive_dir.exists() {
            fs::create_dir_all(&archive_dir)?;
        }
        Ok(())
    }

    /// Archive a session to a compressed file
    ///
    /// This creates an archive file containing all session metadata and messages.
    /// The session status is set to Archived, making it read-only.
    ///
    /// # Arguments
    /// * `session_id` - The session to archive
    /// * `delete_history` - If true, delete the history file after archiving (keeps metadata)
    ///
    /// # Returns
    /// Archive result with path and stats
    pub fn archive_session(
        &self,
        session_id: &str,
        delete_history: bool,
    ) -> Result<ArchiveResult, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        let mut session = self.get_session(session_id)?;

        // Don't archive if already archived
        if session.status == SessionStatus::Archived {
            return Err(SessionStoreError::AlreadyArchived(session_id.to_string()));
        }

        // Acquire lock on the history file for the duration of archiving
        let history_path = self.session_history_path(session_id)?;
        let _lock =
            FileLock::acquire(&history_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        self.ensure_archive_dir()?;

        let raw_messages = self.read_raw_history_messages(&history_path, session_id)?;
        let message_count = raw_messages
            .iter()
            .filter(|message| !Self::is_protected_inbound_dedupe_marker(message))
            .count();

        // Create archive structure
        let archived = ArchivedSession {
            session: session.clone(),
            messages: raw_messages.clone(),
            archived_at: now_millis(),
            version: 1,
        };

        let archive_path = self.archive_path(session_id)?;
        self.write_archive_file(&archive_path, session_id, &archived)?;

        // Get archive size
        let archive_size = fs::metadata(&archive_path).map(|m| m.len()).unwrap_or(0);

        // Update session status to archived
        session.status = SessionStatus::Archived;
        session.updated_at = now_millis();
        self.write_session_meta(&session)?;

        // Optionally delete history file to save space
        if delete_history {
            let history_path = self.session_history_path(session_id)?;
            if history_path.exists() {
                fs::remove_file(&history_path)?;
            }
            self.delete_history_hmac(&history_path, session_id)?;
            self.clear_history_hmac_state(session_id);
            self.invalidate_protected_inbound_dedupe_cache(session_id);
            self.mark_history_file_current(session_id);
        } else if self.encryption_active() {
            self.rewrite_history_file_preserving_inbound_dedupe(
                &history_path,
                session_id,
                archived.messages.clone(),
                &raw_messages,
                &[],
            )?;
        }

        // Update cache
        {
            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session = session;
                cached.dirty = false;
            }
        }

        Ok(ArchiveResult {
            session_id: session_id.to_string(),
            archive_path: archive_path.display().to_string(),
            message_count,
            archive_size,
            archived_at: archived.archived_at,
            history_deleted: delete_history,
        })
    }

    /// Restore a session from archive
    ///
    /// This reads the archive file and restores the session history.
    /// The session status is set back to Active.
    ///
    /// # Arguments
    /// * `session_id` - The session to restore
    ///
    /// # Returns
    /// Restore result with stats
    pub fn restore_session(&self, session_id: &str) -> Result<RestoreResult, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        let mut session = self.get_session(session_id)?;

        // Only restore archived sessions
        if session.status != SessionStatus::Archived {
            return Err(SessionStoreError::NotArchived(session_id.to_string()));
        }

        // Read archive file
        let archive_path = self.archive_path(session_id)?;
        if !archive_path.exists() {
            return Err(SessionStoreError::ArchiveNotFound(session_id.to_string()));
        }

        let archive_content = fs::read(&archive_path)?;
        self.verify_archive_integrity(&archive_path, &archive_content)
            .map_err(|err| Self::archive_read_error(session_id, "restore archive", err))?;
        let archived = self
            .decode_archive(session_id, &archive_content)
            .map_err(|err| Self::archive_read_error(session_id, "restore archive", err))?;

        let history_path = self.session_history_path(session_id)?;
        let _history_lock = FileLock::acquire(&history_path)?;
        self.rewrite_history_file_preserving_inbound_dedupe(
            &history_path,
            session_id,
            archived.messages.clone(),
            &archived.messages,
            &[],
        )?;

        let message_count = archived
            .messages
            .iter()
            .filter(|message| !Self::is_protected_inbound_dedupe_marker(message))
            .count();

        // Update session status to active
        session.status = SessionStatus::Active;
        session.message_count = message_count;
        session.updated_at = now_millis();
        session.last_activity_at = Some(now_millis());
        self.write_session_meta(&session)?;

        // Update cache
        {
            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session = session;
                cached.dirty = false;
            }
        }

        Ok(RestoreResult {
            session_id: session_id.to_string(),
            message_count,
            restored_at: now_millis(),
        })
    }

    /// List all archived sessions
    ///
    /// Returns sessions with Archived status along with archive file info
    pub fn list_archived_sessions(&self) -> Result<Vec<(Session, Option<u64>)>, SessionStoreError> {
        self.load_sessions_from_disk()?;

        let sessions = self.sessions.read();
        let mut result = Vec::new();

        for cached in sessions.values() {
            if cached.session.status == SessionStatus::Archived {
                // Skip sessions with invalid IDs (shouldn't happen, but defensive)
                let archive_path = match self.archive_path(&cached.session.id) {
                    Ok(p) => p,
                    Err(_) => continue,
                };
                let archive_size = if archive_path.exists() {
                    fs::metadata(&archive_path).ok().map(|m| m.len())
                } else {
                    None
                };
                result.push((cached.session.clone(), archive_size));
            }
        }

        // Sort by updated_at descending
        result.sort_by(|a, b| b.0.updated_at.cmp(&a.0.updated_at));

        Ok(result)
    }

    /// Delete an archive file without restoring
    ///
    /// This removes the archive file but keeps the session metadata.
    /// Use this to clean up archives that are no longer needed.
    pub fn delete_archive(&self, session_id: &str) -> Result<bool, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        let _ = self.get_session(session_id)?;
        let archive_path = self.archive_path(session_id)?;
        let existed = archive_path.exists();
        if existed {
            fs::remove_file(&archive_path)?;
        }
        self.delete_archive_hmac(&archive_path, session_id)?;
        self.clear_archive_file_current(session_id);
        Ok(existed)
    }

    /// Get archive info for a session
    pub fn get_archive_info(
        &self,
        session_id: &str,
    ) -> Result<Option<(PathBuf, u64, i64)>, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        let archive_path = self.archive_path(session_id)?;
        if !archive_path.exists() {
            return Ok(None);
        }

        let metadata = fs::metadata(&archive_path)?;
        let size = metadata.len();

        // Read archived_at from the file
        let content = fs::read(&archive_path)?;
        self.verify_archive_integrity(&archive_path, &content)
            .map_err(|err| Self::archive_read_error(session_id, "read archive info", err))?;
        let archived = self
            .decode_archive(session_id, &content)
            .map_err(|err| Self::archive_read_error(session_id, "read archive info", err))?;

        Ok(Some((archive_path, size, archived.archived_at)))
    }

    /// Archive old inactive sessions automatically
    ///
    /// Archives sessions that haven't been updated within the given threshold.
    ///
    /// # Arguments
    /// * `inactive_days` - Archive sessions not updated in this many days
    /// * `delete_history` - Whether to delete history after archiving
    ///
    /// # Returns
    /// Number of sessions archived
    pub fn archive_inactive_sessions(
        &self,
        inactive_days: u32,
        delete_history: bool,
    ) -> Result<Vec<ArchiveResult>, SessionStoreError> {
        self.load_sessions_from_disk()?;

        let cutoff = now_millis() - (inactive_days as i64 * 24 * 60 * 60 * 1000);
        let to_archive: Vec<String> = {
            let sessions = self.sessions.read();
            sessions
                .values()
                .filter(|c| {
                    c.session.status == SessionStatus::Active && c.session.updated_at < cutoff
                })
                .map(|c| c.session.id.clone())
                .collect()
        };

        let mut results = Vec::new();
        for session_id in to_archive {
            match self.archive_session(&session_id, delete_history) {
                Ok(result) => results.push(result),
                Err(e) => {
                    // Log but continue with other sessions
                    warn!(
                        error_kind = session_store_error_kind(&e),
                        "failed to archive inactive session"
                    );
                }
            }
        }

        Ok(results)
    }

    /// Load a session from disk
    fn load_session(&self, session_id: &str) -> Result<Session, SessionStoreError> {
        self.load_session_with_locked_tracking(session_id, true)
    }

    fn load_session_with_locked_tracking(
        &self,
        session_id: &str,
        update_locked_scan_state: bool,
    ) -> Result<Session, SessionStoreError> {
        self.ensure_required_encryption_available()?;
        let meta_path = self.session_meta_path(session_id)?;

        if !meta_path.exists() {
            return Err(SessionStoreError::NotFound(session_id.to_string()));
        }

        // SECURITY: route the session-meta read through the FIFO-safe,
        // no-symlink-follow, capped helper. The bare `fs::read` had
        // no cap: a same-uid attacker who plants a multi-GB file at
        // the meta path OOMs the daemon at the next session load.
        // Session metadata is a single session record (id, key,
        // channel info, last-message envelope) — realistic size is
        // a few KB; 4 MiB is a generous cap that still refuses the
        // planted-multi-GB attack.
        let content = match crate::paths::read_to_vec_no_hang_no_follow_capped(
            &meta_path,
            SESSION_META_FILE_MAX_BYTES,
        ) {
            Ok(Some(content)) => content,
            Ok(None) => return Err(SessionStoreError::NotFound(session_id.to_string())),
            Err(err) => return Err(SessionStoreError::Io(err.to_string())),
        };
        let meta_was_encrypted = super::crypto::has_encrypted_payload_prefix(&content);

        // Verify session integrity if HMAC key is configured
        self.verify_integrity_bytes(&content, &meta_path)?;

        let session = match self.decode_session_metadata(session_id, &content) {
            Ok(session) => session,
            Err(SessionStoreError::Locked(message)) => {
                if update_locked_scan_state {
                    self.record_locked_session_entry(
                        session_id.to_string(),
                        Self::file_updated_at(&meta_path),
                    );
                }
                return Err(SessionStoreError::Locked(message));
            }
            Err(err @ SessionStoreError::Crypto(_)) if meta_was_encrypted => {
                if update_locked_scan_state {
                    self.record_locked_session_entry(
                        session_id.to_string(),
                        Self::file_updated_at(&meta_path),
                    );
                }
                return Err(err);
            }
            Err(err) => return Err(err),
        };

        self.ensure_session_artifacts_current(&session)?;

        if update_locked_scan_state {
            self.clear_locked_session_entry(session_id);
        }

        // Update caches
        {
            let mut sessions = self.sessions.write();
            let mut key_map = self.key_to_id.write();
            key_map.insert(session.session_key.clone(), session.id.clone());
            sessions.insert(
                session.id.clone(),
                self.new_cached_session(session.clone(), false),
            );
        }
        self.mark_history_file_current(&session.id);

        Ok(session)
    }

    /// Load all sessions from disk into cache
    fn load_sessions_from_disk(&self) -> Result<(), SessionStoreError> {
        self.ensure_required_encryption_available()?;
        if !self.base_path.exists() {
            self.replace_locked_session_entries(HashMap::new());
            return Ok(());
        }

        let entries = fs::read_dir(&self.base_path)?;
        let mut locked_session_entries = HashMap::new();

        for entry in entries {
            let entry = entry?;
            let path = entry.path();

            if path.extension().and_then(|s| s.to_str()) == Some("json") {
                if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
                    // Skip if already cached
                    {
                        let sessions = self.sessions.read();
                        if sessions.contains_key(stem) {
                            continue;
                        }
                    }

                    match self.load_session_with_locked_tracking(stem, false) {
                        Ok(_) => {}
                        Err(SessionStoreError::Locked(_)) => {
                            locked_session_entries.insert(
                                stem.to_string(),
                                SessionListEntry::locked(
                                    stem.to_string(),
                                    Self::file_updated_at(&path),
                                ),
                            );
                        }
                        Err(
                            err @ (SessionStoreError::DecryptionFailed(_)
                            | SessionStoreError::ManifestIntegrityFailed(_)),
                        ) if self.crypto.is_some() => {
                            let manifest_locked = self
                                .crypto
                                .as_ref()
                                .is_some_and(|crypto| !crypto.manifest_integrity_valid());
                            if manifest_locked {
                                return Err(err);
                            }
                            locked_session_entries.insert(
                                stem.to_string(),
                                SessionListEntry::locked(
                                    stem.to_string(),
                                    Self::file_updated_at(&path),
                                ),
                            );
                            tracing::error!(
                                error_kind = session_store_error_kind(&err),
                                "skipping encrypted session metadata that failed authentication during scan"
                            );
                        }
                        Err(err @ SessionStoreError::Crypto(_)) if self.crypto.is_some() => {
                            locked_session_entries.insert(
                                stem.to_string(),
                                SessionListEntry::locked(
                                    stem.to_string(),
                                    Self::file_updated_at(&path),
                                ),
                            );
                            tracing::warn!(
                                error_kind = session_store_error_kind(&err),
                                "skipping unreadable encrypted session metadata during scan"
                            );
                        }
                        Err(err) => {
                            tracing::warn!(
                                error_kind = session_store_error_kind(&err),
                                "failed to load session metadata from disk"
                            );
                        }
                    }
                }
            }
        }

        self.replace_locked_session_entries(locked_session_entries);
        Ok(())
    }

    /// Write session metadata to disk
    fn write_session_meta(&self, session: &Session) -> Result<(), SessionStoreError> {
        self.ensure_base_dir()?;
        self.ensure_required_encryption_available()?;

        let meta_path = self.session_meta_path(&session.id)?;
        let _lock =
            FileLock::acquire(&meta_path).map_err(|e| SessionStoreError::Io(e.to_string()))?;

        let temp_path = crate::paths::atomic_tmp_path(&meta_path, "json");

        // Serialize to bytes so we can reuse for HMAC
        let serialized = self.encode_session_metadata(session)?;

        let write_result = (|| -> Result<(), SessionStoreError> {
            // Write to temp file first, then sync
            {
                let file = Self::create_private_output_file_create_new(&temp_path)?;
                let mut writer = BufWriter::new(file);
                writer.write_all(&serialized)?;
                writer.flush()?;
                writer
                    .into_inner()
                    .map_err(|e| std::io::Error::other(e.to_string()))?
                    .sync_all()?;
            }

            if let Some(ref key) = self.hmac_key {
                super::integrity::prepare_pending_hmac_file(key, &serialized, &meta_path).map_err(
                    |e| {
                        SessionStoreError::Io(format!(
                            "failed to stage HMAC sidecar for session {}: {}",
                            session.id, e
                        ))
                    },
                )?;
            }

            // Atomic rename
            fs::rename(&temp_path, &meta_path)?;
            #[cfg(test)]
            if TEST_FORCE_SESSION_META_PARENT_SYNC_FAIL.swap(false, Ordering::SeqCst) {
                return Err(SessionStoreError::Io(format!(
                    "failed to fsync parent directory after writing session {} metadata: forced session metadata parent fsync failure",
                    session.id
                )));
            }
            crate::paths::sync_parent_dir_blocking(&meta_path).map_err(|err| {
                SessionStoreError::Io(format!(
                    "failed to fsync parent directory after writing session {} metadata: {}",
                    session.id, err
                ))
            })?;

            // Commit HMAC sidecar if integrity is enabled.
            if self.hmac_key.is_some() {
                super::integrity::commit_pending_hmac_sidecar(&meta_path).map_err(|e| {
                    SessionStoreError::Io(format!(
                        "failed to commit HMAC sidecar for session {}: {}",
                        session.id, e
                    ))
                })?;
            }

            Ok(())
        })();
        if write_result.is_err() {
            let _ = fs::remove_file(&temp_path);
        }
        write_result
    }

    /// Flush a session's changes to disk
    fn flush_session(&self, session_id: &str) -> Result<(), SessionStoreError> {
        let session = {
            let sessions = self.sessions.read();
            sessions.get(session_id).map(|c| c.session.clone())
        };

        if let Some(session) = session {
            self.write_session_meta(&session)?;

            let mut sessions = self.sessions.write();
            if let Some(cached) = sessions.get_mut(session_id) {
                cached.dirty = false;
            }
        }

        Ok(())
    }

    /// Increment message count for a session
    fn increment_message_count(&self, session_id: &str) -> Result<(), SessionStoreError> {
        // Update message count in a scoped block to ensure write lock is released
        // before attempting to acquire read lock for periodic flush
        {
            let mut sessions = self.sessions.write();

            if let Some(cached) = sessions.get_mut(session_id) {
                cached.session.message_count += 1;
                cached.session.updated_at = now_millis();
                cached.session.last_activity_at = Some(now_millis());
                cached.dirty = true;
            } else {
                drop(sessions);

                // Load session if not cached
                let mut session = self.load_session(session_id)?;
                session.message_count += 1;
                session.updated_at = now_millis();
                session.last_activity_at = Some(now_millis());

                let mut sessions = self.sessions.write();
                if let Some(cached) = sessions.get_mut(session_id) {
                    cached.session = session;
                    cached.dirty = true;
                } else {
                    sessions.insert(
                        session_id.to_string(),
                        self.new_cached_session(session, true),
                    );
                }
            }
        }

        // Periodic flush (every 10 messages)
        {
            let sessions = self.sessions.read();
            if let Some(cached) = sessions.get(session_id) {
                if cached.session.message_count % 10 == 0 {
                    drop(sessions);
                    self.flush_session(session_id)?;
                }
            }
        }

        Ok(())
    }

    /// Flush all dirty sessions to disk
    pub fn flush_all(&self) -> Result<(), SessionStoreError> {
        let dirty_ids: Vec<String> = {
            let sessions = self.sessions.read();
            sessions
                .iter()
                .filter(|(_, c)| c.dirty)
                .map(|(id, _)| id.clone())
                .collect()
        };

        for id in dirty_ids {
            self.flush_session(&id)?;
        }

        Ok(())
    }

    /// Get the base path for session storage
    pub fn base_path(&self) -> &PathBuf {
        &self.base_path
    }

    /// Get session count
    pub fn session_count(&self) -> usize {
        self.sessions.read().len()
    }
}

/// Generate a session key from agent ID and metadata
fn generate_session_key(agent_id: &str, metadata: &SessionMetadata) -> String {
    let channel = metadata.channel.as_deref().unwrap_or("default");
    let chat_id = metadata.chat_id.as_deref().unwrap_or("default");

    format!("{}:{}:{}", agent_id, channel, chat_id)
}

fn canonical_inbound_event_id(value: &str) -> String {
    const V2_PREFIX: &str = "matrix-event-v2-sha256:";
    const V3_PREFIX: &str = "matrix-event-v3-sha256:";
    if let Some(rest) = value.strip_prefix(V2_PREFIX) {
        return format!("{V3_PREFIX}{rest}");
    }
    value.to_string()
}

fn metadata_string(metadata: &Value, key: &str) -> Option<String> {
    metadata
        .get(key)
        .and_then(Value::as_str)
        .map(ToString::to_string)
}

/// Get current time in milliseconds since Unix epoch
fn now_millis() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sessions::crypto::{self, EncryptionMode, SessionCryptoContext};
    use crate::sessions::integrity;
    use tempfile::TempDir;

    static TEST_SESSION_FAILURE_FLAGS_LOCK: std::sync::LazyLock<std::sync::Mutex<()>> =
        std::sync::LazyLock::new(|| std::sync::Mutex::new(()));

    #[cfg(unix)]
    fn assert_private_mode(path: &Path) {
        use std::os::unix::fs::PermissionsExt;

        let mode = fs::metadata(path).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600);
    }

    fn create_test_store() -> (SessionStore, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf());
        (store, temp_dir)
    }

    fn create_encrypted_store(password: &[u8]) -> (SessionStore, TempDir) {
        create_encrypted_store_with_mode(password, EncryptionMode::IfPassword)
    }

    fn create_encrypted_store_with_mode(
        password: &[u8],
        mode: EncryptionMode,
    ) -> (SessionStore, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let store = reopen_store_with_encryption(temp_dir.path(), Some(password), mode);
        (store, temp_dir)
    }

    fn create_encrypted_store_without_hmac(password: &[u8]) -> (SessionStore, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let crypto = SessionCryptoContext::load_or_create(temp_dir.path(), password).unwrap();
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_encryption_mode(EncryptionMode::IfPassword)
            .with_crypto_context(Arc::new(crypto));
        (store, temp_dir)
    }

    #[test]
    fn test_inbound_event_index_returns_original_run_id_on_duplicate() {
        let (store, _temp) = create_test_store();
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();

        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        let first_outcome = store
            .append_message_if_new_inbound(first, "$event:example.com")
            .unwrap();
        assert!(first_outcome.appended);

        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let duplicate_outcome = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .unwrap();
        assert!(!duplicate_outcome.appended);
        assert_eq!(
            duplicate_outcome.original_run_id.as_deref(),
            Some("run-original")
        );
    }

    #[test]
    fn test_protected_inbound_dedupe_signature_ignores_mtime_drift() {
        let (store, _temp) = create_test_store();
        let session = create_matrix_test_session(&store, "mtime-dedupe");
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        let first_outcome = store
            .append_message_if_new_inbound(first, "$event:example.com")
            .unwrap();
        assert!(first_outcome.appended);

        let history_path = store.session_history_path(&session.id).unwrap();
        let cached_history_signature = store
            .protected_inbound_dedupe_cache
            .read()
            .get(&session.id)
            .and_then(|cache| cache.history_signature.clone())
            .expect("first inbound append should seed protected dedupe cache");
        filetime::set_file_mtime(
            &history_path,
            filetime::FileTime::from_unix_time(1_700_000_000, 0),
        )
        .unwrap();
        assert_eq!(
            Some(cached_history_signature),
            SessionStore::protected_inbound_dedupe_history_signature(&history_path),
            "history mtime drift must not invalidate exact-byte protected dedupe cache identity"
        );

        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let duplicate_outcome = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .unwrap();
        assert!(!duplicate_outcome.appended);
        assert_eq!(
            duplicate_outcome.original_run_id.as_deref(),
            Some("run-original")
        );
    }

    #[test]
    fn test_protected_inbound_dedupe_cache_hit_is_content_bound_before_append_fast_path() {
        let _guard = TEST_SESSION_FAILURE_FLAGS_LOCK.lock().unwrap();
        TEST_PROTECTED_INBOUND_DEDUPE_HISTORY_SIGNATURE_READS.store(0, Ordering::SeqCst);
        let (store, _temp) = create_test_store();
        let session = create_matrix_test_session(&store, "hot-dedupe");
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event-one:example.com",
                "inbound_run_id": "run-one",
            }));
        assert!(
            store
                .append_message_if_new_inbound(first, "$event-one:example.com")
                .unwrap()
                .appended
        );

        TEST_PROTECTED_INBOUND_DEDUPE_HISTORY_SIGNATURE_READS.store(0, Ordering::SeqCst);
        let duplicate =
            ChatMessage::user(session.id.clone(), "duplicate").with_metadata(serde_json::json!({
                "inbound_event_id": "$event-one:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        assert!(
            !store
                .append_message_if_new_inbound(duplicate, "$event-one:example.com")
                .unwrap()
                .appended
        );
        assert_eq!(
            TEST_PROTECTED_INBOUND_DEDUPE_HISTORY_SIGNATURE_READS.load(Ordering::SeqCst),
            1,
            "protected dedupe cache hits must re-read history bytes before trusting cached entries"
        );

        let second =
            ChatMessage::user(session.id.clone(), "second").with_metadata(serde_json::json!({
                "inbound_event_id": "$event-two:example.com",
                "inbound_run_id": "run-two",
            }));
        assert!(
            store
                .append_message_if_new_inbound(second, "$event-two:example.com")
                .unwrap()
                .appended
        );
        assert_eq!(
            TEST_PROTECTED_INBOUND_DEDUPE_HISTORY_SIGNATURE_READS.load(Ordering::SeqCst),
            2,
            "new inbound events also re-read protected history before extending cached hash state"
        );
    }

    #[test]
    fn test_protected_inbound_dedupe_same_length_rewrite_invalidates_cache() {
        let (store, _temp) = create_test_store();
        let session = create_matrix_test_session(&store, "content-bound-dedupe");
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event-a:example.com",
                "inbound_run_id": "run-original",
            }));
        assert!(
            store
                .append_message_if_new_inbound(first, "$event-a:example.com")
                .unwrap()
                .appended
        );

        let history_path = store.session_history_path(&session.id).unwrap();
        let seeded_signature = store
            .protected_inbound_dedupe_cache
            .read()
            .get(&session.id)
            .and_then(|cache| cache.history_signature.clone())
            .expect("first append should seed content-bound cache");
        filetime::set_file_mtime(
            &history_path,
            filetime::FileTime::from_unix_time(1_700_000_000, 0),
        )
        .unwrap();
        let rewritten = fs::read_to_string(&history_path)
            .unwrap()
            .replace("$event-a:example.com", "$event-b:example.com");
        fs::write(&history_path, rewritten).unwrap();
        filetime::set_file_mtime(
            &history_path,
            filetime::FileTime::from_unix_time(1_700_000_000, 0),
        )
        .unwrap();

        let replayed_original = ChatMessage::user(session.id.clone(), "fresh original")
            .with_metadata(serde_json::json!({
                "inbound_event_id": "$event-a:example.com",
                "inbound_run_id": "run-after-rewrite",
            }));
        let outcome = store
            .append_message_if_new_inbound(replayed_original, "$event-a:example.com")
            .unwrap();
        assert!(
            outcome.appended,
            "same-length history rewrite must invalidate stale cached entries by content signature"
        );
        let cache = store.protected_inbound_dedupe_cache.read();
        let session_cache = cache.get(&session.id).expect("cache rebuilt after append");
        let rebuilt_signature = session_cache
            .history_signature
            .as_ref()
            .expect("cache remains content-bound after rebuild");
        assert_ne!(
            rebuilt_signature, &seeded_signature,
            "cache snapshot must advance after same-length content rewrite"
        );
        assert_eq!(
            session_cache
                .entries
                .get("$event-a:example.com")
                .and_then(|entry| entry.run_id.as_deref()),
            Some("run-after-rewrite"),
            "rebuilt cache must contain the appended event, not the stale pre-rewrite entry"
        );
    }

    #[test]
    fn test_protected_inbound_dedupe_cache_rejects_unsigned_complete_state() {
        let (store, _temp) = create_test_store();
        let session = create_matrix_test_session(&store, "unsigned-dedupe-cache");
        let mut entries = HashMap::new();
        entries.insert(
            "$event:example.com".to_string(),
            InboundEventIndexEntry {
                event_id: "$event:example.com".to_string(),
                run_id: Some("run-original".to_string()),
            },
        );
        store.replace_protected_inbound_dedupe_cache(&session.id, None, entries);

        let cache = store.protected_inbound_dedupe_cache.read();
        let session_cache = cache.get(&session.id).expect("cache entry");
        assert!(
            !session_cache.complete,
            "cache entries without a protected history signature must be incomplete"
        );
        assert!(
            session_cache.entries.is_empty(),
            "unsigned protected dedupe cache entries must not be retained"
        );
    }

    #[test]
    fn test_protected_inbound_dedupe_cache_ignores_marker_appends() {
        let (store, temp) = create_test_store();
        let session = create_matrix_test_session(&store, "marker-cache");
        let history_path = store
            .session_history_path(&session.id)
            .expect("history path");
        let appended = b"{\"role\":\"system\",\"content\":\"\"}\n";
        std::fs::create_dir_all(history_path.parent().expect("history parent")).unwrap();
        std::fs::write(&history_path, appended).unwrap();
        store.replace_protected_inbound_dedupe_cache(&session.id, None, HashMap::new());
        let before = {
            let cache = store.protected_inbound_dedupe_cache.read();
            let session_cache = cache.get(&session.id).expect("cache entry");
            (
                session_cache.complete,
                session_cache.history_signature.clone(),
                session_cache.history_hasher.is_some(),
                session_cache.entries.len(),
            )
        };

        let entry = InboundEventIndexEntry {
            event_id: "$marker-event:example.com".to_string(),
            run_id: None,
        };
        let marker = SessionStore::protected_inbound_dedupe_marker(&session.id, &entry);
        store.cache_protected_inbound_event_from_message(&marker, &history_path, appended);

        let cache = store.protected_inbound_dedupe_cache.read();
        let session_cache = cache.get(&session.id).expect("cache entry");
        assert_eq!(
            (
                session_cache.complete,
                session_cache.history_signature.clone(),
                session_cache.history_hasher.is_some(),
                session_cache.entries.len(),
            ),
            before,
            "dedupe marker appends must not be re-cached as fresh inbound events"
        );
        drop(cache);
        drop(temp);
    }

    #[test]
    fn test_protected_inbound_dedupe_marker_requires_exact_system_empty_shape() {
        let entry = InboundEventIndexEntry {
            event_id: "$marker-event:example.com".to_string(),
            run_id: Some("run-original".to_string()),
        };
        let marker = SessionStore::protected_inbound_dedupe_marker("session", &entry);
        assert!(SessionStore::is_protected_inbound_dedupe_marker(&marker));

        let mut user_visible = marker.clone();
        user_visible.role = MessageRole::User;
        assert!(
            !SessionStore::is_protected_inbound_dedupe_marker(&user_visible),
            "visible user messages must not become protected markers by metadata bit alone"
        );

        let mut non_empty_system = marker.clone();
        non_empty_system.content = "operator-visible".to_string();
        assert!(
            !SessionStore::is_protected_inbound_dedupe_marker(&non_empty_system),
            "system messages with visible content must not be hidden as dedupe markers"
        );

        let mut extra_metadata = marker.clone();
        extra_metadata.metadata = Some(serde_json::json!({
            "inbound_event_id": "$marker-event:example.com",
            "inbound_run_id": "run-original",
            "inbound_dedupe_marker": true,
            "unexpected": "value",
        }));
        assert!(
            !SessionStore::is_protected_inbound_dedupe_marker(&extra_metadata),
            "protected marker metadata must contain only the store-generated marker fields"
        );
    }

    #[test]
    fn test_inbound_event_index_rejects_corrupt_protected_history_lines() {
        let (store, _temp) = create_test_store();
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        store
            .append_message_if_new_inbound(first, "$event:example.com")
            .unwrap();
        let history_path = store.session_history_path(&session.id).unwrap();
        let mut file = OpenOptions::new().append(true).open(&history_path).unwrap();
        file.write_all(b"{not-json\n").unwrap();
        file.sync_all().unwrap();

        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let err = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .expect_err("corrupt protected history must fail closed");
        assert!(
            matches!(err, SessionStoreError::HistoryCorrupt(_)),
            "corrupt protected history must return the typed HistoryCorrupt error"
        );
    }

    #[test]
    fn test_inbound_event_index_rejects_history_hmac_failure_as_corrupt() {
        let temp_dir = TempDir::new().unwrap();
        let hmac_key = Zeroizing::new(integrity::derive_hmac_key(b"history-dedupe-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(hmac_key)
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let session = create_matrix_test_session(&store, "hmac-dedupe");
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        store
            .append_message_if_new_inbound(first, "$event:example.com")
            .unwrap();
        let history_path = store.session_history_path(&session.id).unwrap();
        let mut file = OpenOptions::new().append(true).open(&history_path).unwrap();
        file.write_all(b"\n").unwrap();
        file.sync_all().unwrap();

        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let err = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .expect_err("history HMAC failure must fail closed during dedupe");

        assert!(
            matches!(err, SessionStoreError::HistoryCorrupt(ref message)
                if message.contains("integrity verification failed")),
            "HMAC failures during dedupe must be typed HistoryCorrupt, got {err:?}"
        );
    }

    #[test]
    fn test_inbound_event_index_rejects_encrypted_line_decrypt_failure_as_corrupt() {
        let password = test_key_material();
        let (store, _temp) = create_encrypted_store_without_hmac(&password);
        let session = create_matrix_test_session(&store, "encrypted-dedupe");
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        store
            .append_message_if_new_inbound(first, "$event:example.com")
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let cached_history_signature = store
            .protected_inbound_dedupe_cache
            .read()
            .get(&session.id)
            .and_then(|cache| cache.history_signature.clone())
            .expect("first inbound append should seed protected dedupe cache");
        let encrypted_line = fs::read_to_string(&history_path)
            .unwrap()
            .lines()
            .next()
            .unwrap()
            .to_string();
        let prefix = std::str::from_utf8(crypto::encrypted_payload_prefix()).unwrap();
        let mut envelope: serde_json::Value =
            serde_json::from_str(encrypted_line.strip_prefix(prefix).unwrap()).unwrap();
        let ciphertext = envelope["c"].as_str().unwrap();
        let replacement = if ciphertext.ends_with('A') { 'B' } else { 'A' };
        let mut tampered_ciphertext = ciphertext.to_string();
        tampered_ciphertext.pop();
        tampered_ciphertext.push(replacement);
        envelope["c"] = serde_json::Value::String(tampered_ciphertext);
        fs::write(
            &history_path,
            format!("{prefix}{}\n", serde_json::to_string(&envelope).unwrap()),
        )
        .unwrap();
        assert_ne!(
            Some(cached_history_signature),
            SessionStore::protected_inbound_dedupe_history_signature(&history_path),
            "same-length encrypted history rewrites must invalidate protected dedupe cache identity"
        );

        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let err = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .expect_err("encrypted history authentication failure must fail closed");

        assert!(
            matches!(err, SessionStoreError::HistoryCorrupt(ref message)
                if message.contains("encrypted session history authentication failed")),
            "encrypted decrypt failures during dedupe must be typed HistoryCorrupt, got {err:?}"
        );
    }

    #[test]
    fn test_inbound_event_index_ignores_corrupt_legacy_index_line() {
        let (store, _temp) = create_test_store();
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();
        let index_path = store.session_inbound_event_index_path(&session.id).unwrap();
        fs::write(&index_path, b"{not-json\n").unwrap();
        let message =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));

        let outcome = store
            .append_message_if_new_inbound(message, "$event:example.com")
            .unwrap();
        assert!(outcome.appended);
        assert_eq!(outcome.corrupt_index_lines, 1);
        assert!(
            !index_path.exists(),
            "legacy plaintext dedupe sidecar must be removed after protected history append"
        );
    }

    #[test]
    fn test_legacy_only_inbound_dedupe_deletes_sidecar_and_writes_protected_marker() {
        let (store, _temp) = create_test_store();
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();
        let index_path = store.session_inbound_event_index_path(&session.id).unwrap();
        let legacy_entry = InboundEventIndexEntry {
            event_id: "$event:example.com".to_string(),
            run_id: Some("legacy-run".to_string()),
        };
        fs::write(
            &index_path,
            format!("{}\n", serde_json::to_string(&legacy_entry).unwrap()),
        )
        .unwrap();
        let message =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "new-run",
            }));
        seed_protected_inbound_dedupe_cache_for_test(&store, &session.id);

        let outcome = store
            .append_message_if_new_inbound(message, "$event:example.com")
            .unwrap();

        assert!(!outcome.appended);
        assert_eq!(
            outcome.original_run_id.as_deref(),
            Some("legacy-run"),
            "legacy sidecar matches should preserve the best-known original run id"
        );
        assert!(
            !index_path.exists(),
            "legacy plaintext dedupe sidecar must be removed after a legacy-only match"
        );
        assert_protected_inbound_dedupe_cache_cleared(&store, &session.id);
        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "new-run-2",
            }));
        let duplicate_outcome = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .unwrap();
        assert!(
            !duplicate_outcome.appended,
            "protected marker must suppress redelivery after legacy sidecar deletion"
        );
        let history = store.get_history(&session.id, None, None).unwrap();
        assert!(
            history.is_empty(),
            "dedupe-only markers must not be returned as chat history"
        );
    }

    #[test]
    fn test_inbound_event_dedupe_does_not_write_plaintext_sidecar() {
        let password = test_key_material();
        let (store, _temp) = create_encrypted_store(&password);
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();
        let index_path = store.session_inbound_event_index_path(&session.id).unwrap();
        let message =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "matrix-event-v3-sha256:abc",
                "inbound_run_id": "run-original",
            }));

        store
            .append_message_if_new_inbound(message, "matrix-event-v3-sha256:abc")
            .unwrap();

        assert!(
            !index_path.exists(),
            "inbound dedupe must be stored in encrypted history, not a plaintext sidecar"
        );
        let history_path = store.session_history_path(&session.id).unwrap();
        let history_bytes = fs::read(history_path).unwrap();
        assert!(
            !String::from_utf8_lossy(&history_bytes).contains("matrix-event-v3-sha256:abc"),
            "encrypted history must not expose Matrix idempotency keys in plaintext"
        );
    }

    #[test]
    fn test_inbound_event_dedupe_normalizes_matrix_v2_to_v3() {
        let (store, _temp) = create_test_store();
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "matrix-event-v2-sha256:abc123",
                "inbound_run_id": "run-original",
            }));
        store
            .append_message_if_new_inbound(first, "matrix-event-v2-sha256:abc123")
            .unwrap();
        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "matrix-event-v3-sha256:abc123",
                "inbound_run_id": "run-duplicate",
            }));

        let outcome = store
            .append_message_if_new_inbound(duplicate, "matrix-event-v3-sha256:abc123")
            .unwrap();

        assert!(!outcome.appended);
        assert_eq!(outcome.original_run_id.as_deref(), Some("run-original"));
    }

    #[test]
    fn test_inbound_event_dedupe_cache_invalidates_on_clear_history() {
        let (store, _temp) = create_test_store();
        let session = store
            .get_or_create_session(
                "matrix:!room:example.com",
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some("!room:example.com".to_string()),
                    ..Default::default()
                },
            )
            .unwrap();
        let first =
            ChatMessage::user(session.id.clone(), "hello").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        store
            .append_message_if_new_inbound(first, "$event:example.com")
            .unwrap();
        let duplicate =
            ChatMessage::user(session.id.clone(), "hello again").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        assert!(
            !store
                .append_message_if_new_inbound(duplicate, "$event:example.com")
                .unwrap()
                .appended,
            "duplicate should be served from protected dedupe state before clear"
        );

        store.clear_history(&session.id).unwrap();
        let after_clear =
            ChatMessage::user(session.id.clone(), "after clear").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-after-clear",
            }));
        let outcome = store
            .append_message_if_new_inbound(after_clear, "$event:example.com")
            .unwrap();

        assert!(
            outcome.appended,
            "clearing history must invalidate protected inbound dedupe cache"
        );
    }

    #[test]
    fn test_inbound_event_dedupe_cache_invalidates_on_history_mutations() {
        let (store, _temp) = create_test_store();

        let append_session = create_matrix_test_session(&store, "append");
        seed_protected_inbound_dedupe_cache_for_test(&store, &append_session.id);
        store
            .append_message(ChatMessage::user(append_session.id.clone(), "append"))
            .unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &append_session.id);

        let batch_session = create_matrix_test_session(&store, "batch");
        seed_protected_inbound_dedupe_cache_for_test(&store, &batch_session.id);
        store
            .append_messages(&[
                ChatMessage::user(batch_session.id.clone(), "one"),
                ChatMessage::user(batch_session.id.clone(), "two"),
            ])
            .unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &batch_session.id);

        let reset_session = create_matrix_test_session(&store, "reset");
        seed_protected_inbound_dedupe_cache_for_test(&store, &reset_session.id);
        store.reset_session(&reset_session.id).unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &reset_session.id);

        let delete_session = create_matrix_test_session(&store, "delete");
        seed_protected_inbound_dedupe_cache_for_test(&store, &delete_session.id);
        store.delete_session(&delete_session.id).unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &delete_session.id);

        let compact_session = create_matrix_test_session(&store, "compact");
        store
            .append_messages(&[
                ChatMessage::user(compact_session.id.clone(), "one"),
                ChatMessage::user(compact_session.id.clone(), "two"),
            ])
            .unwrap();
        seed_protected_inbound_dedupe_cache_for_test(&store, &compact_session.id);
        store
            .compact_session(&compact_session.id, 1, |_| "summary".to_string())
            .unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &compact_session.id);

        let archive_session = create_matrix_test_session(&store, "archive");
        store
            .append_message(ChatMessage::user(archive_session.id.clone(), "archive"))
            .unwrap();
        seed_protected_inbound_dedupe_cache_for_test(&store, &archive_session.id);
        store.archive_session(&archive_session.id, true).unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &archive_session.id);

        let restore_session = create_matrix_test_session(&store, "restore");
        store
            .append_message(ChatMessage::user(restore_session.id.clone(), "restore"))
            .unwrap();
        store.archive_session(&restore_session.id, false).unwrap();
        seed_protected_inbound_dedupe_cache_for_test(&store, &restore_session.id);
        store.restore_session(&restore_session.id).unwrap();
        assert_protected_inbound_dedupe_cache_cleared(&store, &restore_session.id);
    }

    #[test]
    fn test_compaction_preserves_dedupe_for_compacted_inbound_messages() {
        let (store, _temp) = create_test_store();
        let session = create_matrix_test_session(&store, "compact-dedupe");
        let inbound = ChatMessage::user(session.id.clone(), "compacted inbound").with_metadata(
            serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }),
        );
        store
            .append_message_if_new_inbound(inbound, "$event:example.com")
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "assistant"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "kept"))
            .unwrap();

        store
            .compact_session(&session.id, 1, |_| "summary".to_string())
            .unwrap();

        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 2);
        assert!(
            history
                .iter()
                .all(|message| message.content != "compacted inbound"),
            "the original inbound message should be compacted out of visible history"
        );
        let history_path = store.session_history_path(&session.id).unwrap();
        let raw_history = store
            .read_raw_history_messages(&history_path, &session.id)
            .unwrap();
        assert!(
            raw_history
                .last()
                .is_some_and(SessionStore::is_protected_inbound_dedupe_marker),
            "compaction intentionally tail-batches order-independent dedupe markers"
        );

        let duplicate =
            ChatMessage::user(session.id.clone(), "redelivered").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let outcome = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .unwrap();

        assert!(!outcome.appended);
        assert_eq!(outcome.original_run_id.as_deref(), Some("run-original"));
    }

    #[test]
    fn test_compaction_restores_active_status_after_precommit_history_failure() {
        let temp_dir = TempDir::new().unwrap();
        let hmac_key = Zeroizing::new(integrity::derive_hmac_key(b"compaction-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(hmac_key)
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let session = create_matrix_test_session(&store, "compact-failure");
        store
            .append_message(ChatMessage::user(&session.id, "one"))
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "two"))
            .unwrap();
        let history_path = store.session_history_path(&session.id).unwrap();
        std::fs::write(&history_path, b"tampered\n").unwrap();

        let err = store
            .compact_session(&session.id, 1, |_| "summary".to_string())
            .expect_err("tampered history must abort compaction");

        assert!(matches!(err, SessionStoreError::HistoryCorrupt(_)));
        assert_eq!(
            store.get_session(&session.id).unwrap().status,
            SessionStatus::Active
        );
    }

    #[test]
    fn test_write_session_meta_parent_fsync_failure_propagates() {
        let _guard = TEST_SESSION_FAILURE_FLAGS_LOCK.lock().unwrap();
        TEST_FORCE_SESSION_META_PARENT_SYNC_FAIL.store(false, Ordering::SeqCst);
        let (store, _temp) = create_test_store();
        let mut session = create_matrix_test_session(&store, "meta-fsync");
        session.metadata.tags.push("updated".to_string());

        TEST_FORCE_SESSION_META_PARENT_SYNC_FAIL.store(true, Ordering::SeqCst);
        let err = store
            .write_session_meta(&session)
            .expect_err("metadata parent fsync failure must propagate");

        assert!(
            matches!(err, SessionStoreError::Io(ref message)
                if message.contains("forced session metadata parent fsync failure")),
            "unexpected metadata fsync error: {err:?}"
        );
    }

    #[test]
    fn test_permanent_history_corruption_classifies_manifest_and_decrypt_failures() {
        for err in [
            SessionStoreError::ManifestIntegrityFailed("tampered manifest".to_string()),
            SessionStoreError::DecryptionFailed("bad encrypted line".to_string()),
            SessionStoreError::HistoryCorrupt("corrupt protected history".to_string()),
            SessionStoreError::IntegrityRejected("hmac mismatch".to_string()),
        ] {
            assert!(
                err.is_permanent_history_corruption(),
                "{err:?} must block inbound replay until operator repair"
            );
        }
    }

    #[test]
    fn test_compaction_post_rename_failure_surfaces_and_blocks_dedupe_replay() {
        let _guard = TEST_SESSION_FAILURE_FLAGS_LOCK.lock().unwrap();
        TEST_FORCE_HISTORY_PARENT_SYNC_AFTER_RENAME_FAIL.store(false, Ordering::SeqCst);
        let temp_dir = TempDir::new().unwrap();
        let hmac_key = Zeroizing::new(integrity::derive_hmac_key(b"compaction-post-rename"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(hmac_key)
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let session = create_matrix_test_session(&store, "compact-post-rename");
        let inbound =
            ChatMessage::user(session.id.clone(), "inbound").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }));
        store
            .append_message_if_new_inbound(inbound, "$event:example.com")
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "assistant"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "kept"))
            .unwrap();

        TEST_FORCE_HISTORY_PARENT_SYNC_AFTER_RENAME_FAIL.store(true, Ordering::SeqCst);
        let err = store
            .compact_session(&session.id, 1, |_| "summary".to_string())
            .expect_err("post-rename history fsync failure must surface");

        assert!(
            matches!(err, SessionStoreError::Io(ref message)
                if message.contains("forced history parent fsync failure after rename")),
            "unexpected compaction error: {err:?}"
        );
        assert_eq!(
            store.get_session(&session.id).unwrap().status,
            SessionStatus::Active,
            "the transient Compacting marker is rolled back even when history is torn"
        );
        store.protected_inbound_dedupe_cache.write().clear();
        let duplicate =
            ChatMessage::user(session.id.clone(), "redelivered").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let replay = store.append_message_if_new_inbound(duplicate, "$event:example.com");
        match replay {
            Ok(outcome) => assert!(
                !outcome.appended,
                "recovered post-rename state must still preserve protected dedupe markers"
            ),
            Err(err) => assert!(
                err.is_permanent_history_corruption(),
                "unrecovered post-rename state must fail closed, got {err:?}"
            ),
        }
        // Without a full compaction transaction/repair marker, post-rename
        // power-loss recovery may either recover the HMAC sidecar or fail
        // closed. It must never silently replay inbound work.
    }

    #[test]
    fn test_encrypted_archive_and_restore_preserve_compacted_dedupe_markers() {
        let password = test_key_material();
        let (store, _temp) = create_encrypted_store(&password);
        let session = create_matrix_test_session(&store, "archive-dedupe");
        let inbound = ChatMessage::user(session.id.clone(), "compacted inbound").with_metadata(
            serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-original",
            }),
        );
        store
            .append_message_if_new_inbound(inbound, "$event:example.com")
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "assistant"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "kept"))
            .unwrap();
        store
            .compact_session(&session.id, 1, |_| "summary".to_string())
            .unwrap();

        let archive = store.archive_session(&session.id, false).unwrap();
        let history_path = store.session_history_path(&session.id).unwrap();
        let archived_live_history = store
            .read_raw_history_messages(&history_path, &session.id)
            .unwrap();
        assert!(
            archived_live_history
                .iter()
                .any(SessionStore::is_protected_inbound_dedupe_marker),
            "encrypted archive rewrite must keep protected dedupe markers in live history"
        );

        store.restore_session(&archive.session_id).unwrap();
        let duplicate =
            ChatMessage::user(session.id.clone(), "redelivered").with_metadata(serde_json::json!({
                "inbound_event_id": "$event:example.com",
                "inbound_run_id": "run-duplicate",
            }));
        let outcome = store
            .append_message_if_new_inbound(duplicate, "$event:example.com")
            .unwrap();

        assert!(!outcome.appended);
        assert_eq!(outcome.original_run_id.as_deref(), Some("run-original"));
    }

    #[test]
    fn test_protected_inbound_dedupe_cache_over_cap_is_incomplete() {
        let (store, _temp) = create_test_store();
        let session = create_matrix_test_session(&store, "cache-cap");
        let entries = (0..=PROTECTED_INBOUND_DEDUPE_CACHE_MAX_ENTRIES)
            .map(|idx| {
                (
                    format!("event-{idx}"),
                    InboundEventIndexEntry {
                        event_id: format!("$event-{idx}:example.com"),
                        run_id: Some(format!("run-{idx}")),
                    },
                )
            })
            .collect();

        store.replace_protected_inbound_dedupe_cache(&session.id, None, entries);

        let cache = store.protected_inbound_dedupe_cache.read();
        let session_cache = cache
            .get(&session.id)
            .expect("session cache entry should be present");
        assert!(
            !session_cache.complete,
            "over-cap cache must be marked incomplete so lookups rescan protected history"
        );
        assert!(
            session_cache.entries.is_empty(),
            "over-cap cache must not retain unbounded per-session entries"
        );
    }

    fn test_key_material() -> Vec<u8> {
        crate::test_support::secrets::random_test_secret_bytes(32)
    }

    fn create_matrix_test_session(store: &SessionStore, name: &str) -> Session {
        store
            .get_or_create_session(
                format!("matrix:!{name}:example.com"),
                SessionMetadata {
                    channel: Some("matrix".to_string()),
                    chat_id: Some(format!("!{name}:example.com")),
                    ..Default::default()
                },
            )
            .unwrap()
    }

    fn seed_protected_inbound_dedupe_cache_for_test(store: &SessionStore, session_id: &str) {
        let mut entries = HashMap::new();
        entries.insert(
            "seed-event".to_string(),
            InboundEventIndexEntry {
                event_id: "$seed:example.com".to_string(),
                run_id: Some("seed-run".to_string()),
            },
        );
        store.protected_inbound_dedupe_cache.write().insert(
            session_id.to_string(),
            ProtectedInboundDedupeCache {
                complete: true,
                history_signature: None,
                history_hasher: None,
                entries,
            },
        );
    }

    fn assert_protected_inbound_dedupe_cache_cleared(store: &SessionStore, session_id: &str) {
        assert!(
            !store
                .protected_inbound_dedupe_cache
                .read()
                .contains_key(session_id),
            "history mutation should invalidate protected inbound dedupe cache"
        );
    }

    fn hmac_sidecar_path(path: &Path) -> PathBuf {
        let mut sidecar = path.as_os_str().to_owned();
        sidecar.push(".hmac");
        PathBuf::from(sidecar)
    }

    fn reopen_store_with_encryption(
        base_path: &Path,
        password: Option<&[u8]>,
        mode: EncryptionMode,
    ) -> SessionStore {
        let mut store =
            SessionStore::with_base_path(base_path.to_path_buf()).with_encryption_mode(mode);
        if let Some(password) = password {
            let crypto = SessionCryptoContext::load_or_create(base_path, password).unwrap();
            let hmac_key = crypto.integrity_hmac_key();
            store = store.with_crypto_context(Arc::new(crypto));
            if let Some(hmac_key) = hmac_key {
                store = store.with_hmac_key(hmac_key);
            }
        }
        store
    }

    #[test]
    fn test_session_status_default() {
        assert_eq!(SessionStatus::default(), SessionStatus::Active);
    }

    #[test]
    fn test_session_status_display() {
        assert_eq!(SessionStatus::Active.to_string(), "active");
        assert_eq!(SessionStatus::Paused.to_string(), "paused");
        assert_eq!(SessionStatus::Archived.to_string(), "archived");
        assert_eq!(SessionStatus::Compacting.to_string(), "compacting");
    }

    #[test]
    fn test_message_role_default() {
        assert_eq!(MessageRole::default(), MessageRole::User);
    }

    #[test]
    fn test_message_role_display() {
        assert_eq!(MessageRole::User.to_string(), "user");
        assert_eq!(MessageRole::Assistant.to_string(), "assistant");
        assert_eq!(MessageRole::System.to_string(), "system");
        assert_eq!(MessageRole::Tool.to_string(), "tool");
        assert_eq!(MessageRole::Unknown.to_string(), "unknown");
    }

    /// Forward-compat regression: pins that an older binary reading a
    /// history line written by a newer daemon does NOT escalate a
    /// `role: "future_role"` value to `SessionStoreError::Serialization`
    /// → `SessionStoreError::HistoryCorrupt`, which is permanent at
    /// `is_permanent_history_corruption` and would refuse to replay the
    /// affected channel until manual operator repair. Unknown role
    /// resolves to `MessageRole::Unknown`.
    /// Forward-compat regression: pins that an unknown session `status`
    /// value written by a newer daemon does NOT escalate to
    /// SessionStoreError::Serialization and silently drop the session
    /// from the listing. Unknown maps to the `Unknown` sentinel; the
    /// session stays visible and is not treated as Archived/Compacting.
    #[test]
    fn test_session_status_forward_compat_unknown_falls_back_to_unknown() {
        let raw = serde_json::json!({
            "id": "sess-1",
            "session_key": "telegram:1:default",
            "status": "future_status_v2",
            "metadata": {},
            "message_count": 0,
            "created_at": 0i64,
            "updated_at": 0i64,
            "agent_id": null,
            "history_path": null,
            "encryption_mode": "warn",
            "salt": null,
            "manifest_path": null,
            "manifest_version": null
        });
        let session: Session = serde_json::from_value(raw)
            .expect("forward-compat: unknown session status must NOT hard-error session load");
        assert_eq!(
            session.status,
            SessionStatus::Unknown,
            "unknown status must fall back to Unknown sentinel"
        );
        assert_ne!(
            session.status,
            SessionStatus::Archived,
            "Unknown must not match Archived predicates"
        );
        assert_ne!(
            session.status,
            SessionStatus::Compacting,
            "Unknown must not match Compacting predicates"
        );
    }

    #[test]
    fn test_chat_message_role_forward_compat_unknown_falls_back_to_unknown() {
        let raw = serde_json::json!({
            "id": "msg-1",
            "session_id": "sess-1",
            "role": "future_role_v2",
            "content": "future content",
            "created_at": 0i64
        });
        let msg: ChatMessage = serde_json::from_value(raw)
            .expect("forward-compat: unknown role must NOT hard-error history line parse");
        assert_eq!(
            msg.role,
            MessageRole::Unknown,
            "unknown role must fall back to Unknown sentinel"
        );
    }

    #[test]
    fn test_session_creation() {
        let metadata = SessionMetadata {
            name: Some("Test Session".into()),
            channel: Some("telegram".into()),
            chat_id: Some("123456".into()),
            ..Default::default()
        };

        let session = Session::new("default", metadata);

        assert!(!session.id.is_empty());
        assert_eq!(session.session_key, "default:telegram:123456");
        assert_eq!(session.status, SessionStatus::Active);
        assert_eq!(session.metadata.agent_id, Some("default".into()));
        assert_eq!(session.metadata.name, Some("Test Session".into()));
        assert_eq!(session.message_count, 0);
        assert!(session.created_at > 0);
    }

    #[test]
    fn test_session_with_session_key() {
        let session = Session::with_session_key(
            "custom:key:here",
            SessionMetadata {
                name: Some("Custom".into()),
                ..Default::default()
            },
        );

        assert_eq!(session.session_key, "custom:key:here");
    }

    #[test]
    fn test_session_needs_compaction() {
        let mut session = Session::new("agent", SessionMetadata::default());

        assert!(!session.needs_compaction());

        session.message_count = 100;
        assert!(session.needs_compaction());

        session.message_count = 500;
        assert!(session.force_compaction_needed());
    }

    #[test]
    fn test_chat_message_creation() {
        let msg = ChatMessage::user("session-123", "Hello, world!");

        assert!(!msg.id.is_empty());
        assert_eq!(msg.session_id, "session-123");
        assert_eq!(msg.role, MessageRole::User);
        assert_eq!(msg.content, "Hello, world!");
        assert!(msg.created_at > 0);
    }

    #[test]
    fn test_chat_message_builders() {
        let user_msg = ChatMessage::user("s1", "Hi");
        assert_eq!(user_msg.role, MessageRole::User);

        let assistant_msg = ChatMessage::assistant("s1", "Hello!");
        assert_eq!(assistant_msg.role, MessageRole::Assistant);

        let system_msg = ChatMessage::system("s1", "Instructions");
        assert_eq!(system_msg.role, MessageRole::System);

        let tool_msg = ChatMessage::tool("s1", "calculator", "call-1", "42");
        assert_eq!(tool_msg.role, MessageRole::Tool);
        assert_eq!(tool_msg.tool_name, Some("calculator".into()));
        assert_eq!(tool_msg.tool_call_id, Some("call-1".into()));
    }

    #[test]
    fn test_chat_message_with_tokens() {
        let msg = ChatMessage::user("s1", "Hello").with_tokens(10);
        assert_eq!(msg.tokens, Some(10));
    }

    #[test]
    fn test_session_filter() {
        let filter = SessionFilter::new()
            .with_agent_id("agent-1")
            .with_status(SessionStatus::Active)
            .with_channel("telegram")
            .with_limit(10)
            .with_offset(5);

        assert_eq!(filter.agent_id, Some("agent-1".into()));
        assert_eq!(filter.status, Some(SessionStatus::Active));
        assert_eq!(filter.channel, Some("telegram".into()));
        assert_eq!(filter.limit, Some(10));
        assert_eq!(filter.offset, Some(5));
    }

    #[test]
    fn test_session_filter_matches() {
        let session = Session::new(
            "agent-1",
            SessionMetadata {
                channel: Some("telegram".into()),
                ..Default::default()
            },
        );

        let filter = SessionFilter::new()
            .with_agent_id("agent-1")
            .with_channel("telegram");

        assert!(filter.matches(&session));

        let non_matching_filter = SessionFilter::new().with_agent_id("agent-2");
        assert!(!non_matching_filter.matches(&session));
    }

    #[test]
    fn test_store_create_session() {
        let (store, _temp) = create_test_store();

        let metadata = SessionMetadata {
            name: Some("Test".into()),
            ..Default::default()
        };

        let session = store.create_session("agent-1", metadata).unwrap();

        assert!(!session.id.is_empty());
        assert_eq!(session.metadata.agent_id, Some("agent-1".into()));

        // Verify persistence
        let loaded = store.get_session(&session.id).unwrap();
        assert_eq!(loaded.id, session.id);
        assert_eq!(loaded.metadata.name, Some("Test".into()));
    }

    #[test]
    fn test_store_duplicate_session_key() {
        let (store, _temp) = create_test_store();

        let metadata = SessionMetadata {
            channel: Some("telegram".into()),
            chat_id: Some("123".into()),
            ..Default::default()
        };

        store.create_session("agent-1", metadata.clone()).unwrap();

        // Should fail with duplicate key
        let result = store.create_session("agent-1", metadata);
        assert!(matches!(result, Err(SessionStoreError::AlreadyExists(_))));
    }

    #[test]
    fn test_store_get_session_by_key() {
        let (store, _temp) = create_test_store();

        let metadata = SessionMetadata {
            channel: Some("discord".into()),
            chat_id: Some("456".into()),
            ..Default::default()
        };

        let session = store.create_session("agent-2", metadata).unwrap();
        let loaded = store.get_session_by_key(&session.session_key).unwrap();

        assert_eq!(loaded.id, session.id);
    }

    #[test]
    fn test_encrypted_session_store_round_trip() {
        let key_material = test_key_material();
        let (store, _temp) = create_encrypted_store(&key_material);

        let metadata = SessionMetadata {
            name: Some("Secret chat".into()),
            ..Default::default()
        };
        let session = store.create_session("agent-1", metadata).unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "top secret"))
            .unwrap();

        let meta_path = store.session_meta_path(&session.id).unwrap();
        let meta_raw = fs::read(&meta_path).unwrap();
        assert!(crypto::is_encrypted_payload(&meta_raw));
        #[cfg(unix)]
        assert_private_mode(&meta_path);

        let history_path = store.session_history_path(&session.id).unwrap();
        let history_raw = fs::read_to_string(&history_path).unwrap();
        let first_line = history_raw
            .lines()
            .find(|line| !line.trim().is_empty())
            .unwrap();
        assert!(crypto::is_encrypted_payload(first_line.as_bytes()));
        #[cfg(unix)]
        assert_private_mode(&history_path);

        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].content, "top secret");

        let archived = store.archive_session(&session.id, false).unwrap();
        let archive_path = PathBuf::from(&archived.archive_path);
        let archive_raw = fs::read(&archive_path).unwrap();
        assert!(crypto::is_encrypted_payload(&archive_raw));
        #[cfg(unix)]
        assert_private_mode(&archive_path);
    }

    #[test]
    fn test_if_password_without_password_lists_locked_encrypted_sessions() {
        let key_material = test_key_material();
        let (encrypted_store, temp_dir) = create_encrypted_store(&key_material);
        let session = encrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        encrypted_store
            .append_message(ChatMessage::user(&session.id, "hello"))
            .unwrap();

        let locked_store =
            reopen_store_with_encryption(temp_dir.path(), None, EncryptionMode::IfPassword);
        let entries = locked_store
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].session_id, session.id);
        assert_eq!(entries[0].access, SessionAccessState::Locked);
        assert!(entries[0].session.is_none());

        let err = locked_store.get_session(&session.id).unwrap_err();
        assert!(matches!(err, SessionStoreError::Locked(_)));

        let err = locked_store
            .get_session_by_key(&session.session_key)
            .unwrap_err();
        assert!(matches!(err, SessionStoreError::Locked(_)));
    }

    #[test]
    fn test_off_mode_without_password_blocks_encrypted_key_lookup_and_create() {
        let key_material = test_key_material();
        let (encrypted_store, temp_dir) = create_encrypted_store(&key_material);
        let session = encrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        let locked_store = reopen_store_with_encryption(temp_dir.path(), None, EncryptionMode::Off);
        let entries = locked_store
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].session_id(), session.id);
        assert_eq!(entries[0].access(), SessionAccessState::Locked);

        assert!(matches!(
            locked_store.get_session_by_key(&session.session_key),
            Err(SessionStoreError::Locked(_))
        ));
        assert!(matches!(
            locked_store.get_or_create_session(&session.session_key, SessionMetadata::default()),
            Err(SessionStoreError::Locked(_))
        ));
    }

    #[test]
    fn test_reject_mode_without_password_keeps_encrypted_sessions_locked_when_hmac_is_configured() {
        let key_material = test_key_material();
        let server_secret = test_key_material();
        let (encrypted_store, temp_dir) = create_encrypted_store(&key_material);
        let session = encrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        encrypted_store
            .append_message(ChatMessage::user(&session.id, "hello"))
            .unwrap();

        let locked_store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_encryption_mode(EncryptionMode::IfPassword)
            .with_hmac_key(Zeroizing::new(integrity::derive_hmac_key(&server_secret)))
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let entries = locked_store
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].session_id(), session.id);
        assert_eq!(entries[0].access(), SessionAccessState::Locked);

        assert!(matches!(
            locked_store.get_session(&session.id),
            Err(SessionStoreError::Locked(_))
        ));
        assert!(matches!(
            locked_store.get_history(&session.id, None, None),
            Err(SessionStoreError::Locked(_))
        ));
        assert!(matches!(
            locked_store.get_session_by_key(&session.session_key),
            Err(SessionStoreError::Locked(_))
        ));
    }

    #[test]
    fn test_warn_mode_without_password_short_circuits_encrypted_metadata_before_hmac_warning() {
        let key_material = test_key_material();
        let server_secret = test_key_material();
        let (encrypted_store, temp_dir) = create_encrypted_store(&key_material);
        let session = encrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        let meta_path = encrypted_store.session_meta_path(&session.id).unwrap();
        let content = fs::read(&meta_path).unwrap();

        let locked_store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_encryption_mode(EncryptionMode::IfPassword)
            .with_hmac_key(Zeroizing::new(integrity::derive_hmac_key(&server_secret)))
            .with_integrity_action(integrity::IntegrityAction::Warn);

        assert!(matches!(
            locked_store.verify_integrity_bytes(&content, &meta_path),
            Err(SessionStoreError::Locked(_))
        ));
    }

    #[test]
    fn test_required_mode_without_password_fails_closed() {
        let temp_dir = TempDir::new().unwrap();
        let store = reopen_store_with_encryption(temp_dir.path(), None, EncryptionMode::Required);
        let err = store
            .list_session_entries(SessionFilter::default())
            .unwrap_err();
        assert!(matches!(err, SessionStoreError::Locked(_)));
    }

    #[test]
    fn test_required_mode_with_password_round_trip() {
        let key_material = test_key_material();
        let (store, temp_dir) =
            create_encrypted_store_with_mode(&key_material, EncryptionMode::Required);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "hello"))
            .unwrap();

        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 1);

        let entries = store
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].session_id(), session.id);
        assert_eq!(entries[0].access(), SessionAccessState::Available);

        let archived = store.archive_session(&session.id, false).unwrap();
        let restored = store.restore_session(&archived.session_id).unwrap();
        let restored_history = store.get_history(&restored.session_id, None, None).unwrap();
        assert_eq!(restored_history.len(), 1);

        let reopened = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::Required,
        );
        let reopened_history = reopened.get_history(&session.id, None, None).unwrap();
        assert_eq!(reopened_history.len(), 1);
    }

    #[test]
    fn test_append_reuses_encrypted_history_current_check_after_load() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store_without_hmac(&key_material);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "first"))
            .unwrap();

        let crypto = SessionCryptoContext::load_or_create(temp_dir.path(), &key_material).unwrap();
        let reopened = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_encryption_mode(EncryptionMode::IfPassword)
            .with_crypto_context(Arc::new(crypto));
        reopened
            .history_current_file_read_count
            .store(0, Ordering::SeqCst);

        reopened
            .append_message(ChatMessage::assistant(&session.id, "second"))
            .unwrap();
        assert_eq!(
            reopened
                .history_current_file_read_count
                .load(Ordering::SeqCst),
            1
        );

        reopened
            .append_message(ChatMessage::user(&session.id, "third"))
            .unwrap();
        assert_eq!(
            reopened
                .history_current_file_read_count
                .load(Ordering::SeqCst),
            1,
            "history currentness should stay cached for later appends"
        );
    }

    #[test]
    fn test_history_currentness_reader_streams_encrypted_prefixes() {
        let mut history = Vec::new();
        history.extend_from_slice(b"  ");
        history.extend_from_slice(crypto::encrypted_payload_prefix());
        history.resize(history.len() + 128 * 1024, b'a');
        history.extend_from_slice(b"\n\t\r\n");
        history.extend_from_slice(crypto::encrypted_payload_prefix());
        history.extend_from_slice(b"{}");

        SessionStore::ensure_history_reader_has_only_encrypted_payload_lines(std::io::Cursor::new(
            history,
        ))
        .expect("encrypted-prefixed history should scan without size gating");
    }

    #[test]
    fn test_history_currentness_reader_rejects_plaintext_prefix() {
        let err = SessionStore::ensure_history_reader_has_only_encrypted_payload_lines(
            std::io::Cursor::new(b"  plaintext\n".to_vec()),
        )
        .expect_err("plaintext history must not be treated as current encrypted history");

        assert!(matches!(err, SessionStoreError::Locked(message)
            if message.contains("session history is not encrypted")));
    }

    #[test]
    fn test_cached_session_history_currentness_starts_unconfirmed() {
        let key_material = test_key_material();
        let (store, _temp_dir) = create_encrypted_store_without_hmac(&key_material);
        let session = Session::new("agent-1", SessionMetadata::default());

        let cached = store.new_cached_session(session, false);

        assert!(
            !cached.history_file_current_confirmed,
            "cache entries must be promoted only after an explicit history-currentness check"
        );
    }

    #[test]
    fn test_archive_currentness_reuses_verified_signature_across_cold_loads() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store_without_hmac(&key_material);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "before archive"))
            .unwrap();
        store.archive_session(&session.id, false).unwrap();

        let reopened = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        );
        reopened
            .archive_current_file_read_count
            .store(0, Ordering::SeqCst);

        assert_eq!(
            reopened.get_session(&session.id).unwrap().status,
            SessionStatus::Archived
        );
        assert_eq!(
            reopened
                .archive_current_file_read_count
                .load(Ordering::SeqCst),
            1
        );

        reopened.sessions.write().remove(&session.id);
        assert_eq!(
            reopened.get_session(&session.id).unwrap().status,
            SessionStatus::Archived
        );
        assert_eq!(
            reopened
                .archive_current_file_read_count
                .load(Ordering::SeqCst),
            1,
            "unchanged archive signature should avoid repeated archive currentness reads"
        );
    }

    #[test]
    fn test_append_rejects_mixed_encrypted_and_plaintext_history() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store_without_hmac(&key_material);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "encrypted-first"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let plaintext = serde_json::to_vec(&ChatMessage::user(&session.id, "plaintext-tail"))
            .expect("plaintext history fixture should serialize");
        let mut history = OpenOptions::new().append(true).open(&history_path).unwrap();
        history.write_all(&plaintext).unwrap();
        history.write_all(b"\n").unwrap();
        history.sync_all().unwrap();

        let crypto = SessionCryptoContext::load_or_create(temp_dir.path(), &key_material).unwrap();
        let reopened = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_encryption_mode(EncryptionMode::IfPassword)
            .with_crypto_context(Arc::new(crypto));

        let err = reopened
            .append_message(ChatMessage::assistant(&session.id, "should fail"))
            .expect_err("mixed-format history must not be marked current");
        assert!(matches!(
            err,
            SessionStoreError::Locked(message)
                if message.contains("session history is not encrypted")
        ));
    }

    #[test]
    fn test_encrypted_compaction_rewrites_history_as_ciphertext() {
        let key_material = test_key_material();
        let (store, _temp) = create_encrypted_store(&key_material);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        for idx in 0..6 {
            store
                .append_message(ChatMessage::user(&session.id, format!("message-{idx}")))
                .unwrap();
        }

        let metadata = store
            .compact_session(&session.id, 2, |msgs| format!("summary: {}", msgs.len()))
            .unwrap();
        assert_eq!(metadata.messages_compacted, 4);

        let history_path = store.session_history_path(&session.id).unwrap();
        let history_raw = fs::read_to_string(&history_path).unwrap();
        assert!(history_raw
            .lines()
            .filter(|line| !line.trim().is_empty())
            .all(|line| crypto::is_encrypted_payload(line.as_bytes())));

        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 3);
        assert_eq!(history[0].role, MessageRole::System);
        assert_eq!(history[0].content, "summary: 4");
        assert_eq!(history[1].content, "message-4");
        assert_eq!(history[2].content, "message-5");
    }

    #[test]
    fn test_restore_session_rejects_unencrypted_archive_when_encryption_enabled() {
        let temp_dir = TempDir::new().unwrap();
        let unencrypted_store = SessionStore::with_base_path(temp_dir.path().to_path_buf());
        let session = unencrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        unencrypted_store
            .append_message(ChatMessage::user(&session.id, "before-archive"))
            .unwrap();
        unencrypted_store
            .archive_session(&session.id, true)
            .unwrap();

        let key_material = test_key_material();
        let encrypted_store = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        );
        let err = encrypted_store.restore_session(&session.id).unwrap_err();
        assert!(matches!(err, SessionStoreError::Locked(_)));
    }

    #[test]
    fn test_get_history_rejects_unencrypted_history_when_encryption_enabled() {
        let temp_dir = TempDir::new().unwrap();
        let unencrypted_store = SessionStore::with_base_path(temp_dir.path().to_path_buf());
        let session = unencrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        unencrypted_store
            .append_message(ChatMessage::user(&session.id, "plaintext-history"))
            .unwrap();

        let key_material = test_key_material();
        let encrypted_store = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        );
        let err = encrypted_store
            .get_history(&session.id, None, None)
            .unwrap_err();
        assert!(matches!(err, SessionStoreError::Locked(_)));
    }

    /// Regression for R58 H-DC1: every other history-mutating path
    /// holds the per-history FileLock across its destructive ops;
    /// `clear_history` previously did not, so a concurrent inbound
    /// append could race the unlink and write a stale-HMAC line onto
    /// a recreated file.
    #[test]
    fn test_clear_history_waits_for_history_lock() {
        let (store, _temp_dir) = create_test_store();
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "hello"))
            .unwrap();
        let history_path = store.session_history_path(&session.id).unwrap();
        let history_lock = FileLock::acquire(&history_path).unwrap();

        let store = std::sync::Arc::new(store);
        let session_id = session.id.clone();
        let (started_tx, started_rx) = std::sync::mpsc::channel();
        let (result_tx, result_rx) = std::sync::mpsc::channel();
        let clear_store = std::sync::Arc::clone(&store);

        let join = std::thread::spawn(move || {
            started_tx.send(()).unwrap();
            let result = clear_store.clear_history(&session_id);
            result_tx.send(result).unwrap();
        });

        started_rx.recv().unwrap();
        // clear_history must block on the FileLock — if it did not,
        // the history file would already be gone here.
        assert!(result_rx.try_recv().is_err());
        assert!(
            history_path.exists(),
            "history file must still exist while the FileLock is held externally"
        );

        drop(history_lock);

        result_rx
            .recv()
            .unwrap()
            .expect("clear_history must succeed once the lock is released");
        join.join().unwrap();
        assert!(
            !history_path.exists(),
            "clear_history must remove the history file once it completes"
        );
    }

    #[test]
    fn test_restore_session_waits_for_history_lock() {
        let (store, _temp_dir) = create_test_store();
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "before-archive"))
            .unwrap();
        store.archive_session(&session.id, true).unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let history_lock = FileLock::acquire(&history_path).unwrap();

        let store = std::sync::Arc::new(store);
        let session_id = session.id.clone();
        let (started_tx, started_rx) = std::sync::mpsc::channel();
        let (result_tx, result_rx) = std::sync::mpsc::channel();
        let restore_store = std::sync::Arc::clone(&store);

        let join = std::thread::spawn(move || {
            started_tx.send(()).unwrap();
            let result = restore_store.restore_session(&session_id);
            result_tx
                .send(result.map(|restored| restored.session_id))
                .unwrap();
        });

        started_rx.recv().unwrap();
        assert!(result_rx.try_recv().is_err());

        drop(history_lock);

        let restored_session_id = result_rx.recv().unwrap().unwrap();
        join.join().unwrap();
        assert_eq!(restored_session_id, session.id);
    }

    #[test]
    fn test_restore_session_rejects_tampered_unencrypted_archive_under_integrity_reject() {
        let temp_dir = TempDir::new().unwrap();
        let hmac_key = Zeroizing::new(integrity::derive_hmac_key(b"archive-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(hmac_key)
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "before-archive"))
            .unwrap();
        store.archive_session(&session.id, false).unwrap();

        let archive_path = store.archive_path(&session.id).unwrap();
        fs::write(&archive_path, br#"{"tampered":true}"#).unwrap();

        let err = store.restore_session(&session.id).unwrap_err();
        assert!(matches!(err, SessionStoreError::HistoryCorrupt(_)));
    }

    #[test]
    fn test_get_archive_info_rejects_tampered_unencrypted_archive_under_integrity_reject() {
        let temp_dir = TempDir::new().unwrap();
        let hmac_key = Zeroizing::new(integrity::derive_hmac_key(b"archive-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(hmac_key)
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "before-archive"))
            .unwrap();
        store.archive_session(&session.id, false).unwrap();

        let archive_path = store.archive_path(&session.id).unwrap();
        fs::write(&archive_path, br#"{"tampered":true}"#).unwrap();

        let err = store.get_archive_info(&session.id).unwrap_err();
        assert!(matches!(err, SessionStoreError::HistoryCorrupt(_)));
    }

    #[test]
    fn test_wrong_password_returns_manifest_integrity_failed_for_encrypted_session() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store(&key_material);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "hello"))
            .unwrap();

        let wrong_key_material = test_key_material();
        let wrong_store = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&wrong_key_material),
            EncryptionMode::IfPassword,
        );
        let err = wrong_store.get_session(&session.id).unwrap_err();
        assert!(matches!(err, SessionStoreError::ManifestIntegrityFailed(_)));
    }

    #[test]
    fn test_list_sessions_fails_closed_on_wrong_password_for_encrypted_store() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store(&key_material);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "secret"))
            .unwrap();

        let wrong_password = test_key_material();
        let wrong_store = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&wrong_password),
            EncryptionMode::IfPassword,
        );

        let err = wrong_store
            .list_session_entries(SessionFilter::default())
            .unwrap_err();
        assert!(matches!(err, SessionStoreError::ManifestIntegrityFailed(_)));
    }

    #[test]
    fn test_list_sessions_isolates_malformed_prefixed_encrypted_metadata_to_locked_stub() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store(&key_material);
        let healthy = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-a".into()),
                    chat_id: Some("chat-a".into()),
                    user_id: Some("user-a".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let corrupt = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-b".into()),
                    chat_id: Some("chat-b".into()),
                    user_id: Some("user-b".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        store
            .append_message(ChatMessage::user(&healthy.id, "still-readable"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&corrupt.id, "will-corrupt"))
            .unwrap();

        let corrupt_meta_path = store.session_meta_path(&corrupt.id).unwrap();
        let malformed = b"cse1:{\"format\":\"session-enc-v1\"}";
        fs::write(&corrupt_meta_path, malformed).unwrap();
        integrity::write_hmac_file(
            store.hmac_key.as_ref().unwrap(),
            malformed,
            &corrupt_meta_path,
        )
        .unwrap();

        let reopened = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        );
        let entries = reopened
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 2);

        let healthy_entry = entries
            .iter()
            .find(|entry| entry.session_id() == healthy.id)
            .unwrap();
        assert_eq!(healthy_entry.access(), SessionAccessState::Available);

        let corrupt_entry = entries
            .iter()
            .find(|entry| entry.session_id() == corrupt.id)
            .unwrap();
        assert_eq!(corrupt_entry.access(), SessionAccessState::Locked);

        let loaded_healthy = reopened.get_session(&healthy.id).unwrap();
        assert_eq!(loaded_healthy.id, healthy.id);

        let err = reopened.get_session(&corrupt.id).unwrap_err();
        assert!(matches!(err, SessionStoreError::Crypto(_)));

        let err = reopened
            .list_session_entries(SessionFilter::new().with_user_id("user-a"))
            .unwrap_err();
        assert!(matches!(err, SessionStoreError::Locked(_)));
    }

    #[test]
    fn test_list_sessions_isolates_truncated_prefixed_encrypted_metadata_to_locked_stub() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store(&key_material);
        let healthy = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-a".into()),
                    chat_id: Some("chat-a".into()),
                    user_id: Some("user-a".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let corrupt = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-b".into()),
                    chat_id: Some("chat-b".into()),
                    user_id: Some("user-b".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        store
            .append_message(ChatMessage::user(&healthy.id, "still-readable"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&corrupt.id, "will-corrupt"))
            .unwrap();

        let corrupt_meta_path = store.session_meta_path(&corrupt.id).unwrap();
        let encrypted = fs::read(&corrupt_meta_path).unwrap();
        assert!(crypto::has_encrypted_payload_prefix(&encrypted));
        let truncated = encrypted[..12].to_vec();
        fs::write(&corrupt_meta_path, &truncated).unwrap();
        integrity::write_hmac_file(
            store.hmac_key.as_ref().unwrap(),
            &truncated,
            &corrupt_meta_path,
        )
        .unwrap();

        let reopened = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        );
        let entries = reopened
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 2);
        assert_eq!(
            entries
                .iter()
                .find(|entry| entry.session_id() == healthy.id)
                .unwrap()
                .access(),
            SessionAccessState::Available
        );
        assert_eq!(
            entries
                .iter()
                .find(|entry| entry.session_id() == corrupt.id)
                .unwrap()
                .access(),
            SessionAccessState::Locked
        );
        assert!(matches!(
            reopened.get_session(&corrupt.id),
            Err(SessionStoreError::Crypto(_))
        ));
    }

    #[test]
    fn test_list_sessions_isolates_undecryptable_encrypted_metadata_to_locked_stub() {
        let key_material = test_key_material();
        let (store, temp_dir) = create_encrypted_store(&key_material);
        let healthy = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-a".into()),
                    chat_id: Some("chat-a".into()),
                    user_id: Some("user-a".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let corrupt = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-b".into()),
                    chat_id: Some("chat-b".into()),
                    user_id: Some("user-b".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        store
            .append_message(ChatMessage::user(&healthy.id, "still-readable"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&corrupt.id, "will-corrupt"))
            .unwrap();

        let corrupt_meta_path = store.session_meta_path(&corrupt.id).unwrap();
        let encrypted = fs::read(&corrupt_meta_path).unwrap();
        assert!(crypto::has_encrypted_payload_prefix(&encrypted));

        let prefix = b"cse1:";
        let mut envelope: serde_json::Value =
            serde_json::from_slice(encrypted.strip_prefix(prefix).unwrap()).unwrap();
        let ciphertext = envelope.get("c").and_then(|value| value.as_str()).unwrap();
        let mut chars: Vec<char> = ciphertext.chars().collect();
        let flip_idx = chars.iter().position(|ch| *ch != 'A').unwrap_or(0);
        chars[flip_idx] = if chars[flip_idx] == 'B' { 'C' } else { 'B' };
        envelope["c"] = serde_json::Value::String(chars.into_iter().collect());

        let mut tampered = prefix.to_vec();
        tampered.extend(serde_json::to_vec(&envelope).unwrap());
        fs::write(&corrupt_meta_path, &tampered).unwrap();
        integrity::write_hmac_file(
            store.hmac_key.as_ref().unwrap(),
            &tampered,
            &corrupt_meta_path,
        )
        .unwrap();

        let reopened = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        );
        let entries = reopened
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 2);
        assert_eq!(
            entries
                .iter()
                .find(|entry| entry.session_id() == healthy.id)
                .unwrap()
                .access(),
            SessionAccessState::Available
        );
        assert_eq!(
            entries
                .iter()
                .find(|entry| entry.session_id() == corrupt.id)
                .unwrap()
                .access(),
            SessionAccessState::Locked
        );

        let loaded_healthy = reopened.get_session(&healthy.id).unwrap();
        assert_eq!(loaded_healthy.id, healthy.id);
        assert!(matches!(
            reopened.get_session(&corrupt.id),
            Err(SessionStoreError::DecryptionFailed(_))
        ));
        assert!(matches!(
            reopened.list_session_entries(SessionFilter::new().with_user_id("user-a")),
            Err(SessionStoreError::Locked(_))
        ));
    }

    #[test]
    fn test_reject_mode_recovers_pending_history_hmac_after_interrupted_rewrite() {
        let temp_dir = TempDir::new().unwrap();
        let key_material = test_key_material();
        let store = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        )
        .with_integrity_action(integrity::IntegrityAction::Reject);
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "before"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let temp_path = history_path.with_extension("tmp");
        {
            let file = SessionStore::create_private_output_file(&temp_path).unwrap();
            let mut writer = BufWriter::new(file);
            for message in [
                ChatMessage::user(&session.id, "before"),
                ChatMessage::assistant(&session.id, "after"),
            ] {
                let encoded = store.encode_history_message(&message).unwrap();
                writer.write_all(&encoded).unwrap();
                writeln!(writer).unwrap();
            }
            writer.flush().unwrap();
            writer
                .into_inner()
                .map_err(|e| std::io::Error::other(e.to_string()))
                .unwrap()
                .sync_all()
                .unwrap();
        }
        store
            .prepare_history_hmac_for_path(&temp_path, &history_path, &session.id)
            .unwrap();
        fs::rename(&temp_path, &history_path).unwrap();

        let mut pending_sidecar = history_path.as_os_str().to_owned();
        pending_sidecar.push(".hmac.tmp");
        let pending_sidecar = PathBuf::from(pending_sidecar);
        assert!(pending_sidecar.exists());

        let reopened = reopen_store_with_encryption(
            temp_dir.path(),
            Some(&key_material),
            EncryptionMode::IfPassword,
        )
        .with_integrity_action(integrity::IntegrityAction::Reject);
        let history = reopened.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 2);
        assert_eq!(history[0].content, "before");
        assert_eq!(history[1].content, "after");
        assert!(!pending_sidecar.exists());
    }

    #[test]
    fn test_store_get_or_create_session() {
        let (store, _temp) = create_test_store();

        let metadata = SessionMetadata {
            name: Some("First".into()),
            ..Default::default()
        };

        // First call creates
        let session1 = store
            .get_or_create_session("test:key:1", metadata.clone())
            .unwrap();
        assert_eq!(session1.metadata.name, Some("First".into()));

        // Second call gets existing
        let session2 = store
            .get_or_create_session(
                "test:key:1",
                SessionMetadata {
                    name: Some("Second".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        assert_eq!(session2.id, session1.id);
        assert_eq!(session2.metadata.name, Some("First".into())); // Not updated
    }

    #[test]
    fn test_store_get_or_create_session_with_encryption_does_not_deadlock() {
        let key_material = test_key_material();
        let (store, _temp) = create_encrypted_store_without_hmac(&key_material);
        let store = Arc::new(store);
        let worker_store = Arc::clone(&store);
        let (tx, rx) = std::sync::mpsc::channel();

        std::thread::spawn(move || {
            let result = worker_store
                .get_or_create_session("encrypted:key:1", SessionMetadata::default())
                .map(|session| session.id);
            let _ = tx.send(result);
        });

        let session_id = rx
            .recv_timeout(std::time::Duration::from_secs(2))
            .expect("encrypted get_or_create_session should not deadlock")
            .expect("encrypted session should be created");
        assert!(store.history_file_current_confirmed(&session_id));
    }

    /// Regression: `get_session_by_key` previously held
    /// `key_to_id.read()` across `get_session(id)`, which acquires
    /// `sessions.read()`. Write paths in this file take the opposite
    /// order (`sessions.write()` → `key_to_id.write()`). Under
    /// contention parking_lot's writer-queues turn the cycle into a
    /// real deadlock. The fix snapshots the id out of `key_to_id`
    /// under its read guard, drops it, and only then re-enters the
    /// `sessions` lock.
    ///
    /// This test exercises the happy-path resolution from cache to
    /// pin the snapshot-and-release shape. It does not stress the
    /// race directly (which needs many threads to be reliable) but
    /// fails closed if `get_session_by_key` is refactored back to a
    /// held-across pattern, because a quick smoke would still
    /// succeed and only a contention probe would fail.
    #[test]
    fn test_get_session_by_key_does_not_hold_key_lock_across_session_lock() {
        let (store, _temp) = create_test_store();
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .expect("create session");

        // Resolve by the session_key in a worker; the timeout
        // catches any held-across regression that would deadlock
        // when many concurrent get_session_by_key callers race
        // against a concurrent write path.
        let store = Arc::new(store);
        let key = session.session_key.clone();
        let session_id = session.id.clone();
        let mut handles = Vec::new();
        for _ in 0..16 {
            let store = Arc::clone(&store);
            let key = key.clone();
            let expected_id = session_id.clone();
            handles.push(std::thread::spawn(move || {
                let resolved = store
                    .get_session_by_key(&key)
                    .expect("get_session_by_key should resolve");
                assert_eq!(resolved.id, expected_id);
            }));
        }
        for handle in handles {
            handle
                .join()
                .expect("get_session_by_key worker must not deadlock");
        }
    }

    #[test]
    fn test_store_list_sessions() {
        let (store, _temp) = create_test_store();

        store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("telegram".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("discord".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        store
            .create_session(
                "agent-2",
                SessionMetadata {
                    channel: Some("telegram".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        // List all
        let all = store.list_sessions(SessionFilter::new()).unwrap();
        assert_eq!(all.len(), 3);

        // Filter by agent
        let agent1_only = store
            .list_sessions(SessionFilter::new().with_agent_id("agent-1"))
            .unwrap();
        assert_eq!(agent1_only.len(), 2);

        // Filter by channel
        let telegram_only = store
            .list_sessions(SessionFilter::new().with_channel("telegram"))
            .unwrap();
        assert_eq!(telegram_only.len(), 2);
    }

    #[test]
    fn test_store_patch_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        let updated = store
            .patch_session(
                &session.id,
                SessionMetadata {
                    name: Some("Updated Name".into()),
                    agent_id: Some("agent-2".into()),
                    channel: Some("signal".into()),
                    chat_id: Some("123".into()),
                    user_id: Some("user-1".into()),
                    model: Some("claude-3".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        assert_eq!(updated.metadata.name, Some("Updated Name".into()));
        assert_eq!(updated.metadata.agent_id, Some("agent-2".into()));
        assert_eq!(updated.metadata.channel, Some("signal".into()));
        assert_eq!(updated.metadata.chat_id, Some("123".into()));
        assert_eq!(updated.metadata.user_id, Some("user-1".into()));
        assert_eq!(updated.metadata.model, Some("claude-3".into()));
        assert!(updated.updated_at > session.updated_at);
    }

    #[test]
    fn test_store_reset_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Add some messages
        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "Hi there!"))
            .unwrap();

        // Verify messages exist
        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 2);

        // Reset
        let reset = store.reset_session(&session.id).unwrap();
        assert_eq!(reset.message_count, 0);

        // Verify history cleared
        let history = store.get_history(&session.id, None, None).unwrap();
        assert!(history.is_empty());
    }

    #[test]
    fn test_store_delete_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store.delete_session(&session.id).unwrap();

        // Should not be found
        let result = store.get_session(&session.id);
        assert!(matches!(result, Err(SessionStoreError::NotFound(_))));
    }

    #[test]
    fn test_store_append_and_get_history() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Append messages
        store
            .append_message(ChatMessage::user(&session.id, "First"))
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "Second"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "Third"))
            .unwrap();

        // Get all
        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 3);
        assert_eq!(history[0].content, "First");
        assert_eq!(history[1].content, "Second");
        assert_eq!(history[2].content, "Third");

        // Get with limit
        let limited = store.get_history(&session.id, Some(2), None).unwrap();
        assert_eq!(limited.len(), 2);
        assert_eq!(limited[0].content, "Second");
        assert_eq!(limited[1].content, "Third");
    }

    #[test]
    fn test_append_messages_rejects_cross_session_batch() {
        let (store, _temp) = create_test_store();

        let session_a = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-a".into()),
                    chat_id: Some("chat-a".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let session_b = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("channel-b".into()),
                    chat_id: Some("chat-b".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        let err = store
            .append_messages(&[
                ChatMessage::user(&session_a.id, "first"),
                ChatMessage::assistant(&session_b.id, "second"),
            ])
            .unwrap_err();

        assert!(matches!(err, SessionStoreError::InvalidMessageBatch(_)));
        assert!(
            store
                .get_history(&session_a.id, None, None)
                .unwrap()
                .is_empty(),
            "cross-session batch must not partially write the target history file"
        );
        assert!(store
            .get_history(&session_b.id, None, None)
            .unwrap()
            .is_empty());
    }

    /// `append_messages` batch path must apply the same per-call
    /// byte cap as `append_message`. Without it, a batch (e.g.
    /// assistant + tool-result messages from a single agent turn)
    /// could cross `SESSION_HISTORY_FILE_MAX_BYTES` even though each
    /// individual message would have been refused on the
    /// single-message path. The read-side cap then makes the file
    /// unloadable, bricking the session.
    #[test]
    fn test_append_messages_refuses_batch_that_would_exceed_history_cap() {
        let (store, _temp) = create_test_store();
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Pre-fill the history file to just under the cap so a small
        // batch trips the projected-size check without needing 256 MiB
        // of test fixture data.
        let history_path = store.session_history_path(&session.id).unwrap();
        let padding_size = (SESSION_HISTORY_FILE_MAX_BYTES - 32 * 1024) as usize; // leave 32 KiB headroom
        std::fs::write(&history_path, vec![b'x'; padding_size]).unwrap();

        // A batch of two 64 KiB messages adds ~128 KiB; combined with
        // the existing 256 MiB - 32 KiB padding this crosses the cap.
        let big_payload = "x".repeat(64 * 1024);
        let err = store
            .append_messages(&[
                ChatMessage::user(&session.id, &big_payload),
                ChatMessage::assistant(&session.id, &big_payload),
            ])
            .expect_err("oversized batch must be refused");
        assert!(
            matches!(err, SessionStoreError::HistoryFileFull(_)),
            "expected HistoryFileFull (permanent, retry-blocking) error, got: {err:?}"
        );
    }

    /// Storage-layer pin: HistoryFileFull must NOT be classified as
    /// permanent-history-corruption (that helper guards the dedupe-
    /// marker readability invariant, which a full file doesn't break)
    /// AND the positive set (HistoryCorrupt / ManifestIntegrityFailed
    /// / DecryptionFailed / IntegrityRejected) must stay intact. The
    /// inbound-dispatch integration that pins the SessionHistoryCorrupt
    /// routing lives next to the classifier itself
    /// (`test_dispatch_inbound_classifies_history_file_full_as_permanent`).
    #[test]
    fn test_history_file_full_is_not_permanent_history_corruption() {
        let err = SessionStoreError::HistoryFileFull("session full".to_string());
        assert!(
            !err.is_permanent_history_corruption(),
            "HistoryFileFull does not hide dedupe markers — keep is_permanent_history_corruption \
             scoped to the crypto/integrity failures it was built for"
        );
        for variant in [
            SessionStoreError::HistoryCorrupt("x".into()),
            SessionStoreError::ManifestIntegrityFailed("x".into()),
            SessionStoreError::DecryptionFailed("x".into()),
            SessionStoreError::IntegrityRejected("x".into()),
        ] {
            assert!(
                variant.is_permanent_history_corruption(),
                "expected {variant:?} to be permanent history corruption"
            );
        }
    }

    #[test]
    fn test_history_hmac_is_written_and_verified() {
        let temp_dir = TempDir::new().unwrap();
        let key = Zeroizing::new(integrity::derive_hmac_key(b"history-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(key)
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let sidecar = history_path.with_extension("jsonl.hmac");
        assert!(sidecar.exists(), "history HMAC sidecar should exist");

        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 1);

        fs::write(&history_path, "tampered\n").unwrap();
        let result = store.get_history(&session.id, None, None);
        assert!(result.is_err(), "tampered history should be rejected");
    }

    #[test]
    fn test_append_message_rejects_session_with_mismatched_metadata_sidecar() {
        let temp_dir = TempDir::new().unwrap();
        let current_key = integrity::derive_hmac_key(b"current-session-secret");
        let legacy_key = integrity::derive_hmac_key(b"legacy-session-secret");
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(Zeroizing::new(current_key))
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        let meta_path = store.session_meta_path(&session.id).unwrap();
        let metadata = fs::read(&meta_path).unwrap();
        integrity::write_hmac_file(&legacy_key, &metadata, &meta_path).unwrap();

        let reopened = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(Zeroizing::new(current_key))
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let err = reopened
            .append_message(ChatMessage::user(&session.id, "should fail"))
            .expect_err("append should propagate get_session integrity rejection");
        let message = match err {
            SessionStoreError::IntegrityRejected(message) => message,
            other => panic!(
                "expected metadata integrity rejection, got {}",
                session_store_error_kind(&other)
            ),
        };
        assert!(
            message.contains("session integrity verification failed"),
            "expected integrity failure wording"
        );
        assert!(
            message.contains(&meta_path.display().to_string()),
            "expected metadata path in integrity failure"
        );
        let history_path = reopened.session_history_path(&session.id).unwrap();
        assert!(
            !history_path.exists(),
            "failed metadata integrity must prevent history append"
        );
    }

    #[test]
    fn test_append_messages_rejects_session_with_mismatched_metadata_sidecar() {
        let temp_dir = TempDir::new().unwrap();
        let current_key = integrity::derive_hmac_key(b"current-session-secret");
        let legacy_key = integrity::derive_hmac_key(b"legacy-session-secret");
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(Zeroizing::new(current_key))
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        let meta_path = store.session_meta_path(&session.id).unwrap();
        let metadata = fs::read(&meta_path).unwrap();
        integrity::write_hmac_file(&legacy_key, &metadata, &meta_path).unwrap();

        let reopened = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(Zeroizing::new(current_key))
            .with_integrity_action(integrity::IntegrityAction::Reject);
        let messages = vec![ChatMessage::user(&session.id, "should fail")];
        let err = reopened
            .append_messages(&messages)
            .expect_err("batch append should propagate get_session integrity rejection");

        assert!(
            matches!(err, SessionStoreError::IntegrityRejected(ref message)
                if message.contains("session integrity verification failed")
                    && message.contains(&meta_path.display().to_string())),
            "expected metadata integrity rejection"
        );
        let history_path = reopened.session_history_path(&session.id).unwrap();
        assert!(
            !history_path.exists(),
            "failed metadata integrity must prevent batch history append"
        );
    }

    #[test]
    fn test_append_history_hmac_reuses_cached_state_after_first_append() {
        let temp_dir = TempDir::new().unwrap();
        let key = Zeroizing::new(integrity::derive_hmac_key(b"history-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(key)
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        integrity::reset_append_hmac_path_rebuild_count();

        store
            .append_message(ChatMessage::user(&session.id, "first"))
            .unwrap();
        assert_eq!(integrity::append_hmac_path_rebuild_count(), 1);

        store
            .append_message(ChatMessage::assistant(&session.id, "second"))
            .unwrap();
        assert_eq!(
            integrity::append_hmac_path_rebuild_count(),
            1,
            "second append should reuse the cached HMAC state"
        );
    }

    #[test]
    fn test_append_history_hmac_rebuilds_when_sidecar_or_metadata_changes() {
        let temp_dir = TempDir::new().unwrap();
        let key = Zeroizing::new(integrity::derive_hmac_key(b"history-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(key)
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        integrity::reset_append_hmac_path_rebuild_count();

        store
            .append_message(ChatMessage::user(&session.id, "first"))
            .unwrap();
        assert_eq!(integrity::append_hmac_path_rebuild_count(), 1);

        let history_path = store.session_history_path(&session.id).unwrap();
        let sidecar = hmac_sidecar_path(&history_path);
        fs::write(&sidecar, "v1:deadbeef").unwrap();

        store
            .append_message(ChatMessage::assistant(&session.id, "second"))
            .unwrap();
        assert_eq!(
            integrity::append_hmac_path_rebuild_count(),
            2,
            "cache should be invalidated when the committed sidecar no longer matches"
        );
    }

    #[test]
    fn test_reject_mode_fails_closed_on_history_sidecar_io_error() {
        let temp_dir = TempDir::new().unwrap();
        let key = Zeroizing::new(integrity::derive_hmac_key(b"history-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(key)
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let sidecar = hmac_sidecar_path(&history_path);
        fs::remove_file(&sidecar).unwrap();
        fs::create_dir(&sidecar).unwrap();

        let err = store.get_history(&session.id, None, None).unwrap_err();
        assert!(matches!(err, SessionStoreError::HistoryCorrupt(_)));
    }

    #[test]
    fn test_reject_mode_without_password_keeps_locked_stub_on_metadata_sidecar_io_error() {
        let key_material = test_key_material();
        let server_secret = test_key_material();
        let (encrypted_store, temp_dir) = create_encrypted_store(&key_material);
        let session = encrypted_store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        let meta_path = encrypted_store.session_meta_path(&session.id).unwrap();
        let sidecar = hmac_sidecar_path(&meta_path);
        fs::remove_file(&sidecar).unwrap();
        fs::create_dir(&sidecar).unwrap();

        let locked_store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_encryption_mode(EncryptionMode::IfPassword)
            .with_hmac_key(Zeroizing::new(integrity::derive_hmac_key(&server_secret)))
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let entries = locked_store
            .list_session_entries(SessionFilter::default())
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].session_id(), session.id);
        assert_eq!(entries[0].access(), SessionAccessState::Locked);
        assert!(matches!(
            locked_store.get_session(&session.id),
            Err(SessionStoreError::Locked(_))
        ));
    }

    #[test]
    fn test_store_clear_history() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        store.clear_history(&session.id).unwrap();

        let history = store.get_history(&session.id, None, None).unwrap();
        assert!(history.is_empty());
    }

    #[test]
    fn test_get_history_fails_closed_on_invalid_encrypted_line() {
        let key_material = test_key_material();
        let (store, _temp) = create_encrypted_store_without_hmac(&key_material);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        fs::write(
            &history_path,
            b"cse1:{\"format\":\"session-enc-v1\",\"n\":\"%%%\",\"c\":\"bad\"}\n",
        )
        .unwrap();

        let err = store.get_history(&session.id, None, None).unwrap_err();
        assert!(matches!(
            err,
            SessionStoreError::HistoryCorrupt(ref message)
                if message.contains("invalid encrypted session history line")
        ));
    }

    #[test]
    fn test_get_history_ignores_json_format_field_matching_encryption_format() {
        let (store, _temp) = create_test_store();
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        let mut message = serde_json::to_value(ChatMessage::user(&session.id, "hello")).unwrap();
        message.as_object_mut().unwrap().insert(
            "format".to_string(),
            serde_json::Value::String("session-enc-v1".to_string()),
        );

        let history_path = store.session_history_path(&session.id).unwrap();
        fs::write(
            &history_path,
            format!("{}\n", serde_json::to_string(&message).unwrap()),
        )
        .unwrap();

        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].content, "hello");
    }

    #[test]
    fn test_store_compact_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Add many messages
        for i in 0..10 {
            store
                .append_message(ChatMessage::user(&session.id, format!("Message {}", i)))
                .unwrap();
        }

        // Compact, keeping last 3
        let metadata = store
            .compact_session(&session.id, 3, |msgs| {
                format!("Summary of {} messages", msgs.len())
            })
            .unwrap();

        assert_eq!(metadata.messages_compacted, 7);
        assert_eq!(metadata.compaction_count, 1);
        assert!(metadata.last_compacted_at.is_some());

        // Check history has summary + 3 kept messages
        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 4);
        assert_eq!(history[0].role, MessageRole::System);
        assert!(history[0].content.contains("Summary of 7 messages"));
    }

    #[test]
    fn test_store_auto_compact() {
        let (store, _temp) = create_test_store();
        let store = store.with_compact_threshold(5);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Below threshold
        for _ in 0..4 {
            store
                .append_message(ChatMessage::user(&session.id, "msg"))
                .unwrap();
        }

        let result = store
            .auto_compact_if_needed(&session.id, 2, |_| "summary".into())
            .unwrap();
        assert!(result.is_none());

        // At threshold
        store
            .append_message(ChatMessage::user(&session.id, "msg"))
            .unwrap();

        let result = store
            .auto_compact_if_needed(&session.id, 2, |_| "summary".into())
            .unwrap();
        assert!(result.is_some());
    }

    #[test]
    fn test_store_persistence() {
        let temp_dir = TempDir::new().unwrap();
        let base_path = temp_dir.path().to_path_buf();

        // Create and populate store
        let session_id = {
            let store = SessionStore::with_base_path(base_path.clone());
            let session = store
                .create_session(
                    "agent-1",
                    SessionMetadata {
                        name: Some("Persistent".into()),
                        ..Default::default()
                    },
                )
                .unwrap();

            store
                .append_message(ChatMessage::user(&session.id, "Hello"))
                .unwrap();

            session.id
        };

        // Create new store instance and verify data persists
        let store2 = SessionStore::with_base_path(base_path);
        let loaded = store2.get_session(&session_id).unwrap();

        assert_eq!(loaded.metadata.name, Some("Persistent".into()));

        let history = store2.get_history(&session_id, None, None).unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].content, "Hello");
    }

    #[test]
    fn test_store_flush_all() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Append message (makes session dirty via increment_message_count)
        store
            .append_message(ChatMessage::user(&session.id, "Test"))
            .unwrap();

        // Flush all
        store.flush_all().unwrap();

        // Verify no errors
    }

    #[test]
    fn test_chat_message_serialization() {
        let msg = ChatMessage::user("session-1", "Hello, world!")
            .with_tokens(15)
            .with_metadata(serde_json::json!({"key": "value"}));

        let json = serde_json::to_string(&msg).unwrap();
        let parsed: ChatMessage = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.session_id, "session-1");
        assert_eq!(parsed.content, "Hello, world!");
        assert_eq!(parsed.tokens, Some(15));
    }

    #[test]
    fn test_session_serialization() {
        let session = Session::new(
            "agent-1",
            SessionMetadata {
                name: Some("Test Session".into()),
                model: Some("claude-3".into()),
                ..Default::default()
            },
        );

        let json = serde_json::to_string(&session).unwrap();
        let parsed: Session = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.id, session.id);
        assert_eq!(parsed.session_key, session.session_key);
        assert_eq!(parsed.metadata.name, Some("Test Session".into()));
    }

    #[test]
    fn test_compaction_metadata_serialization() {
        let metadata = CompactionMetadata {
            messages_compacted: 50,
            last_compacted_at: Some(1234567890),
            compaction_count: 3,
            last_summary: Some("Summary text".into()),
            original_message_count: 100,
        };

        let json = serde_json::to_string(&metadata).unwrap();
        let parsed: CompactionMetadata = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.messages_compacted, 50);
        assert_eq!(parsed.compaction_count, 3);
    }

    #[test]
    fn test_store_session_count() {
        let (store, _temp) = create_test_store();

        assert_eq!(store.session_count(), 0);

        store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        assert_eq!(store.session_count(), 1);

        store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("different".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        assert_eq!(store.session_count(), 2);
    }

    #[test]
    fn test_thread_safety() {
        use std::sync::{Arc, Barrier};
        use std::thread;

        let temp_dir = TempDir::new().unwrap();
        let store = Arc::new(SessionStore::with_base_path(temp_dir.path().to_path_buf()));

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        let session_id = session.id.clone();

        // Use a barrier so all threads start writing at roughly the same time
        let barrier = Arc::new(Barrier::new(10));
        let mut handles = vec![];

        // Spawn multiple threads appending messages
        for i in 0..10 {
            let store_clone = Arc::clone(&store);
            let sid = session_id.clone();
            let bar = Arc::clone(&barrier);
            handles.push(thread::spawn(move || {
                bar.wait();
                store_clone
                    .append_message(ChatMessage::user(&sid, format!("Message {}", i)))
                    .unwrap();
            }));
        }

        for handle in handles {
            handle.join().unwrap();
        }

        let history = store.get_history(&session_id, None, None).unwrap();
        // All 10 messages should be present. get_history skips corrupt lines,
        // so if concurrent appends interleaved we may see fewer — but append
        // mode on POSIX guarantees atomic writes under PIPE_BUF (4 KiB+).
        assert_eq!(history.len(), 10);
    }

    #[test]
    fn test_get_history_with_before_id() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Add messages
        let msg1 = ChatMessage::user(&session.id, "First");
        let msg1_id = msg1.id.clone();
        store.append_message(msg1).unwrap();

        let msg2 = ChatMessage::user(&session.id, "Second");
        let msg2_id = msg2.id.clone();
        store.append_message(msg2).unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Third"))
            .unwrap();

        // Get after msg1
        let history = store
            .get_history(&session.id, None, Some(&msg1_id))
            .unwrap();
        assert_eq!(history.len(), 2);
        assert_eq!(history[0].content, "Second");
        assert_eq!(history[1].content, "Third");

        // Get after msg2
        let history = store
            .get_history(&session.id, None, Some(&msg2_id))
            .unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].content, "Third");
    }

    #[test]
    fn test_session_filter_date_range() {
        let (store, _temp) = create_test_store();

        let _session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Filter with timestamp range
        let filter = SessionFilter::new().with_agent_id("agent-1").with_limit(10);

        // Created after 0 should match
        let mut f = filter.clone();
        f.created_after = Some(0);
        let result = store.list_sessions(f).unwrap();
        assert_eq!(result.len(), 1);

        // Created after future timestamp should not match
        let mut f = filter.clone();
        f.created_after = Some(i64::MAX);
        let result = store.list_sessions(f).unwrap();
        assert!(result.is_empty());

        // Created before now should match
        let mut f = filter.clone();
        f.created_before = Some(now_millis() + 1000);
        let result = store.list_sessions(f).unwrap();
        assert_eq!(result.len(), 1);
    }

    // ============== Archive Tests ==============

    #[test]
    fn test_archive_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    name: Some("Archive Test".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        // Add some messages
        for i in 0..5 {
            store
                .append_message(ChatMessage::user(&session.id, format!("Message {}", i)))
                .unwrap();
        }

        // Archive the session
        let result = store.archive_session(&session.id, false).unwrap();

        assert_eq!(result.session_id, session.id);
        assert_eq!(result.message_count, 5);
        assert!(result.archive_size > 0);
        assert!(!result.history_deleted);

        // Verify session status changed
        let archived = store.get_session(&session.id).unwrap();
        assert_eq!(archived.status, SessionStatus::Archived);

        // Verify archive file exists
        let archive_path = store.archive_path(&session.id).unwrap();
        assert!(archive_path.exists());
    }

    #[test]
    fn test_archive_session_with_history_deletion() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        // Archive with history deletion
        let result = store.archive_session(&session.id, true).unwrap();
        assert!(result.history_deleted);

        // Verify history file is deleted
        let history_path = store.session_history_path(&session.id).unwrap();
        assert!(!history_path.exists());
    }

    #[test]
    fn test_archive_already_archived() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        // Archive first time
        store.archive_session(&session.id, false).unwrap();

        // Try to archive again
        let result = store.archive_session(&session.id, false);
        assert!(matches!(result, Err(SessionStoreError::AlreadyArchived(_))));
    }

    #[test]
    fn test_append_message_rejected_for_archived_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Before archive"))
            .unwrap();

        // Archive the session
        store.archive_session(&session.id, false).unwrap();

        // Attempting to append to an archived session must fail
        let result = store.append_message(ChatMessage::user(&session.id, "After archive"));
        assert!(
            matches!(result, Err(SessionStoreError::AlreadyArchived(_))),
            "append_message should reject writes to archived sessions"
        );

        // Verify history was not modified (still just the one pre-archive message)
        let history = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].content, "Before archive");
    }

    #[test]
    fn test_restore_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Add messages
        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&session.id, "Hi there!"))
            .unwrap();

        // Archive with history deletion
        store.archive_session(&session.id, true).unwrap();

        // Verify history is gone
        let history_before = store.get_history(&session.id, None, None).unwrap();
        assert!(history_before.is_empty());

        // Restore
        let result = store.restore_session(&session.id).unwrap();
        assert_eq!(result.session_id, session.id);
        assert_eq!(result.message_count, 2);

        // Verify session status is active again
        let restored = store.get_session(&session.id).unwrap();
        assert_eq!(restored.status, SessionStatus::Active);

        // Verify history is restored
        let history_after = store.get_history(&session.id, None, None).unwrap();
        assert_eq!(history_after.len(), 2);
        assert_eq!(history_after[0].content, "Hello");
        assert_eq!(history_after[1].content, "Hi there!");
    }

    #[test]
    fn test_restore_not_archived() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Try to restore non-archived session
        let result = store.restore_session(&session.id);
        assert!(matches!(result, Err(SessionStoreError::NotArchived(_))));
    }

    #[test]
    fn test_list_archived_sessions() {
        let (store, _temp) = create_test_store();

        // Create and archive some sessions
        let session1 = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        let session2 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("discord".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let _session3 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("telegram".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        store.archive_session(&session1.id, false).unwrap();
        store.archive_session(&session2.id, false).unwrap();
        // session3 not archived

        let archived = store.list_archived_sessions().unwrap();
        assert_eq!(archived.len(), 2);
    }

    #[test]
    fn test_delete_archive() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        store.archive_session(&session.id, false).unwrap();

        // Verify archive exists
        let archive_path = store.archive_path(&session.id).unwrap();
        assert!(archive_path.exists());

        // Delete archive
        let deleted = store.delete_archive(&session.id).unwrap();
        assert!(deleted);

        // Verify archive is gone
        assert!(!archive_path.exists());

        // Delete again returns false
        let deleted_again = store.delete_archive(&session.id).unwrap();
        assert!(!deleted_again);
    }

    #[test]
    fn test_delete_archive_removes_archive_hmac_sidecar() {
        let key_material = test_key_material();
        let (store, _temp) = create_encrypted_store(&key_material);

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();
        store.archive_session(&session.id, false).unwrap();

        let archive_path = store.archive_path(&session.id).unwrap();
        let sidecar_path = hmac_sidecar_path(&archive_path);
        assert!(archive_path.exists());
        assert!(sidecar_path.exists());

        let deleted = store.delete_archive(&session.id).unwrap();
        assert!(deleted);
        assert!(!archive_path.exists());
        assert!(!sidecar_path.exists());
    }

    #[test]
    fn test_get_archive_info() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        // No archive yet
        let info = store.get_archive_info(&session.id).unwrap();
        assert!(info.is_none());

        // Archive
        store.archive_session(&session.id, false).unwrap();

        // Get archive info
        let info = store.get_archive_info(&session.id).unwrap();
        assert!(info.is_some());
        let (path, size, archived_at) = info.unwrap();
        assert!(path.exists());
        assert!(size > 0);
        assert!(archived_at > 0);
    }

    #[test]
    fn test_archive_result_serialization() {
        let result = ArchiveResult {
            session_id: "session-1".to_string(),
            archive_path: "/path/to/archive.json".to_string(),
            message_count: 42,
            archive_size: 12345,
            archived_at: 1234567890,
            history_deleted: false,
        };

        let json = serde_json::to_string(&result).unwrap();
        let parsed: ArchiveResult = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.session_id, "session-1");
        assert_eq!(parsed.message_count, 42);
        assert_eq!(parsed.archive_size, 12345);
    }

    #[test]
    fn test_archived_session_serialization() {
        let session = Session::new("agent-1", SessionMetadata::default());
        let messages = vec![
            ChatMessage::user(&session.id, "Hello"),
            ChatMessage::assistant(&session.id, "Hi!"),
        ];

        let archived = ArchivedSession {
            session: session.clone(),
            messages,
            archived_at: 1234567890,
            version: 1,
        };

        let json = serde_json::to_string(&archived).unwrap();
        let parsed: ArchivedSession = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.session.id, session.id);
        assert_eq!(parsed.messages.len(), 2);
        assert_eq!(parsed.archived_at, 1234567890);
        assert_eq!(parsed.version, 1);
    }

    /// `#[serde(deny_unknown_fields)]` on ArchivedSession refuses
    /// parses that contain top-level fields the binary doesn't know.
    /// Without this, a future v1.5 daemon could add a field without
    /// bumping `version`, and this binary would silently drop the
    /// unknown field on re-archive — bypassing the version-check
    /// guard that prevents downgrade-corruption.
    #[test]
    fn test_archived_session_rejects_unknown_fields() {
        let json = serde_json::json!({
            "session": serde_json::to_value(Session::new("agent-1", SessionMetadata::default())).unwrap(),
            "messages": [],
            "archived_at": 0,
            "version": 1,
            "future_field": "should-be-rejected"
        });
        let result = serde_json::from_value::<ArchivedSession>(json);
        assert!(
            result.is_err(),
            "unknown top-level field must be rejected by deny_unknown_fields"
        );
    }

    /// ChatMessage's `#[serde(flatten)] extra` field round-trips
    /// unknown top-level fields rather than silently dropping them on
    /// re-encode. This is the inverse posture from ArchivedSession:
    /// per-line forward-compat must preserve unknowns, never reject
    /// them, because rejecting a single ChatMessage would corrupt the
    /// entire session history.
    #[test]
    fn test_chat_message_round_trips_unknown_top_level_field() {
        let raw = serde_json::json!({
            "id": "msg-1",
            "session_id": "sess-1",
            "role": "user",
            "content": "hi",
            "created_at": 1234,
            "reasoning_blocks": [{"type": "text", "value": "preserve me"}]
        });
        let parsed: ChatMessage = serde_json::from_value(raw.clone()).expect("parses");
        let reserialized = serde_json::to_value(&parsed).expect("serializes");
        assert_eq!(
            reserialized.get("reasoning_blocks").cloned(),
            Some(serde_json::json!([{"type": "text", "value": "preserve me"}])),
            "unknown field must round-trip via `extra`",
        );
    }

    #[test]
    fn test_archive_empty_session() {
        let (store, _temp) = create_test_store();

        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();

        // Archive with no messages
        let result = store.archive_session(&session.id, false).unwrap();
        assert_eq!(result.message_count, 0);

        // Restore should work
        let restored = store.restore_session(&session.id).unwrap();
        assert_eq!(restored.message_count, 0);
    }

    #[test]
    fn test_archive_inactive_sessions() {
        let (store, _temp) = create_test_store();

        // Create sessions (all will be "recent" since created now)
        let _s1 = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        let _s2 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    channel: Some("ch2".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        // Try to archive sessions older than 30 days (none should match)
        let results = store.archive_inactive_sessions(30, false).unwrap();
        assert!(results.is_empty());
    }

    // ==================== Path Traversal Security Tests ====================

    #[test]
    fn test_validate_session_id_rejects_path_traversal() {
        // Path traversal with ..
        assert!(SessionStore::validate_session_id("../etc/passwd").is_err());
        assert!(SessionStore::validate_session_id("foo/../bar").is_err());
        assert!(SessionStore::validate_session_id("..").is_err());

        // Path traversal with slashes
        assert!(SessionStore::validate_session_id("/etc/passwd").is_err());
        assert!(SessionStore::validate_session_id("foo/bar").is_err());
        assert!(SessionStore::validate_session_id("foo\\bar").is_err());

        // Empty ID
        assert!(SessionStore::validate_session_id("").is_err());

        // Invalid characters
        assert!(SessionStore::validate_session_id("foo bar").is_err());
        assert!(SessionStore::validate_session_id("foo@bar").is_err());
        assert!(SessionStore::validate_session_id("foo:bar").is_err());
    }

    #[test]
    fn test_validate_session_id_accepts_valid_ids() {
        // UUIDs
        assert!(SessionStore::validate_session_id("550e8400-e29b-41d4-a716-446655440000").is_ok());

        // Alphanumeric slugs
        assert!(SessionStore::validate_session_id("my-session-123").is_ok());
        assert!(SessionStore::validate_session_id("session_with_underscores").is_ok());
        assert!(SessionStore::validate_session_id("ABC123").is_ok());
    }

    #[test]
    fn test_get_session_rejects_path_traversal() {
        let (store, _temp) = create_test_store();

        // Create a valid session first
        let session = store
            .create_session("agent-1", SessionMetadata::default())
            .unwrap();
        assert!(store.get_session(&session.id).is_ok());

        // Path traversal attempts should fail with InvalidSessionKey error
        let result = store.get_session("../malicious");
        assert!(matches!(
            result,
            Err(SessionStoreError::InvalidSessionKey(_))
        ));

        let result = store.get_session("foo/../bar");
        assert!(matches!(
            result,
            Err(SessionStoreError::InvalidSessionKey(_))
        ));

        let result = store.get_session("/etc/passwd");
        assert!(matches!(
            result,
            Err(SessionStoreError::InvalidSessionKey(_))
        ));
    }

    #[test]
    fn test_delete_session_rejects_path_traversal() {
        let (store, _temp) = create_test_store();

        // Path traversal attempts should fail
        let result = store.delete_session("../malicious");
        assert!(matches!(
            result,
            Err(SessionStoreError::InvalidSessionKey(_))
        ));
    }

    #[test]
    fn test_get_history_rejects_path_traversal() {
        let (store, _temp) = create_test_store();

        // Path traversal attempts should fail
        let result = store.get_history("../malicious", None, None);
        assert!(matches!(
            result,
            Err(SessionStoreError::InvalidSessionKey(_))
        ));
    }

    // ==================== Export / Purge User Data Tests ====================

    #[test]
    fn test_export_user_data_empty() {
        let (store, _temp) = create_test_store();

        let result = store.export_user_data("nonexistent-user").unwrap();

        assert_eq!(result["user_id"], "nonexistent-user");
        assert_eq!(result["session_count"], 0);
        assert!(result["sessions"].as_array().unwrap().is_empty());
        assert!(result["warnings"].as_array().unwrap().is_empty());
        assert!(result["exported_at"].as_str().is_some());
    }

    #[test]
    fn test_export_user_data_with_sessions() {
        let (store, _temp) = create_test_store();

        // Create two sessions for user-1
        let s1 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-1".into()),
                    channel: Some("telegram".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        let s2 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-1".into()),
                    channel: Some("discord".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        // Add messages to both sessions
        store
            .append_message(ChatMessage::user(&s1.id, "Hello from session 1"))
            .unwrap();
        store
            .append_message(ChatMessage::assistant(&s1.id, "Hi there!"))
            .unwrap();
        store
            .append_message(ChatMessage::user(&s2.id, "Hello from session 2"))
            .unwrap();

        let result = store.export_user_data("user-1").unwrap();

        assert_eq!(result["user_id"], "user-1");
        assert_eq!(result["session_count"], 2);

        let sessions = result["sessions"].as_array().unwrap();
        assert_eq!(sessions.len(), 2);

        // Verify each exported session has both "session" and "messages" keys
        for exported in sessions {
            assert!(exported["session"].is_object());
            assert!(exported["messages"].is_array());
        }

        assert!(result["warnings"].as_array().unwrap().is_empty());
    }

    #[test]
    fn test_export_user_data_redacts_error_detail_in_warnings() {
        let temp_dir = TempDir::new().unwrap();
        let key = Zeroizing::new(integrity::derive_hmac_key(b"history-secret"));
        let store = SessionStore::with_base_path(temp_dir.path().to_path_buf())
            .with_hmac_key(key)
            .with_integrity_action(integrity::IntegrityAction::Reject);

        let session = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-1".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "Hello"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        fs::write(&history_path, "tampered\n").unwrap();

        let result = store.export_user_data("user-1").unwrap();
        let warnings = result["warnings"].as_array().unwrap();
        assert_eq!(warnings.len(), 1);
        let warning = warnings[0].as_str().unwrap();
        assert!(warning.contains("failed to export session"));
        assert!(warning.ends_with(": session history is corrupt or failed integrity verification"));
        assert!(!warning.contains("Session store is locked"));
        assert!(!warning.contains("history-secret"));
    }

    #[test]
    fn test_export_user_data_warning_uses_encryption_gate_reason() {
        let key_material = test_key_material();
        let (store, _temp_dir) = create_encrypted_store_without_hmac(&key_material);
        let session = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-1".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        store
            .append_message(ChatMessage::user(&session.id, "encrypted-first"))
            .unwrap();

        let history_path = store.session_history_path(&session.id).unwrap();
        let plaintext = serde_json::to_vec(&ChatMessage::user(&session.id, "plaintext-history"))
            .expect("plaintext history fixture should serialize");
        fs::write(&history_path, [plaintext.as_slice(), b"\n"].concat()).unwrap();

        let result = store.export_user_data("user-1").unwrap();
        let warnings = result["warnings"].as_array().unwrap();
        assert_eq!(warnings.len(), 1);
        let warning = warnings[0].as_str().unwrap();
        assert!(warning.contains("failed to export session"));
        assert!(warning.ends_with(
            "session history is not encrypted; current session encryption requires encrypted artifacts"
        ));
        assert!(!warning.contains("without the config password"));
    }

    #[test]
    fn test_purge_user_data_empty() {
        let (store, _temp) = create_test_store();

        let (deleted, total) = store.purge_user_data("nonexistent-user").unwrap();
        assert_eq!(deleted, 0);
        assert_eq!(total, 0);
    }

    #[test]
    fn test_purge_user_data_deletes_user_sessions() {
        let (store, _temp) = create_test_store();

        // Create sessions for user-1
        let s1 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-1".into()),
                    channel: Some("telegram".into()),
                    chat_id: Some("u1-chat1".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-1".into()),
                    channel: Some("discord".into()),
                    chat_id: Some("u1-chat2".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        // Create session for user-2
        let s3 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user-2".into()),
                    channel: Some("telegram".into()),
                    chat_id: Some("u2-chat1".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        // Add messages
        store
            .append_message(ChatMessage::user(&s1.id, "Hello"))
            .unwrap();

        // Purge user-1
        let (deleted, total) = store.purge_user_data("user-1").unwrap();
        assert_eq!(deleted, 2);
        assert_eq!(total, 2);

        // Verify user-1 sessions are gone
        let user1_sessions = store
            .list_sessions(SessionFilter::new().with_user_id("user-1"))
            .unwrap();
        assert!(user1_sessions.is_empty());

        // Verify user-2 sessions are still intact
        let user2_sessions = store
            .list_sessions(SessionFilter::new().with_user_id("user-2"))
            .unwrap();
        assert_eq!(user2_sessions.len(), 1);
        assert_eq!(user2_sessions[0].id, s3.id);
    }

    #[test]
    fn test_purge_user_data_returns_total() {
        let (store, _temp) = create_test_store();

        // Create 3 sessions for user-1
        for i in 0..3 {
            store
                .create_session(
                    "agent-1",
                    SessionMetadata {
                        user_id: Some("user-1".into()),
                        channel: Some(format!("ch-{}", i)),
                        ..Default::default()
                    },
                )
                .unwrap();
        }

        let (deleted, total) = store.purge_user_data("user-1").unwrap();
        assert_eq!(total, 3);
        assert_eq!(
            deleted, total,
            "all sessions should be successfully deleted"
        );
    }

    #[test]
    fn test_export_user_data_rejects_empty_user_id() {
        let (store, _temp) = create_test_store();
        let result = store.export_user_data("");
        assert!(result.is_err());
        assert!(matches!(result, Err(SessionStoreError::InvalidUserId(_))));
    }

    #[test]
    fn test_export_user_data_rejects_whitespace_user_id() {
        let (store, _temp) = create_test_store();
        let result = store.export_user_data("   ");
        assert!(result.is_err());
        assert!(matches!(result, Err(SessionStoreError::InvalidUserId(_))));
    }

    #[test]
    fn test_purge_user_data_rejects_empty_user_id() {
        let (store, _temp) = create_test_store();
        let result = store.purge_user_data("");
        assert!(result.is_err());
        assert!(matches!(result, Err(SessionStoreError::InvalidUserId(_))));
    }

    #[test]
    fn test_purge_user_data_rejects_whitespace_user_id() {
        let (store, _temp) = create_test_store();
        let result = store.purge_user_data("  \t  ");
        assert!(result.is_err());
        assert!(matches!(result, Err(SessionStoreError::InvalidUserId(_))));
    }

    #[test]
    fn test_export_user_data_returns_sessions() {
        let (store, _temp) = create_test_store();

        // Create sessions for the user
        let _s1 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user123".into()),
                    channel: Some("channel1".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let _s2 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user123".into()),
                    channel: Some("channel2".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        // Create session for different user
        let _s3 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("other_user".into()),
                    channel: Some("channel1".into()),
                    chat_id: Some("other".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        let export = store.export_user_data("user123").unwrap();
        assert_eq!(export["session_count"], 2);
        assert_eq!(export["user_id"], "user123");
        assert!(export["exported_at"].is_string());
        assert!(export["warnings"].as_array().unwrap().is_empty());
    }

    #[test]
    fn test_export_user_data_empty_user() {
        let (store, _temp) = create_test_store();

        let result = store.export_user_data("");
        assert!(result.is_err());
    }

    #[test]
    fn test_export_user_data_whitespace_user() {
        let (store, _temp) = create_test_store();

        let result = store.export_user_data("   ");
        assert!(result.is_err());
    }

    #[test]
    fn test_export_user_data_no_sessions() {
        let (store, _temp) = create_test_store();

        let export = store.export_user_data("nonexistent").unwrap();
        assert_eq!(export["session_count"], 0);
        assert!(export["sessions"].as_array().unwrap().is_empty());
    }

    #[test]
    fn test_purge_user_data_deletes_sessions() {
        let (store, _temp) = create_test_store();

        let _s1 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user123".into()),
                    channel: Some("channel1".into()),
                    chat_id: Some("purge-c1".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let _s2 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("user123".into()),
                    channel: Some("channel2".into()),
                    chat_id: Some("purge-c2".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        let _s3 = store
            .create_session(
                "agent-1",
                SessionMetadata {
                    user_id: Some("other_user".into()),
                    channel: Some("channel1".into()),
                    chat_id: Some("purge-other".into()),
                    ..Default::default()
                },
            )
            .unwrap();

        let (deleted, total) = store.purge_user_data("user123").unwrap();
        assert_eq!(deleted, 2);
        assert_eq!(total, 2);

        // Verify user123's sessions are gone
        let export = store.export_user_data("user123").unwrap();
        assert_eq!(export["session_count"], 0);

        // Verify other_user's sessions remain
        let export = store.export_user_data("other_user").unwrap();
        assert_eq!(export["session_count"], 1);
    }

    #[test]
    fn test_purge_user_data_empty_user() {
        let (store, _temp) = create_test_store();

        let result = store.purge_user_data("");
        assert!(result.is_err());
    }

    #[test]
    fn test_purge_user_data_no_sessions() {
        let (store, _temp) = create_test_store();

        let (deleted, total) = store.purge_user_data("nonexistent").unwrap();
        assert_eq!(deleted, 0);
        assert_eq!(total, 0);
    }
}
