falsegreen_agent/

event.rs

use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use thiserror::Error;

use crate::genui::{ActionKind, ActionSourceType, action_kind_source_compatible};

#[derive(Debug, Error)]
pub enum EventError {
    #[error("event-store I/O error: {0}")]
    Io(#[from] std::io::Error),
    #[error("SQLite event-store error: {0}")]
    Sqlite(#[from] rusqlite::Error),
    #[error("invalid event kind in durable history: {0}")]
    InvalidKind(String),
    #[error("invalid event JSON in durable history: {0}")]
    InvalidJson(#[from] serde_json::Error),
    #[error("system clock is before the Unix epoch")]
    InvalidClock,
    #[error("terminal session {0} is immutable")]
    TerminalSessionImmutable(String),
    #[error("predecessor session history changed while creating its replacement")]
    PredecessorHistoryChanged,
    #[error("session {predecessor} already has replacement {replacement}")]
    ReplacementAlreadyExists {
        predecessor: String,
        replacement: String,
    },
    #[error("replacement session chains are not supported")]
    ReplacementChainUnsupported,
    #[error("malformed durable replacement provenance: {0}")]
    MalformedReplacement(String),
    #[error("GenUI action {0} is not confirmed")]
    GenUiActionNotConfirmed(String),
    #[error("GenUI action {0} has already started or completed")]
    GenUiActionAlreadyStarted(String),
    #[error("GenUI action {0} has an unknown external outcome")]
    GenUiActionOutcomeUnknown(String),
    #[error("GenUI lifecycle events must use the typed transition API")]
    GenUiEventMustUseLifecycle,
    #[error("invalid GenUI lifecycle event: {0}")]
    InvalidGenUiEvent(String),
    #[error("illegal GenUI lifecycle transition for action {action}: {from} -> {to}")]
    GenUiIllegalTransition {
        action: String,
        from: String,
        to: String,
    },
    #[error("trusted GenUI state generation cannot rewind")]
    GenUiStateGenerationRewind,
    #[error("trusted GenUI state identity changed without advancing generation")]
    GenUiStateIdentityChanged,
}

pub(crate) const GENUI_EVENT_SCHEMA_VERSION: u16 = 1;

/// Strict, versioned payload shared by every durable GenUI lifecycle event.
/// Result/reconciliation fields are intentionally optional because they only
/// apply to terminal outcomes.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct GenUiActionEventPayload {
    pub schema_version: u16,
    pub action_id: String,
    pub action_kind: String,
    /// Explicit executable transport identity. Older durable events omitted
    /// this V9 field; they are read as MCP for compatibility, while all new
    /// bindings persist the authoritative source selected by the catalog.
    #[serde(default = "default_genui_source_type")]
    pub source_type: String,
    pub label_digest: String,
    pub surface_id: String,
    pub surface_digest: String,
    pub state_digest: String,
    pub state_generation: u64,
    pub session_id: String,
    pub principal: String,
    pub authorization_context: String,
    pub provider_id: String,
    pub server_id: String,
    pub tool_name: String,
    pub remote_tool_name: String,
    pub schema_digest: String,
    pub policy_version: u64,
    pub policy_digest: String,
    pub requires_confirmation: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub payload_digest: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result_digest: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ok: Option<bool>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
}

impl GenUiActionEventPayload {
    fn authority_eq(&self, other: &Self) -> bool {
        self.schema_version == other.schema_version
            && self.action_id == other.action_id
            && self.action_kind == other.action_kind
            && self.source_type == other.source_type
            && self.label_digest == other.label_digest
            && self.surface_id == other.surface_id
            && self.surface_digest == other.surface_digest
            && self.state_digest == other.state_digest
            && self.state_generation == other.state_generation
            && self.session_id == other.session_id
            && self.principal == other.principal
            && self.authorization_context == other.authorization_context
            && self.provider_id == other.provider_id
            && self.server_id == other.server_id
            && self.tool_name == other.tool_name
            && self.remote_tool_name == other.remote_tool_name
            && self.schema_digest == other.schema_digest
            && self.policy_version == other.policy_version
            && self.policy_digest == other.policy_digest
            && self.requires_confirmation == other.requires_confirmation
    }

    fn binding_eq(&self, other: &Self) -> bool {
        self.authority_eq(other) && self.payload_digest == other.payload_digest
    }
}

fn default_genui_source_type() -> String {
    "mcp".to_owned()
}

fn parse_genui_action_kind(value: &str) -> Option<ActionKind> {
    serde_json::from_value(Value::String(value.to_owned())).ok()
}

fn parse_genui_source_type(value: &str) -> Option<ActionSourceType> {
    serde_json::from_value(Value::String(value.to_owned())).ok()
}

type GenUiActionState = (EventKind, GenUiActionEventPayload);

fn validate_sha256_digest(label: &str, value: &str) -> Result<(), EventError> {
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(EventError::InvalidGenUiEvent(format!(
            "{label} must be an exact SHA-256 hex digest"
        )));
    }
    Ok(())
}

fn validate_genui_payload(
    session_id: &str,
    kind: EventKind,
    payload: &GenUiActionEventPayload,
) -> Result<(), EventError> {
    if !kind.is_genui()
        || payload.schema_version != GENUI_EVENT_SCHEMA_VERSION
        || payload.session_id != session_id
        || payload.action_id.is_empty()
    {
        return Err(EventError::InvalidGenUiEvent(
            "schema version, session, or action identity is invalid".to_owned(),
        ));
    }
    for (label, value) in [
        ("action_id", &payload.action_id),
        ("action_kind", &payload.action_kind),
        ("label_digest", &payload.label_digest),
        ("surface_id", &payload.surface_id),
        ("state_digest", &payload.state_digest),
        ("session_id", &payload.session_id),
        ("principal", &payload.principal),
        ("authorization_context", &payload.authorization_context),
        ("provider_id", &payload.provider_id),
        ("server_id", &payload.server_id),
        ("tool_name", &payload.tool_name),
        ("remote_tool_name", &payload.remote_tool_name),
    ] {
        if value.is_empty() {
            return Err(EventError::InvalidGenUiEvent(format!(
                "{label} must not be empty"
            )));
        }
    }
    if !matches!(
        payload.action_kind.as_str(),
        "local_presentation" | "form_value_update" | "navigation" | "mcp_tool" | "consequential"
    ) {
        return Err(EventError::InvalidGenUiEvent(
            "action_kind is not in the declared protocol subset".to_owned(),
        ));
    }
    if !matches!(payload.source_type.as_str(), "mcp" | "host_local") {
        return Err(EventError::InvalidGenUiEvent(
            "source_type is not in the declared protocol subset".to_owned(),
        ));
    }
    let action_kind = parse_genui_action_kind(&payload.action_kind).ok_or_else(|| {
        EventError::InvalidGenUiEvent("action_kind is not a known executable kind".to_owned())
    })?;
    let source_type = parse_genui_source_type(&payload.source_type).ok_or_else(|| {
        EventError::InvalidGenUiEvent("source_type is not a known executable source".to_owned())
    })?;
    if !action_kind_source_compatible(action_kind, source_type) {
        return Err(EventError::InvalidGenUiEvent(format!(
            "invalid action kind/source type pair: {action_kind:?}/{source_type:?}"
        )));
    }
    if payload.requires_confirmation != (action_kind == ActionKind::Consequential) {
        return Err(EventError::InvalidGenUiEvent(
            "confirmation requirement does not match action kind".to_owned(),
        ));
    }
    validate_sha256_digest("surface_digest", &payload.surface_digest)?;
    validate_sha256_digest("schema_digest", &payload.schema_digest)?;
    validate_sha256_digest("policy_digest", &payload.policy_digest)?;
    if let Some(value) = &payload.payload_digest {
        validate_sha256_digest("payload_digest", value)?;
    }
    if let Some(value) = &payload.result_digest {
        validate_sha256_digest("result_digest", value)?;
    }
    match kind {
        EventKind::GenUiActionPresented => {
            if payload.payload_digest.is_some()
                || payload.result_digest.is_some()
                || payload.ok.is_some()
                || payload.reason.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "presented event cannot carry a payload or outcome".to_owned(),
                ));
            }
        }
        EventKind::GenUiActionConfirmationRequested => {
            if !payload.requires_confirmation
                || payload.payload_digest.is_none()
                || payload.result_digest.is_some()
                || payload.ok.is_some()
                || payload.reason.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "confirmation request must carry only a consequential payload binding"
                        .to_owned(),
                ));
            }
        }
        EventKind::GenUiActionConfirmed => {
            if !payload.requires_confirmation
                || payload.payload_digest.is_none()
                || payload.result_digest.is_some()
                || payload.ok.is_some()
                || payload.reason.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "confirmation must carry only a consequential payload binding".to_owned(),
                ));
            }
        }
        EventKind::GenUiActionExecutionStarted => {
            if payload.payload_digest.is_none()
                || payload.result_digest.is_some()
                || payload.ok.is_some()
                || payload.reason.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "execution start must carry an exact payload binding".to_owned(),
                ));
            }
        }
        EventKind::GenUiActionExecutionCompleted => {
            if payload.payload_digest.is_none()
                || payload.ok != Some(true)
                || payload.result_digest.is_none()
                || payload.reason.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "completed event must carry ok=true and an exact result digest".to_owned(),
                ));
            }
        }
        EventKind::GenUiActionExecutionFailed => {
            if payload.payload_digest.is_none()
                || payload.ok != Some(false)
                || payload.result_digest.is_none()
                || payload.reason.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "failed event must carry ok=false and an exact result digest".to_owned(),
                ));
            }
        }
        EventKind::GenUiActionExecutionUnknown => {
            if payload.payload_digest.is_none()
                || payload.reason.as_deref().is_none_or(str::is_empty)
                || payload.result_digest.is_some()
                || payload.ok.is_some()
            {
                return Err(EventError::InvalidGenUiEvent(
                    "unknown event must carry only a payload binding and reconciliation reason"
                        .to_owned(),
                ));
            }
        }
        _ => unreachable!("validate_genui_payload only receives GenUI events"),
    }
    Ok(())
}

