use fs2::FileExt;
use ring::digest::{digest, SHA256};
use serde::{Deserialize, Serialize};
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use tauri::AppHandle;
use uuid::Uuid;

const SESSION_SCHEMA: &str = "hololake.direct-local-session/v1";
const MAX_ID_BYTES: usize = 128;

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OpenSessionInput {
    pub account_id: String,
    pub lane_id: String,
    pub client_instance_id: String,
    pub discovery_ticket: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct IssueDiscoveryTicketInput {
    pub account_id: String,
    pub lane_id: String,
    pub client_instance_id: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DiscoveryTicketReceipt {
    pub schema: &'static str,
    pub state: &'static str,
    pub account_key: String,
    pub lane_id: String,
    pub client_instance_id: String,
    pub discovery_ticket: String,
    pub issued_at_unix_ms: u128,
    pub terminal_transport: &'static str,
    pub connector_executable: String,
    pub connector_arguments: Vec<&'static str>,
    pub heartbeat_interval_ms: u64,
    pub environment_refresh_interval_ms: u64,
    pub required_bootstrap_operations: Vec<&'static str>,
    pub receipt_id: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ResumeSessionInput {
    pub account_id: String,
    pub session_id: String,
    pub client_instance_id: String,
    pub resume_secret: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AppendSessionEventInput {
    pub account_id: String,
    pub session_id: String,
    pub resume_secret: String,
    pub idempotency_key: String,
    pub event_kind: String,
    pub payload_sha256: String,
    pub expected_previous_sequence: u64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AuthenticateSessionInput {
    pub account_id: String,
    pub session_id: String,
    pub resume_secret: String,
}

#[derive(Clone, Debug)]
pub(crate) struct AuthenticatedSessionContext {
    pub account_key: String,
    pub session_id: String,
    pub lane_id: String,
    pub client_instance_id: String,
    pub observed_at_unix_ms: u128,
    pub last_event_sequence: u64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DirectSessionReceipt {
    pub schema: &'static str,
    pub state: &'static str,
    pub account_key: String,
    pub session_id: String,
    pub lane_id: String,
    pub client_instance_id: String,
    pub opened_at_unix_ms: u128,
    pub observed_at_unix_ms: u128,
    pub last_event_sequence: u64,
    pub resume_secret: Option<String>,
    pub continuity_owner: &'static str,
    pub transport: &'static str,
    pub mcp_role: &'static str,
    pub required_next_operation: &'static str,
    pub receipt_id: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionEventReceipt {
    pub schema: &'static str,
    pub state: &'static str,
    pub session_id: String,
    pub sequence: u64,
    pub idempotency_key: String,
    pub event_kind: String,
    pub payload_sha256: String,
    pub observed_at_unix_ms: u128,
    pub receipt_id: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DirectSessionProjection {
    pub session_id: String,
    pub lane_id: String,
    pub client_instance_id: String,
    pub state: &'static str,
    pub opened_at_unix_ms: u128,
    pub observed_at_unix_ms: u128,
    pub last_event_sequence: u64,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct SessionRecord {
    schema: String,
    account_key: String,
    session_id: String,
    lane_id: String,
    client_instance_id: String,
    resume_secret_sha256: String,
    discovery_ticket_sha256: String,
    opened_at_unix_ms: u128,
    observed_at_unix_ms: u128,
    last_event_sequence: u64,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct EventRecord {
    schema: String,
    session_id: String,
    sequence: u64,
    idempotency_key: String,
    event_kind: String,
    payload_sha256: String,
    observed_at_unix_ms: u128,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct DiscoveryTicketRecord {
    schema: String,
    account_key: String,
    lane_id: String,
    client_instance_id: String,
    discovery_ticket_sha256: String,
    issued_at_unix_ms: u128,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct ActiveSessionRecord {
    schema: String,
    account_key: String,
    session_id: String,
    lane_id: String,
}

pub async fn issue_direct_local_discovery_ticket(
    app: AppHandle,
    input: IssueDiscoveryTicketInput,
) -> Result<DiscoveryTicketReceipt, String> {
    let root = direct_session_root(&app)?;
    tauri::async_runtime::spawn_blocking(move || issue_ticket_at(&root, input))
        .await
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}

pub async fn open_direct_local_session(
    app: AppHandle,
    input: OpenSessionInput,
) -> Result<DirectSessionReceipt, String> {
    let root = direct_session_root(&app)?;
    tauri::async_runtime::spawn_blocking(move || open_at(&root, input))
        .await
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}

pub async fn resume_direct_local_session(
    app: AppHandle,
    input: ResumeSessionInput,
) -> Result<DirectSessionReceipt, String> {
    let root = direct_session_root(&app)?;
    tauri::async_runtime::spawn_blocking(move || resume_at(&root, input))
        .await
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}

pub async fn append_direct_local_session_event(
    app: AppHandle,
    input: AppendSessionEventInput,
) -> Result<SessionEventReceipt, String> {
    let root = direct_session_root(&app)?;
    tauri::async_runtime::spawn_blocking(move || append_event_at(&root, input))
        .await
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}

pub async fn heartbeat_direct_local_session(
    app: AppHandle,
    input: AuthenticateSessionInput,
) -> Result<DirectSessionReceipt, String> {
    let root = direct_session_root(&app)?;
    tauri::async_runtime::spawn_blocking(move || heartbeat_at(&root, &input))
        .await
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_JOIN_FAILED: {error}"))?
}

pub(crate) fn direct_session_root(app: &AppHandle) -> Result<PathBuf, String> {
    let root = crate::authenticated_storage::account_storage_root(app, "direct-local-session-v1")?;
    fs::create_dir_all(&root)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    root.canonicalize()
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))
}

pub(crate) fn active_session_count_at(root: &Path) -> Result<usize, String> {
    let accounts = root.join("accounts");
    if !accounts.exists() {
        return Ok(0);
    }
    let mut count = 0;
    for entry in fs::read_dir(&accounts)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?
    {
        let entry = entry
            .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
        let kind = entry
            .file_type()
            .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
        if !kind.is_dir() || kind.is_symlink() {
            continue;
        }
        let active_path = entry.path().join("active-session.json");
        if !active_path.exists() {
            continue;
        }
        let metadata = fs::symlink_metadata(&active_path)
            .map_err(|error| format!("HOLOLAKE_ACTIVE_SESSION_INVALID: {error}"))?;
        if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
            return Err("HOLOLAKE_ACTIVE_SESSION_INVALID".into());
        }
        let active: ActiveSessionRecord = read_json(&active_path, "ACTIVE_SESSION")?;
        if active.schema != SESSION_SCHEMA
            || active.account_key != entry.file_name().to_string_lossy()
        {
            return Err("HOLOLAKE_ACTIVE_SESSION_INVALID".into());
        }
        count += 1;
    }
    Ok(count)
}

pub(crate) fn active_session_projections_at(
    root: &Path,
) -> Result<Vec<DirectSessionProjection>, String> {
    let accounts = root.join("accounts");
    if !accounts.exists() {
        return Ok(Vec::new());
    }
    let now = now_unix_ms()?;
    let mut projections = Vec::new();
    for entry in fs::read_dir(&accounts)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?
    {
        let entry = entry
            .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
        if !entry
            .file_type()
            .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?
            .is_dir()
        {
            continue;
        }
        let active_path = entry.path().join("active-session.json");
        if !active_path.exists() {
            continue;
        }
        let active: ActiveSessionRecord = read_json(&active_path, "ACTIVE_SESSION")?;
        let record = read_session(&session_path(root, &active.account_key, &active.session_id))?;
        let state = if now.saturating_sub(record.observed_at_unix_ms) <= 45_000 {
            "LIVE"
        } else {
            "RESUMABLE"
        };
        projections.push(DirectSessionProjection {
            session_id: record.session_id,
            lane_id: record.lane_id,
            client_instance_id: record.client_instance_id,
            state,
            opened_at_unix_ms: record.opened_at_unix_ms,
            observed_at_unix_ms: record.observed_at_unix_ms,
            last_event_sequence: record.last_event_sequence,
        });
    }
    projections.sort_by(|left, right| right.observed_at_unix_ms.cmp(&left.observed_at_unix_ms));
    Ok(projections)
}

pub(crate) fn open_at(
    root: &Path,
    input: OpenSessionInput,
) -> Result<DirectSessionReceipt, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.lane_id, "LANE")?;
    validate_identifier(&input.client_instance_id, "CLIENT")?;
    validate_secret(&input.discovery_ticket, "DISCOVERY_TICKET")?;

    let account_key = sha256_hex(input.account_id.as_bytes());
    let _account_lock = lock_account(root, &account_key)?;
    let active_path = active_session_path(root, &account_key);
    let tickets_root = root.join("discovery-tickets");
    let used_tickets_root = root.join("used-discovery-tickets");
    fs::create_dir_all(&used_tickets_root)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    let discovery_ticket_sha256 = sha256_hex(input.discovery_ticket.as_bytes());
    let ticket_path = tickets_root.join(format!("{discovery_ticket_sha256}.json"));
    let used_ticket_path = used_tickets_root.join(format!("{discovery_ticket_sha256}.json"));
    if used_ticket_path.exists() {
        return Err("HOLOLAKE_DISCOVERY_TICKET_ALREADY_USED".into());
    }
    if active_path.exists() {
        let active: ActiveSessionRecord = read_json(&active_path, "ACTIVE_SESSION")?;
        if active.schema != SESSION_SCHEMA || active.account_key != account_key {
            return Err("HOLOLAKE_ACTIVE_SESSION_INVALID".into());
        }
        return Err(format!(
            "HOLOLAKE_ACCOUNT_ALREADY_HAS_ACTIVE_SESSION:{}:{}",
            active.lane_id, active.session_id
        ));
    }
    let ticket: DiscoveryTicketRecord = read_json(&ticket_path, "DISCOVERY_TICKET")?;
    if ticket.account_key != account_key
        || ticket.lane_id != input.lane_id
        || ticket.client_instance_id != input.client_instance_id
        || ticket.discovery_ticket_sha256 != discovery_ticket_sha256
    {
        return Err("HOLOLAKE_DISCOVERY_TICKET_NOT_AUTHORIZED".into());
    }
    fs::rename(&ticket_path, &used_ticket_path).map_err(|error| {
        if error.kind() == std::io::ErrorKind::NotFound && used_ticket_path.exists() {
            "HOLOLAKE_DISCOVERY_TICKET_ALREADY_USED".to_string()
        } else {
            format!("HOLOLAKE_DISCOVERY_TICKET_CONSUME_FAILED: {error}")
        }
    })?;

    let session_id = Uuid::new_v4().to_string();
    let resume_secret = format!("{}{}", Uuid::new_v4(), Uuid::new_v4());
    let observed_at_unix_ms = now_unix_ms()?;
    let record = SessionRecord {
        schema: SESSION_SCHEMA.into(),
        account_key: account_key.clone(),
        session_id: session_id.clone(),
        lane_id: input.lane_id,
        client_instance_id: input.client_instance_id,
        resume_secret_sha256: sha256_hex(resume_secret.as_bytes()),
        discovery_ticket_sha256,
        opened_at_unix_ms: observed_at_unix_ms,
        observed_at_unix_ms,
        last_event_sequence: 0,
    };
    write_record_atomic(&session_path(root, &account_key, &session_id), &record)?;
    write_record_create_new(
        &active_path,
        &ActiveSessionRecord {
            schema: SESSION_SCHEMA.into(),
            account_key: account_key.clone(),
            session_id: session_id.clone(),
            lane_id: record.lane_id.clone(),
        },
        "ACTIVE_SESSION",
    )?;
    Ok(session_receipt("OPENED", &record, Some(resume_secret)))
}

pub(crate) fn issue_ticket_at(
    root: &Path,
    input: IssueDiscoveryTicketInput,
) -> Result<DiscoveryTicketReceipt, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.lane_id, "LANE")?;
    validate_identifier(&input.client_instance_id, "CLIENT")?;
    let account_key = sha256_hex(input.account_id.as_bytes());
    let discovery_ticket = format!("{}{}", Uuid::new_v4(), Uuid::new_v4());
    let discovery_ticket_sha256 = sha256_hex(discovery_ticket.as_bytes());
    let issued_at_unix_ms = now_unix_ms()?;
    let record = DiscoveryTicketRecord {
        schema: SESSION_SCHEMA.into(),
        account_key: account_key.clone(),
        lane_id: input.lane_id.clone(),
        client_instance_id: input.client_instance_id.clone(),
        discovery_ticket_sha256: discovery_ticket_sha256.clone(),
        issued_at_unix_ms,
    };
    let path = root
        .join("discovery-tickets")
        .join(format!("{discovery_ticket_sha256}.json"));
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)
            .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    }
    write_record_create_new(&path, &record, "DISCOVERY_TICKET")?;
    Ok(DiscoveryTicketReceipt {
        schema: SESSION_SCHEMA,
        state: "ISSUED",
        account_key,
        lane_id: input.lane_id,
        client_instance_id: input.client_instance_id,
        discovery_ticket,
        issued_at_unix_ms,
        terminal_transport: "HOLOLAKE_TERMINAL_LINK/3",
        connector_executable: std::env::current_exe()
            .map(|path| path.to_string_lossy().into_owned())
            .unwrap_or_else(|_| "HoloLake".into()),
        connector_arguments: vec!["--connector"],
        heartbeat_interval_ms: 15_000,
        environment_refresh_interval_ms: 30_000,
        required_bootstrap_operations: vec![
            "OPEN_SESSION",
            "ACQUIRE_DEVELOPMENT_WRITE_LANE",
            "GET_WORK_ENVIRONMENT",
        ],
        receipt_id: sha256_hex(
            format!("ISSUED\n{discovery_ticket_sha256}\n{issued_at_unix_ms}").as_bytes(),
        ),
    })
}

pub(crate) fn resume_at(
    root: &Path,
    input: ResumeSessionInput,
) -> Result<DirectSessionReceipt, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.session_id, "SESSION")?;
    validate_identifier(&input.client_instance_id, "CLIENT")?;
    validate_secret(&input.resume_secret, "RESUME_SECRET")?;
    let account_key = sha256_hex(input.account_id.as_bytes());
    require_active_session(root, &account_key, &input.session_id)?;
    let path = session_path(root, &account_key, &input.session_id);
    let _lock = lock_session(&path)?;
    let mut record = read_session(&path)?;
    authorize(
        &record,
        &account_key,
        &input.session_id,
        &input.resume_secret,
    )?;
    record.client_instance_id = input.client_instance_id;
    record.observed_at_unix_ms = now_unix_ms()?;
    write_record_atomic(&path, &record)?;
    Ok(session_receipt("RESUMED", &record, None))
}

pub(crate) fn append_event_at(
    root: &Path,
    input: AppendSessionEventInput,
) -> Result<SessionEventReceipt, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.session_id, "SESSION")?;
    validate_identifier(&input.idempotency_key, "IDEMPOTENCY")?;
    validate_identifier(&input.event_kind, "EVENT_KIND")?;
    validate_sha256(&input.payload_sha256)?;
    validate_secret(&input.resume_secret, "RESUME_SECRET")?;
    let account_key = sha256_hex(input.account_id.as_bytes());
    require_active_session(root, &account_key, &input.session_id)?;
    let path = session_path(root, &account_key, &input.session_id);
    let _lock = lock_session(&path)?;
    let mut record = read_session(&path)?;
    authorize(
        &record,
        &account_key,
        &input.session_id,
        &input.resume_secret,
    )?;

    let events_root = path
        .parent()
        .ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?
        .join("events");
    fs::create_dir_all(&events_root)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    let idempotency_path = events_root.join(format!("idempotency-{}.json", input.idempotency_key));
    if idempotency_path.exists() {
        let existing: EventRecord = read_json(&idempotency_path, "EVENT")?;
        if existing.event_kind == input.event_kind
            && existing.payload_sha256 == input.payload_sha256
        {
            return Ok(event_receipt("DUPLICATE_CONFIRMED", &existing));
        }
        return Err("HOLOLAKE_EVENT_IDEMPOTENCY_CONFLICT".into());
    }
    if input.expected_previous_sequence != record.last_event_sequence {
        return Err("HOLOLAKE_EVENT_CURSOR_CONFLICT".into());
    }

    let observed_at_unix_ms = now_unix_ms()?;
    let event = EventRecord {
        schema: SESSION_SCHEMA.into(),
        session_id: record.session_id.clone(),
        sequence: record.last_event_sequence + 1,
        idempotency_key: input.idempotency_key,
        event_kind: input.event_kind,
        payload_sha256: input.payload_sha256,
        observed_at_unix_ms,
    };
    write_record_create_new(&idempotency_path, &event, "EVENT")?;
    record.last_event_sequence = event.sequence;
    record.observed_at_unix_ms = observed_at_unix_ms;
    write_record_atomic(&path, &record)?;
    Ok(event_receipt("APPENDED", &event))
}

