store.rs

//! In-memory SessionStore: create sessions, record events, and trigger persistence.
//!
//! All durable session-map mutations flow through `SessionStoreInner` helpers.
//! Callers outside this module use `SessionStore` methods only.
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};
use std::io;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use webcodex_core::runner_job_lifecycle::RunnerJobLifecycle;
use webcodex_core::validation_identity::{
    assertion_validation_identity, is_validation_execution_identity,
};
use webcodex_core::workflow_session_contract::{is_safe_job_id, PermissionDecision, SessionMode};
use webcodex_tool_contracts::{
    runtime_tool_activity_semantics, runtime_tool_session_evidence_policy, ToolActivityKind,
    ToolSessionLifecycleEffect,
};

use super::assignment::{
    assignment_fence_fingerprint, assignment_fence_from_state, current_assignment_state,
    open_assignment_state, validate_assignment_fence,
};
use super::console::{
    build_detail as build_console_detail, build_list_item as build_console_list_item,
    normalize_console_activity_limit, normalize_console_session_limit, CachedConsoleList,
    ConsoleValidationHooks, WorkflowSessionConsoleDetail, WorkflowSessionConsoleList,
    WorkflowSessionConsoleListItem,
};
use super::events::{
    actual_failure_kind_for_tool_result, changed_paths_for_tool_call,
    changed_paths_for_tool_result, classify_error_message, classify_failure_expectation,
    context_result_summary_for_tool_result, diff_review_like_for_tool, extract_job_id,
    extract_project, is_valid_session_id, observed_input_paths_for_tool,
    observed_paths_for_successful_result, persistent_shell_event_evidence_for_tool_result,
    sanitize_tool_execution_state, session_input_summary_for_tool,
    validation_output_summary_for_tool_result, SessionToolContract,
};
#[cfg(any(test, feature = "root-test-support"))]
use super::model::SESSION_LEDGER_VERSION;
use super::model::{
    CodingSessionError, CodingSessionOutcome, CodingSessionRequest, ColdSessionRecord,
    CompleteSessionMessageInput, CompleteSessionMessageOutcome, PersistedSessionRecord,
    PersistentShellEventEvidence, PostSessionMessageInput, ReplaceSessionMessageInput,
    ReplaceSessionMessageOutcome, SessionCloseError, SessionCloseOutcome, SessionCounts,
    SessionCreateOptions, SessionDiscoveryItem, SessionDiscoveryPage, SessionEvent,
    SessionExecutionContext, SessionExecutionContextUpdateError,
    SessionExecutionContextUpdateOutcome, SessionGuardDenial, SessionGuards, SessionLifecycle,
    SessionLifecycleDenial, SessionMessage, SessionMessageClosureKind, SessionMessageDelivery,
    SessionMessageDeliveryOutcome, SessionMessageDeliveryReplay, SessionMessageError,
    SessionMessagePriority, SessionMessageStatus, SessionRecord, SessionRetentionTombstone,
    SessionStoreStatus, SessionSummary, SessionTransport, StoredSession, ToolCallExpectation,
    ToolCallRecorderMetadata, ToolCallStart, ToolEffectEventEvidence,
    WithdrawSessionMessageOutcome, CALL_ID_PREFIX, DEFAULT_MAX_SESSIONS,
    DEFAULT_RECENT_EVENT_LIMIT, DEFAULT_SESSION_CACHE_BYTES, DEFAULT_SUMMARY_LIMIT,
    EVENT_ID_PREFIX, MAX_CODING_INSTRUCTION_CHARS, MAX_INPUT_ARRAY_ITEMS,
    MAX_MATERIALIZED_VALIDATION_JOB_IDS, MAX_MESSAGE_DELIVERY_KEY_CHARS, MAX_SUMMARY_LIMIT,
    MESSAGE_ID_PREFIX, SESSION_ID_PREFIX,
};
#[cfg(test)]
use super::model::{
    PersistedRetentionTombstoneRow, PersistedSessionLedger, PersistedSessionSnapshot,
};
use super::persistence::{cold_session_from_persisted, materialize_cold_session};
use super::query::{
    build_messages_summary, is_valid_completion_id, validate_message_tags, validate_message_text,
    validate_resolution_text,
};
use super::util::{
    bound_event_error_summary, bound_summary_string, now_ts, redact_and_bound_instruction,
    redact_and_bound_value,
};

#[cfg(test)]
#[path = "identifier_tests.rs"]
mod identifier_tests;

mod recency;
mod records;
mod residency;
use crate::persistence::database::{SessionDatabase, SessionWrite};
use records::SessionRecords;

use recency::SessionRecency;
mod writer;
use writer::LedgerWriterGuard;

#[cfg(test)]
mod borrowed_projection_tests;
#[cfg(test)]
mod event_tests;
#[cfg(test)]
mod scale_tests;

#[derive(Debug, Clone)]
pub struct SessionStore {
    /// Shared session map and LRU metadata.
    /// `pub(super)` so sibling modules can lock and call `SessionStoreInner`
    /// transition helpers without touching the maps directly.
    pub(super) inner: Arc<Mutex<SessionStoreInner>>,
    persistence_write_mutex: Arc<Mutex<()>>,
    /// At most one blocking durability write when the dedicated writer is unavailable.
    persistence_fallback: Arc<tokio::sync::Semaphore>,
    /// Background writer; absent for memory-only stores or a failed thread spawn.
    /// Ordinary mutations retain synchronous fallback; durable requests admit
    /// one blocking write before asynchronously waiting for its confirmation.
    writer: Option<Arc<LedgerWriterGuard>>,
    /// Process-local wake signal only; durable observation truth remains in the
    /// persisted per-Session revision bookkeeping.
    pub(super) message_observation_notify: tokio::sync::watch::Sender<u64>,
    #[cfg(any(test, feature = "root-test-support"))]
    fail_next_coding_continuity_precommit: Arc<std::sync::atomic::AtomicBool>,
}

/// Owns admission to a durability barrier, not the business mutation itself.
pub(super) struct SessionDurability<'a> {
    store: &'a SessionStore,
    fallback_permit: Option<tokio::sync::OwnedSemaphorePermit>,
}

impl SessionDurability<'_> {
    pub(super) async fn persist(self) -> Result<(), ()> {
        if let Some(writer) = &self.store.writer {
            let generation = self
                .store
                .inner
                .lock()
                .expect("Session store mutex poisoned")
                .sessions
                .generation;
            writer.flush_through(generation).await
        } else if let Some(permit) = self.fallback_permit {
            let store = self.store.clone();
            // The task owns the permit and write after it is scheduled. Dropping
            // the requesting future cannot cancel or replay that write.
            tokio::task::spawn_blocking(move || {
                let _permit = permit;
                writer::persist_batch(&store.inner, &store.persistence_write_mutex)
                    .1
                    .map(|_| ())
                    .map_err(|_| ())
            })
            .await
            .map_err(|_| ())?
        } else {
            Ok(())
        }
    }
}

#[derive(Debug)]
pub(super) struct SessionStoreInner {
    /// Durable workflow sessions. Mutated only via the helpers below.
    sessions: SessionRecords,
    lru: SessionRecency,
    hot_session_capacity_target: usize,
    cache_target_bytes: usize,
    /// Imported legacy deletion proofs; no new automatic expiry or pruning.
    tombstones: BTreeMap<u64, SessionRetentionTombstone>,
    tombstone_ids: HashMap<String, u64>,
    recent_event_limit: usize,
    persistence: Option<SessionPersistence>,
}

#[derive(Debug, Clone)]
struct SessionPersistence {
    path: PathBuf,
    database: Arc<SessionDatabase>,
    restored_sessions: usize,
    last_persist_error: Option<String>,
}

impl Default for SessionStore {
    fn default() -> Self {
        Self::new_in_memory(DEFAULT_MAX_SESSIONS, DEFAULT_RECENT_EVENT_LIMIT)
    }
}

impl SessionStore {
    pub fn new(max_sessions: usize, recent_event_limit: usize) -> Self {
        Self::new_in_memory(max_sessions, recent_event_limit)
    }

    /// Memory-only embedder/fixture; its Hot count target never deletes identity.
    pub fn new_in_memory(hot_session_capacity_target: usize, recent_event_limit: usize) -> Self {
        let (message_observation_notify, _) = tokio::sync::watch::channel(0_u64);
        Self {
            inner: Arc::new(Mutex::new(SessionStoreInner {
                sessions: SessionRecords::default(),
                lru: SessionRecency::default(),
                hot_session_capacity_target,
                cache_target_bytes: DEFAULT_SESSION_CACHE_BYTES,
                tombstones: BTreeMap::new(),
                tombstone_ids: HashMap::new(),
                recent_event_limit,
                persistence: None,
            })),
            persistence_write_mutex: Arc::new(Mutex::new(())),
            persistence_fallback: Arc::new(tokio::sync::Semaphore::new(1)),
            writer: None,
            message_observation_notify,
            #[cfg(any(test, feature = "root-test-support"))]
            fail_next_coding_continuity_precommit: Arc::new(std::sync::atomic::AtomicBool::new(
                false,
            )),
        }
    }