fn legal_genui_transition(
    previous: Option<&GenUiActionState>,
    kind: EventKind,
    payload: &GenUiActionEventPayload,
) -> bool {
    match previous {
        None => kind == EventKind::GenUiActionPresented,
        Some((previous_kind, previous_payload)) => match (*previous_kind, kind) {
            (EventKind::GenUiActionPresented, EventKind::GenUiActionConfirmationRequested) => {
                payload.requires_confirmation && payload.payload_digest.is_some()
            }
            (EventKind::GenUiActionConfirmationRequested, EventKind::GenUiActionConfirmed) => {
                previous_payload.binding_eq(payload)
            }
            (EventKind::GenUiActionPresented, EventKind::GenUiActionExecutionStarted) => {
                !payload.requires_confirmation && payload.payload_digest.is_some()
            }
            (EventKind::GenUiActionConfirmed, EventKind::GenUiActionExecutionStarted) => {
                previous_payload.binding_eq(payload)
            }
            (
                EventKind::GenUiActionExecutionStarted,
                EventKind::GenUiActionExecutionCompleted
                | EventKind::GenUiActionExecutionFailed
                | EventKind::GenUiActionExecutionUnknown,
            ) => previous_payload.binding_eq(payload),
            _ => false,
        },
    }
}

fn replay_genui_history(
    events: &[Event],
    session_id: &str,
) -> Result<BTreeMap<String, GenUiActionState>, EventError> {
    let mut states: BTreeMap<String, GenUiActionState> = BTreeMap::new();
    for event in events.iter().filter(|event| event.kind.is_genui()) {
        if event.session_id != session_id {
            return Err(EventError::InvalidGenUiEvent(
                "lifecycle event belongs to another session".to_owned(),
            ));
        }
        let payload = serde_json::from_value::<GenUiActionEventPayload>(event.payload.clone())
            .map_err(|error| EventError::InvalidGenUiEvent(error.to_string()))?;
        validate_genui_payload(session_id, event.kind, &payload)?;
        if let Some(previous) = states.get(&payload.action_id)
            && !previous.1.authority_eq(&payload)
        {
            return Err(EventError::GenUiIllegalTransition {
                action: payload.action_id.clone(),
                from: previous.0.as_str().to_owned(),
                to: event.kind.as_str().to_owned(),
            });
        }
        let previous = states.get(&payload.action_id);
        if !legal_genui_transition(previous, event.kind, &payload) {
            return Err(EventError::GenUiIllegalTransition {
                action: payload.action_id.clone(),
                from: previous
                    .map_or_else(|| "none".to_owned(), |state| state.0.as_str().to_owned()),
                to: event.kind.as_str().to_owned(),
            });
        }
        states.insert(payload.action_id.clone(), (event.kind, payload));
    }
    Ok(states)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EventKind {
    SessionCreated,
    UserGoal,
    ModelRequest,
    ModelResponse,
    ToolRequest,
    ToolResult,
    FileMutation,
    CommandExecution,
    GitState,
    CandidateReady,
    FalsegreenResult,
    RepairStarted,
    SessionReplaced,
    Checkpoint,
    TerminalState,
    GenUiActionPresented,
    GenUiActionConfirmationRequested,
    GenUiActionConfirmed,
    GenUiActionExecutionStarted,
    GenUiActionExecutionCompleted,
    GenUiActionExecutionFailed,
    GenUiActionExecutionUnknown,
    CompositionRequested,
    ModelProposalReceived,
    CompositionRejected,
    CompositionAccepted,
    TrustedHandleResolved,
    CompositionFallbackUsed,
}

impl EventKind {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::SessionCreated => "session_created",
            Self::UserGoal => "user_goal",
            Self::ModelRequest => "model_request",
            Self::ModelResponse => "model_response",
            Self::ToolRequest => "tool_request",
            Self::ToolResult => "tool_result",
            Self::FileMutation => "file_mutation",
            Self::CommandExecution => "command_execution",
            Self::GitState => "git_state",
            Self::CandidateReady => "candidate_ready",
            Self::FalsegreenResult => "falsegreen_result",
            Self::RepairStarted => "repair_started",
            Self::SessionReplaced => "session_replaced",
            Self::Checkpoint => "checkpoint",
            Self::TerminalState => "terminal_state",
            Self::GenUiActionPresented => "genui_action_presented",
            Self::GenUiActionConfirmationRequested => "genui_action_confirmation_requested",
            Self::GenUiActionConfirmed => "genui_action_confirmed",
            Self::GenUiActionExecutionStarted => "genui_action_execution_started",
            Self::GenUiActionExecutionCompleted => "genui_action_execution_completed",
            Self::GenUiActionExecutionFailed => "genui_action_execution_failed",
            Self::GenUiActionExecutionUnknown => "genui_action_execution_unknown",
            Self::CompositionRequested => "composition_requested",
            Self::ModelProposalReceived => "model_proposal_received",
            Self::CompositionRejected => "composition_rejected",
            Self::CompositionAccepted => "composition_accepted",
            Self::TrustedHandleResolved => "trusted_handle_resolved",
            Self::CompositionFallbackUsed => "composition_fallback_used",
        }
    }

    fn parse(value: &str) -> Result<Self, EventError> {
        match value {
            "session_created" => Ok(Self::SessionCreated),
            "user_goal" => Ok(Self::UserGoal),
            "model_request" => Ok(Self::ModelRequest),
            "model_response" => Ok(Self::ModelResponse),
            "tool_request" => Ok(Self::ToolRequest),
            "tool_result" => Ok(Self::ToolResult),
            "file_mutation" => Ok(Self::FileMutation),
            "command_execution" => Ok(Self::CommandExecution),
            "git_state" => Ok(Self::GitState),
            "candidate_ready" => Ok(Self::CandidateReady),
            "falsegreen_result" => Ok(Self::FalsegreenResult),
            "repair_started" => Ok(Self::RepairStarted),
            "session_replaced" => Ok(Self::SessionReplaced),
            "checkpoint" => Ok(Self::Checkpoint),
            "terminal_state" => Ok(Self::TerminalState),
            "genui_action_presented" => Ok(Self::GenUiActionPresented),
            "genui_action_confirmation_requested" => Ok(Self::GenUiActionConfirmationRequested),
            "genui_action_confirmed" => Ok(Self::GenUiActionConfirmed),
            "genui_action_execution_started" => Ok(Self::GenUiActionExecutionStarted),
            "genui_action_execution_completed" => Ok(Self::GenUiActionExecutionCompleted),
            "genui_action_execution_failed" => Ok(Self::GenUiActionExecutionFailed),
            "genui_action_execution_unknown" => Ok(Self::GenUiActionExecutionUnknown),
            "composition_requested" => Ok(Self::CompositionRequested),
            "model_proposal_received" => Ok(Self::ModelProposalReceived),
            "composition_rejected" => Ok(Self::CompositionRejected),
            "composition_accepted" => Ok(Self::CompositionAccepted),
            "trusted_handle_resolved" => Ok(Self::TrustedHandleResolved),
            "composition_fallback_used" => Ok(Self::CompositionFallbackUsed),
            other => Err(EventError::InvalidKind(other.to_owned())),
        }
    }

    #[must_use]
    pub(crate) const fn is_genui(self) -> bool {
        matches!(
            self,
            Self::GenUiActionPresented
                | Self::GenUiActionConfirmationRequested
                | Self::GenUiActionConfirmed
                | Self::GenUiActionExecutionStarted
                | Self::GenUiActionExecutionCompleted
                | Self::GenUiActionExecutionFailed
                | Self::GenUiActionExecutionUnknown
        )
    }

    #[must_use]
    pub(crate) const fn is_composition(self) -> bool {
        matches!(
            self,
            Self::CompositionRequested
                | Self::ModelProposalReceived
                | Self::CompositionRejected
                | Self::CompositionAccepted
                | Self::TrustedHandleResolved
                | Self::CompositionFallbackUsed
        )
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Event {
    pub sequence: u64,
    pub session_id: String,
    pub kind: EventKind,
    pub created_at_ms: u64,
    pub payload: Value,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionReplacementRecord {
    pub predecessor_session_id: String,
    pub replacement_session_id: String,
    pub predecessor_state: String,
    pub predecessor_history_sha256: String,
    pub candidate_sha256: String,
    pub candidate_event_sequence: u64,
    pub source_git_state_sequence: u64,
    pub falsegreen_task_id: String,
    pub created_at_ms: u64,
}

#[derive(Debug)]
pub struct EventStore {
    connection: Connection,
    path: Option<PathBuf>,
}

impl EventStore {
    pub fn open(path: impl AsRef<Path>) -> Result<Self, EventError> {
        let path = path.as_ref();
        let connection = Connection::open(path)?;
        connection.busy_timeout(Duration::from_secs(5))?;
        connection.pragma_update(None, "journal_mode", "WAL")?;
        connection.execute_batch(
            "CREATE TABLE IF NOT EXISTS events (
                sequence INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT NOT NULL,
                kind TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL,
                payload_json TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS events_session_sequence
                ON events(session_id, sequence);
            CREATE TABLE IF NOT EXISTS session_replacements (
                predecessor_session_id TEXT PRIMARY KEY,
                replacement_session_id TEXT NOT NULL UNIQUE,
                predecessor_state TEXT NOT NULL,
                predecessor_history_sha256 TEXT NOT NULL,
                candidate_sha256 TEXT NOT NULL,
                candidate_event_sequence INTEGER NOT NULL,
                source_git_state_sequence INTEGER NOT NULL,
                falsegreen_task_id TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS genui_current_states (
                session_id TEXT NOT NULL,
                principal TEXT NOT NULL,
                generation INTEGER NOT NULL,
                digest TEXT NOT NULL,
                authorization_context TEXT NOT NULL,
                PRIMARY KEY (session_id, principal)
            );
            CREATE TABLE IF NOT EXISTS authority_generation (
                id INTEGER PRIMARY KEY CHECK (id = 1),
                generation INTEGER NOT NULL
            );
            INSERT INTO authority_generation (id, generation)
                VALUES (1, 0)
                ON CONFLICT(id) DO NOTHING;",
        )?;
        let path = if path == Path::new(":memory:") {
            None
        } else {
            Some(fs::canonicalize(path)?)
        };
        Ok(Self { connection, path })
    }

    pub fn open_memory() -> Result<Self, EventError> {
        Self::open(":memory:")
    }

    #[must_use]
    pub fn path(&self) -> Option<&Path> {
        self.path.as_deref()
    }

    /// Monotonic durable authority generation. It is incremented in the same
    /// SQLite transaction as every event, lifecycle, trusted-state, or
    /// replacement mutation, so a publication fence can use it as a real
    /// mutation-time token rather than inferring change from a later read.
    pub fn authority_generation(&self) -> Result<u64, EventError> {
        let generation = self.connection.query_row(
            "SELECT generation FROM authority_generation WHERE id = 1",
            [],
            |row| row.get::<_, i64>(0),
        )?;
        u64::try_from(generation).map_err(|_| {
            EventError::InvalidGenUiEvent("authority generation exceeds SQLite range".to_owned())
        })
    }

    pub fn append(
        &mut self,
        session_id: &str,
        kind: EventKind,
        payload: &Value,
    ) -> Result<Event, EventError> {
        let _authority_guard = crate::mcp::acquire_publication_guard();
        if kind.is_genui() {
            return Err(EventError::GenUiEventMustUseLifecycle);
        }
        let created_at_ms = now_ms()?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        if session_is_terminal(&transaction, session_id)? && !kind.is_composition() {
            return Err(EventError::TerminalSessionImmutable(session_id.to_owned()));
        }
        let event = insert_event(&transaction, session_id, kind, created_at_ms, payload)?;
        bump_authority_generation(&transaction)?;
        transaction.commit()?;
        Ok(event)
    }

    /// Append one typed GenUI lifecycle event. GenUI payloads are parsed with
    /// a closed schema and the complete action history is reconstructed inside
    /// the same SQLite transaction before the new event is committed.
    pub(crate) fn append_genui_lifecycle(
        &mut self,
        session_id: &str,
        kind: EventKind,
        payload: &Value,
    ) -> Result<Event, EventError> {
        let _authority_guard = crate::mcp::acquire_publication_guard();
        let typed: GenUiActionEventPayload = serde_json::from_value(payload.clone())?;
        validate_genui_payload(session_id, kind, &typed)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        if session_is_terminal(&transaction, session_id)? {
            return Err(EventError::TerminalSessionImmutable(session_id.to_owned()));
        }
        let events = load_events(&transaction, session_id)?;
        let states = replay_genui_history(&events, session_id)?;
        let previous = states.get(&typed.action_id);
        if let Some((previous_kind, previous_payload)) = previous
            && !previous_payload.authority_eq(&typed)
        {
            return Err(EventError::GenUiIllegalTransition {
                action: typed.action_id.clone(),
                from: previous_kind.as_str().to_owned(),
                to: kind.as_str().to_owned(),
            });
        }
        if !legal_genui_transition(previous, kind, &typed) {
            return Err(EventError::GenUiIllegalTransition {
                action: typed.action_id.clone(),
                from: previous
                    .map_or_else(|| "none".to_owned(), |(kind, _)| kind.as_str().to_owned()),
                to: kind.as_str().to_owned(),
            });
        }
        let created_at_ms = now_ms()?;
        let event = insert_event(&transaction, session_id, kind, created_at_ms, payload)?;
        bump_authority_generation(&transaction)?;
        transaction.commit()?;
        Ok(event)
    }

    /// Atomically transition a confirmed GenUI action into execution. The
    /// transition is kept in the append-only session history so concurrent
    /// processes and restarts observe the same lifecycle.
    pub fn start_genui_action(
        &mut self,
        session_id: &str,
        action_id: &str,
        identity: &Value,
        requires_confirmation: bool,
    ) -> Result<Event, EventError> {
        let typed: GenUiActionEventPayload = serde_json::from_value(identity.clone())
            .map_err(|_| EventError::GenUiActionNotConfirmed(action_id.to_owned()))?;
        if typed.action_id != action_id
            || typed.session_id != session_id
            || typed.requires_confirmation != requires_confirmation
        {
            return Err(EventError::GenUiActionNotConfirmed(action_id.to_owned()));
        }
        let result = self.append_genui_lifecycle(
            session_id,
            EventKind::GenUiActionExecutionStarted,
            identity,
        );
        result.map_err(|error| match error {
            EventError::GenUiIllegalTransition { from, .. }
                if from == EventKind::GenUiActionExecutionUnknown.as_str() =>
            {
                EventError::GenUiActionOutcomeUnknown(action_id.to_owned())
            }
            EventError::GenUiIllegalTransition { .. } if requires_confirmation => {
                EventError::GenUiActionAlreadyStarted(action_id.to_owned())
            }
            EventError::GenUiActionOutcomeUnknown(_) => error,
            EventError::GenUiIllegalTransition { .. } => {
                EventError::GenUiActionNotConfirmed(action_id.to_owned())
            }
            other => other,
        })
    }

    pub(crate) fn complete_genui_action(
        &mut self,
        session_id: &str,
        action_id: &str,
        identity: &Value,
        ok: bool,
        result_digest: &str,
    ) -> Result<Event, EventError> {
        let mut payload: GenUiActionEventPayload = serde_json::from_value(identity.clone())?;
        if payload.action_id != action_id || payload.session_id != session_id {
            return Err(EventError::InvalidGenUiEvent(
                "completion identity mismatch".to_owned(),
            ));
        }
        payload.ok = Some(ok);
        payload.result_digest = Some(result_digest.to_owned());
        let kind = if ok {
            EventKind::GenUiActionExecutionCompleted
        } else {
            EventKind::GenUiActionExecutionFailed
        };
        self.append_genui_lifecycle(session_id, kind, &serde_json::to_value(payload)?)
    }

    /// Terminally reconcile a durable execution start when the host authority
    /// changed before transport. This records an explicit unknown outcome so a
    /// later continuation cannot mistake the stale action for an unattempted
    /// execution or replay it against a new workspace.
    pub(crate) fn mark_genui_action_unknown(
        &mut self,
        session_id: &str,
        action_id: &str,
        identity: &Value,
        reason: &str,
    ) -> Result<Event, EventError> {
        let mut payload: GenUiActionEventPayload = serde_json::from_value(identity.clone())?;
        if payload.action_id != action_id || payload.session_id != session_id {
            return Err(EventError::InvalidGenUiEvent(
                "unknown outcome identity mismatch".to_owned(),
            ));
        }
        if reason.is_empty() {
            return Err(EventError::InvalidGenUiEvent(
                "unknown outcome reason must not be empty".to_owned(),
            ));
        }
        payload.reason = Some(reason.to_owned());
        self.append_genui_lifecycle(
            session_id,
            EventKind::GenUiActionExecutionUnknown,
            &serde_json::to_value(payload)?,
        )
    }

    pub fn events(&self, session_id: &str) -> Result<Vec<Event>, EventError> {
        load_events(&self.connection, session_id)
    }

    /// Reconstruct one complete GenUI lifecycle from durable history. Resume
    /// paths must use this method rather than searching for an individual
    /// event kind, so malformed, reordered, truncated, or terminal histories
    /// fail closed in exactly the same validator used for normal appends.
    pub(crate) fn replay_genui_action(
        &self,
        session_id: &str,
        action_id: &str,
    ) -> Result<Option<(EventKind, Value)>, EventError> {
        let events = load_events(&self.connection, session_id)?;
        if events
            .iter()
            .any(|event| event.kind == EventKind::TerminalState)
        {
            return Err(EventError::TerminalSessionImmutable(session_id.to_owned()));
        }
        let states = replay_genui_history(&events, session_id)?;
        Ok(states.get(action_id).map(|(kind, payload)| {
            (
                *kind,
                serde_json::to_value(payload).expect("typed payload serializes"),
            )
        }))
    }

    pub(crate) fn genui_current_state(
        &self,
        session_id: &str,
        principal: &str,
    ) -> Result<Option<(u64, String, String)>, EventError> {
        let state = self
            .connection
            .query_row(
                "SELECT generation, digest, authorization_context
                 FROM genui_current_states WHERE session_id = ?1 AND principal = ?2",
                params![session_id, principal],
                |row| {
                    let generation = row.get::<_, i64>(0)?;
                    let generation = u64::try_from(generation)
                        .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(0, generation))?;
                    Ok((
                        generation,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .optional()
            .map_err(EventError::from)?;
        if let Some((generation, digest, authorization_context)) = &state
            && (*generation == 0 || digest.is_empty() || authorization_context.is_empty())
        {
            return Err(EventError::InvalidGenUiEvent(
                "durable current state is malformed".to_owned(),
            ));
        }
        Ok(state)
    }

    #[cfg(test)]
    pub(crate) fn set_genui_current_state(
        &mut self,
        session_id: &str,
        principal: &str,
        generation: u64,
        digest: &str,
        authorization_context: &str,
    ) -> Result<(), EventError> {
        let _authority_guard = crate::mcp::acquire_publication_guard();
        let generation_sql = i64::try_from(generation).map_err(|_| {
            EventError::InvalidGenUiEvent(
                "trusted state generation exceeds SQLite range".to_owned(),
            )
        })?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        let existing = transaction
            .query_row(
                "SELECT generation, digest, authorization_context
                 FROM genui_current_states WHERE session_id = ?1 AND principal = ?2",
                params![session_id, principal],
                |row| {
                    let generation = row.get::<_, i64>(0)?;
                    let generation = u64::try_from(generation)
                        .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(0, generation))?;
                    Ok((
                        generation,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                    ))
                },
            )
            .optional()?;
        if let Some((previous_generation, previous_digest, previous_context)) = existing {
            if generation < previous_generation {
                return Err(EventError::GenUiStateGenerationRewind);
            }
            if generation == previous_generation
                && (digest != previous_digest || authorization_context != previous_context)
            {
                return Err(EventError::GenUiStateIdentityChanged);
            }
        }
        transaction.execute(
            "INSERT INTO genui_current_states
             (session_id, principal, generation, digest, authorization_context)
             VALUES (?1, ?2, ?3, ?4, ?5)
             ON CONFLICT(session_id, principal) DO UPDATE SET
               generation = excluded.generation,
               digest = excluded.digest,
               authorization_context = excluded.authorization_context",
            params![
                session_id,
                principal,
                generation_sql,
                digest,
                authorization_context
            ],
        )?;
        bump_authority_generation(&transaction)?;
        transaction.commit()?;
        Ok(())
    }

    pub(crate) fn advance_genui_current_state(
        &mut self,
        session_id: &str,
        principal: &str,
        digest: &str,
        authorization_context: &str,
    ) -> Result<u64, EventError> {
        let _authority_guard = crate::mcp::acquire_publication_guard();
        if session_id.is_empty()
            || principal.is_empty()
            || digest.is_empty()
            || authorization_context.is_empty()
        {
            return Err(EventError::InvalidGenUiEvent(
                "durable current state identity must not be empty".to_owned(),
            ));
        }
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        let previous = transaction
            .query_row(
                "SELECT generation FROM genui_current_states
                 WHERE session_id = ?1 AND principal = ?2",
                params![session_id, principal],
                |row| row.get::<_, i64>(0),
            )
            .optional()?;
        let generation = previous.map_or(1, |value| {
            u64::try_from(value).unwrap_or(u64::MAX).saturating_add(1)
        });
        let generation_sql = i64::try_from(generation).map_err(|_| {
            EventError::InvalidGenUiEvent(
                "trusted state generation exceeds SQLite range".to_owned(),
            )
        })?;
        transaction.execute(
            "INSERT INTO genui_current_states
             (session_id, principal, generation, digest, authorization_context)
             VALUES (?1, ?2, ?3, ?4, ?5)
             ON CONFLICT(session_id, principal) DO UPDATE SET
               generation = excluded.generation,
               digest = excluded.digest,
               authorization_context = excluded.authorization_context",
            params![
                session_id,
                principal,
                generation_sql,
                digest,
                authorization_context
            ],
        )?;
        bump_authority_generation(&transaction)?;
        transaction.commit()?;
        Ok(generation)
    }

    /// Qualification-only durable-state mutation seam. Normal callers must
    /// advance GenUI state through the Agent/workspace authority boundary.
    #[doc(hidden)]
    pub fn advance_genui_current_state_for_testing(
        &mut self,
        session_id: &str,
        principal: &str,
        digest: &str,
        authorization_context: &str,
    ) -> Result<u64, EventError> {
        self.advance_genui_current_state(session_id, principal, digest, authorization_context)
    }

    /// Convert any in-flight GenUI execution to an explicit unknown outcome
    /// after a process restart. An external MCP side effect must never be
    /// retried blindly when its prior outcome is not durable.
    pub fn reconcile_genui_actions(&mut self, session_id: &str) -> Result<usize, EventError> {
        let events = self.events(session_id)?;
        let states = replay_genui_history(&events, session_id)?;
        let unresolved = states
            .iter()
            .filter_map(|(action_id, (kind, payload))| {
                (*kind == EventKind::GenUiActionExecutionStarted)
                    .then_some((action_id.clone(), payload.clone()))
            })
            .collect::<Vec<_>>();
        for (_action_id, mut payload) in unresolved {
            payload.reason = Some("process_restart_before_external_outcome_was_durable".to_owned());
            self.append_genui_lifecycle(
                session_id,
                EventKind::GenUiActionExecutionUnknown,
                &serde_json::to_value(payload)?,
            )?;
        }
        Ok(states
            .values()
            .filter(|(kind, _)| *kind == EventKind::GenUiActionExecutionStarted)
            .count())
    }

    pub(crate) fn create_session_replacement(
        &mut self,
        mut record: SessionReplacementRecord,
        event_payloads: &[(EventKind, Value)],
    ) -> Result<SessionReplacementRecord, EventError> {
        let _authority_guard = crate::mcp::acquire_publication_guard();
        record.created_at_ms = now_ms()?;
        validate_replacement_event_payloads(&record, event_payloads)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)?;
        let predecessor_events = load_events(&transaction, &record.predecessor_session_id)?;
        if history_sha256(&predecessor_events)? != record.predecessor_history_sha256 {
            return Err(EventError::PredecessorHistoryChanged);
        }
        if replacement_row_by_replacement(&transaction, &record.predecessor_session_id)?.is_some() {
            return Err(EventError::ReplacementChainUnsupported);
        }
        if let Some(existing) =
            replacement_row_by_predecessor(&transaction, &record.predecessor_session_id)?
        {
            return Err(EventError::ReplacementAlreadyExists {
                predecessor: existing.predecessor_session_id,
                replacement: existing.replacement_session_id,
            });
        }
        if !load_events(&transaction, &record.replacement_session_id)?.is_empty()
            || replacement_row_by_replacement(&transaction, &record.replacement_session_id)?
                .is_some()
        {
            return Err(EventError::MalformedReplacement(
                "generated replacement session ID is already in use".to_owned(),
            ));
        }
        transaction.execute(
            "INSERT INTO session_replacements (
                predecessor_session_id, replacement_session_id, predecessor_state,
                predecessor_history_sha256, candidate_sha256, candidate_event_sequence,
                source_git_state_sequence, falsegreen_task_id, created_at_ms
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
            params![
                record.predecessor_session_id,
                record.replacement_session_id,
                record.predecessor_state,
                record.predecessor_history_sha256,
                record.candidate_sha256,
                record.candidate_event_sequence,
                record.source_git_state_sequence,
                record.falsegreen_task_id,
                record.created_at_ms,
            ],
        )?;
        for (kind, payload) in event_payloads {
            insert_event(
                &transaction,
                &record.replacement_session_id,
                *kind,
                record.created_at_ms,
                payload,
            )?;
        }
        bump_authority_generation(&transaction)?;
        transaction.commit()?;
        Ok(record)
    }

    pub fn replacement_for_session(
        &self,
        replacement_session_id: &str,
    ) -> Result<Option<SessionReplacementRecord>, EventError> {
        replacement_row_by_replacement(&self.connection, replacement_session_id)?
            .map(|record| validate_replacement_record(&self.connection, record))
            .transpose()
    }

    pub fn replacement_for_predecessor(
        &self,
        predecessor_session_id: &str,
    ) -> Result<Option<SessionReplacementRecord>, EventError> {
        replacement_row_by_predecessor(&self.connection, predecessor_session_id)?
            .map(|record| validate_replacement_record(&self.connection, record))
            .transpose()
    }

    /// Return known session IDs from most recently active to least recently active.
    ///
    /// This is intentionally a small event-store query rather than a mutable
    /// "current session" pointer. Resume selection can therefore be rebuilt from
    /// the same append-only history used by the authority boundary.
    pub fn session_ids_by_recency(&self) -> Result<Vec<String>, EventError> {
        let mut statement = self.connection.prepare(
            "SELECT session_id, MAX(sequence) AS latest_sequence
             FROM events
             GROUP BY session_id
             ORDER BY latest_sequence DESC, session_id ASC",
        )?;
        let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
        let mut session_ids = Vec::new();
        for row in rows {
            session_ids.push(row?);
        }
        Ok(session_ids)
    }
}

fn insert_event(
    transaction: &Transaction<'_>,
    session_id: &str,
    kind: EventKind,
    created_at_ms: u64,
    payload: &Value,
) -> Result<Event, EventError> {
    transaction.execute(
        "INSERT INTO events(session_id, kind, created_at_ms, payload_json)
         VALUES (?1, ?2, ?3, ?4)",
        params![
            session_id,
            kind.as_str(),
            created_at_ms,
            serde_json::to_string(payload)?
        ],
    )?;
    let sequence = u64::try_from(transaction.last_insert_rowid())
        .map_err(|_| EventError::InvalidKind("negative SQLite sequence".to_owned()))?;
    Ok(Event {
        sequence,
        session_id: session_id.to_owned(),
        kind,
        created_at_ms,
        payload: payload.clone(),
    })
}

fn session_is_terminal(connection: &Connection, session_id: &str) -> Result<bool, EventError> {
    Ok(connection.query_row(
        "SELECT EXISTS(
            SELECT 1 FROM events WHERE session_id = ?1 AND kind = 'terminal_state'
        )",
        [session_id],
        |row| row.get(0),
    )?)
}

fn load_events(connection: &Connection, session_id: &str) -> Result<Vec<Event>, EventError> {
    let mut statement = connection.prepare(
        "SELECT sequence, session_id, kind, created_at_ms, payload_json
         FROM events WHERE session_id = ?1 ORDER BY sequence ASC",
    )?;
    let rows = statement.query_map([session_id], |row| {
        Ok((
            row.get::<_, u64>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, u64>(3)?,
            row.get::<_, String>(4)?,
        ))
    })?;
    let mut events = Vec::new();
    for row in rows {
        let (sequence, stored_session_id, kind, created_at_ms, payload_json) = row?;
        events.push(Event {
            sequence,
            session_id: stored_session_id,
            kind: EventKind::parse(&kind)?,
            created_at_ms,
            payload: serde_json::from_str(&payload_json)?,
        });
    }
    Ok(events)
}

pub fn history_sha256(events: &[Event]) -> Result<String, EventError> {
    let mut digest = Sha256::new();
    digest.update(serde_json::to_vec(events)?);
    Ok(format!("{:x}", digest.finalize()))
}

fn replacement_row_by_predecessor(
    connection: &Connection,
    predecessor_session_id: &str,
) -> Result<Option<SessionReplacementRecord>, EventError> {
    replacement_row(
        connection,
        "SELECT predecessor_session_id, replacement_session_id, predecessor_state,
                predecessor_history_sha256, candidate_sha256, candidate_event_sequence,
                source_git_state_sequence, falsegreen_task_id, created_at_ms
           FROM session_replacements WHERE predecessor_session_id = ?1",
        predecessor_session_id,
    )
}

fn replacement_row_by_replacement(
    connection: &Connection,
    replacement_session_id: &str,
) -> Result<Option<SessionReplacementRecord>, EventError> {
    replacement_row(
        connection,
        "SELECT predecessor_session_id, replacement_session_id, predecessor_state,
                predecessor_history_sha256, candidate_sha256, candidate_event_sequence,
                source_git_state_sequence, falsegreen_task_id, created_at_ms
           FROM session_replacements WHERE replacement_session_id = ?1",
        replacement_session_id,
    )
}

fn replacement_row(
    connection: &Connection,
    query: &str,
    id: &str,
) -> Result<Option<SessionReplacementRecord>, EventError> {
    Ok(connection
        .query_row(query, [id], |row| {
            Ok(SessionReplacementRecord {
                predecessor_session_id: row.get(0)?,
                replacement_session_id: row.get(1)?,
                predecessor_state: row.get(2)?,
                predecessor_history_sha256: row.get(3)?,
                candidate_sha256: row.get(4)?,
                candidate_event_sequence: row.get(5)?,
                source_git_state_sequence: row.get(6)?,
                falsegreen_task_id: row.get(7)?,
                created_at_ms: row.get(8)?,
            })
        })
        .optional()?)
}

fn validate_replacement_event_payloads(
    record: &SessionReplacementRecord,
    events: &[(EventKind, Value)],
) -> Result<(), EventError> {
    if !replacement_record_shape_is_valid(record) {
        return Err(EventError::MalformedReplacement(
            "replacement record contains invalid identity fields".to_owned(),
        ));
    }
    let matching: Vec<&Value> = events
        .iter()
        .filter_map(|(kind, payload)| (*kind == EventKind::SessionReplaced).then_some(payload))
        .collect();
    if matching.len() != 1 || !replacement_payload_matches(record, matching[0]) {
        return Err(EventError::MalformedReplacement(
            "atomic replacement batch must contain one exact session_replaced event".to_owned(),
        ));
    }
    let authority_bindings: Vec<&Value> = events
        .iter()
        .filter_map(|(kind, payload)| {
            (*kind == EventKind::Checkpoint && payload["checkpoint_kind"] == "acceptance_authority")
                .then_some(payload)
        })
        .collect();
    if authority_bindings.len() != 1
        || authority_bindings[0]["falsegreen_task_id"].as_str() != Some(&record.falsegreen_task_id)
        || authority_bindings[0]["replacement_predecessor_session_id"].as_str()
            != Some(&record.predecessor_session_id)
    {
        return Err(EventError::MalformedReplacement(
            "atomic replacement batch must contain one exact authority binding".to_owned(),
        ));
    }
    Ok(())
}

fn validate_replacement_record(
    connection: &Connection,
    record: SessionReplacementRecord,
) -> Result<SessionReplacementRecord, EventError> {
    if !replacement_record_shape_is_valid(&record) {
        return Err(EventError::MalformedReplacement(
            "replacement record contains invalid identity fields".to_owned(),
        ));
    }
    if replacement_row_by_replacement(connection, &record.predecessor_session_id)?.is_some() {
        return Err(EventError::MalformedReplacement(
            "replacement chains are forbidden".to_owned(),
        ));
    }
    let predecessor_events = load_events(connection, &record.predecessor_session_id)?;
    if predecessor_events.is_empty()
        || history_sha256(&predecessor_events)? != record.predecessor_history_sha256
    {
        return Err(EventError::MalformedReplacement(
            "predecessor history no longer matches its replacement record".to_owned(),
        ));
    }
    let predecessor_terminal = predecessor_events.last().ok_or_else(|| {
        EventError::MalformedReplacement("predecessor history is empty".to_owned())
    })?;
    if predecessor_terminal.kind != EventKind::TerminalState
        || predecessor_terminal.payload["state"].as_str() != Some(&record.predecessor_state)
    {
        return Err(EventError::MalformedReplacement(
            "predecessor terminal state disagrees with replacement record".to_owned(),
        ));
    }
    let candidate_event = predecessor_events
        .iter()
        .find(|event| event.sequence == record.candidate_event_sequence)
        .filter(|event| {
            event.kind == EventKind::CandidateReady
                && event.payload["candidate_sha256"].as_str() == Some(&record.candidate_sha256)
        })
        .ok_or_else(|| {
            EventError::MalformedReplacement(
                "predecessor candidate event disagrees with replacement record".to_owned(),
            )
        })?;
    let source_git_state = predecessor_events
        .iter()
        .find(|event| event.sequence == record.source_git_state_sequence)
        .filter(|event| {
            event.kind == EventKind::GitState && event.sequence > candidate_event.sequence
        })
        .ok_or_else(|| {
            EventError::MalformedReplacement(
                "predecessor Git-state event disagrees with replacement record".to_owned(),
            )
        })?;
    let task_bindings: Vec<&Event> = predecessor_events
        .iter()
        .filter(|event| {
            event.kind == EventKind::Checkpoint
                && event.payload["checkpoint_kind"] == "acceptance_authority"
        })
        .collect();
    if task_bindings.len() != 1
        || task_bindings[0].payload["falsegreen_task_id"].as_str()
            != Some(&record.falsegreen_task_id)
    {
        return Err(EventError::MalformedReplacement(
            "predecessor task authority disagrees with replacement record".to_owned(),
        ));
    }
    let replacement_events = load_events(connection, &record.replacement_session_id)?;
    let matching: Vec<&Value> = replacement_events
        .iter()
        .filter_map(|event| (event.kind == EventKind::SessionReplaced).then_some(&event.payload))
        .collect();
    if matching.len() != 1 || !replacement_payload_matches(&record, matching[0]) {
        return Err(EventError::MalformedReplacement(
            "replacement relation and durable event disagree".to_owned(),
        ));
    }
    let replacement_authority: Vec<&Event> = replacement_events
        .iter()
        .filter(|event| {
            event.kind == EventKind::Checkpoint
                && event.payload["checkpoint_kind"] == "acceptance_authority"
        })
        .collect();
    let mut expected_authority = task_bindings[0].payload.clone();
    expected_authority["replacement_predecessor_session_id"] =
        Value::String(record.predecessor_session_id.clone());
    if replacement_authority.len() != 1 || replacement_authority[0].payload != expected_authority {
        return Err(EventError::MalformedReplacement(
            "replacement task authority disagrees with predecessor and relation".to_owned(),
        ));
    }
    let carried_candidates: Vec<&Event> = replacement_events
        .iter()
        .filter(|event| {
            event.kind == EventKind::CandidateReady
                && event.payload["carried_forward"].as_bool() == Some(true)
        })
        .collect();
    if carried_candidates.len() != 1
        || carried_candidates[0].payload["candidate_sha256"].as_str()
            != Some(&record.candidate_sha256)
        || carried_candidates[0].payload["predecessor_session_id"].as_str()
            != Some(&record.predecessor_session_id)
        || carried_candidates[0].payload["predecessor_candidate_event_sequence"].as_u64()
            != Some(record.candidate_event_sequence)
    {
        return Err(EventError::MalformedReplacement(
            "replacement candidate continuity is incomplete".to_owned(),
        ));
    }
    let carried_git_states: Vec<&Event> = replacement_events
        .iter()
        .filter(|event| event.kind == EventKind::GitState)
        .collect();
    if carried_git_states.len() != 1
        || carried_git_states[0].sequence <= carried_candidates[0].sequence
        || carried_git_states[0].payload != source_git_state.payload
    {
        return Err(EventError::MalformedReplacement(
            "replacement source Git-state provenance changed".to_owned(),
        ));
    }
    let validation_checkpoints: Vec<&Event> = replacement_events
        .iter()
        .filter(|event| {
            event.kind == EventKind::Checkpoint
                && event.payload["checkpoint_kind"] == "replacement_candidate_validation"
        })
        .collect();
    if validation_checkpoints.len() != 1
        || validation_checkpoints[0].payload["state"] != "candidate_ready"
        || validation_checkpoints[0].payload["predecessor_session_id"].as_str()
            != Some(&record.predecessor_session_id)
        || validation_checkpoints[0].payload["expected_candidate_sha256"].as_str()
            != Some(&record.candidate_sha256)
        || validation_checkpoints[0].payload["actual_candidate_sha256"].as_str()
            != Some(&record.candidate_sha256)
        || validation_checkpoints[0].payload["matched"].as_bool() != Some(true)
    {
        return Err(EventError::MalformedReplacement(
            "replacement candidate-validation checkpoint disagrees with relation".to_owned(),
        ));
    }
    Ok(record)
}

fn replacement_record_shape_is_valid(record: &SessionReplacementRecord) -> bool {
    record.predecessor_session_id.starts_with("session_")
        && record.replacement_session_id.starts_with("session_")
        && record.predecessor_session_id != record.replacement_session_id
        && record.predecessor_state == "failed"
        && valid_sha256(&record.predecessor_history_sha256)
        && valid_sha256(&record.candidate_sha256)
        && record.candidate_event_sequence > 0
        && record.source_git_state_sequence > record.candidate_event_sequence
        && !record.falsegreen_task_id.is_empty()
}

fn valid_sha256(value: &str) -> bool {
    value.len() == 64
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

fn replacement_payload_matches(record: &SessionReplacementRecord, payload: &Value) -> bool {
    payload["predecessor_session_id"].as_str() == Some(&record.predecessor_session_id)
        && payload["predecessor_state"].as_str() == Some(&record.predecessor_state)
        && payload["predecessor_history_sha256"].as_str()
            == Some(&record.predecessor_history_sha256)
        && payload["candidate_sha256"].as_str() == Some(&record.candidate_sha256)
        && payload["candidate_event_sequence"].as_u64() == Some(record.candidate_event_sequence)
        && payload["source_git_state_sequence"].as_u64() == Some(record.source_git_state_sequence)
        && payload["falsegreen_task_id"].as_str() == Some(&record.falsegreen_task_id)
        && payload["topology"].as_str() == Some("single_direct_replacement_no_chains")
}

fn now_ms() -> Result<u64, EventError> {
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| EventError::InvalidClock)?
        .as_millis();
    u64::try_from(millis).map_err(|_| EventError::InvalidClock)
}

fn bump_authority_generation(transaction: &Transaction<'_>) -> Result<(), EventError> {
    transaction.execute(
        "UPDATE authority_generation
         SET generation = generation + 1
         WHERE id = 1",
        [],
    )?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use serde_json::json;
    use tempfile::tempdir;

    use super::{EventKind, EventStore};

    #[test]
    fn events_are_strictly_ordered_and_survive_reopen() {
        let directory = tempdir().expect("tempdir");
        let path = directory.path().join("events.db");
        {
            let mut store = EventStore::open(&path).expect("open");
            store
                .append(
                    "s1",
                    EventKind::SessionCreated,
                    &json!({"state": "initializing"}),
                )
                .expect("append");
            store
                .append("s1", EventKind::ToolResult, &json!({"ok": true}))
                .expect("append");
        }
        let store = EventStore::open(&path).expect("reopen");
        let events = store.events("s1").expect("events");
        assert_eq!(events.len(), 2);
        assert!(events[0].sequence < events[1].sequence);
        assert_eq!(events[1].kind, EventKind::ToolResult);
        assert_eq!(events[1].payload, json!({"ok": true}));
    }

    #[test]
    fn lists_sessions_by_latest_activity() {
        let mut store = EventStore::open_memory().expect("store");
        store
            .append("older", EventKind::SessionCreated, &json!({}))
            .expect("older");
        store
            .append("newer", EventKind::SessionCreated, &json!({}))
            .expect("newer");
        store
            .append("older", EventKind::Checkpoint, &json!({}))
            .expect("activity");
        assert_eq!(
            store.session_ids_by_recency().expect("sessions"),
            vec!["older", "newer"]
        );
    }

    #[test]
    fn trusted_genui_state_is_monotonic_and_survives_restart() {
        let directory = tempdir().expect("tempdir");
        let path = directory.path().join("state.db");
        {
            let mut store = EventStore::open(&path).expect("open");
            store
                .set_genui_current_state("s", "p", 1, "state-1", "auth")
                .expect("initial state");
            store
                .set_genui_current_state("s", "p", 2, "state-2", "auth")
                .expect("advance");
            assert!(matches!(
                store.set_genui_current_state("s", "p", 1, "state-1", "auth"),
                Err(super::EventError::GenUiStateGenerationRewind)
            ));
        }
        let store = EventStore::open(&path).expect("reopen");
        assert_eq!(
            store
                .genui_current_state("s", "p")
                .expect("state")
                .expect("present")
                .0,
            2
        );
    }

    #[test]
    fn durable_genui_replay_rejects_invalid_kind_source_pairs() {
        fn payload(action_id: &str, action_kind: &str, source_type: &str) -> serde_json::Value {
            json!({
                "schema_version": 1,
                "action_id": action_id,
                "action_kind": action_kind,
                "source_type": source_type,
                "label_digest": "a".repeat(64),
                "surface_id": "surface",
                "surface_digest": "b".repeat(64),
                "state_digest": "c".repeat(64),
                "state_generation": 1,
                "session_id": "session",
                "principal": "principal",
                "authorization_context": "auth",
                "provider_id": "provider",
                "server_id": "server",
                "tool_name": "tool",
                "remote_tool_name": "remote",
                "schema_digest": "d".repeat(64),
                "policy_version": 1,
                "policy_digest": "e".repeat(64),
                "requires_confirmation": action_kind == "consequential"
            })
        }

        let invalid = [
            ("navigation", "mcp"),
            ("form_value_update", "mcp"),
            ("local_presentation", "mcp"),
            ("mcp_tool", "host_local"),
            ("auto", "mcp"),
            ("mcp_tool", "unknown_source"),
            ("unknown_kind", "mcp"),
        ];
        for (index, (kind, source)) in invalid.into_iter().enumerate() {
            let mut store = EventStore::open_memory().expect("store");
            let result = store.append_genui_lifecycle(
                "session",
                EventKind::GenUiActionPresented,
                &payload(&format!("action-{index}"), kind, source),
            );
            assert!(
                result.is_err(),
                "invalid pair unexpectedly persisted: {kind}/{source}"
            );
        }

        for (index, (kind, source)) in [("mcp_tool", "mcp"), ("navigation", "host_local")]
            .into_iter()
            .enumerate()
        {
            let mut store = EventStore::open_memory().expect("store");
            store
                .append_genui_lifecycle(
                    "session",
                    EventKind::GenUiActionPresented,
                    &payload(&format!("valid-{index}"), kind, source),
                )
                .expect("valid pair persists");
            assert_eq!(store.events("session").expect("events").len(), 1);
        }
    }
}