pub(crate) fn heartbeat_at(
    root: &Path,
    input: &AuthenticateSessionInput,
) -> Result<DirectSessionReceipt, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.session_id, "SESSION")?;
    validate_secret(&input.resume_secret, "RESUME_SECRET")?;
    let account_key = sha256_hex(input.account_id.as_bytes());
    require_active_session(root, &account_key, &input.session_id)?;
    let path = session_path(root, &account_key, &input.session_id);
    let _lock = lock_session(&path)?;
    let mut record = read_session(&path)?;
    authorize(
        &record,
        &account_key,
        &input.session_id,
        &input.resume_secret,
    )?;
    record.observed_at_unix_ms = now_unix_ms()?;
    write_record_atomic(&path, &record)?;
    Ok(session_receipt("HEARTBEAT_ACK", &record, None))
}

pub(crate) fn authenticate_with_lane_at(
    root: &Path,
    input: &AuthenticateSessionInput,
) -> Result<String, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.session_id, "SESSION")?;
    validate_secret(&input.resume_secret, "RESUME_SECRET")?;
    let account_key = sha256_hex(input.account_id.as_bytes());
    require_active_session(root, &account_key, &input.session_id)?;
    let path = session_path(root, &account_key, &input.session_id);
    let _lock = lock_session(&path)?;
    let record = read_session(&path)?;
    authorize(
        &record,
        &account_key,
        &input.session_id,
        &input.resume_secret,
    )?;
    Ok(record.lane_id)
}