    /// Legacy fixture signature retained only to exercise old count thresholds.
    #[cfg(any(test, feature = "root-test-support"))]
    pub fn new_in_memory_with_limits(
        hot: usize,
        _retired_history_limit: usize,
        recent: usize,
    ) -> Self {
        Self::new_in_memory(hot, recent)
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn with_persistence(
        path: impl Into<PathBuf>,
        max_sessions: usize,
        recent_event_limit: usize,
    ) -> Self {
        Self::with_persistence_limits(path, max_sessions, max_sessions, recent_event_limit)
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn with_persistence_limits(
        path: impl Into<PathBuf>,
        _hot_session_capacity_target: usize,
        _historical_session_retention_limit: usize,
        recent_event_limit: usize,
    ) -> Self {
        let legacy = path.into();
        let database = legacy.with_extension("sqlite3");
        Self::open_database(
            database,
            Some(&legacy),
            DEFAULT_SESSION_CACHE_BYTES,
            recent_event_limit,
        )
        .expect("open Workflow Session database")
    }

    /// Open the SQLite domain in the configured Server database. Startup fails
    /// closed on storage or migration errors; it never substitutes an empty store.
    pub fn open_database(
        path: impl Into<PathBuf>,
        legacy: Option<&std::path::Path>,
        cache_target_bytes: usize,
        recent_event_limit: usize,
    ) -> io::Result<Self> {
        let path = path.into();
        let database = SessionDatabase::open(&path)?;
        if let Some(legacy) = legacy {
            database.import_legacy(legacy)?;
        }
        let inventory = database.inventory()?;
        let mut lru = SessionRecency::default();
        let mut sessions = HashMap::new();
        let mut generation = 0;
        for record in inventory {
            generation = generation.max(record.database_revision);
            lru.touch(
                &record.session_id,
                record.lifecycle == SessionLifecycle::Closed,
            );
            sessions.insert(record.session_id.clone(), StoredSession::Cold(record));
        }
        let mut tombstones = BTreeMap::new();
        let mut tombstone_ids = HashMap::new();
        for (index, mut tombstone) in database.tombstones()?.into_iter().enumerate() {
            let ordinal = index as u64 + 1;
            tombstone.expiry_ordinal = ordinal;
            tombstone_ids.insert(tombstone.session_id.clone(), ordinal);
            tombstones.insert(ordinal, tombstone);
        }
        let restored_sessions = sessions.len();
        let inner = Arc::new(Mutex::new(SessionStoreInner {
            sessions: SessionRecords {
                generation,
                ..SessionRecords::restored(sessions)
            },
            lru,
            hot_session_capacity_target: DEFAULT_MAX_SESSIONS,
            cache_target_bytes,
            tombstones,
            tombstone_ids,
            recent_event_limit: recent_event_limit.max(1),
            persistence: Some(SessionPersistence {
                path,
                database,
                restored_sessions,
                last_persist_error: None,
            }),
        }));
        let persistence_write_mutex = Arc::new(Mutex::new(()));
        let writer =
            LedgerWriterGuard::spawn(Arc::clone(&inner), Arc::clone(&persistence_write_mutex));
        let (message_observation_notify, _) = tokio::sync::watch::channel(0_u64);
        Ok(Self {
            inner,
            persistence_write_mutex,
            persistence_fallback: Arc::new(tokio::sync::Semaphore::new(1)),
            writer,
            message_observation_notify,
            #[cfg(any(test, feature = "root-test-support"))]
            fail_next_coding_continuity_precommit: Arc::new(std::sync::atomic::AtomicBool::new(
                false,
            )),
        })
    }

    /// Synchronous fixture barrier. Production async callers use `flush` and
    /// propagate its error; inspecting status does not itself confirm durability.
    #[cfg(any(test, feature = "root-test-support"))]
    pub fn flush_persistence(&self) {
        if let Some(writer) = &self.writer {
            writer.flush();
        }
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn set_persistence_failure_for_test(&self, fail: bool) {
        let database = self
            .inner
            .lock()
            .unwrap()
            .persistence
            .as_ref()
            .unwrap()
            .database
            .clone();
        database.fail_writes(fail);
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn persisted_json_for_test(&self) -> String {
        self.flush_persistence();
        let database = self
            .inner
            .lock()
            .unwrap()
            .persistence
            .as_ref()
            .unwrap()
            .database
            .clone();
        let rows = database
            .inventory()
            .unwrap()
            .into_iter()
            .map(|cold| database.load(&cold, usize::MAX).unwrap())
            .collect::<Vec<_>>();
        serde_json::to_string(
            &serde_json::json!({"version": SESSION_LEDGER_VERSION, "sessions": rows}),
        )
        .unwrap()
    }

    pub fn status(&self) -> SessionStoreStatus {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        let (persistence, restored_sessions, last_persist_error) = match &inner.persistence {
            Some(persistence) => (
                "enabled".to_string(),
                persistence.restored_sessions,
                persistence.last_persist_error.clone(),
            ),
            None => ("disabled".to_string(), 0, None),
        };
        let active_sessions = inner
            .sessions
            .values()
            .filter(|session| session.lifecycle() == SessionLifecycle::Active)
            .count();
        let closed_sessions = inner.sessions.len().saturating_sub(active_sessions);
        let hot_sessions = inner
            .sessions
            .values()
            .filter(|session| matches!(session, StoredSession::Hot(_)))
            .count();
        let cold_sessions = inner.sessions.len().saturating_sub(hot_sessions);
        SessionStoreStatus {
            persistence,
            restored_sessions,
            retained_sessions: inner.sessions.len(),
            active_sessions,
            closed_sessions,
            hot_sessions,
            cold_sessions,
            active_cold_sessions: inner.sessions.values().filter(|entry| {
                matches!(entry, StoredSession::Cold(cold) if cold.lifecycle.allows_mutation())
            }).count(),
            cold_payload_bytes: inner.sessions.values().filter_map(|entry| {
                match entry {
                    StoredSession::Cold(cold) => Some(cold.raw.get().len()),
                    StoredSession::Hot(_) => None,
                }
            }).sum(),
            cache_target_bytes: inner.persistence.as_ref().map(|_| inner.cache_target_bytes),
            retention_tombstones: inner.tombstones.len(),
            recent_event_limit: inner.recent_event_limit,
            last_persist_error,
        }
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn active_session_count_for_test(&self, project: Option<&str>) -> usize {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner
            .sessions
            .values()
            .filter(|record| {
                record.lifecycle() == SessionLifecycle::Active
                    && project.is_none_or(|project| record.project() == Some(project))
            })
            .count()
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn hot_payload_entry_count_for_test(&self, session_id: &str) -> Option<usize> {
        self.inner
            .lock()
            .expect("session store mutex poisoned")
            .sessions
            .get(session_id)?
            .hot()
            .map(|record| record.events.len() + record.messages.len())
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn cold_payload_bytes_for_test(&self, session_id: &str) -> Option<usize> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        match inner.sessions.get(session_id)? {
            StoredSession::Hot(_) => None,
            StoredSession::Cold(record) => Some(record.raw.get().len()),
        }
    }

    /// Thin convenience wrapper — creation always goes through
    /// [`Self::start_session_with_options`].
    #[cfg(any(test, feature = "root-test-support"))]
    pub fn start_session(&self, project: Option<String>, title: Option<String>) -> SessionSummary {
        self.start_session_with_guards(
            project,
            title,
            SessionMode::Normal,
            SessionGuards::default(),
        )
    }

    /// Thin convenience wrapper — creation always goes through
    /// [`Self::start_session_with_options`].
    #[cfg(any(test, feature = "root-test-support"))]
    pub fn start_session_with_guards(
        &self,
        project: Option<String>,
        title: Option<String>,
        mode: SessionMode,
        guards: SessionGuards,
    ) -> SessionSummary {
        self.start_session_with_options(SessionCreateOptions::new(project, title, mode, guards))
            .expect("default Session execution context must be valid")
    }

    /// Sole create entry point for workflow sessions.
    ///
    /// Stores session-creation inputs (including project instructions) on the
    /// `SessionRecord`. Convenience wrappers above all delegate here.
    pub fn start_session_with_options(
        &self,
        mut opts: SessionCreateOptions,
    ) -> Result<SessionSummary, String> {
        opts.execution_context = opts.execution_context.validated()?;
        #[cfg(any(test, feature = "root-test-support"))]
        if opts.owner_authority_fingerprint.is_none() {
            // Direct store fixtures do not have an AuthContext. Give them an
            // explicit cfg(test)-only authority so production invariants never
            // depend on an optional creation-time fingerprint.
            opts.owner_authority_fingerprint =
                Some(super::TEST_ONLY_PROJECT_SESSION_AUTHORITY_FINGERPRINT.to_string());
        }
        if opts.project.is_none() && !opts.execution_context.is_empty() {
            return Err(
                "execution_context requires a Workflow Session bound to a registered project"
                    .to_string(),
            );
        }
        let summary = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            let session_id = inner
                .allocate_session_id(webcodex_core::compact::random_suffix::<12>)
                .ok_or_else(|| "session_id_allocation_exhausted".to_string())?;
            let now = now_ts();
            let guards = SessionGuards::effective(opts.mode, opts.guards);
            let owner_authority_fingerprint =
                opts.owner_authority_fingerprint.ok_or_else(|| {
                    "Workflow Session creation requires a canonical authority fingerprint"
                        .to_string()
                })?;
            let record = SessionRecord {
                current_instruction: None,
                history_is_durable: false,
                event_history_start: 0,
                pending_events: Default::default(),
                durable: None,
                session_id: session_id.clone(),
                project: opts.project,
                project_identity: opts.project_identity,
                owner_authority_fingerprint,
                title: opts.title,
                mode: opts.mode,
                guards,
                execution_context: opts.execution_context,
                // Create always yields Active; only explicit close transitions later.
                lifecycle: SessionLifecycle::Active,
                created_at: now,
                updated_at: now,
                messages: VecDeque::new(),
                message_delivery_replays: Default::default(),
                events: VecDeque::new(),
                events_observed: 0,
                git_baseline_tree: None,
                repository_edit_observed: false,
                materialized_validation_job_ids: VecDeque::new(),
                message_observation_revision: 0,
                message_observation_floor: 0,
                message_observation_revisions: Default::default(),
                assignment_history_floors: Default::default(),
                assignment_history_tracking_complete: true,
                completion_assignment_fence_fingerprints: Default::default(),
                completion_assignment_fence_tracking_complete: true,
                project_instructions: opts.project_instructions,
            };
            inner.insert_session(record)
        };
        self.persist_after_mutation();
        Ok(summary)
    }

    /// Commit the coding context and accepted instruction under the in-memory
    /// store lock. Persistent stores then queue the updated JSON ledger to the
    /// background writer; success does not imply disk flush.
    ///
    /// Every fallible check happens before the in-memory commit. Once mutation
    /// begins, session creation or capability update and the instruction event
    /// are applied under the same store lock and enter one persistence generation.
    pub fn ensure_coding_session(
        &self,
        request: CodingSessionRequest,
    ) -> Result<CodingSessionOutcome, CodingSessionError> {
        self.ensure_coding_session_with_git_baseline(request, None)
    }

    /// Coding bootstrap variant used by `work_on_project` to bind one fresh
    /// startup Git tree. Exact continuation ignores the supplied candidate and
    /// preserves the durable Session baseline already recorded at creation.
    pub fn ensure_coding_session_with_git_baseline(
        &self,
        request: CodingSessionRequest,
        git_baseline_tree: Option<String>,
    ) -> Result<CodingSessionOutcome, CodingSessionError> {
        let explicit_resume_session_id = match request.resume_session_id.as_deref() {
            Some(session_id)
                if session_id != session_id.trim() || !is_valid_session_id(session_id) =>
            {
                return Err(CodingSessionError::InvalidResumeSessionId);
            }
            Some(session_id) => Some(session_id.to_string()),
            None => None,
        };
        let explicit_resume = explicit_resume_session_id.is_some();
        let now = now_ts();
        let new_event_id = format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple());
        let requested_guards = SessionGuards::effective(request.mode, request.guards);
        let requested_execution_context = request
            .execution_context
            .clone()
            .map(SessionExecutionContext::validated)
            .transpose()
            .map_err(CodingSessionError::InvalidExecutionContext)?;
        let instruction = request
            .instruction
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(|value| redact_and_bound_instruction(value, MAX_CODING_INSTRUCTION_CHARS));

        let outcome = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            let reusable_session_id = if let Some(session_id) = explicit_resume_session_id {
                let Some(stored) = inner.sessions.get(&session_id) else {
                    if inner
                        .owned_retention_incarnation(
                            &session_id,
                            request.authority_fingerprint.as_str(),
                        )
                        .is_some()
                    {
                        return Err(CodingSessionError::ResumeRetentionExpired { session_id });
                    }
                    return Err(CodingSessionError::UnknownResumeSession { session_id });
                };
                let lifecycle = stored.lifecycle();
                if !lifecycle.allows_mutation() {
                    return Err(CodingSessionError::ResumeSessionNotActive {
                        session_id,
                        lifecycle,
                    });
                }
                if stored.project() != Some(request.project.as_str()) {
                    return Err(CodingSessionError::ResumeProjectMismatch {
                        session_id,
                        session_project: stored.project().map(str::to_string),
                        request_project: request.project.clone(),
                    });
                }
                if stored.project_identity() != request.project_identity.as_deref() {
                    return Err(CodingSessionError::ResumeProjectIdentityChanged { session_id });
                }
                Some(session_id)
            } else {
                None
            };

            if let Some(session_id) = reusable_session_id.as_deref() {
                let stored_fingerprint = inner
                    .sessions
                    .get(session_id)
                    .expect("reusable session must exist")
                    .owner_authority_fingerprint();
                #[cfg(any(test, feature = "root-test-support"))]
                let synthetic_test_fixture =
                    stored_fingerprint == super::TEST_ONLY_PROJECT_SESSION_AUTHORITY_FINGERPRINT;
                #[cfg(not(any(test, feature = "root-test-support")))]
                let synthetic_test_fixture = false;
                if !synthetic_test_fixture
                    && stored_fingerprint != request.authority_fingerprint.as_str()
                {
                    return Err(CodingSessionError::ResumeAuthorityMismatch {
                        session_id: session_id.to_string(),
                    });
                }
            }

            if let Some(session_id) = reusable_session_id {
                if !inner.prepare_record_mutation(&session_id) {
                    return Err(CodingSessionError::CommitFailed);
                }

                let (previous_mode, previous_guards, previous_execution_context) = {
                    let record = inner
                        .sessions
                        .get(&session_id)
                        .and_then(StoredSession::hot)
                        .expect("active reusable session must stay hot");
                    (record.mode, record.guards, record.execution_context.clone())
                };
                // Repeating a mode preserves any stricter explicit guard from
                // the root task. A real mode transition applies the newly
                // requested effective guard profile.
                let next_guards = if previous_mode == request.mode {
                    SessionGuards {
                        deny_write_tools: previous_guards.deny_write_tools
                            || requested_guards.deny_write_tools,
                        deny_shell_tools: previous_guards.deny_shell_tools
                            || requested_guards.deny_shell_tools,
                    }
                } else {
                    requested_guards
                };
                let capability_changed =
                    previous_mode != request.mode || previous_guards != next_guards;
                let next_execution_context = requested_execution_context
                    .clone()
                    .unwrap_or_else(|| previous_execution_context.clone());
                let execution_context_changed =
                    next_execution_context != previous_execution_context;
                if previous_guards.deny_write_tools
                    && !next_guards.deny_write_tools
                    && !request.write_scope_verified
                {
                    return Err(CodingSessionError::WriteScopeRequired);
                }
                if self.take_coding_continuity_fault() {
                    return Err(CodingSessionError::CommitFailed);
                }

                // Snapshot the previous attempt *before* appending the new
                // task_instruction. Continuation feedback projects over this so
                // it describes what the previous attempt did, not the empty new
                // attempt. Use the per-session evidence cap so a long previous
                // attempt's work is retained for the projection.
                let pre_instruction_summary = inner
                    .summary(&session_id, Some(DEFAULT_RECENT_EVENT_LIMIT))
                    .expect("reusable session must summarize before instruction append");

                let event = coding_instruction_event(
                    &new_event_id,
                    &session_id,
                    &request.project,
                    instruction.clone(),
                    request.transport,
                    request.mode,
                    Some(previous_mode),
                    requested_guards,
                    Some(previous_guards),
                    capability_changed,
                    request.context_refreshed,
                    requested_execution_context.clone(),
                    Some(previous_execution_context.clone()),
                    execution_context_changed,
                    true,
                    explicit_resume,
                    now,
                );
                {
                    let max_events = inner.recent_event_limit;
                    let record = inner
                        .sessions
                        .get_mut(&session_id)
                        .and_then(StoredSession::hot_mut)
                        .expect("active reusable session must stay hot before commit");
                    record.mode = request.mode;
                    record.guards = next_guards;
                    record.execution_context = next_execution_context;
                    record.updated_at = now;
                    if let Some(project_instructions) = request.project_instructions {
                        record.project_instructions = Some(
                            project_instructions
                                .retain_unavailable_scopes(record.project_instructions.as_ref()),
                        );
                    }
                    append_event(record, event, max_events);
                }
                inner.touch(&session_id);
                let summary = inner
                    .summary(&session_id, Some(DEFAULT_SUMMARY_LIMIT))
                    .expect("continued session must summarize");
                CodingSessionOutcome {
                    project_instructions: inner
                        .sessions
                        .get(&session_id)
                        .and_then(StoredSession::hot)
                        .and_then(|record| record.project_instructions.clone()),
                    summary,
                    pre_instruction_summary: Some(pre_instruction_summary),
                    reused: true,
                    previous_mode: Some(previous_mode),
                    previous_guards: Some(previous_guards),
                    capability_changed,
                    execution_context_changed,
                }
            } else {
                if self.take_coding_continuity_fault() {
                    return Err(CodingSessionError::CommitFailed);
                }
                let new_session_id = inner
                    .allocate_session_id(webcodex_core::compact::random_suffix::<12>)
                    .ok_or(CodingSessionError::CommitFailed)?;
                let execution_context = requested_execution_context.clone().unwrap_or_default();
                let execution_context_changed = !execution_context.is_empty();
                let event = coding_instruction_event(
                    &new_event_id,
                    &new_session_id,
                    &request.project,
                    instruction.clone(),
                    request.transport,
                    request.mode,
                    None,
                    requested_guards,
                    None,
                    false,
                    request.context_refreshed,
                    requested_execution_context,
                    None,
                    execution_context_changed,
                    false,
                    false,
                    now,
                );
                let record = SessionRecord {
                    current_instruction: None,
                    history_is_durable: false,
                    event_history_start: 0,
                    pending_events: Default::default(),
                    durable: None,
                    session_id: new_session_id.clone(),
                    project: Some(request.project.clone()),
                    project_identity: request.project_identity.clone(),
                    owner_authority_fingerprint: request.authority_fingerprint.clone(),
                    // The first accepted instruction remains the root title.
                    // Follow-up instructions never overwrite it.
                    title: instruction,
                    mode: request.mode,
                    guards: requested_guards,
                    execution_context,
                    lifecycle: SessionLifecycle::Active,
                    created_at: now,
                    updated_at: now,
                    messages: VecDeque::new(),
                    message_delivery_replays: Default::default(),
                    events: VecDeque::from([Arc::new(event)]),
                    events_observed: 1,
                    git_baseline_tree,
                    repository_edit_observed: false,
                    materialized_validation_job_ids: VecDeque::new(),
                    message_observation_revision: 0,
                    message_observation_floor: 0,
                    message_observation_revisions: Default::default(),
                    assignment_history_floors: Default::default(),
                    assignment_history_tracking_complete: true,
                    completion_assignment_fence_fingerprints: Default::default(),
                    completion_assignment_fence_tracking_complete: true,
                    project_instructions: request.project_instructions,
                };
                let project_instructions = record.project_instructions.clone();
                let summary = inner.insert_session(record);
                CodingSessionOutcome {
                    project_instructions,
                    summary,
                    pre_instruction_summary: None,
                    reused: false,
                    previous_mode: None,
                    previous_guards: None,
                    capability_changed: false,
                    execution_context_changed,
                }
            }
        };
        self.persist_after_mutation();
        Ok(outcome)
    }

    #[cfg(any(test, feature = "root-test-support"))]
    /// Inject a failure before any in-memory continuity mutation. This does
    /// not model background ledger persistence failure or rollback.
    pub fn fail_next_coding_continuity_precommit_for_test(&self) {
        self.fail_next_coding_continuity_precommit
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    #[cfg(any(test, feature = "root-test-support"))]
    fn take_coding_continuity_fault(&self) -> bool {
        self.fail_next_coding_continuity_precommit
            .swap(false, std::sync::atomic::Ordering::SeqCst)
    }

    #[cfg(not(any(test, feature = "root-test-support")))]
    fn take_coding_continuity_fault(&self) -> bool {
        false
    }

    pub fn summary(&self, session_id: &str, limit: Option<usize>) -> Option<SessionSummary> {
        self.with_record_for_query(session_id, |record, cold| {
            summarize_record(record, limit, cold)
        })
    }

    /// Exact retained changed-path evidence for recovery attribution. Unlike
    /// `summary`, this scans the full bounded durable event ledger instead of the
    /// model-facing 200-event tail, so ordinary presentation truncation does not
    /// masquerade as history loss. The boolean is true only when the retained
    /// ledger and each event's changed-path projection are known complete.
    pub fn retained_changed_path_evidence(&self, session_id: &str) -> Option<(Vec<String>, bool)> {
        self.with_record_for_query(session_id, |record, _cold| {
            let mut paths = BTreeSet::new();
            let mut complete = record.events_observed <= record.events.len() as u64;
            // Every persisted event sanitizes changed_paths to MAX_INPUT_ARRAY_ITEMS,
            // and runtime audit arguments can already be bounded before the Store
            // observes them. Equality to that durable bound therefore cannot prove
            // the original path set was complete, even while the Session is hot.
            for event in super::events::canonical_tool_call_finished_event_refs(
                record.events.iter().map(Arc::as_ref),
            ) {
                // Path attribution is consequence evidence, not an attempted-write
                // list. A failed/no-op edit can name the same path without proving
                // that this Session caused the current dirty state.
                if !event_observes_repository_edit(event) {
                    continue;
                }
                if event.changed_paths.len() >= MAX_INPUT_ARRAY_ITEMS {
                    complete = false;
                }
                paths.extend(event.changed_paths.iter().cloned());
            }
            (paths.into_iter().collect(), complete)
        })
    }

    /// Latest retained task-instruction identity at an exact Session event-count
    /// snapshot. Unlike `summary`, this scans the full bounded durable ledger so
    /// the model-facing 200-event tail cannot erase the current attempt identity.
    /// `through_events_total` fences the lookup against a later instruction racing
    /// the caller's already-observed Session snapshot. Durable eviction remains
    /// fail-closed and therefore returns `None`.
    pub fn retained_task_instruction_event_id_at(
        &self,
        session_id: &str,
        through_events_total: usize,
    ) -> Option<String> {
        self.with_record_for_query(session_id, |record, _| {
            let total = usize::try_from(record.events_observed).ok()?;
            if through_events_total > total {
                return None;
            }
            if through_events_total == total {
                return record
                    .current_instruction
                    .as_ref()
                    .map(|event| event.event_id.clone());
            }
            let first = total.saturating_sub(record.events.len());
            let tail = record
                .events
                .iter()
                .take(through_events_total.saturating_sub(first))
                .rev()
                .find(|event| event.kind == "task_instruction")
                .map(|event| event.event_id.clone());
            tail.or_else(|| {
                record
                    .durable
                    .as_ref()?
                    .database
                    .as_ref()?
                    .instruction_at(session_id, through_events_total)
                    .ok()
                    .flatten()
            })
        })
        .flatten()
    }

    /// Discover only identities in the already-authorized exact Project and
    /// creation-time authority group. Filter before counting/pagination; foreign
    /// Sessions cannot affect totals or hide authorized rows behind a page limit.
    /// Inventory ordering is presentation only, never an implicit resume rule.
    pub fn discover_sessions(
        &self,
        project: &str,
        owner_authority_fingerprint: &str,
        lifecycle: Option<SessionLifecycle>,
        offset: usize,
        limit: usize,
    ) -> SessionDiscoveryPage {
        let limit = limit.clamp(1, 20);
        let mut candidates = {
            let inner = self.inner.lock().expect("session store mutex poisoned");
            inner
                .sessions
                .values()
                .filter(|record| {
                    record.project() == Some(project)
                        && record.owner_authority_fingerprint() == owner_authority_fingerprint
                        && lifecycle.is_none_or(|state| record.lifecycle() == state)
                })
                .map(|record| (record.session_id().to_string(), record.updated_at()))
                .collect::<Vec<_>>()
        };
        candidates.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| right.0.cmp(&left.0)));
        let total = candidates.len();
        let end = offset.saturating_add(limit).min(total);
        let sessions = candidates
            .into_iter()
            .skip(offset)
            .take(limit)
            .filter_map(|(session_id, _)| {
                self.with_record_for_query(&session_id, |record, _| {
                    if record.project.as_deref() != Some(project)
                        || record.owner_authority_fingerprint != owner_authority_fingerprint
                        || lifecycle.is_some_and(|state| record.lifecycle != state)
                    {
                        return None;
                    }
                    let title = record.title.as_deref().map(|title| {
                        redact_and_bound_instruction(title, 240)
                            .chars()
                            .take(240)
                            .collect::<String>()
                    });
                    let title_truncated = title.as_deref() != record.title.as_deref();
                    Some(SessionDiscoveryItem {
                        session_id: record.session_id.clone(),
                        title,
                        title_truncated,
                        lifecycle: record.lifecycle,
                        created_at: record.created_at,
                        updated_at: record.updated_at,
                    })
                })
                .flatten()
            })
            .collect();
        SessionDiscoveryPage {
            total,
            offset,
            next_offset: (end < total).then_some(end),
            sessions,
        }
    }

    /// Bounded, read-only Workflow Session rows for one exact runtime project.
    /// The project is authoritative caller context, never request-controlled UI state.
    pub fn console_list_for_project(
        &self,
        project: &str,
        limit: Option<usize>,
        validation: ConsoleValidationHooks,
    ) -> WorkflowSessionConsoleList {
        self.console_lists_for_projects(&[project], limit, validation)
            .remove(project)
            .expect("requested project has a console list")
    }

    /// Scan retained identities once for an already-authorized set of projects.
    /// Each project keeps its own ordering and limit. Only immutable Closed
    /// records may reuse a small base Console projection across repeated reads.
    pub fn console_lists_for_projects(
        &self,
        projects: &[&str],
        limit: Option<usize>,
        validation: ConsoleValidationHooks,
    ) -> HashMap<String, WorkflowSessionConsoleList> {
        let limit = normalize_console_session_limit(limit);
        let mut candidates: HashMap<&str, Vec<(String, i64)>> = projects
            .iter()
            .map(|project| (*project, Vec::new()))
            .collect();
        if candidates.is_empty() {
            return HashMap::new();
        }
        {
            let inner = self.inner.lock().expect("session store mutex poisoned");
            for session in inner.sessions.values() {
                if let Some(rows) = session
                    .project()
                    .and_then(|project| candidates.get_mut(project))
                {
                    rows.push((session.session_id().to_string(), session.updated_at()));
                }
            }
        }
        // Sort and materialize outside the inventory lock. Cold records retain
        // their ordinary query path, including the exact-project recheck.
        candidates
            .into_iter()
            .map(|(project, mut candidates)| {
                candidates
                    .sort_by(|left, right| right.1.cmp(&left.1).then_with(|| right.0.cmp(&left.0)));
                let total = candidates.len();
                candidates.truncate(limit);
                let sessions = candidates
                    .into_iter()
                    .filter_map(|(session_id, _)| {
                        self.console_list_item_for_project(&session_id, project, validation)
                    })
                    .collect::<Vec<_>>();
                (
                    project.to_string(),
                    WorkflowSessionConsoleList {
                        returned: sessions.len(),
                        truncated: total > limit,
                        total,
                        sessions,
                    },
                )
            })
            .collect()
    }

    /// Reuse only a Closed Session's immutable base row (never a live Job
    /// overlay). Cache identity includes the cold raw allocation and both
    /// validation hooks, so replacements and alternate validation projections
    /// cannot return stale evidence. One row maximum per retained Closed identity.
    fn console_list_item_for_project(
        &self,
        session_id: &str,
        project: &str,
        validation: ConsoleValidationHooks,
    ) -> Option<WorkflowSessionConsoleListItem> {
        if self
            .inner
            .lock()
            .expect("session store mutex poisoned")
            .persistence
            .is_some()
        {
            return self
                .with_record_for_query(session_id, |record, _| {
                    (record.project.as_deref() == Some(project))
                        .then(|| build_console_list_item(record, project, validation))
                })
                .flatten();
        }
        let (cold, recent_event_limit) = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            inner.touch(session_id);
            let recent_event_limit = inner.recent_event_limit;
            match inner.sessions.get(session_id)? {
                StoredSession::Hot(record) => {
                    return (record.project.as_deref() == Some(project))
                        .then(|| build_console_list_item(record, project, validation));
                }
                StoredSession::Cold(record) => (record.clone(), recent_event_limit),
            }
        };

        if cold.project.as_deref() != Some(project) {
            return None;
        }
        let immutable = cold.lifecycle == SessionLifecycle::Closed;
        if immutable {
            let cache = cold
                .console_list_cache
                .lock()
                .expect("cold Console projection cache mutex poisoned");
            if let Some(cached) = cache.as_ref() {
                if Arc::ptr_eq(&cached.raw, &cold.raw) && cached.hooks.same_callbacks(validation) {
                    return Some(cached.row.clone());
                }
            }
        }

        let record = materialize_cold_session(&cold, recent_event_limit)?;
        if record.project.as_deref() != Some(project) {
            return None;
        }
        let row = build_console_list_item(&record, project, validation);
        if immutable {
            *cold
                .console_list_cache
                .lock()
                .expect("cold Console projection cache mutex poisoned") = Some(CachedConsoleList {
                raw: cold.raw.clone(),
                hooks: validation,
                row: row.clone(),
            });
        }
        Some(row)
    }

    /// Bounded, read-only human timeline for one exact project-scoped Session.
    /// Unknown and wrong-project ids intentionally collapse to the same `None`.
    pub fn console_detail_for_project(
        &self,
        project: &str,
        session_id: &str,
        limit: Option<usize>,
        validation: ConsoleValidationHooks,
    ) -> Option<WorkflowSessionConsoleDetail> {
        let allowed = {
            let inner = self.inner.lock().expect("session store mutex poisoned");
            inner
                .sessions
                .get(session_id)
                .is_some_and(|session| session.project() == Some(project))
        };
        if !allowed {
            return None;
        }
        let limit = normalize_console_activity_limit(limit);
        self.with_record_for_query(session_id, |record, _| {
            (record.project.as_deref() == Some(project))
                .then(|| build_console_detail(record, project, limit, validation))
        })
        .flatten()
    }

    pub(super) fn with_record_for_query<T>(
        &self,
        session_id: &str,
        query: impl FnOnce(&SessionRecord, Option<&ColdSessionRecord>) -> T,
    ) -> Option<T> {
        let mut inner = self.inner.lock().expect("session store mutex poisoned");
        inner.touch(session_id);
        match inner.sessions.get(session_id)? {
            StoredSession::Hot(record) => Some(query(record, None)),
            StoredSession::Cold(cold) => {
                let record = materialize_cold_session(cold, inner.recent_event_limit)?;
                Some(query(&record, Some(cold)))
            }
        }
    }

    fn coldify_closed_session(&self, session_id: &str) {
        {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            if inner.persistence.is_some() {
                if !inner.sessions.dirty.contains_key(session_id) {
                    let cold = inner
                        .sessions
                        .get(session_id)
                        .and_then(StoredSession::hot)
                        .and_then(|record| record.durable.clone());
                    if let Some(cold) = cold {
                        inner
                            .sessions
                            .replace_residency(session_id.to_owned(), StoredSession::Cold(cold));
                    }
                }
                return;
            }
        }
        let (snapshot, project_instructions) = {
            let inner = self.inner.lock().expect("session store mutex poisoned");
            let Some(StoredSession::Hot(record)) = inner.sessions.get(session_id) else {
                return;
            };
            if record.lifecycle.allows_mutation() {
                return;
            }
            (
                PersistedSessionRecord::from_record(record, inner.recent_event_limit),
                record
                    .project_instructions
                    .as_ref()
                    .map(|instructions| instructions.to_summary()),
            )
        };
        let cold = match cold_session_from_persisted(&snapshot, project_instructions) {
            Ok(cold) => cold,
            Err(err) => {
                tracing::warn!("session cold serialization failed: {err}");
                return;
            }
        };
        let mut inner = self.inner.lock().expect("session store mutex poisoned");
        let Some(stored) = inner.sessions.get_mut(session_id) else {
            return;
        };
        let Some(record) = stored.hot() else {
            return;
        };
        if !snapshot.still_matches_record(record) {
            return;
        }
        debug_assert!(!cold.lifecycle.allows_mutation());
        *stored = StoredSession::Cold(cold);
    }

    pub fn contains_session(&self, session_id: &str) -> bool {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.contains_session(session_id)
    }

    /// Internal consistency fence for assembling recovery evidence. Event mutations
    /// advance events_observed; collaboration mutations advance
    /// message_observation_revision. Never serialized or accepted from a caller.
    pub fn handoff_revision(&self, session_id: &str) -> Option<(u64, u64)> {
        self.retained_revisions(session_id)
    }

    /// Cold JSON is canonical and immutable after admission, so scalar
    /// revisions need not materialize its entire (up to 2,000-event) body.
    /// Keep the same LRU read touch as with_record_for_query. If an internal
    /// repair/test replaced raw bytes, fall through to the original checked
    /// materialization path rather than returning stale revision metadata.
    pub(super) fn retained_revisions(&self, session_id: &str) -> Option<(u64, u64)> {
        {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            inner.touch(session_id);
            match inner.sessions.get(session_id)? {
                StoredSession::Hot(record) => {
                    return Some((record.events_observed, record.message_observation_revision));
                }
                StoredSession::Cold(cold)
                    if std::ptr::eq(
                        std::sync::Weak::as_ptr(&cold.revision_raw),
                        Arc::as_ptr(&cold.raw),
                    ) =>
                {
                    return Some((
                        cold.retained_events_observed,
                        cold.retained_message_revision,
                    ));
                }
                StoredSession::Cold(_) => {}
            }
        }
        self.with_record_for_query(session_id, |record, _| {
            (record.events_observed, record.message_observation_revision)
        })
    }

    /// Immutable incarnation metadata never requires loading event/message history.
    pub fn session_created_at(&self, session_id: &str) -> Option<i64> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        Some(match inner.sessions.get(session_id)? {
            StoredSession::Hot(record) => record.created_at,
            StoredSession::Cold(record) => record.created_at,
        })
    }