pub(crate) fn authenticate_context_at(
    root: &Path,
    input: &AuthenticateSessionInput,
) -> Result<AuthenticatedSessionContext, String> {
    validate_identifier(&input.account_id, "ACCOUNT")?;
    validate_identifier(&input.session_id, "SESSION")?;
    validate_secret(&input.resume_secret, "RESUME_SECRET")?;
    let account_key = sha256_hex(input.account_id.as_bytes());
    require_active_session(root, &account_key, &input.session_id)?;
    let path = session_path(root, &account_key, &input.session_id);
    let _lock = lock_session(&path)?;
    let record = read_session(&path)?;
    authorize(
        &record,
        &account_key,
        &input.session_id,
        &input.resume_secret,
    )?;
    if record.lane_id == "visitor-expression-only" {
        return Err("HOLOLAKE_VISITOR_SESSION_HAS_NO_SYSTEM_AUTHORITY".into());
    }
    Ok(AuthenticatedSessionContext {
        account_key,
        session_id: record.session_id,
        lane_id: record.lane_id,
        client_instance_id: record.client_instance_id,
        observed_at_unix_ms: record.observed_at_unix_ms,
        last_event_sequence: record.last_event_sequence,
    })
}

fn session_path(root: &Path, account_key: &str, session_id: &str) -> PathBuf {
    root.join("accounts")
        .join(account_key)
        .join("sessions")
        .join(session_id)
        .join("session.json")
}