    /// Async admission backpressure for runtime callers. This bounds deferred
    /// changes independently of total persisted Session/history size. Already
    /// running effects may still append their terminal evidence.
    pub async fn prepare_request(&self) -> Result<(), String> {
        let needs_flush = {
            let inner = self.inner.lock().expect("session store mutex poisoned");
            let Some(persistence) = &inner.persistence else {
                return Ok(());
            };
            let pending = inner
                .sessions
                .dirty
                .keys()
                .filter_map(|id| inner.sessions.get(id).and_then(StoredSession::hot))
                .fold(inner.sessions.dirty.len() as u64, |total, record| {
                    total
                        .saturating_add(record.pending_events.len() as u64)
                        .saturating_add(
                            record.message_observation_revision.saturating_sub(
                                record
                                    .durable
                                    .as_ref()
                                    .map_or(0, |cold| cold.retained_message_revision),
                            ),
                        )
                });
            persistence.last_persist_error.is_some() || pending >= 512
        };
        if needs_flush {
            self.flush().await
        } else {
            Ok(())
        }
    }

    /// Successful return confirms the observed generation committed in SQLite.
    pub async fn flush(&self) -> Result<(), String> {
        self.persist_after_mutation_durable()
            .await
            .map_err(|_| "Workflow Session persistence unavailable".to_owned())
    }

    pub fn session_project(&self, session_id: &str) -> Option<Option<String>> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.session_project(session_id)
    }

    pub fn session_project_identity(&self, session_id: &str) -> Option<Option<String>> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner
            .sessions
            .get(session_id)
            .map(|s| s.project_identity().map(str::to_string))
    }

    pub fn session_target_authority(&self, session_id: &str) -> Option<(Option<String>, String)> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.session_target_authority(session_id)
    }

    /// Synchronous revalidation boundary for an already authorized current-work
    /// target. Identity and creation authority remain explicit; this closure
    /// cannot close/reassign a Session or confer authority through a Window.
    /// The callback must not reenter SessionStore or perform asynchronous work.
    pub fn with_active_session_authority_fence<T>(
        &self,
        session_id: &str,
        expected_project: &str,
        expected_owner_authority_fingerprint: &str,
        commit: impl FnOnce() -> T,
    ) -> Option<T> {
        let inner = self.inner.lock().ok()?;
        let record = inner.sessions.get(session_id)?;
        if !record.lifecycle().allows_mutation()
            || record.project() != Some(expected_project)
            || record.owner_authority_fingerprint() != expected_owner_authority_fingerprint
        {
            return None;
        }
        Some(commit())
    }

    /// Return inherited defaults only for an active Session whose registered
    /// project exactly matches the already-resolved request project.
    pub fn execution_context_for_project(
        &self,
        session_id: &str,
        resolved_project: &str,
    ) -> Option<SessionExecutionContext> {
        self.with_record_for_query(session_id, |record, _| {
            (record.lifecycle.allows_mutation()
                && record.project.as_deref() == Some(resolved_project))
            .then(|| record.execution_context.clone())
        })
        .flatten()
    }

    pub fn guard_state(&self, session_id: &str) -> Option<(SessionMode, SessionGuards)> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.guard_state(session_id)
    }

    pub fn lifecycle_state(&self, session_id: &str) -> Option<SessionLifecycle> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.lifecycle_state(session_id)
    }

    /// Incarnation fingerprint of a retention tombstone owned by this principal.
    /// A different principal and a missing tombstone are both `None`.
    pub fn owned_retention_incarnation(
        &self,
        session_id: &str,
        owner_authority_fingerprint: &str,
    ) -> Option<String> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.owned_retention_incarnation(session_id, owner_authority_fingerprint)
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn retention_tombstone_for_test(
        &self,
        session_id: &str,
    ) -> Option<SessionRetentionTombstone> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        let ordinal = inner.tombstone_ids.get(session_id).copied()?;
        inner.tombstones.get(&ordinal).cloned()
    }

    /// Try one fixed suffix sixteen times. A retained tombstone occupies that id.
    #[cfg(any(test, feature = "root-test-support"))]
    pub fn allocate_fixed_session_suffix_for_test(&self, suffix: &str) -> Option<String> {
        let inner = self.inner.lock().expect("session store mutex poisoned");
        inner.allocate_session_id(|| suffix.to_string())
    }

    /// Explicit close: `Active → Closed`. Idempotent for already-closed sessions.
    ///
    /// Never creates a session for an unknown id. Emits a single
    /// `session_closed` ledger event only on a real `Active → Closed` transition.
    pub fn close_session(
        &self,
        session_id: &str,
    ) -> Result<SessionCloseOutcome, SessionCloseError> {
        let session_id = session_id.trim();
        if session_id.is_empty() || !is_valid_session_id(session_id) {
            return Err(SessionCloseError::UnknownSession);
        }
        let committed = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            inner.close_session(session_id)?
        };
        let outcome = committed;
        self.coldify_closed_session(session_id);
        self.persist_after_mutation();
        Ok(outcome)
    }

    /// Replace the complete execution context and append one safe metadata
    /// event together under the in-memory store lock. Persistent stores then
    /// queue the changed state to the SQLite writer. `{}` clears all execution defaults.
    pub fn update_execution_context(
        &self,
        session_id: &str,
        execution_context: SessionExecutionContext,
        transport: SessionTransport,
    ) -> Result<SessionExecutionContextUpdateOutcome, SessionExecutionContextUpdateError> {
        let session_id = session_id.trim();
        if session_id.is_empty() || !is_valid_session_id(session_id) {
            return Err(SessionExecutionContextUpdateError::UnknownSession);
        }
        let outcome = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            if !inner.prepare_record_mutation(session_id) {
                return Err(SessionExecutionContextUpdateError::UnknownSession);
            }
            let stored = inner
                .sessions
                .get(session_id)
                .ok_or(SessionExecutionContextUpdateError::UnknownSession)?;
            let lifecycle = stored.lifecycle();
            if !lifecycle.allows_mutation() {
                return Err(SessionExecutionContextUpdateError::SessionNotActive { lifecycle });
            }
            let record = stored
                .hot()
                .expect("active execution-context session must stay hot");
            let project = record
                .project
                .clone()
                .ok_or(SessionExecutionContextUpdateError::SessionHasNoProject)?;
            let previous_execution_context = record.execution_context.clone();
            let execution_context = execution_context
                .validated()
                .map_err(SessionExecutionContextUpdateError::InvalidExecutionContext)?;
            let changed = execution_context != previous_execution_context;
            let now = now_ts();
            let event = session_execution_context_updated_event(
                session_id,
                &project,
                transport,
                execution_context.clone(),
                previous_execution_context.clone(),
                changed,
                now,
            );
            let max_events = inner.recent_event_limit;
            let record = inner
                .sessions
                .get_mut(session_id)
                .and_then(StoredSession::hot_mut)
                .expect("validated active Session must stay hot under store lock");
            record.execution_context = execution_context;
            record.updated_at = now;
            append_event(record, event, max_events);
            inner.touch(session_id);
            let summary = inner
                .summary(session_id, Some(DEFAULT_SUMMARY_LIMIT))
                .expect("updated Session must summarize");
            SessionExecutionContextUpdateOutcome {
                summary,
                previous_execution_context,
                changed,
            }
        };
        self.persist_after_mutation();
        Ok(outcome)
    }

    /// Authoritative lifecycle check used by dispatch/kernel before mutation.
    ///
    /// Closed sessions deny write-like, shell-like, and a small set of
    /// session-local mutations (messages, checkpoint create/restore/delete).
    /// Query tools and pure reads remain allowed. `close_session` itself is
    /// never denied so repeated close stays idempotent.
    pub fn lifecycle_denial(
        &self,
        session_id: &str,
        tool_name: &str,
        contract: SessionToolContract,
    ) -> Option<SessionLifecycleDenial> {
        let lifecycle = self.lifecycle_state(session_id)?;
        if lifecycle.allows_mutation() {
            return None;
        }
        let lifecycle_effect = runtime_tool_session_evidence_policy(tool_name).lifecycle;
        if lifecycle_effect == ToolSessionLifecycleEffect::IdempotentClose {
            return None;
        }
        if lifecycle_blocks_tool(contract, lifecycle_effect) {
            return Some(SessionLifecycleDenial { lifecycle });
        }
        None
    }

    /// Authoritative guard check used by dispatch/kernel before mutation.
    pub fn guard_denial(
        &self,
        session_id: &str,
        contract: SessionToolContract,
    ) -> Option<SessionGuardDenial> {
        let (mode, guards) = self.guard_state(session_id)?;
        let classification = contract;
        if guards.deny_write_tools && classification.write_like {
            return Some(SessionGuardDenial {
                mode,
                guard: "deny_write_tools",
            });
        }
        if guards.deny_shell_tools && classification.shell_like {
            return Some(SessionGuardDenial {
                mode,
                guard: "deny_shell_tools",
            });
        }
        None
    }

    #[cfg(any(test, feature = "root-test-support"))]
    pub fn record_tool_call_started(
        &self,
        session_id: Option<&str>,
        transport: SessionTransport,
        tool_name: &str,
        arguments: &Value,
        contract: SessionToolContract,
    ) -> Option<ToolCallStart> {
        self.record_tool_call_started_with_options(
            session_id, transport, tool_name, arguments, None, contract,
        )
    }

    pub fn record_tool_call_started_with_options(
        &self,
        session_id: Option<&str>,
        transport: SessionTransport,
        tool_name: &str,
        arguments: &Value,
        resolved_project: Option<String>,
        contract: SessionToolContract,
    ) -> Option<ToolCallStart> {
        self.record_tool_call_started_with_metadata(
            session_id,
            transport,
            tool_name,
            arguments,
            resolved_project,
            ToolCallRecorderMetadata::from_business_arguments(arguments),
            contract,
        )
    }

    /// Sole entry for appending a `tool_call_started` ledger event.
    pub fn record_tool_call_started_with_metadata(
        &self,
        session_id: Option<&str>,
        transport: SessionTransport,
        tool_name: &str,
        arguments: &Value,
        resolved_project: Option<String>,
        metadata: ToolCallRecorderMetadata,
        contract: SessionToolContract,
    ) -> Option<ToolCallStart> {
        let session_id = session_id?.trim();
        if !is_valid_session_id(session_id) {
            return None;
        }
        // Preserve the existing fail-closed Session boundary. Evicted or unknown
        // Sessions cannot be revived by appending a tool event.
        if !self.contains_session(session_id) {
            return None;
        }
        let now = now_ts();
        let event_id = format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple());
        let project = extract_project(arguments);
        let call_id = format!("{CALL_ID_PREFIX}{}", uuid::Uuid::new_v4().simple());
        let classification = contract;
        let risk_class = classification.risk_class.to_string();
        let changed_paths = changed_paths_for_tool_call(contract, arguments);
        let observed_paths = observed_input_paths_for_tool(tool_name, contract, arguments);
        let diff_review_like = diff_review_like_for_tool(tool_name, arguments);
        let input_summary = Some(session_input_summary_for_tool(tool_name, arguments));
        let expectation = metadata.expectation;
        let start = ToolCallStart {
            event_id,
            call_id,
            logical_invocation_id: metadata.logical_invocation_id,
            logical_invocation_role: metadata.logical_invocation_role,
            session_id: session_id.to_string(),
            transport,
            tool_name: tool_name.to_string(),
            project,
            resolved_project,
            risk_class,
            read_like: classification.read_like,
            write_like: classification.write_like,
            shell_like: classification.shell_like,
            git_like: classification.git_like,
            change_summary_like: classification.change_summary_like,
            diff_review_like,
            changed_paths,
            observed_paths,
            started_at: now,
            started_instant: Instant::now(),
            permission: None,
            expectation,
        };
        self.push_event(SessionEvent {
            input_summary,
            ..tool_call_event(start.clone(), "tool_call_started", now)
        });
        Some(start)
    }

    pub fn record_permission_decision(
        &self,
        start: &mut ToolCallStart,
        permission: PermissionDecision,
    ) {
        start.permission = Some(permission.clone());
        let persisted = self.set_event_permission(&start.session_id, &start.event_id, permission);
        if persisted {
            self.persist_after_mutation();
        }
    }

    /// Attach the bounded outcome of automatic persistent-shell cleanup to the
    /// single `session_closed` system event. Phase one permits at most one
    /// active shell per Session, so this remains a scalar evidence record.
    pub fn record_session_close_persistent_shell_evidence(
        &self,
        session_id: &str,
        shell_id: &str,
        shell_state: &str,
        execution_state: &str,
        error_code: Option<&str>,
        already_closed: bool,
    ) {
        let evidence = persistent_shell_event_evidence_for_tool_result(
            "close_session",
            &serde_json::json!({
                "shell_id": shell_id,
                "shell_state": shell_state,
                "execution_state": execution_state,
                "error_code": error_code,
                "command_started": false,
                "command_completed": false,
                "already_closed": already_closed,
            }),
        );
        let Some(evidence) = evidence else {
            return;
        };
        let persisted = self.set_session_close_persistent_shell_evidence(session_id, evidence);
        if persisted {
            self.persist_after_mutation();
        }
    }

    fn tool_effect_event_evidence_for_result(output: &Value) -> Option<ToolEffectEventEvidence> {
        let state_changed = output.get("state_changed").and_then(Value::as_bool);
        let command_started = output.get("command_started").and_then(Value::as_bool);
        let command_completed = output.get("command_completed").and_then(Value::as_bool);
        let execution_state = output
            .get("execution_state")
            .and_then(Value::as_str)
            .and_then(sanitize_tool_execution_state);
        if state_changed.is_none()
            && command_started.is_none()
            && command_completed.is_none()
            && execution_state.is_none()
        {
            return None;
        }
        Some(ToolEffectEventEvidence {
            state_changed,
            command_started,
            command_completed,
            execution_state,
        })
    }

    /// Append a finished tool-call ledger event through the canonical event path.
    pub fn record_tool_call_finished(
        &self,
        start: Option<ToolCallStart>,
        success: bool,
        output: &Value,
        error: Option<&str>,
        error_kind: Option<&str>,
    ) -> Option<String> {
        let event = Self::tool_call_finished_event(start, success, output, error, error_kind)?;
        let event_id = event.event_id.clone();
        self.push_event(event);
        Some(event_id)
    }

    fn tool_call_finished_event(
        start: Option<ToolCallStart>,
        success: bool,
        output: &Value,
        error: Option<&str>,
        error_kind: Option<&str>,
    ) -> Option<SessionEvent> {
        let start = start?;
        let finished_at = now_ts();
        let duration_ms = start
            .started_instant
            .elapsed()
            .as_millis()
            .min(u64::MAX as u128) as u64;
        let event_id = format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple());
        let failure_kind = output
            .get("failure_kind")
            .and_then(Value::as_str)
            .map(str::to_string);
        // A caller-supplied kind or output failure_kind stays authoritative.
        // Only a retention-expired identity replaces the generic runtime_error
        // default, so evidence on another live Session keeps that exact kind.
        let classified_retention = error
            .map(classify_error_message)
            .filter(|kind| kind == "session_retention_expired");
        let error_kind = error_kind
            .or_else(|| error.and_then(|_| output.get("failure_kind").and_then(Value::as_str)))
            .or(classified_retention.as_deref());
        let actual_failure_kind = actual_failure_kind_for_tool_result(output, error, error_kind);
        let failure_expectation_result = classify_failure_expectation(
            success,
            &start.expectation,
            actual_failure_kind.as_deref(),
            output,
        );
        let warning_kind = output
            .get("warning_kind")
            .and_then(Value::as_str)
            .map(str::to_string);
        let session_project = output
            .get("session_project")
            .and_then(Value::as_str)
            .map(str::to_string);
        let request_project = output
            .get("request_project")
            .and_then(Value::as_str)
            .map(str::to_string);
        let error_message_summary =
            error.map(|message| bound_event_error_summary(message, start.shell_like));
        let validation_output_summary =
            validation_output_summary_for_tool_result(&start.tool_name, output);
        let observed_paths = if success {
            observed_paths_for_successful_result(
                &start.tool_name,
                start.observed_paths.clone(),
                output,
            )
        } else {
            Vec::new()
        };
        let persistent_shell =
            persistent_shell_event_evidence_for_tool_result(&start.tool_name, output);
        let effect_evidence = Self::tool_effect_event_evidence_for_result(output);
        let mut changed_paths = start.changed_paths.clone();
        for path in changed_paths_for_tool_result(&start.tool_name, output) {
            if !changed_paths.iter().any(|existing| existing == &path) {
                changed_paths.push(path);
            }
        }
        let event = SessionEvent {
            event_id,
            context_result_summary: context_result_summary_for_tool_result(
                &start.tool_name,
                output,
            ),
            finished_at: Some(finished_at),
            duration_ms: Some(duration_ms),
            status: Some(if success { "succeeded" } else { "failed" }.to_string()),
            exit_code: output.get("exit_code").and_then(Value::as_i64),
            failure_kind,
            error_kind: error.map(|_| error_kind.unwrap_or("runtime_error").to_string()),
            actual_failure_kind,
            failure_expectation_result: Some(failure_expectation_result.to_string()),
            warning_kind,
            session_project,
            request_project,
            error_message_summary,
            changed_paths,
            observed_paths,
            job_id: extract_job_id(output),
            persistent_shell,
            effect_evidence,
            validation_output_summary,
            ..tool_call_event(start, "tool_call_finished", finished_at)
        };
        Some(event)
    }

    #[allow(clippy::too_many_arguments)]
    pub fn record_validation_job_terminal(
        &self,
        session_id: &str,
        job_id: &str,
        retained_terminal_job_ids: &[&str],
        tool_name: &str,
        contract: SessionToolContract,
        project: Option<String>,
        validation_target_id: &str,
        assertion_name: Option<&str>,
        job_status: &str,
        exit_code: Option<i64>,
        validation_passed: Option<bool>,
        started_at: Option<i64>,
        finished_at: Option<i64>,
        duration_ms: Option<u64>,
        validation_output_summary: Option<Value>,
    ) -> bool {
        let session_id = session_id.trim();
        let job_id = job_id.trim();
        let valid_target = is_validation_execution_identity(validation_target_id);
        let assertion_name = assertion_name
            .and_then(|value| super::events::safe_model_facing_assertion_name(tool_name, value))
            .filter(|value| assertion_validation_identity(value) == validation_target_id);
        let Some(timestamp) = finished_at else {
            // Reconciliation must never substitute wall-clock read time for
            // authoritative execution activity.
            return false;
        };
        let Ok(job_lifecycle) = RunnerJobLifecycle::from_wire(job_status) else {
            return false;
        };
        if !is_valid_session_id(session_id)
            || !is_safe_job_id(job_id)
            || retained_terminal_job_ids.len() > MAX_MATERIALIZED_VALIDATION_JOB_IDS
            || retained_terminal_job_ids
                .iter()
                .any(|candidate| *candidate != candidate.trim() || !is_safe_job_id(candidate))
            || !retained_terminal_job_ids.contains(&job_id)
            || !matches!(
                tool_name,
                "project_validate"
                    | "cargo_fmt"
                    | "cargo_check"
                    | "cargo_test"
                    | "go_test"
                    | "run_process"
                    | "run_script"
                    | "run_shell"
                    | "run_job"
            )
            || !valid_target
            || !job_lifecycle.is_terminal()
        {
            return false;
        }
        let inherited_expectation = self
            .summary(session_id, Some(MAX_SUMMARY_LIMIT))
            .and_then(|summary| {
                summary
                    .events
                    .iter()
                    .rev()
                    .find(|event| {
                        event.kind == "tool_call_finished"
                            && event.job_id.as_deref() == Some(job_id)
                            && event.tool_name == tool_name
                            && event
                                .resolved_project
                                .as_deref()
                                .or(event.project.as_deref())
                                == project.as_deref()
                    })
                    .map(|event| ToolCallExpectation {
                        expected_failure: event.expected_failure == Some(true),
                        expected_failure_kind: event.expected_failure_kind.clone(),
                        result_expectation: event.result_expectation.clone(),
                        accepted_exit_codes: event.accepted_exit_codes.clone(),
                        assertion_name: event.assertion_name.clone(),
                    })
            })
            .unwrap_or_default();
        let process_succeeded =
            job_lifecycle == RunnerJobLifecycle::Completed && exit_code == Some(0);
        let succeeded = process_succeeded && validation_passed.unwrap_or(true);
        let failure_kind = (!succeeded).then(|| match job_lifecycle {
            RunnerJobLifecycle::Timeout | RunnerJobLifecycle::TimedOut => "timeout".to_string(),
            RunnerJobLifecycle::Stopped | RunnerJobLifecycle::Cancelled => "cancelled".to_string(),
            RunnerJobLifecycle::Lost => "execution_lost".to_string(),
            _ if process_succeeded => "validation_failed".to_string(),
            _ => "command_exit_nonzero".to_string(),
        });
        let terminal_execution_state = match job_lifecycle {
            RunnerJobLifecycle::Completed | RunnerJobLifecycle::Failed => "completed",
            RunnerJobLifecycle::Timeout | RunnerJobLifecycle::TimedOut => "timed_out",
            RunnerJobLifecycle::Stopped | RunnerJobLifecycle::Cancelled => "cancelled",
            RunnerJobLifecycle::Lost => "outcome_unknown",
            _ => "outcome_unknown",
        };
        let expectation_output = serde_json::json!({
            "job_id": job_id,
            "exit_code": exit_code,
            "execution_state": terminal_execution_state,
            "command_completed": matches!(
                job_lifecycle,
                RunnerJobLifecycle::Completed | RunnerJobLifecycle::Failed
            ) && exit_code.is_some(),
        });
        let failure_expectation_result = classify_failure_expectation(
            succeeded,
            &inherited_expectation,
            failure_kind.as_deref(),
            &expectation_output,
        );
        let classification = contract;
        let mut input_summary = serde_json::json!({
            "execution_identity": validation_target_id,
        });
        if validation_target_id.starts_with("target:") {
            input_summary["validation_target_id"] = serde_json::json!(validation_target_id);
        }
        if let Some(validation_tool) = validation_output_summary
            .as_ref()
            .and_then(|summary| summary.get("validation_tool"))
            .and_then(Value::as_str)
            .filter(|tool| matches!(*tool, "cargo_fmt" | "cargo_check" | "cargo_test"))
        {
            input_summary["validation_tool"] = serde_json::json!(validation_tool);
        }
        let inherited_expected_failure = inherited_expectation.expected_failure;
        let inherited_expected_failure_kind = inherited_expectation.expected_failure_kind.clone();
        let inherited_result_expectation = inherited_expectation.result_expectation.clone();
        let inherited_accepted_exit_codes = inherited_expectation.accepted_exit_codes.clone();
        let inherited_assertion_name = inherited_expectation.assertion_name.clone();
        let event = SessionEvent {
            project: project.clone(),
            resolved_project: project.clone(),
            read_like: classification.read_like,
            write_like: classification.write_like,
            shell_like: classification.shell_like,
            git_like: classification.git_like,
            change_summary_like: classification.change_summary_like,
            started_at,
            finished_at: Some(timestamp),
            duration_ms,
            status: Some(if succeeded { "succeeded" } else { "failed" }.to_string()),
            exit_code,
            failure_kind: failure_kind.clone(),
            expected_failure: inherited_expected_failure.then_some(true),
            expected_failure_kind: inherited_expected_failure_kind,
            result_expectation: inherited_result_expectation,
            accepted_exit_codes: inherited_accepted_exit_codes,
            assertion_name: assertion_name.or(inherited_assertion_name),
            actual_failure_kind: failure_kind,
            failure_expectation_result: Some(failure_expectation_result.to_string()),
            job_id: Some(job_id.to_string()),
            input_summary: Some(input_summary),
            validation_output_summary,
            ..session_event(
                format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple()),
                session_id.to_string(),
                "validation_job_terminal".to_string(),
                timestamp,
                "job_terminal".to_string(),
                tool_name.to_string(),
                classification.risk_class.to_string(),
            )
        };

        let (hot_closed, recorded) = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            if !inner.prepare_record_mutation(session_id) {
                return false;
            }
            let max_events = inner.recent_event_limit;
            let Some(record) = inner
                .sessions
                .get_mut(session_id)
                .and_then(StoredSession::hot_mut)
            else {
                return false;
            };
            let recorded = Self::append_validation_job_terminal_to_record(
                record,
                job_id,
                retained_terminal_job_ids,
                project.as_deref(),
                &event,
                timestamp,
                max_events,
            );
            (!record.lifecycle.allows_mutation(), recorded)
        };
        if !recorded {
            return false;
        }
        if hot_closed {
            self.coldify_closed_session(session_id);
        }
        self.persist_after_mutation();
        true
    }

    fn append_validation_job_terminal_to_record(
        record: &mut SessionRecord,
        job_id: &str,
        retained_terminal_job_ids: &[&str],
        project: Option<&str>,
        event: &SessionEvent,
        timestamp: i64,
        max_events: usize,
    ) -> bool {
        if record.project.as_deref() != project
            || record
                .materialized_validation_job_ids
                .iter()
                .any(|materialized| materialized == job_id)
        {
            return false;
        }

        // The Runner can retain at most 64 authoritative terminal Jobs. Keep an
        // exact marker for every Job still present in this reconciliation snapshot;
        // if stale markers fill the bound, discard one that the authoritative
        // snapshot can no longer name before inserting the new identity.
        while record.materialized_validation_job_ids.len() >= MAX_MATERIALIZED_VALIDATION_JOB_IDS {
            let Some(stale_index) =
                record
                    .materialized_validation_job_ids
                    .iter()
                    .position(|materialized| {
                        !retained_terminal_job_ids.contains(&materialized.as_str())
                    })
            else {
                // A complete valid terminal snapshot cannot name more than the
                // bound. Fail closed rather than evicting a still-retained Job.
                return false;
            };
            record.materialized_validation_job_ids.remove(stale_index);
        }
        record
            .materialized_validation_job_ids
            .push_back(job_id.to_string());
        record.updated_at = record.updated_at.max(timestamp);
        append_event(record, event.clone(), max_events);
        true
    }

    /// Append bounded recorder-only CodingAgentRun lifecycle evidence. The
    /// explicit Workflow Session is provenance only: this path grants no Run
    /// authority and intentionally stores no prompt, ACP session id, event body,
    /// reasoning, tool payload, credential, or idempotency key.
    pub fn record_coding_agent_lifecycle_evidence(
        &self,
        session_id: &str,
        project: &str,
        run_id: &str,
        provider_id: &str,
        kind: &str,
        state: &str,
        execution_state: &str,
        terminal_stop_reason: Option<&str>,
        terminal_error_code: Option<&str>,
    ) -> bool {
        let project_matches = self
            .with_record_for_query(session_id, |record, _| {
                record.project.as_deref() == Some(project)
            })
            .unwrap_or(false);
        if !project_matches {
            return false;
        }
        let now = now_ts();
        self.push_event(coding_agent_lifecycle_event(
            session_id,
            project,
            run_id,
            provider_id,
            kind,
            state,
            execution_state,
            terminal_stop_reason,
            terminal_error_code,
            now,
        ));
        true
    }

    /// Sole entry for appending a session ledger event.
    fn push_event(&self, event: SessionEvent) {
        let session_id = event.session_id.clone();
        let closed = {
            let mut inner = self.inner.lock().expect("session store mutex poisoned");
            if !inner.prepare_record_mutation(&session_id) {
                return;
            }
            let recent_event_limit = inner.recent_event_limit;
            let Some(record) = inner
                .sessions
                .get_mut(&session_id)
                .and_then(StoredSession::hot_mut)
            else {
                return;
            };
            record.updated_at = now_ts();
            if event_observes_repository_edit(&event) {
                record.repository_edit_observed = true;
            }
            append_event(record, event, recent_event_limit);
            let closed = !record.lifecycle.allows_mutation();
            inner.touch(&session_id);
            closed
        };
        if closed {
            self.coldify_closed_session(&session_id);
        }
        self.persist_after_mutation();
    }

    fn set_event_permission(
        &self,
        session_id: &str,
        event_id: &str,
        permission: PermissionDecision,
    ) -> bool {
        self.update_retained_record(session_id, permission, |record, permission| {
            let Some(event) = record
                .events
                .iter_mut()
                .rev()
                .find(|event| event.event_id == event_id)
            else {
                return false;
            };
            Arc::make_mut(event).permission = Some(permission);
            record.pending_events.push_back(Arc::clone(event));
            true
        })
    }

    fn set_session_close_persistent_shell_evidence(
        &self,
        session_id: &str,
        evidence: PersistentShellEventEvidence,
    ) -> bool {
        self.update_retained_record(session_id, evidence, |record, evidence| {
            let Some(event) =
                record.events.iter_mut().rev().find(|event| {
                    event.kind == "session_closed" && event.tool_name == "close_session"
                })
            else {
                return false;
            };
            Arc::make_mut(event).persistent_shell = Some(evidence);
            record.pending_events.push_back(Arc::clone(event));
            record.updated_at = now_ts();
            true
        })
    }

    pub(super) fn persist_after_mutation(&self) {
        if let Some(writer) = &self.writer {
            // Fire-and-forget: the dedicated writer thread serializes and
            // writes. Narrow restart-safe operations use
            // `persist_after_mutation_durable` instead.
            let generation = self
                .inner
                .lock()
                .expect("Session store mutex poisoned")
                .sessions
                .generation;
            writer.notify_generation(generation);
            return;
        }
        let _ = writer::persist_batch(&self.inner, &self.persistence_write_mutex).1;
    }

    /// Reserve the fallback I/O lane before changing state. Cancellation while
    /// waiting for this lane therefore cannot leave an unqueued mutation behind.
    /// The normal background writer needs no permit and no blocking task.
    pub(super) async fn prepare_durable_mutation(&self) -> SessionDurability<'_> {
        let fallback = self.writer.is_none()
            && self
                .inner
                .lock()
                .expect("session store mutex poisoned")
                .persistence
                .is_some();
        let fallback_permit = if fallback {
            Some(
                Arc::clone(&self.persistence_fallback)
                    .acquire_owned()
                    .await
                    .expect("session persistence semaphore closed"),
            )
        } else {
            None
        };
        SessionDurability {
            store: self,
            fallback_permit,
        }
    }

    /// Fence an observation without mutating state. Mutation callers reserve
    /// their durability lane before the mutation and persist that reservation.
    pub(super) async fn persist_after_mutation_durable(&self) -> Result<(), ()> {
        self.prepare_durable_mutation().await.persist().await
    }
}