fn active_session_path(root: &Path, account_key: &str) -> PathBuf {
    root.join("accounts")
        .join(account_key)
        .join("active-session.json")
}

fn require_active_session(root: &Path, account_key: &str, session_id: &str) -> Result<(), String> {
    let active: ActiveSessionRecord =
        read_json(&active_session_path(root, account_key), "ACTIVE_SESSION")?;
    if active.schema != SESSION_SCHEMA
        || active.account_key != account_key
        || active.session_id != session_id
    {
        return Err("HOLOLAKE_SESSION_IS_NOT_ACCOUNT_ACTIVE_SESSION".into());
    }
    Ok(())
}

fn lock_account(root: &Path, account_key: &str) -> Result<std::fs::File, String> {
    lock_file(
        &root.join("accounts").join(account_key).join("account.lock"),
        "ACCOUNT",
    )
}

fn read_session(path: &Path) -> Result<SessionRecord, String> {
    let record: SessionRecord = read_json(path, "SESSION")?;
    if record.schema != SESSION_SCHEMA {
        return Err("HOLOLAKE_DIRECT_SESSION_SCHEMA_UNSUPPORTED".into());
    }
    Ok(record)
}

fn lock_session(path: &Path) -> Result<std::fs::File, String> {
    let parent = path
        .parent()
        .ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?;
    fs::create_dir_all(parent)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    lock_file(&parent.join("session.lock"), "SESSION")
}