/// Authoritative in-memory transitions for workflow session state.
///
/// Map fields stay private; sibling modules must use these helpers so create,
/// message, and event mutations cannot bypass the store.
/// Tools blocked on Closed workflow sessions.
///
/// Query and pure-read tools remain allowed. Message-board mutations and
/// session-scoped checkpoint mutations are blocked even when metadata marks
/// them read-only (they still change durable session/project evidence).
fn lifecycle_blocks_tool(
    classification: SessionToolContract,
    lifecycle_effect: ToolSessionLifecycleEffect,
) -> bool {
    classification.write_like
        || classification.shell_like
        || lifecycle_effect == ToolSessionLifecycleEffect::Mutation
}

/// Construct a new event without inventing identity, time, or outcome facts.
/// This is deliberately private: the public/persisted model keeps its existing
/// required fields and serde compatibility rules.
fn session_event(
    event_id: String,
    session_id: String,
    kind: String,
    timestamp: i64,
    transport: String,
    tool_name: String,
    risk_class: String,
) -> SessionEvent {
    SessionEvent {
        event_id,
        call_id: None,
        logical_invocation_id: None,
        logical_invocation_role: None,
        session_id,
        kind,
        legacy_context_revision: None,
        context_result_summary: None,
        timestamp,
        transport,
        tool_name,
        project: None,
        resolved_project: None,
        risk_class,
        read_like: false,
        write_like: false,
        shell_like: false,
        git_like: false,
        change_summary_like: false,
        diff_review_like: false,
        started_at: None,
        finished_at: None,
        duration_ms: None,
        status: None,
        exit_code: None,
        failure_kind: None,
        error_kind: None,
        expected_failure: None,
        expected_failure_kind: None,
        result_expectation: None,
        accepted_exit_codes: Vec::new(),
        assertion_name: None,
        actual_failure_kind: None,
        failure_expectation_result: None,
        warning_kind: None,
        session_project: None,
        request_project: None,
        error_message_summary: None,
        changed_paths: Vec::new(),
        observed_paths: Vec::new(),
        job_id: None,
        persistent_shell: None,
        effect_evidence: None,
        input_summary: None,
        validation_output_summary: None,
        permission: None,
        instruction: None,
        requested_mode: None,
        previous_mode: None,
        requested_guards: None,
        previous_guards: None,
        capability_changed: None,
        context_refreshed: None,
        execution_context: None,
        previous_execution_context: None,
        execution_context_changed: None,
    }
}

/// Both halves of a tool call inherit the same accepted metadata. Completion
/// supplies its new event id and result evidence at the call site.
fn tool_call_event(start: ToolCallStart, kind: &str, timestamp: i64) -> SessionEvent {
    SessionEvent {
        call_id: Some(start.call_id),
        logical_invocation_id: start.logical_invocation_id,
        logical_invocation_role: start.logical_invocation_role,
        project: start.project,
        resolved_project: start.resolved_project,
        read_like: start.read_like,
        write_like: start.write_like,
        shell_like: start.shell_like,
        git_like: start.git_like,
        change_summary_like: start.change_summary_like,
        diff_review_like: start.diff_review_like,
        started_at: Some(start.started_at),
        expected_failure: start.expectation.expected_failure.then_some(true),
        expected_failure_kind: start.expectation.expected_failure_kind,
        result_expectation: start.expectation.result_expectation,
        accepted_exit_codes: start.expectation.accepted_exit_codes,
        assertion_name: start.expectation.assertion_name,
        changed_paths: start.changed_paths,
        observed_paths: start.observed_paths,
        permission: start.permission,
        ..session_event(
            start.event_id,
            start.session_id,
            kind.to_string(),
            timestamp,
            start.transport.as_str().to_string(),
            start.tool_name,
            start.risk_class,
        )
    }
}

/// Caller retains the store lock and owns activity/persistence ordering.
fn append_event(record: &mut SessionRecord, event: SessionEvent, max_events: usize) {
    let event = Arc::new(event);
    if event.kind == "task_instruction" {
        record.current_instruction = Some(Arc::clone(&event));
    }
    record.pending_events.push_back(Arc::clone(&event));
    record.events.push_back(event);
    record.events_observed = record.events_observed.saturating_add(1);
    while record.events.len() > max_events {
        record.events.pop_front();
    }
}

#[allow(clippy::too_many_arguments)]
fn coding_instruction_event(
    event_id: &str,
    session_id: &str,
    project: &str,
    instruction: Option<String>,
    transport: SessionTransport,
    requested_mode: SessionMode,
    previous_mode: Option<SessionMode>,
    requested_guards: SessionGuards,
    previous_guards: Option<SessionGuards>,
    capability_changed: bool,
    context_refreshed: bool,
    execution_context: Option<SessionExecutionContext>,
    previous_execution_context: Option<SessionExecutionContext>,
    execution_context_changed: bool,
    reused: bool,
    explicit_resume: bool,
    now: i64,
) -> SessionEvent {
    SessionEvent {
        project: Some(project.to_string()),
        resolved_project: Some(project.to_string()),
        read_like: true,
        started_at: Some(now),
        finished_at: Some(now),
        duration_ms: Some(0),
        status: Some("succeeded".to_string()),
        input_summary: Some(redact_and_bound_value(&serde_json::json!({
            "requested_mode": requested_mode.as_str(),
            "requested_guards": requested_guards,
            "capability_changed": capability_changed,
            "context_refreshed": context_refreshed,
            "execution_context_provided": execution_context.is_some(),
            "execution_context_changed": execution_context_changed,
            "execution_context": execution_context,
            "previous_execution_context": previous_execution_context,
            "session_reused": reused,
            "explicit_resume": explicit_resume,
        }))),
        instruction,
        requested_mode: Some(requested_mode.as_str().to_string()),
        previous_mode: previous_mode.map(|mode| mode.as_str().to_string()),
        requested_guards: Some(requested_guards),
        previous_guards,
        capability_changed: Some(capability_changed),
        context_refreshed: Some(context_refreshed),
        execution_context,
        previous_execution_context,
        execution_context_changed: Some(execution_context_changed),
        ..session_event(
            event_id.to_string(),
            session_id.to_string(),
            "task_instruction".to_string(),
            now,
            transport.as_str().to_string(),
            "work_on_project".to_string(),
            "read_only".to_string(),
        )
    }
}

fn coding_agent_lifecycle_event(
    session_id: &str,
    project: &str,
    run_id: &str,
    provider_id: &str,
    kind: &str,
    state: &str,
    execution_state: &str,
    terminal_stop_reason: Option<&str>,
    terminal_error_code: Option<&str>,
    now: i64,
) -> SessionEvent {
    let input_summary = serde_json::json!({
        "run_id": bound_summary_string(run_id),
        "provider_id": bound_summary_string(provider_id),
        "state": bound_summary_string(state),
        "execution_state": bound_summary_string(execution_state),
        "terminal_stop_reason": terminal_stop_reason.map(bound_summary_string),
        "terminal_error_code": terminal_error_code.map(bound_summary_string),
    });
    SessionEvent {
        project: Some(project.to_string()),
        resolved_project: Some(project.to_string()),
        started_at: Some(now),
        finished_at: Some(now),
        duration_ms: Some(0),
        status: Some(bound_summary_string(state)),
        error_kind: terminal_error_code.map(bound_summary_string),
        input_summary: Some(input_summary),
        ..session_event(
            format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple()),
            session_id.to_string(),
            bound_summary_string(kind),
            now,
            "system".to_string(),
            "start_coding_agent".to_string(),
            "job_run".to_string(),
        )
    }
}

fn session_closed_system_event(session_id: &str, now: i64) -> SessionEvent {
    SessionEvent {
        read_like: true,
        started_at: Some(now),
        finished_at: Some(now),
        duration_ms: Some(0),
        status: Some("succeeded".to_string()),
        ..session_event(
            format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple()),
            session_id.to_string(),
            "session_closed".to_string(),
            now,
            "system".to_string(),
            "close_session".to_string(),
            "read_only".to_string(),
        )
    }
}

fn session_execution_context_updated_event(
    session_id: &str,
    project: &str,
    transport: SessionTransport,
    execution_context: SessionExecutionContext,
    previous_execution_context: SessionExecutionContext,
    changed: bool,
    now: i64,
) -> SessionEvent {
    SessionEvent {
        project: Some(project.to_string()),
        resolved_project: Some(project.to_string()),
        read_like: true,
        started_at: Some(now),
        finished_at: Some(now),
        duration_ms: Some(0),
        status: Some("succeeded".to_string()),
        input_summary: Some(redact_and_bound_value(&serde_json::json!({
            "execution_context": execution_context,
            "previous_execution_context": previous_execution_context,
            "execution_context_changed": changed,
        }))),
        execution_context: Some(execution_context),
        previous_execution_context: Some(previous_execution_context),
        execution_context_changed: Some(changed),
        ..session_event(
            format!("{EVENT_ID_PREFIX}{}", uuid::Uuid::new_v4().simple()),
            session_id.to_string(),
            "session_execution_context_updated".to_string(),
            now,
            transport.as_str().to_string(),
            "update_session_context".to_string(),
            "read_only".to_string(),
        )
    }
}

fn event_observes_repository_edit(event: &SessionEvent) -> bool {
    event.kind == "tool_call_finished"
        && event.status.as_deref() == Some("succeeded")
        && event
            .effect_evidence
            .as_ref()
            .and_then(|evidence| evidence.state_changed)
            == Some(true)
        && runtime_tool_activity_semantics(&event.tool_name).kind == ToolActivityKind::Edit
}

fn summarize_record(
    record: &SessionRecord,
    limit: Option<usize>,
    cold: Option<&ColdSessionRecord>,
) -> SessionSummary {
    let limit = limit
        .unwrap_or(DEFAULT_SUMMARY_LIMIT)
        .clamp(0, MAX_SUMMARY_LIMIT);
    let finished_events = super::events::canonical_tool_call_finished_event_refs(
        record.events.iter().map(Arc::as_ref),
    );
    let counts = SessionCounts {
        tool_calls: finished_events.len(),
        succeeded: finished_events
            .iter()
            .filter(|event| event.status.as_deref() == Some("succeeded"))
            .count(),
        failed: finished_events
            .iter()
            .filter(|event| event.status.as_deref() == Some("failed"))
            .count(),
        read_like: finished_events
            .iter()
            .filter(|event| event.read_like)
            .count(),
        write_like: finished_events
            .iter()
            .filter(|event| event.write_like)
            .count(),
        shell_like: finished_events
            .iter()
            .filter(|event| event.shell_like)
            .count(),
        git_like: finished_events
            .iter()
            .filter(|event| event.git_like)
            .count(),
        change_summary_like: finished_events
            .iter()
            .filter(|event| event.change_summary_like)
            .count(),
    };
    let retained_total = record.events.len();
    let observed_total = record.events_observed.max(retained_total as u64) as usize;
    let skip = retained_total.saturating_sub(limit);
    let events: Vec<SessionEvent> = record
        .events
        .iter()
        .skip(skip)
        .map(|event| event.as_ref().clone())
        .collect();
    let events_returned = events.len();
    let project_instructions = match cold {
        Some(cold) => cold.project_instructions.clone(),
        None => record
            .project_instructions
            .as_ref()
            .map(|snapshot| snapshot.to_summary()),
    };
    SessionSummary {
        session_id: record.session_id.clone(),
        project: record.project.clone(),
        title: record.title.clone(),
        mode: record.mode,
        guards: record.guards,
        execution_context: record.execution_context.clone(),
        lifecycle: record.lifecycle,
        git_baseline_tree: record.git_baseline_tree.clone(),
        repository_edit_observed: record.repository_edit_observed,
        created_at: record.created_at,
        updated_at: record.updated_at,
        counts,
        current_instruction: record.current_instruction.as_deref().cloned(),
        events,
        events_total: observed_total,
        events_retained: if record.history_is_durable {
            observed_total.saturating_sub(record.event_history_start as usize)
        } else {
            retained_total
        },
        events_evicted: if record.history_is_durable {
            record.event_history_start as usize
        } else {
            observed_total.saturating_sub(retained_total)
        },
        retention_truncated: if record.history_is_durable {
            record.event_history_start > 0
        } else {
            observed_total > retained_total
        },
        ledger_first_retained_sequence: if record.history_is_durable {
            record.event_history_start as usize
        } else {
            observed_total.saturating_sub(retained_total)
        },
        events_returned,
        events_truncated: observed_total > events_returned,
        first_retained_sequence: observed_total.saturating_sub(events_returned),
        // Computed over the full retained history, not the bounded tail
        // above: an unfinished mutation older than the summary window must
        // still block a resume with `unconfirmed_tool_execution`.
        unclosed_mutating_call: super::continuation::has_unclosed_mutating_call(
            record.events.iter().map(Arc::as_ref),
            &record.session_id,
        ),
        project_instructions,
        messages: build_messages_summary(record),
    }
}

fn session_message_delivery_identity(
    delivery: &SessionMessageDelivery,
    kind: super::model::SessionMessageKind,
    message: &str,
    tags: &[String],
    reply_to: Option<&str>,
    priority: SessionMessagePriority,
    requires_ack: bool,
) -> Result<(String, String), SessionMessageError> {
    if delivery.sender_scope.len() != 64
        || !delivery
            .sender_scope
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(SessionMessageError::InvalidInput(
            "message delivery sender scope is unavailable".to_string(),
        ));
    }
    let delivery_key = delivery.delivery_key.trim();
    if delivery_key.is_empty() || delivery_key.chars().count() > MAX_MESSAGE_DELIVERY_KEY_CHARS {
        return Err(SessionMessageError::InvalidInput(format!(
            "delivery_key must contain 1..={MAX_MESSAGE_DELIVERY_KEY_CHARS} characters"
        )));
    }
    let mut key_hasher = Sha256::new();
    key_hasher.update(b"webcodex.session-message-delivery-key.v1\0");
    key_hasher.update(delivery_key.as_bytes());
    let scope_key = format!("{}:{:x}", delivery.sender_scope, key_hasher.finalize());

    fn hash_field(hasher: &mut Sha256, value: &[u8]) {
        hasher.update((value.len() as u64).to_be_bytes());
        hasher.update(value);
    }
    let mut payload = Sha256::new();
    payload.update(b"webcodex.session-message-delivery-payload.v1\0");
    hash_field(&mut payload, kind.as_str().as_bytes());
    hash_field(&mut payload, message.as_bytes());
    payload.update((tags.len() as u64).to_be_bytes());
    for tag in tags {
        hash_field(&mut payload, tag.as_bytes());
    }
    hash_field(&mut payload, reply_to.unwrap_or_default().as_bytes());
    let priority = match priority {
        SessionMessagePriority::Low => "low",
        SessionMessagePriority::Normal => "normal",
        SessionMessagePriority::High => "high",
    };
    hash_field(&mut payload, priority.as_bytes());
    payload.update([u8::from(requires_ack)]);
    Ok((scope_key, format!("{:x}", payload.finalize())))
}

impl SessionStoreInner {
    // --- create / lifecycle ---

    /// Sole map-insert path for a newly created session.
    // Called under the store mutex; every fresh-session path uses this allocator.
    pub(super) fn allocate_session_id(&self, mut suffix: impl FnMut() -> String) -> Option<String> {
        for _ in 0..16 {
            let id = format!("{SESSION_ID_PREFIX}{}", suffix());
            if !self.session_id_occupied(&id) {
                return Some(id);
            }
        }
        None
    }