fn lock_file(lock_path: &Path, kind: &str) -> Result<std::fs::File, String> {
    let parent = lock_path
        .parent()
        .ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?;
    fs::create_dir_all(parent)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(lock_path)
        .map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_LOCK_FAILED: {error}"))?;
    file.lock_exclusive()
        .map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_LOCK_FAILED: {error}"))?;
    Ok(file)
}

fn authorize(
    record: &SessionRecord,
    account_key: &str,
    session_id: &str,
    secret: &str,
) -> Result<(), String> {
    if record.account_key != account_key
        || record.session_id != session_id
        || record.resume_secret_sha256 != sha256_hex(secret.as_bytes())
    {
        return Err("HOLOLAKE_DIRECT_SESSION_NOT_AUTHORIZED".into());
    }
    Ok(())
}

fn write_record_atomic<T: Serialize>(path: &Path, record: &T) -> Result<(), String> {
    let parent = path
        .parent()
        .ok_or("HOLOLAKE_DIRECT_SESSION_PATH_INVALID")?;
    fs::create_dir_all(parent)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_STORAGE_UNAVAILABLE: {error}"))?;
    let temporary = parent.join(format!(".{}.tmp", Uuid::new_v4()));
    write_bytes_create_new(&temporary, record, "SESSION")?;
    fs::rename(&temporary, path)
        .map_err(|error| format!("HOLOLAKE_DIRECT_SESSION_WRITE_FAILED: {error}"))
}

fn write_record_create_new<T: Serialize>(
    path: &Path,
    record: &T,
    kind: &str,
) -> Result<(), String> {
    write_bytes_create_new(path, record, kind)
}

fn write_bytes_create_new<T: Serialize>(path: &Path, record: &T, kind: &str) -> Result<(), String> {
    let bytes = serde_json::to_vec_pretty(record)
        .map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_INVALID: {error}"))?;
    let mut file = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(path)
        .map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_WRITE_FAILED: {error}"))?;
    file.write_all(&bytes)
        .and_then(|_| file.sync_all())
        .map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_WRITE_FAILED: {error}"))
}

fn read_json<T: for<'de> Deserialize<'de>>(path: &Path, kind: &str) -> Result<T, String> {
    let bytes = fs::read(path).map_err(|error| {
        if error.kind() == std::io::ErrorKind::NotFound {
            format!("HOLOLAKE_DIRECT_{kind}_NOT_FOUND")
        } else {
            format!("HOLOLAKE_DIRECT_{kind}_UNREADABLE: {error}")
        }
    })?;
    serde_json::from_slice(&bytes)
        .map_err(|error| format!("HOLOLAKE_DIRECT_{kind}_INVALID: {error}"))
}

fn session_receipt(
    state: &'static str,
    record: &SessionRecord,
    resume_secret: Option<String>,
) -> DirectSessionReceipt {
    DirectSessionReceipt {
        schema: SESSION_SCHEMA,
        state,
        account_key: record.account_key.clone(),
        session_id: record.session_id.clone(),
        lane_id: record.lane_id.clone(),
        client_instance_id: record.client_instance_id.clone(),
        opened_at_unix_ms: record.opened_at_unix_ms,
        observed_at_unix_ms: record.observed_at_unix_ms,
        last_event_sequence: record.last_event_sequence,
        resume_secret,
        continuity_owner: "HOLOLAKE",
        transport: "HOLOLAKE_TERMINAL_LINK/3",
        mcp_role: "DISCOVERY_RECOVERY_COMPATIBILITY_ONLY",
        required_next_operation: "GET_WORK_ENVIRONMENT",
        receipt_id: sha256_hex(
            format!(
                "{state}\n{}\n{}\n{}\n{}",
                record.account_key, record.session_id, record.lane_id, record.observed_at_unix_ms
            )
            .as_bytes(),
        ),
    }
}