    pub(super) fn insert_session(&mut self, mut record: SessionRecord) -> SessionSummary {
        assert!(
            !self.session_id_occupied(&record.session_id),
            "Session allocation must not replace an existing ledger"
        );
        record.history_is_durable = self.persistence.is_some();
        record.current_instruction = record
            .events
            .iter()
            .rev()
            .find(|event| event.kind == "task_instruction")
            .cloned();
        let session_id = record.session_id.clone();
        self.sessions
            .insert(session_id.clone(), StoredSession::Hot(record));
        self.touch(&session_id);
        self.compact_hot_sessions(&session_id);

        self.summary(&session_id, Some(DEFAULT_SUMMARY_LIMIT))
            .expect("newly inserted session must summarize")
    }

    /// Explicit lifecycle close. Unknown ids fail without create.
    pub(super) fn close_session(
        &mut self,
        session_id: &str,
    ) -> Result<SessionCloseOutcome, SessionCloseError> {
        self.touch(session_id);
        if !self.prepare_record_mutation(session_id) {
            return Err(SessionCloseError::UnknownSession);
        }

        let lifecycle = self
            .sessions
            .get(session_id)
            .map(StoredSession::lifecycle)
            .ok_or(SessionCloseError::UnknownSession)?;
        match lifecycle {
            SessionLifecycle::Closed => {
                let record = self
                    .sessions
                    .get(session_id)
                    .and_then(StoredSession::hot)
                    .expect("prepared Closed Session is materialized");
                Ok(SessionCloseOutcome {
                    summary: summarize_record(record, Some(DEFAULT_SUMMARY_LIMIT), None),
                    already_closed: true,
                })
            }
            SessionLifecycle::Active => {
                let now = now_ts();
                let event = session_closed_system_event(session_id, now);
                let max_events = self.recent_event_limit;
                {
                    let record = self
                        .sessions
                        .get_mut(session_id)
                        .and_then(StoredSession::hot_mut)
                        .expect("active session must stay hot through close commit");
                    record.lifecycle = SessionLifecycle::Closed;
                    record.updated_at = now;
                    append_event(record, event, max_events);
                }
                // Update eviction eligibility in the same locked lifecycle commit.
                self.touch(session_id);
                let record = self
                    .sessions
                    .get(session_id)
                    .and_then(StoredSession::hot)
                    .expect("just-closed session remains hot until outer coldification");
                Ok(SessionCloseOutcome {
                    summary: summarize_record(record, Some(DEFAULT_SUMMARY_LIMIT), None),
                    already_closed: false,
                })
            }
        }
    }

    // --- messages ---

    pub(super) fn post_message(
        &mut self,
        input: PostSessionMessageInput,
        requires_ack: bool,
        delivery: Option<SessionMessageDelivery>,
    ) -> Result<SessionMessageDeliveryOutcome, SessionMessageError> {
        self.touch(&input.session_id);
        if !self.prepare_record_mutation(&input.session_id) {
            return Err(SessionMessageError::InvalidObservationState);
        }
        let Some(stored) = self.sessions.get_mut(&input.session_id) else {
            return Err(SessionMessageError::UnknownSession);
        };
        let lifecycle = stored.lifecycle();
        if !lifecycle.allows_mutation() {
            return Err(SessionMessageError::SessionClosed { lifecycle });
        }
        let record = stored
            .hot_mut()
            .expect("active touched session message mutation must stay hot");
        let message = validate_message_text(input.message)?;
        let tags = validate_message_tags(input.tags)?;
        if let Some(reply_to) = input.reply_to.as_deref() {
            let found = record
                .messages
                .iter()
                .any(|message| message.message_id == reply_to);
            if !found {
                return Err(SessionMessageError::UnknownMessage);
            }
        }
        let delivery_identity = delivery
            .as_ref()
            .map(|delivery| {
                session_message_delivery_identity(
                    delivery,
                    input.kind,
                    &message,
                    &tags,
                    input.reply_to.as_deref(),
                    input.priority,
                    requires_ack,
                )
            })
            .transpose()?;
        if let Some((scope_key, payload_fingerprint)) = delivery_identity.as_ref() {
            if let Some(replay) = record.message_delivery_replays.get(scope_key) {
                if replay.payload_fingerprint != *payload_fingerprint {
                    return Err(SessionMessageError::DeliveryKeyConflict);
                }
                let Some(message) = record
                    .messages
                    .iter()
                    .find(|message| message.message_id == replay.message_id)
                else {
                    return Err(SessionMessageError::DeliveryKeyConflict);
                };
                return Ok(SessionMessageDeliveryOutcome {
                    message: message.as_ref().clone(),
                    replayed: true,
                    state_changed: false,
                });
            }
        }
        let now = now_ts();
        let message_id = allocate_message_id(record, webcodex_core::compact::random_suffix::<12>)?;
        let message = SessionMessage {
            message_id,
            session_id: input.session_id.clone(),
            created_at: now,
            kind: input.kind,
            status: SessionMessageStatus::Open,
            priority: input.priority,
            message,
            tags,
            reply_to: input.reply_to,
            requires_ack,
            first_ack_observed_at: None,
            author_session_id: None,
            resolved_at: None,
            resolution: None,
            closure_kind: None,
            superseded_by_message_id: None,
            supersedes_message_id: None,
            resolved_by_message_id: None,
            completion_id: None,
        };
        let revision = Self::next_message_observation_revision(record)?;
        record.updated_at = now;
        record.messages.push_back(Arc::new(message.clone()));
        record
            .message_observation_revisions
            .insert(message.message_id.clone(), revision);
        if let Some((scope_key, payload_fingerprint)) = delivery_identity {
            record.message_delivery_replays.insert(
                scope_key,
                SessionMessageDeliveryReplay {
                    payload_fingerprint,
                    message_id: message.message_id.clone(),
                },
            );
        }
        Ok(SessionMessageDeliveryOutcome {
            message,
            replayed: false,
            state_changed: true,
        })
    }

    pub(super) fn observe_message_acks(
        &mut self,
        session_id: &str,
        message_ids: &[String],
    ) -> super::model::SessionAckObservation {
        if !self.prepare_record_mutation(session_id) {
            return super::model::SessionAckObservation {
                ignored_count: message_ids.len(),
                ..Default::default()
            };
        }

        let Some(stored) = self.sessions.get_mut(session_id) else {
            return super::model::SessionAckObservation {
                ignored_count: message_ids.len(),
                ..Default::default()
            };
        };
        let Some(record) = stored.hot_mut() else {
            return super::model::SessionAckObservation {
                ignored_count: message_ids.len(),
                ..Default::default()
            };
        };
        let mut outcome = super::model::SessionAckObservation::default();
        let mut seen = std::collections::HashSet::new();
        let now = now_ts();
        for message_id in message_ids {
            if !seen.insert(message_id.as_str()) {
                continue;
            }
            let Some(index) = record.messages.iter().position(|message| {
                message.message_id == *message_id
                    && message.status == SessionMessageStatus::Open
                    && message.requires_ack
            }) else {
                outcome.ignored_count += 1;
                continue;
            };
            outcome.accepted_count += 1;
            outcome.accepted_ids.push(message_id.clone());
            if record.messages[index].first_ack_observed_at.is_none() {
                let Ok(revision) = Self::next_message_observation_revision(record) else {
                    outcome.accepted_ids.pop();
                    outcome.accepted_count = outcome.accepted_count.saturating_sub(1);
                    outcome.ignored_count += 1;
                    continue;
                };
                let message = Arc::make_mut(&mut record.messages[index]);
                message.first_ack_observed_at = Some(now);
                record
                    .message_observation_revisions
                    .insert(message.message_id.clone(), revision);
                record.updated_at = now;
                outcome.first_observed_count += 1;
            }
        }
        outcome
    }

    pub(super) fn withdraw_message(
        &mut self,
        session_id: &str,
        message_id: &str,
    ) -> Result<WithdrawSessionMessageOutcome, SessionMessageError> {
        self.touch(session_id);
        if !self.prepare_record_mutation(session_id) {
            return Err(SessionMessageError::InvalidObservationState);
        }
        let Some(stored) = self.sessions.get_mut(session_id) else {
            return Err(SessionMessageError::UnknownSession);
        };
        let lifecycle = stored.lifecycle();
        if !lifecycle.allows_mutation() {
            return Err(SessionMessageError::SessionClosed { lifecycle });
        }
        let record = stored
            .hot_mut()
            .expect("active session message mutation must stay hot");
        let Some(message_index) = record
            .messages
            .iter()
            .position(|message| message.message_id == message_id)
        else {
            return Err(SessionMessageError::UnknownMessage);
        };
        let snapshot = record.messages[message_index].as_ref().clone();
        if snapshot.status == SessionMessageStatus::Resolved {
            if snapshot.closure_kind == Some(SessionMessageClosureKind::Withdrawn) {
                return Ok(WithdrawSessionMessageOutcome {
                    message: snapshot,
                    replayed: true,
                });
            }
            return Err(SessionMessageError::MessageNotOpen);
        }
        if !matches!(
            snapshot.kind,
            super::model::SessionMessageKind::Note
                | super::model::SessionMessageKind::Guidance
                | super::model::SessionMessageKind::Question
                | super::model::SessionMessageKind::Todo
        ) {
            return Err(SessionMessageError::InvalidInput(
                "message kind cannot be withdrawn by the Runtime Console".to_string(),
            ));
        }
        if snapshot.closure_kind.is_some()
            || snapshot.superseded_by_message_id.is_some()
            || snapshot.completion_id.is_some()
            || snapshot.resolved_by_message_id.is_some()
        {
            return Err(SessionMessageError::MessageNotOpen);
        }
        let revision = Self::next_message_observation_revision(record)?;
        let now = now_ts();
        let message = Arc::make_mut(&mut record.messages[message_index]);
        message.status = SessionMessageStatus::Resolved;
        message.resolved_at = Some(now);
        message.closure_kind = Some(SessionMessageClosureKind::Withdrawn);
        record
            .message_observation_revisions
            .insert(message.message_id.clone(), revision);
        record.updated_at = now;
        Ok(WithdrawSessionMessageOutcome {
            message: message.clone(),
            replayed: false,
        })
    }

    pub(super) fn replace_message(
        &mut self,
        input: ReplaceSessionMessageInput,
    ) -> Result<ReplaceSessionMessageOutcome, SessionMessageError> {
        self.touch(&input.session_id);
        if !self.prepare_record_mutation(&input.session_id) {
            return Err(SessionMessageError::InvalidObservationState);
        }
        let Some(stored) = self.sessions.get_mut(&input.session_id) else {
            return Err(SessionMessageError::UnknownSession);
        };
        let lifecycle = stored.lifecycle();
        if !lifecycle.allows_mutation() {
            return Err(SessionMessageError::SessionClosed { lifecycle });
        }
        let replacement_text = validate_message_text(input.message)?;
        let record = stored
            .hot_mut()
            .expect("active session message mutation must stay hot");
        let Some(original_index) = record
            .messages
            .iter()
            .position(|message| message.message_id == input.message_id)
        else {
            return Err(SessionMessageError::UnknownMessage);
        };
        let original_snapshot = record.messages[original_index].as_ref().clone();
        if original_snapshot.status == SessionMessageStatus::Resolved {
            if original_snapshot.closure_kind != Some(SessionMessageClosureKind::Superseded) {
                return Err(SessionMessageError::MessageNotOpen);
            }
            let replacement_id = original_snapshot
                .superseded_by_message_id
                .as_deref()
                .ok_or(SessionMessageError::IdempotencyConflict)?;
            let replacement = record
                .messages
                .iter()
                .find(|message| message.message_id == replacement_id)
                .ok_or(SessionMessageError::IdempotencyConflict)?;
            let relation_is_canonical = replacement.supersedes_message_id.as_deref()
                == Some(input.message_id.as_str())
                && replacement.kind == original_snapshot.kind
                && replacement.priority == original_snapshot.priority
                && replacement.tags == original_snapshot.tags
                && replacement.reply_to == original_snapshot.reply_to
                && replacement.requires_ack == original_snapshot.requires_ack;
            if !relation_is_canonical || replacement.message != replacement_text {
                return Err(SessionMessageError::IdempotencyConflict);
            }
            return Ok(ReplaceSessionMessageOutcome {
                original: original_snapshot,
                replacement: replacement.as_ref().clone(),
                replayed: true,
            });
        }
        if !matches!(
            original_snapshot.kind,
            super::model::SessionMessageKind::Note
                | super::model::SessionMessageKind::Guidance
                | super::model::SessionMessageKind::Question
                | super::model::SessionMessageKind::Todo
        ) {
            return Err(SessionMessageError::InvalidInput(
                "message kind cannot be replaced by the Runtime Console".to_string(),
            ));
        }
        if original_snapshot.closure_kind.is_some()
            || original_snapshot.superseded_by_message_id.is_some()
            || original_snapshot.completion_id.is_some()
            || original_snapshot.resolved_by_message_id.is_some()
        {
            return Err(SessionMessageError::MessageNotOpen);
        }

        let message_id = allocate_message_id(record, webcodex_core::compact::random_suffix::<12>)?;
        let original_revision = record
            .message_observation_revision
            .checked_add(1)
            .ok_or(SessionMessageError::InvalidObservationState)?;
        let replacement_revision = original_revision
            .checked_add(1)
            .ok_or(SessionMessageError::InvalidObservationState)?;
        record.message_observation_revision = replacement_revision;
        let now = now_ts();
        let replacement = SessionMessage {
            message_id,
            session_id: input.session_id.clone(),
            created_at: now,
            kind: original_snapshot.kind,
            status: SessionMessageStatus::Open,
            priority: original_snapshot.priority,
            message: replacement_text,
            tags: original_snapshot.tags.clone(),
            reply_to: original_snapshot.reply_to.clone(),
            requires_ack: original_snapshot.requires_ack,
            first_ack_observed_at: None,
            author_session_id: None,
            resolved_at: None,
            resolution: None,
            closure_kind: None,
            superseded_by_message_id: None,
            supersedes_message_id: Some(input.message_id.clone()),
            resolved_by_message_id: None,
            completion_id: None,
        };
        {
            let original = Arc::make_mut(&mut record.messages[original_index]);
            original.status = SessionMessageStatus::Resolved;
            original.resolved_at = Some(now);
            original.closure_kind = Some(SessionMessageClosureKind::Superseded);
            original.superseded_by_message_id = Some(replacement.message_id.clone());
        }
        record.messages.push_back(Arc::new(replacement.clone()));
        record
            .message_observation_revisions
            .insert(input.message_id.clone(), original_revision);
        record
            .message_observation_revisions
            .insert(replacement.message_id.clone(), replacement_revision);
        record.updated_at = now;
        let original = record
            .messages
            .iter()
            .find(|message| message.message_id == input.message_id)
            .expect("superseded source retained")
            .as_ref()
            .clone();
        Ok(ReplaceSessionMessageOutcome {
            original,
            replacement,
            replayed: false,
        })
    }