fn event_receipt(state: &'static str, event: &EventRecord) -> SessionEventReceipt {
    SessionEventReceipt {
        schema: SESSION_SCHEMA,
        state,
        session_id: event.session_id.clone(),
        sequence: event.sequence,
        idempotency_key: event.idempotency_key.clone(),
        event_kind: event.event_kind.clone(),
        payload_sha256: event.payload_sha256.clone(),
        observed_at_unix_ms: event.observed_at_unix_ms,
        receipt_id: sha256_hex(
            format!(
                "{}\n{}\n{}\n{}",
                event.session_id, event.sequence, event.idempotency_key, event.payload_sha256
            )
            .as_bytes(),
        ),
    }
}

fn validate_identifier(value: &str, kind: &str) -> Result<(), String> {
    if value.is_empty()
        || value.len() > MAX_ID_BYTES
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
    {
        return Err(format!("HOLOLAKE_{kind}_ID_INVALID"));
    }
    Ok(())
}

fn validate_secret(value: &str, kind: &str) -> Result<(), String> {
    if value.len() < 16 || value.len() > 512 {
        return Err(format!("HOLOLAKE_{kind}_INVALID"));
    }
    Ok(())
}

fn validate_sha256(value: &str) -> Result<(), String> {
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    {
        return Err("HOLOLAKE_PAYLOAD_SHA256_INVALID".into());
    }
    Ok(())
}

fn now_unix_ms() -> Result<u128, String> {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_millis())
        .map_err(|error| format!("HOLOLAKE_SYSTEM_CLOCK_INVALID: {error}"))
}

fn sha256_hex(value: &[u8]) -> String {
    digest(&SHA256, value)
        .as_ref()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn open_input(ticket: &str) -> OpenSessionInput {
        OpenSessionInput {
            account_id: "human-BS-0001".into(),
            lane_id: "DEV-001".into(),
            client_instance_id: "codex-1".into(),
            discovery_ticket: ticket.into(),
        }
    }

    fn issued_open_input(temp: &TempDir, suffix: &str) -> OpenSessionInput {
        let issued = issue_ticket_at(
            temp.path(),
            IssueDiscoveryTicketInput {
                account_id: "human-BS-0001".into(),
                lane_id: "DEV-001".into(),
                client_instance_id: "codex-1".into(),
            },
        )
        .unwrap();
        let mut input = open_input(&issued.discovery_ticket);
        input.client_instance_id = format!("codex-{suffix}");
        if suffix != "1" {
            let issued = issue_ticket_at(
                temp.path(),
                IssueDiscoveryTicketInput {
                    account_id: input.account_id.clone(),
                    lane_id: input.lane_id.clone(),
                    client_instance_id: input.client_instance_id.clone(),
                },
            )
            .unwrap();
            input.discovery_ticket = issued.discovery_ticket;
        }
        input
    }

    #[test]
    fn discovery_ticket_is_single_use_and_secret_is_not_stored_plaintext() {
        let temp = TempDir::new().unwrap();
        let input = issued_open_input(&temp, "1");
        let ticket = input.discovery_ticket.clone();
        let opened = open_at(temp.path(), input).unwrap();
        assert_eq!(opened.state, "OPENED");
        assert!(open_at(temp.path(), open_input(&ticket))
            .unwrap_err()
            .contains("ALREADY_USED"));
        let stored = fs::read_to_string(session_path(
            temp.path(),
            &opened.account_key,
            &opened.session_id,
        ))
        .unwrap();
        assert!(!stored.contains(opened.resume_secret.as_ref().unwrap()));
    }

    #[test]
    fn ai_restart_resumes_the_same_session_and_cursor() {
        let temp = TempDir::new().unwrap();
        let opened = open_at(temp.path(), issued_open_input(&temp, "2")).unwrap();
        let event = append_event_at(
            temp.path(),
            AppendSessionEventInput {
                account_id: "human-BS-0001".into(),
                session_id: opened.session_id.clone(),
                resume_secret: opened.resume_secret.clone().unwrap(),
                idempotency_key: "event-1".into(),
                event_kind: "TASK_OBSERVED".into(),
                payload_sha256: "a".repeat(64),
                expected_previous_sequence: 0,
            },
        )
        .unwrap();
        assert_eq!(event.sequence, 1);
        let resumed = resume_at(
            temp.path(),
            ResumeSessionInput {
                account_id: "human-BS-0001".into(),
                session_id: opened.session_id,
                client_instance_id: "codex-2".into(),
                resume_secret: opened.resume_secret.unwrap(),
            },
        )
        .unwrap();
        assert_eq!(resumed.state, "RESUMED");
        assert_eq!(resumed.last_event_sequence, 1);
        assert_eq!(resumed.client_instance_id, "codex-2");
    }

    #[test]
    fn event_retries_are_idempotent_but_conflicts_fail_closed() {
        let temp = TempDir::new().unwrap();
        let opened = open_at(temp.path(), issued_open_input(&temp, "3")).unwrap();
        let secret = opened.resume_secret.unwrap();
        let input = AppendSessionEventInput {
            account_id: "human-BS-0001".into(),
            session_id: opened.session_id,
            resume_secret: secret,
            idempotency_key: "event-1".into(),
            event_kind: "TASK_OBSERVED".into(),
            payload_sha256: "b".repeat(64),
            expected_previous_sequence: 0,
        };
        assert_eq!(
            append_event_at(temp.path(), input.clone()).unwrap().state,
            "APPENDED"
        );
        assert_eq!(
            append_event_at(temp.path(), input.clone()).unwrap().state,
            "DUPLICATE_CONFIRMED"
        );
        let conflicting = AppendSessionEventInput {
            payload_sha256: "c".repeat(64),
            ..input
        };
        assert_eq!(
            append_event_at(temp.path(), conflicting).unwrap_err(),
            "HOLOLAKE_EVENT_IDEMPOTENCY_CONFLICT"
        );
    }

    #[test]
    fn wrong_secret_and_stale_cursor_fail_closed() {
        let temp = TempDir::new().unwrap();
        let opened = open_at(temp.path(), issued_open_input(&temp, "4")).unwrap();
        let wrong = resume_at(
            temp.path(),
            ResumeSessionInput {
                account_id: "human-BS-0001".into(),
                session_id: opened.session_id.clone(),
                client_instance_id: "codex-2".into(),
                resume_secret: "wrong-resume-secret-0000".into(),
            },
        );
        assert_eq!(wrong.unwrap_err(), "HOLOLAKE_DIRECT_SESSION_NOT_AUTHORIZED");
        let stale = append_event_at(
            temp.path(),
            AppendSessionEventInput {
                account_id: "human-BS-0001".into(),
                session_id: opened.session_id,
                resume_secret: opened.resume_secret.unwrap(),
                idempotency_key: "event-2".into(),
                event_kind: "TASK_OBSERVED".into(),
                payload_sha256: "d".repeat(64),
                expected_previous_sequence: 9,
            },
        );
        assert_eq!(stale.unwrap_err(), "HOLOLAKE_EVENT_CURSOR_CONFLICT");
    }

    #[test]
    fn caller_cannot_invent_a_discovery_ticket() {
        let temp = TempDir::new().unwrap();
        assert_eq!(
            open_at(temp.path(), open_input("invented-ticket-0000")).unwrap_err(),
            "HOLOLAKE_DIRECT_DISCOVERY_TICKET_NOT_FOUND"
        );
    }

    #[test]
    fn one_account_cannot_open_two_direct_write_sessions() {
        let temp = TempDir::new().unwrap();
        let first = open_at(temp.path(), issued_open_input(&temp, "1")).unwrap();
        let second = open_at(temp.path(), issued_open_input(&temp, "2")).unwrap_err();
        assert!(second.contains("ACCOUNT_ALREADY_HAS_ACTIVE_SESSION"));
        assert!(second.contains(&first.session_id));
    }
}