    /// Resolve an open message. Already-resolved messages stay resolved
    /// (status is not reopened); an optional new resolution text may update.
    pub(super) fn resolve_message(
        &mut self,
        session_id: &str,
        message_id: &str,
        resolution: Option<String>,
    ) -> Result<(SessionMessage, bool), SessionMessageError> {
        self.touch(session_id);
        if !self.prepare_record_mutation(session_id) {
            return Err(SessionMessageError::InvalidObservationState);
        }
        let Some(stored) = self.sessions.get_mut(session_id) else {
            return Err(SessionMessageError::UnknownSession);
        };
        let lifecycle = stored.lifecycle();
        if !lifecycle.allows_mutation() {
            return Err(SessionMessageError::SessionClosed { lifecycle });
        }
        let record = stored
            .hot_mut()
            .expect("active session message mutation must stay hot");
        let Some(message_index) = record
            .messages
            .iter()
            .position(|message| message.message_id == message_id)
        else {
            return Err(SessionMessageError::UnknownMessage);
        };
        if record.messages[message_index].closure_kind.is_some() {
            return Err(SessionMessageError::MessageNotOpen);
        }
        let resolution = match resolution {
            Some(value) => Some(validate_resolution_text(value)?),
            None => None,
        };
        let changed = record.messages[message_index].status == SessionMessageStatus::Open
            || resolution.as_ref().is_some_and(|resolution| {
                record.messages[message_index].resolution.as_ref() != Some(resolution)
            });
        let revision = if changed {
            Some(Self::next_message_observation_revision(record)?)
        } else {
            None
        };
        let message = Arc::make_mut(&mut record.messages[message_index]);
        if message.status == SessionMessageStatus::Open {
            message.status = SessionMessageStatus::Resolved;
            message.resolved_at = Some(now_ts());
        }
        if resolution.is_some() {
            message.resolution = resolution;
        }
        record.updated_at = now_ts();
        if let Some(revision) = revision {
            record
                .message_observation_revisions
                .insert(message.message_id.clone(), revision);
        }
        Ok((message.clone(), changed))
    }

    pub(super) fn resolve_message_from_wrapper(
        &mut self,
        session_id: &str,
        message_id: &str,
        resolution: String,
        current_request_acknowledged: bool,
    ) -> Result<(SessionMessage, bool), SessionMessageError> {
        self.touch(session_id);
        if !self.prepare_record_mutation(session_id) {
            return Err(SessionMessageError::InvalidObservationState);
        }
        let Some(stored) = self.sessions.get_mut(session_id) else {
            return Err(SessionMessageError::UnknownSession);
        };
        let lifecycle = stored.lifecycle();
        if !lifecycle.allows_mutation() {
            return Err(SessionMessageError::SessionClosed { lifecycle });
        }
        let resolution = validate_resolution_text(resolution)?;
        if resolution.is_empty() {
            return Err(SessionMessageError::InvalidInput(
                "session_message_resolution.resolution must not be empty".to_string(),
            ));
        }
        let record = stored
            .hot_mut()
            .expect("active session message mutation must stay hot");
        let Some(message_index) = record
            .messages
            .iter()
            .position(|message| message.message_id == message_id)
        else {
            return Err(SessionMessageError::UnknownMessage);
        };
        let snapshot = record.messages[message_index].as_ref().clone();
        if snapshot.closure_kind.is_some() {
            return Err(SessionMessageError::MessageNotOpen);
        }
        if snapshot.kind == super::model::SessionMessageKind::Todo {
            return Err(SessionMessageError::InvalidInput(
                "todo messages require complete_session_message rather than session_message_resolution"
                    .to_string(),
            ));
        }
        if snapshot.status == SessionMessageStatus::Resolved {
            if snapshot.resolution.as_deref() == Some(resolution.as_str()) {
                return Ok((snapshot, false));
            }
            return Err(SessionMessageError::IdempotencyConflict);
        }
        if snapshot.requires_ack && !current_request_acknowledged {
            return Err(SessionMessageError::InvalidInput(
                "requires_ack message must be acknowledged on the same request before wrapper resolution"
                    .to_string(),
            ));
        }

        let revision = Self::next_message_observation_revision(record)?;
        let now = now_ts();
        let message = Arc::make_mut(&mut record.messages[message_index]);
        message.status = SessionMessageStatus::Resolved;
        message.resolved_at = Some(now);
        message.resolution = Some(resolution);
        record.updated_at = now;
        record
            .message_observation_revisions
            .insert(message.message_id.clone(), revision);
        Ok((message.clone(), true))
    }

    pub(super) fn complete_message(
        &mut self,
        input: CompleteSessionMessageInput,
    ) -> Result<CompleteSessionMessageOutcome, SessionMessageError> {
        self.touch(&input.session_id);
        if !self.prepare_record_mutation(&input.session_id) {
            return Err(SessionMessageError::InvalidObservationState);
        }
        let Some(stored) = self.sessions.get_mut(&input.session_id) else {
            return Err(SessionMessageError::UnknownSession);
        };
        let lifecycle = stored.lifecycle();
        if !lifecycle.allows_mutation() {
            return Err(SessionMessageError::SessionClosed { lifecycle });
        }
        if !is_valid_completion_id(&input.completion_id) {
            return Err(SessionMessageError::InvalidInput(
                "completion identity is invalid".to_string(),
            ));
        }
        if input
            .author_session_id
            .as_deref()
            .is_some_and(|author_session_id| !super::events::is_valid_session_id(author_session_id))
        {
            return Err(SessionMessageError::InvalidInput(
                "author session identity is invalid".to_string(),
            ));
        }
        let answer_text = validate_message_text(input.answer)?;
        let tags = validate_message_tags(input.tags)?;
        let record = stored
            .hot_mut()
            .expect("active session message mutation must stay hot");
        let Some(todo_index) = record
            .messages
            .iter()
            .position(|message| message.message_id == input.message_id)
        else {
            return Err(SessionMessageError::UnknownMessage);
        };
        if record.messages[todo_index].kind != super::model::SessionMessageKind::Todo {
            return Err(SessionMessageError::NotTodo);
        }

        let todo_snapshot = record.messages[todo_index].as_ref().clone();
        validate_assignment_fence(&input.expected_assignment_fence)?;
        let provided_assignment_fence_fingerprint =
            assignment_fence_fingerprint(&input.expected_assignment_fence);
        if todo_snapshot.status == SessionMessageStatus::Resolved {
            match (
                todo_snapshot.completion_id.as_deref(),
                todo_snapshot.resolved_by_message_id.as_deref(),
            ) {
                (None, None) => {
                    return Err(SessionMessageError::AssignmentStale {
                        current: current_assignment_state(record, &input.message_id)?,
                        fresh_assignment_fence: None,
                    });
                }
                (Some(completion_id), Some(answer_message_id)) => {
                    let Some(answer) = record.messages.iter().find(|message| {
                        message.message_id == answer_message_id
                            && message.kind == super::model::SessionMessageKind::Answer
                            && message.reply_to.as_deref() == Some(input.message_id.as_str())
                    }) else {
                        return Err(SessionMessageError::InvalidCompletionState);
                    };
                    if completion_id != input.completion_id {
                        return Err(SessionMessageError::AlreadyCompleted {
                            answer_message_id: Some(answer.message_id.clone()),
                            completion_id: Some(completion_id.to_string()),
                        });
                    }
                    if answer.message != answer_text
                        || answer.tags != tags
                        || answer.priority != input.priority
                    {
                        return Err(SessionMessageError::IdempotencyConflict);
                    }
                    match record
                        .completion_assignment_fence_fingerprints
                        .get(&input.message_id)
                    {
                        Some(Some(stored_fingerprint)) => {
                            if stored_fingerprint != &provided_assignment_fence_fingerprint
                                || answer.author_session_id != input.author_session_id
                            {
                                return Err(SessionMessageError::IdempotencyConflict);
                            }
                        }
                        Some(None) => {
                            // Historical completion with known no-fence provenance.
                            // Current callers must never replay it as if their newly
                            // supplied fence had been part of the original intent.
                            return Err(SessionMessageError::IdempotencyConflict);
                        }
                        None if record.completion_assignment_fence_tracking_complete => {
                            return Err(SessionMessageError::InvalidCompletionState);
                        }
                        None => {
                            // Historical completion predating fence metadata. Preserve
                            // it for query/restore, but live replay cannot prove a match.
                            return Err(SessionMessageError::IdempotencyConflict);
                        }
                    }
                    return Ok(CompleteSessionMessageOutcome {
                        todo: todo_snapshot,
                        answer: answer.as_ref().clone(),
                        replayed: true,
                    });
                }
                _ => return Err(SessionMessageError::InvalidCompletionState),
            }
        }
        if todo_snapshot.completion_id.is_some() || todo_snapshot.resolved_by_message_id.is_some() {
            return Err(SessionMessageError::InvalidCompletionState);
        }
        let current_assignment = open_assignment_state(record, &input.message_id)?;
        let fresh_assignment_fence =
            assignment_fence_from_state(&input.session_id, &input.message_id, &current_assignment);
        if fresh_assignment_fence != input.expected_assignment_fence {
            let current = current_assignment_state(record, &input.message_id)?;
            return Err(SessionMessageError::AssignmentStale {
                current,
                fresh_assignment_fence: Some(fresh_assignment_fence),
            });
        }

        let message_id = allocate_message_id(record, webcodex_core::compact::random_suffix::<12>)?;
        let todo_revision = record
            .message_observation_revision
            .checked_add(1)
            .ok_or(SessionMessageError::InvalidObservationState)?;
        let answer_revision = todo_revision
            .checked_add(1)
            .ok_or(SessionMessageError::InvalidObservationState)?;
        record.message_observation_revision = answer_revision;

        let now = now_ts();
        let answer = SessionMessage {
            message_id,
            session_id: input.session_id.clone(),
            created_at: now,
            kind: super::model::SessionMessageKind::Answer,
            status: SessionMessageStatus::Open,
            priority: input.priority,
            message: answer_text,
            tags,
            reply_to: Some(input.message_id.clone()),
            requires_ack: false,
            first_ack_observed_at: None,
            author_session_id: input.author_session_id,
            resolved_at: None,
            resolution: None,
            closure_kind: None,
            superseded_by_message_id: None,
            supersedes_message_id: None,
            resolved_by_message_id: None,
            completion_id: None,
        };
        {
            let todo = Arc::make_mut(&mut record.messages[todo_index]);
            todo.status = SessionMessageStatus::Resolved;
            todo.resolved_at = Some(now);
            todo.resolved_by_message_id = Some(answer.message_id.clone());
            todo.completion_id = Some(input.completion_id);
        }
        record.messages.push_back(Arc::new(answer.clone()));
        record
            .message_observation_revisions
            .insert(input.message_id.clone(), todo_revision);
        record
            .message_observation_revisions
            .insert(answer.message_id.clone(), answer_revision);
        record.completion_assignment_fence_fingerprints.insert(
            input.message_id.clone(),
            Some(provided_assignment_fence_fingerprint),
        );
        record.updated_at = now;
        let todo = record
            .messages
            .iter()
            .find(|message| message.message_id == input.message_id)
            .expect("completed todo retained")
            .as_ref()
            .clone();
        Ok(CompleteSessionMessageOutcome {
            todo,
            answer,
            replayed: false,
        })
    }

    fn next_message_observation_revision(
        record: &mut SessionRecord,
    ) -> Result<u64, SessionMessageError> {
        let revision = record
            .message_observation_revision
            .checked_add(1)
            .ok_or(SessionMessageError::InvalidObservationState)?;
        record.message_observation_revision = revision;
        Ok(revision)
    }

    // --- reads / housekeeping ---

    pub(super) fn contains_session(&self, session_id: &str) -> bool {
        self.sessions.contains_key(session_id)
    }

    pub(super) fn session_project(&self, session_id: &str) -> Option<Option<String>> {
        self.sessions
            .get(session_id)
            .map(|record| record.project().map(str::to_string))
    }

    pub(super) fn session_target_authority(
        &self,
        session_id: &str,
    ) -> Option<(Option<String>, String)> {
        self.sessions.get(session_id).map(|record| {
            (
                record.project().map(str::to_string),
                record.owner_authority_fingerprint().to_string(),
            )
        })
    }

    pub(super) fn guard_state(&self, session_id: &str) -> Option<(SessionMode, SessionGuards)> {
        self.sessions
            .get(session_id)
            .map(StoredSession::mode_guards)
    }

    pub(super) fn lifecycle_state(&self, session_id: &str) -> Option<SessionLifecycle> {
        self.sessions.get(session_id).map(StoredSession::lifecycle)
    }

    #[cfg(test)]
    fn to_persisted_ledger(&self) -> PersistedSessionLedger {
        let mut sessions: Vec<PersistedSessionSnapshot> = self
            .lru
            .iter()
            .filter_map(|session_id| self.sessions.get(session_id))
            .map(|record| match record {
                StoredSession::Hot(record) => PersistedSessionSnapshot::Hot(
                    PersistedSessionRecord::from_record(record, self.recent_event_limit),
                ),
                StoredSession::Cold(record) => {
                    PersistedSessionSnapshot::Cold(Arc::clone(&record.raw))
                }
            })
            .collect();
        sessions.extend(self.tombstones.values().cloned().map(|tombstone| {
            PersistedSessionSnapshot::Tombstone(PersistedRetentionTombstoneRow {
                retention_tombstone: tombstone,
            })
        }));
        PersistedSessionLedger {
            version: SESSION_LEDGER_VERSION,
            sessions,
        }
    }

    pub(super) fn touch(&mut self, session_id: &str) {
        if let Some(session) = self.sessions.get(session_id) {
            self.lru
                .touch(session_id, session.lifecycle() == SessionLifecycle::Closed);
        }
    }

    fn session_id_occupied(&self, session_id: &str) -> bool {
        self.sessions.contains_key(session_id) || self.tombstone_ids.contains_key(session_id)
    }

    fn owned_retention_incarnation(
        &self,
        session_id: &str,
        owner_authority_fingerprint: &str,
    ) -> Option<String> {
        let ordinal = self.tombstone_ids.get(session_id).copied()?;
        let tombstone = self.tombstones.get(&ordinal)?;
        if tombstone.session_id != session_id
            || tombstone.owner_authority_fingerprint != owner_authority_fingerprint
        {
            return None;
        }
        Some(tombstone.incarnation_fingerprint.clone())
    }

    pub(super) fn summary(&self, session_id: &str, limit: Option<usize>) -> Option<SessionSummary> {
        let record = self.sessions.get(session_id)?.hot()?;
        Some(summarize_record(record, limit, None))
    }
}

// No retained identity or historical retained link may silently retarget.
pub(super) fn allocate_message_id(
    record: &SessionRecord,
    mut suffix: impl FnMut() -> String,
) -> Result<String, SessionMessageError> {
    for _ in 0..16 {
        let id = format!("{MESSAGE_ID_PREFIX}{}", suffix());
        if !record.messages.iter().any(|m| {
            m.message_id == id
                || m.reply_to.as_deref() == Some(&id)
                || m.resolved_by_message_id.as_deref() == Some(&id)
                || m.superseded_by_message_id.as_deref() == Some(&id)
                || m.supersedes_message_id.as_deref() == Some(&id)
        }) {
            return Ok(id);
        }
    }
    Err(SessionMessageError::InvalidObservationState)
}