eidetic-engine 0.15.2

//! JSONL import execution (EE-222).
//!
//! The import path consumes EE JSONL export records, validates their schemas,
//! and imports memories, tags, and their relationships into the local workspace
//! database. Other record families are counted but are not replayed here.

use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};
use std::str::FromStr;

use serde_json::{Value as JsonValue, json};
use uuid::Uuid;

use crate::db::{
    CreateAuditInput, CreateMemoryInput, CreateMemoryLinkInput, CreateSearchIndexJobInput,
    DatabaseConfig, DbConnection, DbError, DbOperation, MemoryLinkRelation, MemoryLinkSource,
    SearchIndexJobType, StoredMemory, StoredMemoryLink,
};
use crate::models::{
    EXPORT_AGENT_SCHEMA_V1, EXPORT_ARTIFACT_SCHEMA_V1, EXPORT_AUDIT_SCHEMA_V1,
    EXPORT_FOOTER_SCHEMA_V1, EXPORT_HEADER_SCHEMA_V1, EXPORT_LINK_SCHEMA_V1,
    EXPORT_MEMORY_SCHEMA_V1, EXPORT_TAG_SCHEMA_V1, EXPORT_WORKSPACE_SCHEMA_V1, ExportFooter,
    ExportHeader, ExportLinkRecord, ExportMemoryRecord, ExportTagRecord, IMPORT_JSONL_SCHEMA_V1,
    ImportSource, MemoryContent, MemoryId, MemoryKind, MemoryLevel, MemoryLinkId, RedactionLevel,
    Tag, TrustClass, TrustLevel, UnitScore,
};
use crate::policy::import_auth::{
    ArtifactContext, EXPORT_ARTIFACT_FAMILY, EXPORT_RECORD_ENCODING_V1, ImportAuthOutcome,
    RecordsRootBuilder, STORE_KEY_NAMESPACE_V1, canonical_record_hash, verify_artifact,
};
use crate::policy::store_auth::{
    MESH_STORE_AUTHENTICATION_UNAVAILABLE_CODE, MacDomain, StoreAuthError, StoreAuthRoot,
    workspace_keys_dir,
};

/// Issue code emitted when a native-source artifact claims `human_explicit`
/// trust but its footer does not authenticate under this store's key
/// (ADR 0086 TC-D14). Closes the spoofable `import_source=native` bypass.
pub const UNAUTHENTICATED_NATIVE_IMPORT_TRUST_CODE: &str = "unauthenticated_native_import_trust";
/// Issue emitted when a verified backup is restored into a fresh store and a
/// source `human_explicit` row is deliberately capped at `agent_validated`.
pub const VERIFIED_BACKUP_TRUST_DOWNGRADED_CODE: &str = "verified_backup_trust_downgraded";
/// JSONL artifacts cannot establish the signed active-member origin required
/// to mint `peer_human_attested`, even when their store-local MAC is valid.
pub const PEER_HUMAN_ATTESTED_IMPORT_PATH_REQUIRED_CODE: &str =
    "peer_human_attested_requires_team_import_path";

const DEFAULT_DB_FILE: &str = "ee.db";
pub(crate) const IMPORT_ACTION: &str = "memory.import.jsonl";

/// Hard cap on the byte length of an `ee import jsonl --source-path` file.
///
/// `import_jsonl_records` previously called `fs::read_to_string(source_path)`
/// directly, which has no upper bound: a multi-GB JSONL file (whether
/// authored maliciously, accumulated from a long-running export, or handed
/// off by another agent) would be slurped into a `String` in one allocation
/// before `parse_jsonl_source` ever ran. That allocation could OOM the
/// process under disk-pressure on the dev host or trip the swap thrashing
/// path described in feedback_hung_subprocess_paralyzes_agent — a benign
/// `ee import jsonl <path>` then becomes a local denial-of-service against
/// the agent that ran it.
///
/// 256 MiB is the same order-of-magnitude as other bulk-import surfaces
/// (e.g. the 4 MiB `.ee/config.toml` reads under `src/core/curate.rs`,
/// `src/config/path_resolver.rs`, etc., scaled up because a JSONL export
/// is bulk material rather than a single config record). At 1 KiB per
/// memory record this is ~262_000 records; at the 65_536-byte memory
/// content CHECK ceiling it is ~4_000 records — generous for ordinary
/// workspace bulk-import. Users with larger exports should stream them in
/// chunks rather than rely on a single mmap-style read.
pub const JSONL_IMPORT_MAX_INPUT_BYTES: u64 = 256 * 1024 * 1024;

/// Options for one `ee import jsonl` run.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JsonlImportOptions {
    pub workspace_path: PathBuf,
    pub database_path: Option<PathBuf>,
    pub source_path: PathBuf,
    pub dry_run: bool,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum NativeTrustPolicy {
    /// Ordinary JSONL imports may preserve `human_explicit` only when the
    /// artifact authenticates under this store and workspace.
    StoreAuthenticatedOnly,
    /// A backup that has already passed manifest and artifact verification may
    /// restore foreign-store rows, but never imports their `human_explicit`
    /// claim. Those rows are capped at the header-derived trust class.
    VerifiedBackupRestore,
}

/// Stable issue severity for JSONL import diagnostics.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum JsonlImportIssueSeverity {
    Info,
    Error,
    Warning,
}

impl JsonlImportIssueSeverity {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Info => "info",
            Self::Error => "error",
            Self::Warning => "warning",
        }
    }
}

/// Validation or import diagnostic for one JSONL record.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JsonlImportIssue {
    pub line: Option<u32>,
    pub code: String,
    pub severity: JsonlImportIssueSeverity,
    pub message: String,
    pub repair: Option<String>,
}

impl JsonlImportIssue {
    fn info(line: Option<u32>, code: &str, message: impl Into<String>) -> Self {
        Self {
            line,
            code: code.to_owned(),
            severity: JsonlImportIssueSeverity::Info,
            message: message.into(),
            repair: None,
        }
    }

    fn error(line: Option<u32>, code: &str, message: impl Into<String>) -> Self {
        let message: String = message.into();
        Self {
            line,
            code: code.to_owned(),
            severity: JsonlImportIssueSeverity::Error,
            message: crate::policy::redact_secret_like_content(&message).content,
            repair: None,
        }
    }

    fn warning(line: Option<u32>, code: &str, message: impl Into<String>) -> Self {
        Self {
            line,
            code: code.to_owned(),
            severity: JsonlImportIssueSeverity::Warning,
            message: message.into(),
            repair: None,
        }
    }
}

/// Error returned by the narrow JSONL header parser used by import validation
/// and fuzzing.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum JsonlHeaderParseError {
    EmptyLine,
    InvalidJson { message: String },
    MissingSchema,
    WrongSchema { schema: String },
    InvalidHeader { message: String },
}

impl fmt::Display for JsonlHeaderParseError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyLine => formatter.write_str("JSONL header line is empty"),
            Self::InvalidJson { message } => {
                write!(formatter, "invalid JSONL header JSON: {message}")
            }
            Self::MissingSchema => {
                formatter.write_str("JSONL header is missing a non-empty schema field")
            }
            Self::WrongSchema { schema } => write!(
                formatter,
                "JSONL header schema must be {EXPORT_HEADER_SCHEMA_V1}, got {schema}"
            ),
            Self::InvalidHeader { message } => write!(formatter, "invalid JSONL header: {message}"),
        }
    }
}

/// Parse one JSONL header line.
///
/// This is intentionally smaller than [`import_jsonl_records`]: fuzzing should
/// exercise the record parser directly without opening files or databases.
pub fn parse_jsonl_header(input: &str) -> Result<ExportHeader, JsonlHeaderParseError> {
    let trimmed = input.trim();
    if trimmed.is_empty() {
        return Err(JsonlHeaderParseError::EmptyLine);
    }

    let value = serde_json::from_str::<JsonValue>(trimmed).map_err(|error| {
        JsonlHeaderParseError::InvalidJson {
            message: error.to_string(),
        }
    })?;
    let schema = value
        .get("schema")
        .and_then(JsonValue::as_str)
        .map(str::trim)
        .filter(|schema| !schema.is_empty())
        .ok_or(JsonlHeaderParseError::MissingSchema)?;

    if schema != EXPORT_HEADER_SCHEMA_V1 {
        return Err(JsonlHeaderParseError::WrongSchema {
            schema: schema.to_owned(),
        });
    }

    let header = serde_json::from_value::<ExportHeader>(value).map_err(|error| {
        JsonlHeaderParseError::InvalidHeader {
            message: error.to_string(),
        }
    })?;
    validate_export_header_required_fields(&header)
        .map_err(|message| JsonlHeaderParseError::InvalidHeader { message })?;
    Ok(header)
}

/// Summary returned by `ee import jsonl`.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JsonlImportReport {
    pub schema: &'static str,
    pub workspace_path: String,
    pub database_path: Option<String>,
    pub source_path: String,
    pub source_id: String,
    pub dry_run: bool,
    pub status: String,
    pub header: Option<JsonlImportHeaderSummary>,
    pub footer: Option<JsonlImportFooterSummary>,
    pub records_total: u32,
    pub memory_records: u32,
    pub tag_records: u32,
    pub link_records: u32,
    pub ignored_records: u32,
    pub memories_imported: u32,
    pub memories_skipped_duplicate: u32,
    pub tags_imported: u32,
    pub links_imported: u32,
    pub links_skipped_duplicate: u32,
    pub links_skipped_conflict: u32,
    pub imported_memory_ids: Vec<String>,
    pub issues: Vec<JsonlImportIssue>,
}

impl JsonlImportReport {
    #[must_use]
    pub fn data_json(&self) -> JsonValue {
        json!({
            "schema": self.schema,
            "command": "import jsonl",
            "workspacePath": self.workspace_path,
            "databasePath": self.database_path,
            "sourcePath": redact_jsonl_import_source_ref(&self.source_path),
            "sourceId": redact_jsonl_import_source_ref(&self.source_id),
            "dryRun": self.dry_run,
            "status": self.status,
            "header": self.header.as_ref().map(JsonlImportHeaderSummary::data_json),
            "footer": self.footer.as_ref().map(JsonlImportFooterSummary::data_json),
            "recordsTotal": self.records_total,
            "memoryRecords": self.memory_records,
            "tagRecords": self.tag_records,
            "linkRecords": self.link_records,
            "ignoredRecords": self.ignored_records,
            "memoriesImported": self.memories_imported,
            "memoriesSkippedDuplicate": self.memories_skipped_duplicate,
            "tagsImported": self.tags_imported,
            "linksImported": self.links_imported,
            "linksSkippedDuplicate": self.links_skipped_duplicate,
            "linksSkippedConflict": self.links_skipped_conflict,
            "importedMemoryIds": self.imported_memory_ids,
            "issues": self.issues.iter().map(|issue| {
                json!({
                    "line": issue.line,
                    "code": issue.code,
                    "severity": issue.severity.as_str(),
                    "message": issue.message,
                    "repair": issue.repair,
                })
            }).collect::<Vec<_>>(),
        })
    }

    /// Response-level degradations caused by a completed import whose derived
    /// search-index publication did not converge. Validation and conflict
    /// diagnostics remain in `data.issues[]`; they are not silently promoted
    /// into unrelated response degradations.
    #[must_use]
    pub fn degraded_json(&self) -> Vec<JsonValue> {
        self.issues
            .iter()
            .filter(|issue| issue.code == "import_index_publish_failed")
            .map(|issue| {
                json!({
                    "code": issue.code,
                    "severity": issue.severity.as_str(),
                    "message": issue.message,
                    "repair": issue.repair,
                })
            })
            .collect()
    }

    #[must_use]
    pub fn human_summary(&self) -> String {
        let mode = if self.dry_run { "DRY RUN: " } else { "" };
        let mut summary = format!(
            "{mode}JSONL import {status}: {imported} memories imported, {skipped} duplicates, {links} links imported, {link_conflicts} link conflicts, {issues} issue(s) from {memories} memory record(s)\n",
            status = self.status,
            imported = self.memories_imported,
            skipped = self.memories_skipped_duplicate,
            issues = self.issues.len(),
            memories = self.memory_records,
            links = self.links_imported,
            link_conflicts = self.links_skipped_conflict,
        );
        for issue in self
            .issues
            .iter()
            .filter(|issue| issue.code == "import_index_publish_failed")
        {
            let message = crate::output::escape_json_string(&issue.message);
            summary.push_str(&format!(
                "  [{}] {}: {}\n",
                issue.severity.as_str(),
                issue.code,
                message
            ));
            if let Some(repair) = issue.repair.as_deref() {
                summary.push_str(&format!("    Repair: {repair}\n"));
            }
        }
        summary
    }
}

fn redact_jsonl_import_source_ref(value: &str) -> String {
    let secret_redacted = crate::policy::redact_secret_like_content(value).content;
    redact_jsonl_import_source_path_segments(&secret_redacted)
}

fn redact_jsonl_import_source_path_segments(value: &str) -> String {
    let mut output = String::with_capacity(value.len());
    let mut cursor = 0;
    while cursor < value.len() {
        let Some((relative_index, _)) = value[cursor..]
            .char_indices()
            .find(|(_, c)| jsonl_import_source_path_separator(*c))
        else {
            output.push_str(&value[cursor..]);
            break;
        };
        let start = cursor + relative_index;
        let Some(redaction_start) = jsonl_import_source_path_redaction_start(value, start) else {
            output.push_str(&value[cursor..=start]);
            cursor = start + 1;
            continue;
        };

        output.push_str(&value[cursor..redaction_start]);
        output.push_str("[REDACTED_PATH]");
        cursor = value[redaction_start..]
            .char_indices()
            .find_map(|(index, c)| {
                jsonl_import_source_path_boundary(c).then_some(redaction_start + index)
            })
            .unwrap_or(value.len());
    }
    output
}

fn jsonl_import_source_path_separator(c: char) -> bool {
    matches!(c, '/' | '\\')
}

fn jsonl_import_source_path_redaction_start(value: &str, separator_start: usize) -> Option<usize> {
    let candidate = &value[separator_start..];
    if jsonl_import_source_path_starts_sensitive_unix_segment(candidate)
        || jsonl_import_source_path_starts_sensitive_windows_segment(candidate)
        || jsonl_import_source_path_starts_unc_path(candidate)
    {
        return Some(
            jsonl_import_source_path_windows_drive_start(value, separator_start)
                .unwrap_or(separator_start),
        );
    }
    None
}

fn jsonl_import_source_path_starts_sensitive_unix_segment(value: &str) -> bool {
    const PREFIXES: &[&str] = &[
        "/Users/",
        "/Volumes/",
        "/private/",
        "/var/",
        "/tmp/",
        "/home/",
        "/data/",
        "/dp/",
        "/workspace/",
        "/repo/",
        "/etc/",
    ];

    PREFIXES.iter().any(|prefix| value.starts_with(prefix))
}

fn jsonl_import_source_path_starts_sensitive_windows_segment(value: &str) -> bool {
    const PREFIXES: &[&str] = &[
        "\\Users\\",
        "\\Volumes\\",
        "\\private\\",
        "\\var\\",
        "\\tmp\\",
        "\\home\\",
        "\\data\\",
        "\\dp\\",
        "\\workspace\\",
        "\\repo\\",
        "\\etc\\",
    ];

    PREFIXES.iter().any(|prefix| value.starts_with(prefix))
}

fn jsonl_import_source_path_starts_unc_path(value: &str) -> bool {
    value.starts_with("\\\\")
}

fn jsonl_import_source_path_windows_drive_start(
    value: &str,
    separator_start: usize,
) -> Option<usize> {
    if separator_start < 2 {
        return None;
    }
    let bytes = value.as_bytes();
    let drive_start = separator_start - 2;
    if !bytes[drive_start].is_ascii_alphabetic() || bytes[drive_start + 1] != b':' {
        return None;
    }
    if drive_start == 0 {
        return Some(drive_start);
    }
    let previous = value[..drive_start].chars().next_back()?;
    jsonl_import_source_path_start_boundary(previous).then_some(drive_start)
}

fn jsonl_import_source_path_start_boundary(c: char) -> bool {
    c.is_whitespace() || matches!(c, '/' | '\\' | '(' | '[' | '{' | '"' | '\'' | '<' | '=')
}

fn jsonl_import_source_path_boundary(c: char) -> bool {
    c.is_whitespace()
        || matches!(
            c,
            '?' | '#' | '"' | '\'' | ')' | ']' | '}' | ',' | ';' | '<' | '>' | '`'
        )
}

/// Stable subset of header metadata exposed by import reports.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JsonlImportHeaderSummary {
    pub export_id: String,
    pub format_version: u32,
    pub export_scope: String,
    pub redaction_level: String,
    pub import_source: String,
    pub trust_level: String,
    pub source_schema_version: Option<String>,
    pub checksum_status: String,
}

impl JsonlImportHeaderSummary {
    fn from_header(header: &ExportHeader) -> Self {
        Self {
            export_id: header.export_id.clone(),
            format_version: header.format_version,
            export_scope: header.export_scope.as_str().to_owned(),
            redaction_level: header.redaction_level.as_str().to_owned(),
            import_source: header.import_source.as_str().to_owned(),
            trust_level: header.trust_level.as_str().to_owned(),
            source_schema_version: header.source_schema_version.clone(),
            checksum_status: if header.checksum.is_some() {
                "present_unverified".to_owned()
            } else {
                "absent".to_owned()
            },
        }
    }

    fn data_json(&self) -> JsonValue {
        json!({
            "exportId": self.export_id,
            "formatVersion": self.format_version,
            "exportScope": self.export_scope,
            "redactionLevel": self.redaction_level,
            "importSource": self.import_source,
            "trustLevel": self.trust_level,
            "sourceSchemaVersion": self.source_schema_version,
            "checksumStatus": self.checksum_status,
        })
    }
}

/// Stable subset of footer metadata exposed by import reports.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JsonlImportFooterSummary {
    pub export_id: String,
    pub total_records: u64,
    pub memory_count: u64,
    pub artifact_count: u64,
    pub tag_count: u64,
    pub success: bool,
}

impl JsonlImportFooterSummary {
    fn from_footer(footer: &ExportFooter) -> Self {
        Self {
            export_id: footer.export_id.clone(),
            total_records: footer.total_records,
            memory_count: footer.memory_count,
            artifact_count: footer.artifact_count,
            tag_count: footer.tag_count,
            success: footer.success,
        }
    }

    fn data_json(&self) -> JsonValue {
        json!({
            "exportId": self.export_id,
            "totalRecords": self.total_records,
            "memoryCount": self.memory_count,
            "artifactCount": self.artifact_count,
            "tagCount": self.tag_count,
            "success": self.success,
        })
    }
}

/// Error produced by JSONL import setup.
#[derive(Debug)]
pub enum JsonlImportError {
    Io { path: PathBuf, message: String },
    Storage(DbError),
}

impl JsonlImportError {
    #[must_use]
    pub const fn repair_hint(&self) -> Option<&'static str> {
        match self {
            Self::Io { .. } => Some("check the JSONL source path and workspace permissions"),
            Self::Storage(_) => {
                Some("ee init --workspace . && ee migrate run --workspace . --json")
            }
        }
    }
}

impl fmt::Display for JsonlImportError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io { path, message } => {
                write!(formatter, "I/O error at {}: {message}", path.display())
            }
            Self::Storage(error) => write!(formatter, "{error}"),
        }
    }
}

impl std::error::Error for JsonlImportError {}

impl From<DbError> for JsonlImportError {
    fn from(error: DbError) -> Self {
        Self::Storage(error)
    }
}

struct ParsedJsonlImport {
    header: Option<ExportHeader>,
    footer: Option<ExportFooter>,
    footer_line: Option<u32>,
    memories: Vec<ExportMemoryRecord>,
    links: Vec<ExportLinkRecord>,
    tags_by_memory: BTreeMap<String, BTreeSet<String>>,
    tag_lines_by_memory: BTreeMap<String, u32>,
    artifact_records: u32,
    tag_records: u32,
    issues: Vec<JsonlImportIssue>,
    records_total: u32,
    ignored_records: u32,
    /// Ordered digest over raw memory, tag, and link line bytes, matching what
    /// the exporter MAC'd (ADR 0086 TC-D14). Verified against the footer
    /// authentication block before native trust is honored.
    records_root: RecordsRootBuilder,
}

impl ParsedJsonlImport {
    fn has_errors(&self) -> bool {
        self.issues
            .iter()
            .any(|issue| issue.severity == JsonlImportIssueSeverity::Error)
    }
}

struct PreparedMemory {
    id: String,
    logical_id: String,
    input: CreateMemoryInput,
    created_at: String,
    updated_at: String,
    tombstoned_at: Option<String>,
    tombstoned_reason: Option<String>,
    bayes_posterior: Option<(f64, f64)>,
    /// bd-multiplicity-aware-trust-p0u7g: attempt-family block restored into
    /// the pointer columns and the family ledger after the memory row lands.
    attempt_family: Option<crate::models::ExportAttemptFamilyRecord>,
    details: String,
    tag_count: u32,
}

/// Store-independent validation shared by dry-run and applied imports. Trust
/// authentication and workspace binding happen only after these fields pass.
struct ValidatedMemory<'a> {
    record: &'a ExportMemoryRecord,
    id: String,
    logical_id: String,
    level: MemoryLevel,
    kind: MemoryKind,
    content: MemoryContent,
    confidence: Option<f32>,
    utility: f32,
    importance: f32,
    bayes_posterior: Option<(f64, f64)>,
}

struct PreparedLink {
    id: String,
    input: CreateMemoryLinkInput,
    created_at: String,
    details: String,
}

/// Fields carried in the exporter's link metadata envelope. Missing fields in
/// minimal/redacted records use the storage defaults; explicit invalid values
/// are rejected rather than silently replaced.
#[derive(Default, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct ImportedLinkMetadata {
    confidence: Option<f64>,
    directed: Option<bool>,
    evidence_count: Option<u32>,
    last_reinforced_at: Option<String>,
    source: Option<String>,
    created_by: Option<String>,
    metadata: Option<JsonValue>,
}

fn prepare_links(parsed: &ParsedJsonlImport) -> Result<Vec<PreparedLink>, Vec<JsonlImportIssue>> {
    if parsed.links.is_empty() {
        return Ok(Vec::new());
    }
    let redaction = parsed
        .header
        .as_ref()
        .map_or(RedactionLevel::None, |header| header.redaction_level);
    let memory_ids = parsed
        .memories
        .iter()
        .map(|memory| import_memory_id(memory, redaction).map(|id| (memory.memory_id.as_str(), id)))
        .collect::<Result<BTreeMap<_, _>, _>>()
        .map_err(|issue| vec![issue])?;
    let mut links = Vec::with_capacity(parsed.links.len());
    let mut issues = Vec::new();
    let mut seen_ids = BTreeSet::new();
    let mut seen_edges = BTreeSet::new();
    for record in &parsed.links {
        match prepare_link(record, &memory_ids, redaction, parsed.header.as_ref()) {
            Ok(link) => {
                let edge = (
                    link.input.src_memory_id.clone(),
                    link.input.dst_memory_id.clone(),
                    link.input.relation.as_str(),
                );
                if !seen_ids.insert(link.id.clone()) || !seen_edges.insert(edge) {
                    issues.push(JsonlImportIssue::error(
                        None,
                        "duplicate_link_record",
                        format!(
                            "link `{}` duplicates an ID or ordered endpoint/relation key",
                            record.link_id
                        ),
                    ));
                } else {
                    links.push(link);
                }
            }
            Err(message) => issues.push(JsonlImportIssue::error(
                None,
                "invalid_link_record",
                format!("link `{}`: {message}", record.link_id),
            )),
        }
    }
    if issues.is_empty() {
        Ok(links)
    } else {
        Err(issues)
    }
}

fn prepare_link(
    record: &ExportLinkRecord,
    memory_ids: &BTreeMap<&str, String>,
    redaction: RedactionLevel,
    header: Option<&ExportHeader>,
) -> Result<PreparedLink, String> {
    let src_memory_id = memory_ids
        .get(record.source_memory_id.as_str())
        .ok_or("source endpoint is absent from the archive's memory records")?
        .clone();
    let dst_memory_id = memory_ids
        .get(record.target_memory_id.as_str())
        .ok_or("target endpoint is absent from the archive's memory records")?
        .clone();
    if src_memory_id == dst_memory_id {
        return Err("a memory link cannot connect a memory to itself".to_owned());
    }
    let id = match record.link_id.parse::<MemoryLinkId>() {
        Ok(_) => record.link_id.clone(),
        Err(_) if redaction.redacts_identifiers() && !record.link_id.trim().is_empty() => {
            MemoryLinkId::from_uuid(stable_uuid(&format!(
                "jsonl-redacted-link:{}:{}:{}:{}:{}",
                record.link_id, src_memory_id, dst_memory_id, record.link_type, record.created_at
            )))
            .to_string()
        }
        Err(error) => return Err(format!("invalid link ID: {error}")),
    };
    let relation = MemoryLinkRelation::parse(&record.link_type)
        .ok_or_else(|| format!("unsupported relation `{}`", record.link_type))?;
    chrono::DateTime::parse_from_rfc3339(&record.created_at)
        .map_err(|error| format!("invalid created_at: {error}"))?;
    let metadata = record
        .metadata
        .as_ref()
        .filter(|value| !value.is_null())
        .map_or_else(
            || Ok(ImportedLinkMetadata::default()),
            |value| serde_json::from_value::<ImportedLinkMetadata>(value.clone()),
        )
        .map_err(|error| format!("invalid link metadata: {error}"))?;
    if let Some(timestamp) = &metadata.last_reinforced_at {
        chrono::DateTime::parse_from_rfc3339(timestamp)
            .map_err(|error| format!("invalid lastReinforcedAt: {error}"))?;
    }
    if metadata
        .created_by
        .as_ref()
        .is_some_and(|value| value.trim().is_empty())
    {
        return Err("createdBy must not be blank".to_owned());
    }
    let source = metadata
        .source
        .as_deref()
        .map_or(Ok(MemoryLinkSource::Import), |source| {
            MemoryLinkSource::parse(source).ok_or_else(|| format!("invalid source `{source}`"))
        })?;
    let record_json = serde_json::to_string(record).map_err(|error| error.to_string())?;
    if crate::policy::redact_secret_like_content(&record_json).redacted {
        return Err("link contains secrets; redact before import".to_owned());
    }
    Ok(PreparedLink {
        id,
        input: CreateMemoryLinkInput {
            src_memory_id,
            dst_memory_id,
            relation,
            weight: score_or_default(record.weight, 1.0)?,
            confidence: score_or_default(metadata.confidence, 1.0)?,
            directed: metadata.directed.unwrap_or(true),
            evidence_count: metadata.evidence_count.unwrap_or(1),
            last_reinforced_at: metadata.last_reinforced_at,
            source,
            created_by: metadata.created_by,
            metadata_json: metadata
                .metadata
                .filter(|value| !value.is_null())
                .map(|value| value.to_string()),
        },
        created_at: record.created_at.clone(),
        details: json!({
            "source": "jsonl_import",
            "sourceExportId": header.map(|header| &header.export_id),
            "sourceLinkId": record.link_id,
            "sourceRecord": record,
        })
        .to_string(),
    })
}

fn link_matches(existing: &StoredMemoryLink, incoming: &PreparedLink) -> bool {
    let input = &incoming.input;
    existing.src_memory_id == input.src_memory_id
        && existing.dst_memory_id == input.dst_memory_id
        && existing.relation == input.relation.as_str()
        && existing.weight == input.weight
        && existing.confidence == input.confidence
        && existing.directed == input.directed
        && existing.evidence_count == input.evidence_count
        && existing.last_reinforced_at == input.last_reinforced_at
        && existing.source == input.source.as_str()
        && existing.created_at == incoming.created_at
        && existing.created_by == input.created_by
        && existing
            .metadata_json
            .as_deref()
            .and_then(|text| serde_json::from_str::<JsonValue>(text).ok())
            == input
                .metadata_json
                .as_deref()
                .and_then(|text| serde_json::from_str::<JsonValue>(text).ok())
}

fn link_conflict_issue(id: &str, reason: &str) -> JsonlImportIssue {
    JsonlImportIssue::warning(
        None,
        "reimport_divergent_existing_link",
        format!("link `{id}` skipped because {reason}; existing state is preserved"),
    )
}

/// Run one JSONL import operation.
///
/// # Errors
///
/// Returns [`JsonlImportError`] for filesystem setup failures or storage errors.
pub fn import_jsonl_records(
    options: &JsonlImportOptions,
) -> Result<JsonlImportReport, JsonlImportError> {
    import_jsonl_records_with_policy(options, NativeTrustPolicy::StoreAuthenticatedOnly)
}

/// Import records from a backup whose manifest and artifacts have already
/// passed [`crate::core::backup::verify_backup`].
///
/// The integrity verification authorizes restoring the content, not carrying
/// a foreign store's `human_explicit` trust across the boundary. Such rows are
/// imported at the header-derived cap and reported as warnings.
pub(crate) fn import_verified_backup_jsonl_records(
    options: &JsonlImportOptions,
) -> Result<JsonlImportReport, JsonlImportError> {
    import_jsonl_records_with_policy(options, NativeTrustPolicy::VerifiedBackupRestore)
}

fn import_jsonl_records_with_policy(
    options: &JsonlImportOptions,
    native_trust_policy: NativeTrustPolicy,
) -> Result<JsonlImportReport, JsonlImportError> {
    let workspace_path = normalize_path(&options.workspace_path);
    ensure_import_source_path_is_regular_file(&options.source_path)?;
    let source_path = normalize_path(&options.source_path);
    let source_id = source_id(&source_path);
    let input = read_jsonl_source_bounded(&source_path)?;

    let parsed = parse_jsonl_source(&input);
    let mut report = report_from_parsed(
        &workspace_path,
        &source_path,
        &source_id,
        options.dry_run,
        &parsed,
    );

    if parsed.has_errors() {
        return Ok(report);
    }
    let validated_memories = match validate_memories(&parsed) {
        Ok(memories) => memories,
        Err(issues) => {
            report.issues.extend(issues);
            report.status = "rejected".to_owned();
            return Ok(report);
        }
    };
    let links = match prepare_links(&parsed) {
        Ok(links) => links,
        Err(issues) => {
            report.issues.extend(issues);
            report.status = "rejected".to_owned();
            return Ok(report);
        }
    };
    if options.dry_run {
        return Ok(report);
    }

    let database_path = database_path(options);
    ensure_database_parent(&database_path)?;
    let connection = DbConnection::open(DatabaseConfig::file(database_path.clone()))?;
    connection.migrate()?;
    let workspace_id = ensure_workspace(&connection, &workspace_path)?;

    let native_auth = native_import_auth_state(&parsed, &workspace_path, &workspace_id);
    let prepared = prepare_memories_with_policy(
        &parsed,
        validated_memories,
        &workspace_id,
        &native_auth,
        native_trust_policy,
    );
    if prepared.has_errors() {
        report.issues.extend(prepared.issues);
        report.status = "rejected".to_owned();
        report.database_path = Some(database_path.to_string_lossy().into_owned());
        return Ok(report);
    }
    report.issues.extend(prepared.issues);

    // Reimport is an idempotent restore: missing rows import, byte-identical
    // rows no-op, and divergent or tombstone-conflicting rows are preserved
    // untouched with an explicit conflict signal — never overwritten or
    // resurrected (ADR 0086 TC-D14).
    let mut to_insert = Vec::new();
    let mut publication_memory_ids = Vec::new();
    let mut conflicting_memory_ids = BTreeSet::new();
    let mut skipped_duplicate = 0_u32;
    // Backup recovery restores the original job ledger and rebuilds the whole
    // staged corpus before publication. Synthesizing import jobs here would
    // collide with recovered jobs when the source itself came from an import.
    let publish_imported_memories = !matches!(
        native_trust_policy,
        NativeTrustPolicy::VerifiedBackupRestore
    );
    connection.with_transaction(|| {
        let lineage_issues = destination_lineage_issues(&connection, &prepared.memories)?;
        if !lineage_issues.is_empty() {
            report.issues.extend(lineage_issues);
            report.status = "rejected".to_owned();
            return Ok(());
        }
        for memory in prepared.memories {
            match connection.get_memory(&memory.id)? {
                Some(existing) => {
                    skipped_duplicate = skipped_duplicate.saturating_add(1);
                    if let Some(issue) = reimport_conflict_issue(&existing, &memory) {
                        report.issues.push(issue);
                        conflicting_memory_ids.insert(memory.id.clone());
                    } else {
                        publication_memory_ids.push(memory.id.clone());
                    }
                }
                None => {
                    publication_memory_ids.push(memory.id.clone());
                    to_insert.push(memory);
                }
            }
        }
        for memory in &to_insert {
            connection.insert_memory_with_timestamps(
                &memory.id,
                &memory.input,
                &memory.created_at,
                &memory.updated_at,
                &memory.logical_id,
            )?;
            if let Some((alpha, beta)) = memory.bayes_posterior {
                connection.update_memory_bayes_posterior(&memory.id, alpha, beta)?;
            }
            // bd-multiplicity-aware-trust-p0u7g: rebuild the family pointer
            // and the attempt-family ledger from the exported block; the
            // ledger keys to the restored row's own logical identity, and the
            // exported origin is preserved for legacy_v094 forensics.
            if let Some(family) = &memory.attempt_family {
                connection.set_memory_attempt_family(
                    &memory.id,
                    &crate::db::MemoryAttemptFamily {
                        family_id: family.family_id.clone(),
                        declared_size: family.declared_size,
                        attempt_index: family.attempt_index,
                        disposition: family.disposition.clone(),
                    },
                )?;
                if let Some(origin) = family.origin.as_deref() {
                    connection.set_attempt_family_origin(
                        &memory.input.workspace_id,
                        &family.family_id,
                        origin,
                    )?;
                }
            }
            if let Some(tombstoned_at) = memory.tombstoned_at.as_deref() {
                connection.restore_imported_memory_tombstone(&memory.id, tombstoned_at)?;
                connection.insert_audit(
                    &crate::db::generate_audit_id(),
                    &CreateAuditInput {
                        workspace_id: Some(memory.input.workspace_id.clone()),
                        actor: Some("ee import jsonl".to_owned()),
                        action: crate::db::audit_actions::MEMORY_TOMBSTONE.to_owned(),
                        target_type: Some("memory".to_owned()),
                        target_id: Some(memory.id.clone()),
                        details: Some(
                            json!({
                                "tombstoned_at": tombstoned_at,
                                "reason": memory.tombstoned_reason.as_deref(),
                                "source": "jsonl_import",
                            })
                            .to_string(),
                        ),
                    },
                )?;
            }
            connection.restore_imported_memory_updated_at(&memory.id, &memory.updated_at)?;
            connection.insert_audit(
                &crate::db::generate_audit_id(),
                &CreateAuditInput {
                    workspace_id: Some(memory.input.workspace_id.clone()),
                    actor: Some("ee import jsonl".to_owned()),
                    action: IMPORT_ACTION.to_owned(),
                    target_type: Some("memory".to_owned()),
                    target_id: Some(memory.id.clone()),
                    details: Some(memory.details.clone()),
                },
            )?;
        }
        for link in &links {
            if conflicting_memory_ids.contains(&link.input.src_memory_id)
                || conflicting_memory_ids.contains(&link.input.dst_memory_id)
            {
                report.links_skipped_conflict += 1;
                report.issues.push(link_conflict_issue(
                    &link.id,
                    "an endpoint conflicts with an existing memory",
                ));
                continue;
            }
            if let Some(existing) = connection.get_memory_link(&link.id)? {
                if link_matches(&existing, link) {
                    report.links_skipped_duplicate += 1;
                } else {
                    report.links_skipped_conflict += 1;
                    report.issues.push(link_conflict_issue(
                        &link.id,
                        "the same link ID already has different fields",
                    ));
                }
                continue;
            }
            if connection
                .get_memory_link_by_edge(
                    &link.input.src_memory_id,
                    &link.input.dst_memory_id,
                    link.input.relation,
                )?
                .is_some()
            {
                report.links_skipped_conflict += 1;
                report.issues.push(link_conflict_issue(
                    &link.id,
                    "the ordered endpoints and relation already have another link ID",
                ));
                continue;
            }
            connection.insert_memory_link_at(&link.id, &link.input, &link.created_at)?;
            connection.insert_audit(
                &crate::db::generate_audit_id(),
                &CreateAuditInput {
                    workspace_id: Some(workspace_id.clone()),
                    actor: Some("ee import jsonl".to_owned()),
                    action: crate::db::audit_actions::MEMORY_LINK_CREATE.to_owned(),
                    target_type: Some("memory_link".to_owned()),
                    target_id: Some(link.id.clone()),
                    details: Some(link.details.clone()),
                },
            )?;
            report.links_imported += 1;
        }
        // bd-index-auto-freshness-m5kwf: every valid imported identity needs
        // durable publication work. Reimports preserve an existing logical
        // job, while legacy duplicates that predate this lane receive their
        // missing deterministic job before the post-commit drain.
        if publish_imported_memories {
            for memory_id in &publication_memory_ids {
                let job_id = import_search_index_job_id(&workspace_id, memory_id);
                if connection.get_search_index_job(&job_id)?.is_none() {
                    connection.insert_search_index_job(
                        &job_id,
                        &CreateSearchIndexJobInput {
                            workspace_id: workspace_id.clone(),
                            job_type: SearchIndexJobType::SingleDocument,
                            document_source: Some("memory".to_owned()),
                            document_id: Some(memory_id.clone()),
                            documents_total: 1,
                        },
                    )?;
                }
            }
        }
        Ok(())
    })?;

    report.database_path = Some(database_path.to_string_lossy().into_owned());
    if report.status == "rejected" {
        return Ok(report);
    }
    report.status = "completed".to_owned();
    report.memories_imported = saturating_len(to_insert.len());
    report.memories_skipped_duplicate = skipped_duplicate;
    report.tags_imported = to_insert.iter().fold(0_u32, |total, memory| {
        total.saturating_add(memory.tag_count)
    });
    report.imported_memory_ids = to_insert.into_iter().map(|memory| memory.id).collect();
    // Ordinary JSONL import retains immediate index convergence; backup
    // recovery owns its full-corpus rebuild after restoring the job ledger.
    if !publication_memory_ids.is_empty() && publish_imported_memories {
        // The rows above are durable; converge the derived index the same way
        // remember and batch remember do. Identical reimports also enter this
        // path so a prior failed deterministic job is requeued and retried.
        // A drain failure downgrades to a truthful non-fatal issue while the
        // durable jobs remain retryable (bd-index-auto-freshness-m5kwf).
        let index_dir = workspace_path
            .join(".ee")
            .join(crate::core::index::DEFAULT_INDEX_SUBDIR);
        let expected_job_ids = publication_memory_ids
            .iter()
            .map(|memory_id| import_search_index_job_id(&workspace_id, memory_id))
            .collect::<Vec<_>>();
        if !jsonl_import_index_is_ready(&workspace_path, &database_path) {
            for job_id in &expected_job_ids {
                connection.requeue_completed_search_index_job_for_repair(job_id)?;
            }
        }
        let publication = crate::core::index::process_pending_index_jobs_coalesced(
            &connection,
            &workspace_id,
            &index_dir,
            None,
        );
        let failure = jsonl_import_publication_failure(
            &connection,
            &expected_job_ids,
            &workspace_path,
            &database_path,
            publication,
        )?;
        if let Some(failure) = failure {
            let repair = jsonl_import_index_repair_command(
                &workspace_path,
                options.database_path.as_deref(),
            );
            report.issues.push(JsonlImportIssue {
                line: None,
                code: "import_index_publish_failed".to_owned(),
                severity: JsonlImportIssueSeverity::Warning,
                message: format!(
                    "Imported memories are durable, but automatic publication of durable search-index jobs did not complete: {failure}. Search may omit imported memories until the durable jobs are retried."
                ),
                repair: Some(repair),
            });
        }
    }
    Ok(report)
}

fn jsonl_import_index_is_ready(workspace_path: &Path, database_path: &Path) -> bool {
    crate::core::index::get_index_status(&crate::core::index::IndexStatusOptions {
        workspace_path: workspace_path.to_path_buf(),
        database_path: Some(database_path.to_path_buf()),
        index_dir: None,
    })
    .is_ok_and(|status| {
        status.health == crate::core::index::IndexHealth::Ready
            && status.db_generation.is_some()
            && status.db_generation == status.index_generation
    })
}

fn jsonl_import_publication_failure(
    connection: &DbConnection,
    expected_job_ids: &[String],
    workspace_path: &Path,
    database_path: &Path,
    publication: Result<
        Vec<crate::core::index::IndexProcessingJobReport>,
        crate::core::index::IndexRebuildError,
    >,
) -> Result<Option<String>, JsonlImportError> {
    let publication_error = publication.as_ref().err().map(ToString::to_string);
    let job_reports = publication.as_ref().ok();
    for job_id in expected_job_ids {
        let Some(job) = connection.get_search_index_job(job_id)? else {
            return Ok(Some(format!(
                "durable search-index job {job_id} disappeared before final verification"
            )));
        };
        if job.status_enum() != Some(crate::db::SearchIndexJobStatus::Completed) {
            let detail = job_reports
                .and_then(|reports| {
                    reports
                        .iter()
                        .find(|report| report.job_id.as_str() == job_id.as_str())
                })
                .and_then(|report| report.error.as_deref())
                .map(str::to_owned)
                .or_else(|| job.error_message.clone())
                .or_else(|| publication_error.clone())
                .unwrap_or_else(|| "publisher did not complete the durable job".to_owned());
            return Ok(Some(format!(
                "durable search-index job {job_id} ended with status {}: {detail}",
                job.status
            )));
        }
    }

    match crate::core::index::get_index_status(&crate::core::index::IndexStatusOptions {
        workspace_path: workspace_path.to_path_buf(),
        database_path: Some(database_path.to_path_buf()),
        index_dir: None,
    }) {
        Ok(status)
            if status.health == crate::core::index::IndexHealth::Ready
                && status.db_generation.is_some()
                && status.db_generation == status.index_generation =>
        {
            Ok(None)
        }
        Ok(status) => Ok(Some(format!(
            "post-drain index status was {:?} (database generation {:?}, index generation {:?})",
            status.health, status.db_generation, status.index_generation
        ))),
        Err(error) => Ok(Some(format!(
            "post-drain index status could not be verified: {error}"
        ))),
    }
}

/// Deterministic import-lane index-job id. Namespaced so a reimport replays
/// the same durable job and can never collide with the remember-lane job id
/// minted for the same memory.
fn import_search_index_job_id(workspace_id: &str, memory_id: &str) -> String {
    let hash = blake3::hash(format!("jsonl_import|{workspace_id}|{memory_id}").as_bytes())
        .to_hex()
        .to_string();
    format!("sidx_{}", &hash[..26])
}

fn jsonl_import_index_repair_command(
    workspace_path: &Path,
    database_path: Option<&Path>,
) -> String {
    let workspace = jsonl_import_shell_quote_arg(workspace_path.to_string_lossy().as_ref());
    match database_path {
        Some(database_path) => {
            let database = jsonl_import_shell_quote_arg(database_path.to_string_lossy().as_ref());
            format!("ee index rebuild --workspace {workspace} --database {database}")
        }
        None => format!("ee index rebuild --workspace {workspace}"),
    }
}

fn jsonl_import_shell_quote_arg(value: &str) -> String {
    if value.is_empty() {
        return "''".to_owned();
    }
    if value.bytes().all(|byte| {
        matches!(
            byte,
            b'A'..=b'Z'
                | b'a'..=b'z'
                | b'0'..=b'9'
                | b'_'
                | b'-'
                | b'.'
                | b'/'
                | b':'
                | b'@'
                | b'+'
                | b'='
        )
    }) {
        value.to_owned()
    } else if value.chars().any(char::is_control) {
        // A literal control byte inside ordinary single quotes would remain
        // shell-valid but could inject terminal lines/sequences into human
        // output. ANSI-C quoting keeps the command copy/pasteable in the
        // supported zsh/bash environments while rendering every control byte
        // visibly and preserving the exact path value when executed.
        let mut quoted = String::with_capacity(value.len() + 3);
        quoted.push_str("$'");
        for character in value.chars() {
            match character {
                '\'' => quoted.push_str("\\'"),
                '\\' => quoted.push_str("\\\\"),
                '\n' => quoted.push_str("\\n"),
                '\r' => quoted.push_str("\\r"),
                '\t' => quoted.push_str("\\t"),
                character if character.is_control() => {
                    let codepoint = character as u32;
                    if codepoint <= u16::MAX.into() {
                        quoted.push_str(&format!("\\u{codepoint:04x}"));
                    } else {
                        quoted.push_str(&format!("\\U{codepoint:08x}"));
                    }
                }
                character => quoted.push(character),
            }
        }
        quoted.push('\'');
        quoted
    } else {
        format!("'{}'", value.replace('\'', "'\\''"))
    }
}

/// A revision archive may fill missing rows in an identical existing chain,
/// but cannot attach new members to divergent or unrepresented local history.
/// Run before the first insert, inside the import transaction.
fn destination_lineage_issues(
    connection: &DbConnection,
    memories: &[PreparedMemory],
) -> Result<Vec<JsonlImportIssue>, DbError> {
    let revision_roots = memories
        .iter()
        .filter(|memory| memory.logical_id != memory.id)
        .map(|memory| memory.logical_id.as_str())
        .collect::<BTreeSet<_>>();
    let mut existing_counts = BTreeMap::<&str, u32>::new();
    let mut issues = Vec::new();
    for memory in memories {
        let Some(existing) = connection.get_memory(&memory.id)? else {
            continue;
        };
        let existing_logical_id = connection.get_memory_logical_id(&memory.id)?;
        let chain_changed = existing_logical_id.as_deref() != Some(memory.logical_id.as_str());
        let fields_changed = revision_roots.contains(memory.logical_id.as_str())
            && reimport_conflict_issue(&existing, memory).is_some();
        if chain_changed || fields_changed {
            issues.push(JsonlImportIssue::error(
                None,
                "reimport_divergent_revision_chain",
                format!(
                    "memory `{}` conflicts with the existing revision chain; no memories or links were imported",
                    memory.id,
                ),
            ));
        }
        *existing_counts.entry(&memory.logical_id).or_default() += 1;
    }
    for root in revision_roots {
        if connection.count_memory_chain(root)? != existing_counts.get(root).copied().unwrap_or(0) {
            issues.push(JsonlImportIssue::error(
                None,
                "reimport_divergent_revision_chain",
                format!(
                    "revision chain `{root}` contains local rows absent from the archive; no memories or links were imported",
                ),
            ));
        }
    }
    Ok(issues)
}

fn reimport_conflict_issue(
    existing: &StoredMemory,
    incoming: &PreparedMemory,
) -> Option<JsonlImportIssue> {
    let mut divergences = Vec::new();
    if existing.workspace_id != incoming.input.workspace_id {
        divergences.push("workspace_id");
    }
    if existing.content != incoming.input.content {
        divergences.push("content");
    }
    if existing.level != incoming.input.level {
        divergences.push("level");
    }
    if existing.kind != incoming.input.kind {
        divergences.push("kind");
    }
    if existing.created_at != incoming.created_at {
        divergences.push("created_at");
    }
    if existing.updated_at != incoming.updated_at {
        divergences.push("updated_at");
    }
    if existing.valid_from.as_deref()
        != Some(
            incoming
                .input
                .valid_from
                .as_deref()
                .unwrap_or(&incoming.created_at),
        )
    {
        divergences.push("valid_from");
    }
    if existing.valid_to != incoming.input.valid_to {
        divergences.push("valid_to");
    }
    if existing.trust_class != incoming.input.trust_class {
        divergences.push("trust_class");
    }
    match (
        existing.tombstoned_at.as_deref(),
        incoming.tombstoned_at.as_deref(),
    ) {
        (Some(_), None) => {
            divergences
                .push("tombstone (existing row is tombstoned; a plain import would resurrect it)");
        }
        (None, Some(_)) => {
            divergences.push("tombstone (import carries a tombstone; the existing row is live)");
        }
        _ => {}
    }
    if divergences.is_empty() {
        return None;
    }
    Some(JsonlImportIssue::warning(
        None,
        "reimport_divergent_existing_row",
        format!(
            "memory `{}` already exists and diverges on {}; the existing row is preserved (reimport never overwrites or resurrects)",
            incoming.id,
            divergences.join(", ")
        ),
    ))
}

fn report_from_parsed(
    workspace_path: &Path,
    source_path: &Path,
    source_id: &str,
    dry_run: bool,
    parsed: &ParsedJsonlImport,
) -> JsonlImportReport {
    let status = if parsed.has_errors() {
        "rejected"
    } else if dry_run {
        "dry_run"
    } else {
        "validated"
    };
    JsonlImportReport {
        schema: IMPORT_JSONL_SCHEMA_V1,
        workspace_path: workspace_path.to_string_lossy().into_owned(),
        database_path: None,
        source_path: source_path.to_string_lossy().into_owned(),
        source_id: source_id.to_owned(),
        dry_run,
        status: status.to_owned(),
        header: parsed
            .header
            .as_ref()
            .map(JsonlImportHeaderSummary::from_header),
        footer: parsed
            .footer
            .as_ref()
            .map(JsonlImportFooterSummary::from_footer),
        records_total: parsed.records_total,
        memory_records: saturating_len(parsed.memories.len()),
        tag_records: parsed.tag_records,
        link_records: saturating_len(parsed.links.len()),
        ignored_records: parsed.ignored_records,
        memories_imported: 0,
        memories_skipped_duplicate: 0,
        tags_imported: 0,
        links_imported: 0,
        links_skipped_duplicate: 0,
        links_skipped_conflict: 0,
        imported_memory_ids: Vec::new(),
        issues: parsed.issues.clone(),
    }
}

fn parse_jsonl_source(input: &str) -> ParsedJsonlImport {
    let mut parsed = ParsedJsonlImport {
        header: None,
        footer: None,
        footer_line: None,
        memories: Vec::new(),
        links: Vec::new(),
        tags_by_memory: BTreeMap::new(),
        tag_lines_by_memory: BTreeMap::new(),
        artifact_records: 0,
        tag_records: 0,
        issues: Vec::new(),
        records_total: 0,
        ignored_records: 0,
        records_root: RecordsRootBuilder::new(),
    };
    let mut first_schema: Option<(u32, String)> = None;
    let mut seen_memory_ids = BTreeSet::new();

    for (index, line) in input.lines().enumerate() {
        let line_number = u32::try_from(index + 1).unwrap_or(u32::MAX);
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        parsed.records_total = parsed.records_total.saturating_add(1);

        let value = match serde_json::from_str::<JsonValue>(trimmed) {
            Ok(value) => value,
            Err(error) => {
                parsed.issues.push(JsonlImportIssue::error(
                    Some(line_number),
                    "invalid_json",
                    error.to_string(),
                ));
                continue;
            }
        };
        let Some(schema) = value
            .get("schema")
            .and_then(JsonValue::as_str)
            .filter(|schema| !schema.trim().is_empty())
        else {
            parsed.issues.push(JsonlImportIssue::error(
                Some(line_number),
                "missing_schema",
                "record is missing a non-empty schema field",
            ));
            continue;
        };

        if first_schema.is_none() {
            first_schema = Some((line_number, schema.to_owned()));
        }

        if parsed.footer.is_some() && schema != EXPORT_FOOTER_SCHEMA_V1 {
            parsed.issues.push(JsonlImportIssue::error(
                Some(line_number),
                "footer_not_last",
                "JSONL footer must be the final non-empty record",
            ));
            continue;
        }

        match schema {
            EXPORT_HEADER_SCHEMA_V1 => parse_header_record(&mut parsed, line_number, value),
            EXPORT_MEMORY_SCHEMA_V1 => {
                // Fold the raw trimmed line bytes at this ordinal — the exact
                // bytes the exporter hashed — so tampering, reordering, or
                // truncation diverges the recomputed root from the MAC'd one.
                if let Some(memory_id) = value.get("memory_id").and_then(JsonValue::as_str) {
                    parsed
                        .records_root
                        .push(memory_id, &canonical_record_hash(trimmed.as_bytes()));
                }
                parse_memory_record(&mut parsed, &mut seen_memory_ids, line_number, value);
            }
            EXPORT_TAG_SCHEMA_V1 => {
                if let Some(memory_id) = value.get("memory_id").and_then(JsonValue::as_str) {
                    parsed
                        .records_root
                        .push(memory_id, &canonical_record_hash(trimmed.as_bytes()));
                }
                parse_tag_record(&mut parsed, line_number, value);
            }
            EXPORT_LINK_SCHEMA_V1 => {
                if let Some(link_id) = value.get("link_id").and_then(JsonValue::as_str) {
                    parsed
                        .records_root
                        .push(link_id, &canonical_record_hash(trimmed.as_bytes()));
                }
                match serde_json::from_value::<ExportLinkRecord>(value) {
                    Ok(link) => parsed.links.push(link),
                    Err(error) => parsed.issues.push(JsonlImportIssue::error(
                        Some(line_number),
                        "invalid_link_record",
                        error.to_string(),
                    )),
                }
            }
            EXPORT_FOOTER_SCHEMA_V1 => parse_footer_record(&mut parsed, line_number, value),
            EXPORT_ARTIFACT_SCHEMA_V1 => {
                parsed.artifact_records = parsed.artifact_records.saturating_add(1);
                parsed.ignored_records = parsed.ignored_records.saturating_add(1);
            }
            EXPORT_AGENT_SCHEMA_V1 | EXPORT_AUDIT_SCHEMA_V1 | EXPORT_WORKSPACE_SCHEMA_V1 => {
                parsed.ignored_records = parsed.ignored_records.saturating_add(1);
            }
            _ => parsed.issues.push(JsonlImportIssue::error(
                Some(line_number),
                "unsupported_schema",
                format!("unsupported JSONL record schema `{schema}`"),
            )),
        }
    }

    validate_header_and_footer(&mut parsed, first_schema);
    parsed
}

fn parse_header_record(parsed: &mut ParsedJsonlImport, line_number: u32, value: JsonValue) {
    if parsed.header.is_some() {
        parsed.issues.push(JsonlImportIssue::error(
            Some(line_number),
            "duplicate_header",
            "JSONL import accepts exactly one header record",
        ));
        return;
    }
    match serde_json::from_value::<ExportHeader>(value)
        .map_err(|error| error.to_string())
        .and_then(|header| {
            validate_export_header_required_fields(&header)?;
            Ok(header)
        }) {
        Ok(header) => parsed.header = Some(header),
        Err(error) => parsed.issues.push(JsonlImportIssue::error(
            Some(line_number),
            "invalid_header",
            error,
        )),
    }
}

fn validate_export_header_required_fields(header: &ExportHeader) -> Result<(), String> {
    for (field, value) in [
        ("schema", header.schema.as_str()),
        ("created_at", header.created_at.as_str()),
        ("ee_version", header.ee_version.as_str()),
        ("export_id", header.export_id.as_str()),
    ] {
        if value.trim().is_empty() {
            return Err(format!("header field `{field}` must not be blank"));
        }
    }
    if header.schema != EXPORT_HEADER_SCHEMA_V1 {
        return Err(format!(
            "header field `schema` must be {EXPORT_HEADER_SCHEMA_V1}"
        ));
    }
    Ok(())
}

fn parse_memory_record(
    parsed: &mut ParsedJsonlImport,
    seen_memory_ids: &mut BTreeSet<String>,
    line_number: u32,
    value: JsonValue,
) {
    match serde_json::from_value::<ExportMemoryRecord>(value) {
        Ok(memory) => {
            if !seen_memory_ids.insert(memory.memory_id.clone()) {
                parsed.issues.push(JsonlImportIssue::error(
                    Some(line_number),
                    "duplicate_memory_id",
                    format!("duplicate memory id `{}` in JSONL source", memory.memory_id),
                ));
            }
            if memory.redacted || memory.redaction_reason.is_some() {
                parsed.issues.push(JsonlImportIssue::info(
                    Some(line_number),
                    "redaction_round_trip_marker_preserved",
                    format!(
                        "redaction marker preserved for imported memory `{}`",
                        memory.memory_id
                    ),
                ));
            }
            parsed.memories.push(memory);
        }
        Err(error) => parsed.issues.push(JsonlImportIssue::error(
            Some(line_number),
            "invalid_memory",
            error.to_string(),
        )),
    }
}

fn parse_tag_record(parsed: &mut ParsedJsonlImport, line_number: u32, value: JsonValue) {
    match serde_json::from_value::<ExportTagRecord>(value) {
        Ok(tag) => {
            parsed.tag_records = parsed.tag_records.saturating_add(1);
            match Tag::parse(&tag.tag) {
                Ok(canonical) => {
                    parsed
                        .tag_lines_by_memory
                        .entry(tag.memory_id.clone())
                        .or_insert(line_number);
                    parsed
                        .tags_by_memory
                        .entry(tag.memory_id)
                        .or_default()
                        .insert(canonical.to_string());
                }
                Err(error) => parsed.issues.push(JsonlImportIssue::error(
                    Some(line_number),
                    "invalid_tag",
                    error.to_string(),
                )),
            }
        }
        Err(error) => parsed.issues.push(JsonlImportIssue::error(
            Some(line_number),
            "invalid_tag_record",
            error.to_string(),
        )),
    }
}

fn parse_footer_record(parsed: &mut ParsedJsonlImport, line_number: u32, value: JsonValue) {
    if parsed.footer.is_some() {
        parsed.issues.push(JsonlImportIssue::error(
            Some(line_number),
            "duplicate_footer",
            "JSONL import accepts at most one footer record",
        ));
        return;
    }
    match serde_json::from_value::<ExportFooter>(value)
        .map_err(|error| error.to_string())
        .and_then(|footer| {
            validate_export_footer_required_fields(&footer)?;
            Ok(footer)
        }) {
        Ok(footer) => {
            parsed.footer = Some(footer);
            parsed.footer_line = Some(line_number);
        }
        Err(error) => parsed.issues.push(JsonlImportIssue::error(
            Some(line_number),
            "invalid_footer",
            error.to_string(),
        )),
    }
}

fn validate_export_footer_required_fields(footer: &ExportFooter) -> Result<(), String> {
    for (field, value) in [
        ("schema", footer.schema.as_str()),
        ("export_id", footer.export_id.as_str()),
        ("completed_at", footer.completed_at.as_str()),
    ] {
        if value.trim().is_empty() {
            return Err(format!("footer field `{field}` must not be blank"));
        }
    }
    if footer.schema != EXPORT_FOOTER_SCHEMA_V1 {
        return Err(format!(
            "footer field `schema` must be {EXPORT_FOOTER_SCHEMA_V1}"
        ));
    }
    Ok(())
}

fn validate_header_and_footer(parsed: &mut ParsedJsonlImport, first_schema: Option<(u32, String)>) {
    match &parsed.header {
        Some(header) => {
            if header.format_version != crate::models::EXPORT_FORMAT_VERSION {
                parsed.issues.push(JsonlImportIssue::error(
                    None,
                    "unsupported_format_version",
                    format!(
                        "unsupported JSONL export format version {}",
                        header.format_version
                    ),
                ));
            }
        }
        None => parsed.issues.push(JsonlImportIssue::error(
            None,
            "missing_header",
            "JSONL import requires an ee.export.header.v1 header record",
        )),
    }

    if parsed.footer.is_none() {
        parsed.issues.push(JsonlImportIssue::error(
            None,
            "missing_footer",
            "JSONL import requires an ee.export.footer.v1 footer record",
        ));
    }

    if let Some((line, schema)) = first_schema {
        if schema != EXPORT_HEADER_SCHEMA_V1 {
            parsed.issues.push(JsonlImportIssue::error(
                Some(line),
                "header_not_first",
                "the first non-empty JSONL record must be ee.export.header.v1",
            ));
        }
    }

    let memory_ids = parsed
        .memories
        .iter()
        .map(|memory| memory.memory_id.as_str())
        .collect::<BTreeSet<_>>();
    for memory_id in parsed.tags_by_memory.keys() {
        if !memory_ids.contains(memory_id.as_str()) {
            parsed.issues.push(JsonlImportIssue::error(
                parsed.tag_lines_by_memory.get(memory_id).copied(),
                "orphaned_tag_record",
                format!("tag record references missing memory `{memory_id}`"),
            ));
        }
    }

    if let Some(footer) = &parsed.footer {
        if let Some(header) = &parsed.header
            && footer.export_id != header.export_id
        {
            parsed.issues.push(JsonlImportIssue::error(
                parsed.footer_line,
                "footer_export_id_mismatch",
                format!(
                    "footer export_id `{}` does not match header export_id `{}`",
                    footer.export_id, header.export_id
                ),
            ));
        }
        let parsed_artifact_count = u64::from(parsed.artifact_records);
        let parsed_tag_count = u64::from(parsed.tag_records);
        let parsed_record_count = u64::from(parsed.records_total);
        if footer.total_records != parsed_record_count {
            parsed.issues.push(JsonlImportIssue::warning(
                None,
                "footer_total_records_mismatch",
                format!(
                    "footer total_records {} does not match parsed JSONL records {}",
                    footer.total_records, parsed_record_count
                ),
            ));
        }
        if footer.artifact_count != parsed_artifact_count {
            parsed.issues.push(JsonlImportIssue::warning(
                None,
                "footer_artifact_count_mismatch",
                format!(
                    "footer artifact_count {} does not match parsed artifact records {}",
                    footer.artifact_count, parsed_artifact_count
                ),
            ));
        }
        if footer.tag_count != parsed_tag_count {
            parsed.issues.push(JsonlImportIssue::warning(
                None,
                "footer_tag_count_mismatch",
                format!(
                    "footer tag_count {} does not match parsed tag records {}",
                    footer.tag_count, parsed_tag_count
                ),
            ));
        }
        if footer.link_count != parsed.links.len() as u64 {
            parsed.issues.push(JsonlImportIssue::warning(
                None,
                "footer_link_count_mismatch",
                format!(
                    "footer link_count {} does not match parsed link records {}",
                    footer.link_count,
                    parsed.links.len()
                ),
            ));
        }
        if !footer.success {
            parsed.issues.push(JsonlImportIssue::warning(
                None,
                "source_export_incomplete",
                "footer marks the source export as unsuccessful",
            ));
        }
        if footer.memory_count != parsed.memories.len() as u64 {
            parsed.issues.push(JsonlImportIssue::warning(
                None,
                "footer_memory_count_mismatch",
                format!(
                    "footer memory_count {} does not match parsed memory records {}",
                    footer.memory_count,
                    parsed.memories.len()
                ),
            ));
        }
    }
}

struct PreparedMemories {
    memories: Vec<PreparedMemory>,
    issues: Vec<JsonlImportIssue>,
}

impl PreparedMemories {
    fn has_errors(&self) -> bool {
        self.issues
            .iter()
            .any(|issue| issue.severity == JsonlImportIssueSeverity::Error)
    }
}

#[cfg(test)]
fn prepare_memories(
    parsed: &ParsedJsonlImport,
    workspace_id: &str,
    native_auth: &NativeAuthState,
) -> PreparedMemories {
    let validated = match validate_memories(parsed) {
        Ok(memories) => memories,
        Err(issues) => {
            return PreparedMemories {
                memories: Vec::new(),
                issues,
            };
        }
    };
    prepare_memories_with_policy(
        parsed,
        validated,
        workspace_id,
        native_auth,
        NativeTrustPolicy::StoreAuthenticatedOnly,
    )
}

fn prepare_memories_with_policy(
    parsed: &ParsedJsonlImport,
    validated: Vec<ValidatedMemory<'_>>,
    workspace_id: &str,
    native_auth: &NativeAuthState,
    native_trust_policy: NativeTrustPolicy,
) -> PreparedMemories {
    let trust_class = trust_class_for_header(parsed.header.as_ref());
    let trust_subclass = trust_subclass_for_header(parsed.header.as_ref());
    let mut memories = Vec::with_capacity(parsed.memories.len());
    let mut issues = Vec::new();

    for validated_memory in validated {
        let memory = validated_memory.record;
        match prepare_memory(
            validated_memory,
            workspace_id,
            trust_class,
            &trust_subclass,
            parsed,
            native_auth,
            native_trust_policy,
        ) {
            Ok(prepared) => {
                if native_trust_policy == NativeTrustPolicy::VerifiedBackupRestore
                    && memory.trust_class.as_deref() == Some(TrustClass::HumanExplicit.as_str())
                    && !matches!(native_auth, NativeAuthState::Authenticated)
                {
                    issues.push(JsonlImportIssue::warning(
                        None,
                        VERIFIED_BACKUP_TRUST_DOWNGRADED_CODE,
                        format!(
                            "memory `{}` restored from a verified backup at {} instead of carrying foreign-store human_explicit trust",
                            memory.memory_id,
                            trust_class.as_str(),
                        ),
                    ));
                }
                memories.push(prepared);
            }
            Err(issue) => issues.push(issue),
        }
    }

    PreparedMemories { memories, issues }
}

fn validate_memories(
    parsed: &ParsedJsonlImport,
) -> Result<Vec<ValidatedMemory<'_>>, Vec<JsonlImportIssue>> {
    let redaction = parsed
        .header
        .as_ref()
        .map_or(RedactionLevel::None, |header| header.redaction_level);
    let mut memories = Vec::with_capacity(parsed.memories.len());
    let mut issues = Vec::new();
    for memory in &parsed.memories {
        match validate_memory(memory, redaction) {
            Ok(memory) => memories.push(memory),
            Err(issue) => issues.push(issue),
        }
    }
    if issues.is_empty() {
        let by_id = memories
            .iter()
            .map(|memory| (memory.record.memory_id.as_str(), memory))
            .collect::<BTreeMap<_, _>>();
        let mut logical_ids = Vec::with_capacity(memories.len());
        let mut live_heads = BTreeSet::new();
        for memory in &memories {
            let record = memory.record;
            let root_id = record.logical_id.as_deref().unwrap_or(&record.memory_id);
            let root = by_id.get(root_id);
            let message = match root {
                None => Some("revision root is absent from the archive"),
                Some(root) if root.record.workspace_id != record.workspace_id => {
                    Some("revision root belongs to a different source workspace")
                }
                Some(root) if root.record.logical_id.as_deref().unwrap_or(root_id) != root_id => {
                    Some("revision root must identify itself, not another revision")
                }
                Some(root) => {
                    logical_ids.push(root.id.clone());
                    if record
                        .valid_to
                        .as_ref()
                        .or(record.expires_at.as_ref())
                        .is_none()
                        && record.tombstoned_at.is_none()
                        && !live_heads.insert(root.id.clone())
                    {
                        Some("revision chain has more than one live head")
                    } else {
                        None
                    }
                }
            };
            if let Some(message) = message {
                issues.push(JsonlImportIssue::error(
                    None,
                    "invalid_memory_lineage",
                    format!("memory `{}`: {message}", record.memory_id),
                ));
            }
        }
        if issues.is_empty() {
            for (memory, logical_id) in memories.iter_mut().zip(logical_ids) {
                memory.logical_id = logical_id;
            }
        }
    }
    if issues.is_empty() {
        Ok(memories)
    } else {
        Err(issues)
    }
}

fn validate_memory(
    memory: &ExportMemoryRecord,
    redaction: RedactionLevel,
) -> Result<ValidatedMemory<'_>, JsonlImportIssue> {
    let id = import_memory_id(memory, redaction)?;
    let level: MemoryLevel = memory.level.parse().map_err(|error| {
        JsonlImportIssue::error(
            None,
            "invalid_memory_level",
            format!("memory `{}` has invalid level: {error}", memory.memory_id),
        )
    })?;
    let kind: MemoryKind = memory.kind.parse().map_err(|error| {
        JsonlImportIssue::error(
            None,
            "invalid_memory_kind",
            format!("memory `{}` has invalid kind: {error}", memory.memory_id),
        )
    })?;
    let content = MemoryContent::parse(&memory.content).map_err(|error| {
        JsonlImportIssue::error(
            None,
            "invalid_memory_content",
            format!("memory `{}` has invalid content: {error}", memory.memory_id),
        )
    })?;
    let redaction_report = crate::policy::redact_secret_like_content(content.as_str());
    if redaction_report.redacted {
        return Err(JsonlImportIssue::error(
            None,
            "memory_contains_secret",
            format!(
                "memory `{}` contains secrets ({}); redact before import",
                memory.memory_id,
                redaction_report.redacted_reasons.join(", ")
            ),
        ));
    }
    // A missing confidence depends on the authenticated trust class. Validate
    // explicit values now without assigning that default before authentication.
    let confidence = memory
        .confidence
        .map(|value| score_or_default(Some(value), 0.0))
        .transpose()
        .map_err(|message| {
            JsonlImportIssue::error(
                None,
                "invalid_memory_confidence",
                format!("memory `{}` {message}", memory.memory_id),
            )
        })?;
    let utility = score_or_default(memory.utility, 0.5).map_err(|message| {
        JsonlImportIssue::error(
            None,
            "invalid_memory_utility",
            format!("memory `{}` {message}", memory.memory_id),
        )
    })?;
    let importance = score_or_default(memory.importance, 0.5).map_err(|message| {
        JsonlImportIssue::error(
            None,
            "invalid_memory_importance",
            format!("memory `{}` {message}", memory.memory_id),
        )
    })?;
    let bayes_posterior = exported_bayes_posterior(memory)?;
    for (field, value) in [
        ("created_at", Some(memory.created_at.as_str())),
        ("updated_at", memory.updated_at.as_deref()),
        ("tombstoned_at", memory.tombstoned_at.as_deref()),
        ("valid_from", memory.valid_from.as_deref()),
        ("valid_to", memory.valid_to.as_deref()),
        ("expires_at", memory.expires_at.as_deref()),
    ] {
        if let Some(value) = value {
            chrono::DateTime::parse_from_rfc3339(value).map_err(|_| {
                JsonlImportIssue::error(
                    None,
                    "invalid_memory_timestamp",
                    format!(
                        "memory `{}` has invalid {field}; expected an RFC 3339 timestamp",
                        memory.memory_id,
                    ),
                )
            })?;
        }
    }

    Ok(ValidatedMemory {
        record: memory,
        logical_id: id.clone(),
        id,
        level,
        kind,
        content,
        confidence,
        utility,
        importance,
        bayes_posterior,
    })
}

fn prepare_memory(
    validated: ValidatedMemory<'_>,
    workspace_id: &str,
    trust_class: TrustClass,
    trust_subclass: &str,
    parsed: &ParsedJsonlImport,
    native_auth: &NativeAuthState,
    native_trust_policy: NativeTrustPolicy,
) -> Result<PreparedMemory, JsonlImportIssue> {
    let memory = validated.record;
    let import_source = parsed
        .header
        .as_ref()
        .map(|header| header.import_source)
        .unwrap_or(ImportSource::Unknown);
    let trust_class = trust_class_for_memory(
        memory,
        trust_class,
        import_source,
        native_auth,
        native_trust_policy,
    )?;
    let trust_subclass = trust_subclass_for_memory(memory, trust_subclass);
    let tags = parsed
        .tags_by_memory
        .get(&memory.memory_id)
        .map(|tags| tags.iter().cloned().collect::<Vec<_>>())
        .unwrap_or_default();
    let tag_count = saturating_len(tags.len());

    Ok(PreparedMemory {
        id: validated.id,
        logical_id: validated.logical_id,
        created_at: memory.created_at.clone(),
        updated_at: memory
            .updated_at
            .as_ref()
            .or(memory.tombstoned_at.as_ref())
            .unwrap_or(&memory.created_at)
            .clone(),
        input: CreateMemoryInput {
            workspace_id: workspace_id.to_owned(),
            level: validated.level.as_str().to_owned(),
            kind: validated.kind.as_str().to_owned(),
            content: validated.content.as_str().to_owned(),
            workflow_id: None,
            confidence: validated
                .confidence
                .unwrap_or_else(|| trust_class.initial_confidence()),
            utility: validated.utility,
            importance: validated.importance,
            provenance_uri: memory.provenance_uri.clone().or_else(|| {
                Some(format!(
                    "jsonl-import://{}",
                    memory.source_agent.as_deref().unwrap_or("unknown")
                ))
            }),
            trust_class: trust_class.as_str().to_owned(),
            trust_subclass,
            tags,
            valid_from: memory.valid_from.clone(),
            valid_to: memory
                .valid_to
                .clone()
                .or_else(|| memory.expires_at.clone()),
        },
        tombstoned_at: memory.tombstoned_at.clone(),
        tombstoned_reason: memory.tombstoned_reason.clone(),
        bayes_posterior: validated.bayes_posterior,
        attempt_family: memory.attempt_family.clone(),
        details: json!({
            "schema": IMPORT_JSONL_SCHEMA_V1,
            "sourceMemoryId": memory.memory_id,
            "sourceLogicalId": memory.logical_id,
            "sourceWorkspaceId": memory.workspace_id,
            "sourceCreatedAt": memory.created_at,
            "sourceUpdatedAt": memory.updated_at,
            "sourceTombstonedAt": memory.tombstoned_at.as_deref(),
            "sourceTombstonedReason": memory.tombstoned_reason.as_deref(),
            "sourceValidFrom": memory.valid_from.as_deref(),
            "sourceValidTo": memory.valid_to.clone().or_else(|| memory.expires_at.clone()),
            "redacted": memory.redacted,
            "redactionReason": memory.redaction_reason,
            "sourceGraphFields": source_graph_fields_json(memory),
        })
        .to_string(),
        tag_count,
    })
}

fn exported_bayes_posterior(
    memory: &ExportMemoryRecord,
) -> Result<Option<(f64, f64)>, JsonlImportIssue> {
    match (memory.bayes_alpha, memory.bayes_beta) {
        (Some(alpha), Some(beta)) => {
            if !positive_finite(alpha) || !positive_finite(beta) {
                return Err(JsonlImportIssue::error(
                    None,
                    "invalid_memory_bayes_posterior",
                    format!(
                        "memory `{}` bayes_alpha and bayes_beta must be positive finite values",
                        memory.memory_id
                    ),
                ));
            }
            Ok(Some((alpha, beta)))
        }
        (None, None) => Ok(None),
        _ => Err(JsonlImportIssue::error(
            None,
            "invalid_memory_bayes_posterior",
            format!(
                "memory `{}` must include both bayes_alpha and bayes_beta when importing an exported posterior",
                memory.memory_id
            ),
        )),
    }
}

fn positive_finite(value: f64) -> bool {
    value.is_finite() && value > 0.0
}

fn source_graph_fields_json(memory: &ExportMemoryRecord) -> JsonValue {
    let mut fields = serde_json::Map::new();
    insert_optional_json(&mut fields, "pagerank_score", memory.pagerank_score);
    insert_optional_json(&mut fields, "betweenness_score", memory.betweenness_score);
    insert_optional_json(&mut fields, "hits_authority", memory.hits_authority);
    insert_optional_json(&mut fields, "hits_hub", memory.hits_hub);
    insert_optional_json(&mut fields, "onion_layer", memory.onion_layer);
    insert_optional_json(&mut fields, "k_truss_max", memory.k_truss_max);
    insert_optional_json(&mut fields, "articulation_point", memory.articulation_point);
    insert_optional_json(&mut fields, "bayes_alpha", memory.bayes_alpha);
    insert_optional_json(&mut fields, "bayes_beta", memory.bayes_beta);
    JsonValue::Object(fields)
}

fn insert_optional_json<T>(
    fields: &mut serde_json::Map<String, JsonValue>,
    key: &str,
    value: Option<T>,
) where
    T: serde::Serialize,
{
    if let Some(value) = value
        && let Ok(json_value) = serde_json::to_value(value)
    {
        fields.insert(key.to_owned(), json_value);
    }
}

pub(super) fn import_memory_id(
    memory: &ExportMemoryRecord,
    redaction_level: RedactionLevel,
) -> Result<String, JsonlImportIssue> {
    match memory.memory_id.parse::<MemoryId>() {
        Ok(_) => Ok(memory.memory_id.clone()),
        Err(_) if redaction_level.redacts_identifiers() => {
            Ok(stable_redacted_memory_id(memory).to_string())
        }
        Err(error) => Err(JsonlImportIssue::error(
            None,
            "invalid_memory_id",
            format!("memory id `{}` is invalid: {error}", memory.memory_id),
        )),
    }
}

fn stable_redacted_memory_id(memory: &ExportMemoryRecord) -> MemoryId {
    MemoryId::from_uuid(stable_uuid(&format!(
        "jsonl-redacted-memory:{}:{}:{}:{}",
        memory.memory_id, memory.level, memory.kind, memory.created_at
    )))
}

fn score_or_default(value: Option<f64>, default: f32) -> Result<f32, String> {
    let score = match value {
        Some(score) => {
            if !score.is_finite() || !(0.0..=1.0).contains(&score) {
                return Err(format!(
                    "score is invalid: value {score} is not finite or outside 0.0..=1.0"
                ));
            }
            score as f32
        }
        None => default,
    };
    UnitScore::parse(score)
        .map(UnitScore::into_inner)
        .map_err(|error| format!("score is invalid: {error}"))
}

/// Whether the artifact authenticates under this store's key for native-trust
/// admission (ADR 0086 TC-D14). Computed once per import, then consulted for
/// every record-level `human_explicit` claim.
#[derive(Clone, Debug, Eq, PartialEq)]
enum NativeAuthState {
    /// The footer MAC verified against the local store key, the local
    /// workspace scope, and the records root recomputed from the received
    /// lines.
    Authenticated,
    /// The artifact carries no valid authentication for this store; the
    /// reason is a secret-free explanation for the refusal message.
    Unauthenticated { reason: String },
    /// The store-local authentication root itself is unavailable. Fail
    /// closed: native trust is refused with
    /// [`MESH_STORE_AUTHENTICATION_UNAVAILABLE_CODE`].
    StoreUnavailable { error: StoreAuthError },
}

fn native_import_auth_state(
    parsed: &ParsedJsonlImport,
    workspace_path: &Path,
    local_workspace_id: &str,
) -> NativeAuthState {
    let Some(authentication) = parsed
        .footer
        .as_ref()
        .and_then(|footer| footer.authentication.as_ref())
    else {
        return NativeAuthState::Unauthenticated {
            reason: "the artifact footer carries no store-local authentication block".to_owned(),
        };
    };
    let root = match StoreAuthRoot::open(workspace_keys_dir(workspace_path)) {
        Ok(root) => root,
        Err(error) => return NativeAuthState::StoreUnavailable { error },
    };
    let context = ArtifactContext {
        artifact_family: EXPORT_ARTIFACT_FAMILY,
        record_encoding_version: EXPORT_RECORD_ENCODING_V1,
        source_key_namespace: STORE_KEY_NAMESPACE_V1,
        workspace_scope: local_workspace_id,
    };
    match verify_artifact(
        &root,
        MacDomain::NativeImportRecordsRoot,
        &context,
        authentication,
        &parsed.records_root.finalize(),
        parsed.records_root.count(),
    ) {
        Ok(ImportAuthOutcome::Authenticated { .. }) => NativeAuthState::Authenticated,
        Ok(ImportAuthOutcome::RecordsMismatch) => NativeAuthState::Unauthenticated {
            reason: "the received records disagree with the MAC-authenticated records root/count \
                     (tampered, reordered, truncated, or padded)"
                .to_owned(),
        },
        Ok(ImportAuthOutcome::MacMismatch) => NativeAuthState::Unauthenticated {
            reason: "the footer MAC does not verify under this store's key and this workspace's \
                     binding context (foreign workspace, surface, or edited header)"
                .to_owned(),
        },
        Ok(ImportAuthOutcome::KeyOutsideWindow) => NativeAuthState::Unauthenticated {
            reason: "the footer names a key outside this store's verification window (foreign \
                     store or rotated-out key)"
                .to_owned(),
        },
        Ok(ImportAuthOutcome::SchemaMismatch) => NativeAuthState::Unauthenticated {
            reason: "the footer authentication block has an unsupported schema".to_owned(),
        },
        Ok(ImportAuthOutcome::Malformed) => NativeAuthState::Unauthenticated {
            reason: "the footer authentication block is malformed".to_owned(),
        },
        Err(error) => NativeAuthState::StoreUnavailable { error },
    }
}

fn trust_class_for_header(header: Option<&ExportHeader>) -> TrustClass {
    let Some(header) = header else {
        return TrustClass::LegacyImport;
    };
    match header.import_source {
        ImportSource::CassImport => TrustClass::CassEvidence,
        ImportSource::LegacyScan | ImportSource::ExternalImport | ImportSource::Unknown => {
            TrustClass::LegacyImport
        }
        ImportSource::Native => match header.trust_level {
            TrustLevel::Validated | TrustLevel::Verified => TrustClass::AgentValidated,
            TrustLevel::Untrusted | TrustLevel::Quarantined => TrustClass::AgentAssertion,
        },
    }
}

fn trust_class_for_memory(
    memory: &ExportMemoryRecord,
    fallback: TrustClass,
    import_source: ImportSource,
    native_auth: &NativeAuthState,
    native_trust_policy: NativeTrustPolicy,
) -> Result<TrustClass, JsonlImportIssue> {
    let Some(raw) = memory.trust_class.as_deref() else {
        return Ok(fallback);
    };
    let raw = raw.trim();
    if raw.is_empty() {
        return Err(JsonlImportIssue::error(
            None,
            "invalid_memory_trust_class",
            format!("memory `{}` has blank trust_class", memory.memory_id),
        ));
    }
    let trust_class = TrustClass::from_str(raw).map_err(|error| {
        JsonlImportIssue::error(
            None,
            "invalid_memory_trust_class",
            format!(
                "memory `{}` has invalid trust_class: {error}",
                memory.memory_id
            ),
        )
    })?;
    if trust_class == TrustClass::PeerHumanAttested {
        return Err(JsonlImportIssue::error(
            None,
            PEER_HUMAN_ATTESTED_IMPORT_PATH_REQUIRED_CODE,
            format!(
                "memory `{}` cannot import as peer_human_attested through JSONL; only the signed active-member admission path may assign that local class",
                memory.memory_id
            ),
        ));
    }
    if trust_class == TrustClass::HumanExplicit {
        if import_source.is_external() {
            return Err(JsonlImportIssue::error(
                None,
                "external_import_human_explicit_trust_class",
                format!(
                    "memory `{}` from {} cannot import as human_explicit; use agent_assertion or agent_validated for peer or external material",
                    memory.memory_id,
                    import_source.as_str()
                ),
            ));
        }
        // Native trust must be authenticated, not merely claimed: a spoofable
        // `import_source=native` header no longer admits human_explicit rows
        // (ADR 0086 TC-D14).
        match native_auth {
            NativeAuthState::Authenticated => {}
            NativeAuthState::Unauthenticated { .. }
                if native_trust_policy == NativeTrustPolicy::VerifiedBackupRestore =>
            {
                return Ok(fallback);
            }
            NativeAuthState::StoreUnavailable { .. }
                if native_trust_policy == NativeTrustPolicy::VerifiedBackupRestore =>
            {
                return Ok(fallback);
            }
            NativeAuthState::Unauthenticated { reason } => {
                return Err(JsonlImportIssue::error(
                    None,
                    UNAUTHENTICATED_NATIVE_IMPORT_TRUST_CODE,
                    format!(
                        "memory `{}` claims human_explicit but the artifact does not authenticate under this store: {reason}. Re-export from this workspace (ee backup create / ee export) so the footer carries a valid store-local MAC, or import the rows at agent_validated or lower",
                        memory.memory_id
                    ),
                ));
            }
            NativeAuthState::StoreUnavailable { error } => {
                return Err(JsonlImportIssue::error(
                    None,
                    MESH_STORE_AUTHENTICATION_UNAVAILABLE_CODE,
                    format!(
                        "memory `{}` claims human_explicit but the store-local authentication root is unavailable: {} Repair: {}",
                        memory.memory_id,
                        error.message(),
                        error.repair()
                    ),
                ));
            }
        }
    }
    Ok(trust_class)
}

fn trust_subclass_for_memory(memory: &ExportMemoryRecord, fallback: &str) -> Option<String> {
    let record_subclass = memory
        .trust_subclass
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_owned);
    if record_subclass.is_some() {
        return record_subclass;
    }
    if memory
        .trust_class
        .as_deref()
        .is_some_and(|value| !value.trim().is_empty())
    {
        return None;
    }
    Some(fallback.to_owned())
}

fn trust_subclass_for_header(header: Option<&ExportHeader>) -> String {
    header.map_or_else(
        || "jsonl:missing-header".to_owned(),
        |header| {
            format!(
                "jsonl:{}:{}",
                header.import_source.as_str(),
                header.trust_level.as_str()
            )
        },
    )
}

fn ensure_workspace(connection: &DbConnection, workspace_path: &Path) -> Result<String, DbError> {
    crate::core::workspace::ensure_bound_workspace(
        connection,
        &crate::core::workspace::stable_workspace_id(workspace_path),
        &[workspace_path],
    )
    .map_err(|error| DbError::MalformedRow {
        operation: DbOperation::Execute,
        message: error.message(),
    })
}

fn database_path(options: &JsonlImportOptions) -> PathBuf {
    options.database_path.clone().unwrap_or_else(|| {
        options
            .workspace_path
            .join(crate::config::WORKSPACE_MARKER)
            .join(DEFAULT_DB_FILE)
    })
}

fn ensure_database_parent(path: &Path) -> Result<(), JsonlImportError> {
    ensure_import_database_path_is_safe_for_write(path)?;
    let Some(parent) = path.parent() else {
        return Ok(());
    };
    std::fs::create_dir_all(parent).map_err(|error| JsonlImportError::Io {
        path: parent.to_path_buf(),
        message: error.to_string(),
    })?;
    ensure_import_database_path_is_safe_for_write(path)
}

fn ensure_import_database_path_is_safe_for_write(path: &Path) -> Result<(), JsonlImportError> {
    if let Some(symlink_path) =
        super::path_safety::first_existing_symlink_component(path).map_err(|error| {
            JsonlImportError::Io {
                path: path.to_path_buf(),
                message: error.to_string(),
            }
        })?
    {
        return Err(JsonlImportError::Io {
            path: path.to_path_buf(),
            message: format!(
                "refusing to import JSONL records into database through symlinked path component `{}`",
                symlink_path.display()
            ),
        });
    }
    match fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_file() => Ok(()),
        Ok(_) => Err(JsonlImportError::Io {
            path: path.to_path_buf(),
            message: format!(
                "refusing to import JSONL records into non-regular database path `{}`",
                path.display()
            ),
        }),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(JsonlImportError::Io {
            path: path.to_path_buf(),
            message: error.to_string(),
        }),
    }
}

/// Read a JSONL import source into a `String` with a hard byte cap.
///
/// Mirrors the bounded-read pattern that `src/cli/mod.rs::read_reflection_result_file`
/// uses for the much smaller `REFLECTION_RESULT_MAX_JSON_BYTES` surface:
/// `fs::metadata` rejects obviously oversized files up front, then the
/// actual read goes through `File::open` + `.take(MAX + 1)` so the
/// allocation is bounded even under TOCTOU growth between the metadata
/// stat and the open. The `+ 1` byte lets the post-read length check
/// distinguish "exactly at the limit" from "grew past the limit during
/// the read" and emit a clear oversize error in the latter case.
///
/// `read_to_string` is preserved (rather than `read_to_end` followed by
/// `String::from_utf8`) because the downstream `parse_jsonl_source`
/// requires UTF-8 input and would otherwise silently coerce or fail
/// later; rejecting invalid UTF-8 here keeps the error close to its
/// cause.
fn read_jsonl_source_bounded(source_path: &Path) -> Result<String, JsonlImportError> {
    use std::io::Read;

    let metadata = fs::metadata(source_path).map_err(|error| JsonlImportError::Io {
        path: source_path.to_path_buf(),
        message: error.to_string(),
    })?;
    if metadata.len() > JSONL_IMPORT_MAX_INPUT_BYTES {
        return Err(JsonlImportError::Io {
            path: source_path.to_path_buf(),
            message: format!(
                "JSONL source is too large: {} bytes exceeds the {} byte limit",
                metadata.len(),
                JSONL_IMPORT_MAX_INPUT_BYTES,
            ),
        });
    }
    let file = fs::File::open(source_path).map_err(|error| JsonlImportError::Io {
        path: source_path.to_path_buf(),
        message: error.to_string(),
    })?;
    let mut input = String::new();
    let mut bounded = file.take(JSONL_IMPORT_MAX_INPUT_BYTES + 1);
    bounded
        .read_to_string(&mut input)
        .map_err(|error| JsonlImportError::Io {
            path: source_path.to_path_buf(),
            message: error.to_string(),
        })?;
    if input.len() as u64 > JSONL_IMPORT_MAX_INPUT_BYTES {
        return Err(JsonlImportError::Io {
            path: source_path.to_path_buf(),
            message: format!(
                "JSONL source is too large: read {} bytes exceeds the {} byte limit",
                input.len(),
                JSONL_IMPORT_MAX_INPUT_BYTES,
            ),
        });
    }
    Ok(input)
}

fn ensure_import_source_path_is_regular_file(path: &Path) -> Result<(), JsonlImportError> {
    if let Some(symlink_path) =
        super::path_safety::first_existing_symlink_component(path).map_err(|error| {
            JsonlImportError::Io {
                path: path.to_path_buf(),
                message: error.to_string(),
            }
        })?
    {
        return Err(JsonlImportError::Io {
            path: path.to_path_buf(),
            message: format!(
                "refusing to import JSONL source through symlinked path component `{}`",
                symlink_path.display()
            ),
        });
    }
    let metadata = fs::symlink_metadata(path).map_err(|error| JsonlImportError::Io {
        path: path.to_path_buf(),
        message: error.to_string(),
    })?;
    if !metadata.file_type().is_file() {
        return Err(JsonlImportError::Io {
            path: path.to_path_buf(),
            message: format!(
                "refusing to import JSONL source from non-regular path `{}`",
                path.display()
            ),
        });
    }
    Ok(())
}

fn normalize_path(path: &Path) -> PathBuf {
    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
}

fn source_id(source_path: &Path) -> String {
    format!("jsonl://{}", source_path.to_string_lossy())
}

fn stable_uuid(input: &str) -> Uuid {
    let hash = blake3::hash(input.as_bytes());
    let mut bytes = [0_u8; 16];
    bytes.copy_from_slice(&hash.as_bytes()[..16]);
    Uuid::from_bytes(bytes)
}

fn saturating_len(len: usize) -> u32 {
    u32::try_from(len).unwrap_or(u32::MAX)
}

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

    type TestResult = Result<(), String>;

    fn ensure<T>(actual: T, expected: T, context: &str) -> TestResult
    where
        T: std::fmt::Debug + PartialEq,
    {
        if actual == expected {
            Ok(())
        } else {
            Err(format!("{context}: expected {expected:?}, got {actual:?}"))
        }
    }

    fn unauthenticated() -> NativeAuthState {
        NativeAuthState::Unauthenticated {
            reason: "test artifact without authentication".to_owned(),
        }
    }

    fn authenticated() -> NativeAuthState {
        NativeAuthState::Authenticated
    }

    fn sample_jsonl() -> String {
        [
            r#"{"schema":"ee.export.header.v1","format_version":1,"created_at":"2026-04-30T00:00:00Z","workspace_id":"wsp_01234567890123456789012345","workspace_path":"/source","export_scope":"memories","redaction_level":"none","record_count":3,"ee_version":"0.1.0","hostname":null,"export_id":"exp-001","import_source":"native","trust_level":"validated","checksum":null,"signature":null,"source_schema_version":null}"#,
            r#"{"schema":"ee.export.memory.v1","memory_id":"mem_01234567890123456789012345","workspace_id":"wsp_01234567890123456789012345","level":"procedural","kind":"rule","content":"Run cargo fmt --check before release.","importance":0.8,"confidence":0.9,"utility":0.7,"created_at":"2026-04-30T00:00:00Z","updated_at":null,"expires_at":null,"source_agent":"MistySalmon","provenance_uri":"ee-export://fixture","superseded_by":null,"supersedes":null,"redacted":false,"redaction_reason":null}"#,
            r#"{"schema":"ee.export.tag.v1","memory_id":"mem_01234567890123456789012345","tag":"Release","created_at":"2026-04-30T00:00:00Z"}"#,
            r#"{"schema":"ee.export.footer.v1","export_id":"exp-001","completed_at":"2026-04-30T00:01:00Z","total_records":4,"memory_count":1,"link_count":0,"tag_count":1,"audit_count":0,"checksum":null,"success":true,"error_message":null}"#,
        ]
        .join("\n")
    }

    fn sample_jsonl_with_graph_fields() -> String {
        sample_jsonl().replace(
            r#""utility":0.7,"created_at""#,
            r#""utility":0.7,"pagerank_score":0.12,"betweenness_score":0.34,"hits_authority":0.56,"hits_hub":0.78,"onion_layer":3,"k_truss_max":4,"articulation_point":true,"bayes_alpha":2.5,"bayes_beta":1.5,"created_at""#,
        )
    }

    fn linked_jsonl_values() -> Result<Vec<JsonValue>, String> {
        let mut records = sample_jsonl()
            .lines()
            .map(serde_json::from_str::<JsonValue>)
            .collect::<Result<Vec<_>, _>>()
            .map_err(|error| error.to_string())?;
        let mut target = records[1].clone();
        target["memory_id"] = json!(MemoryId::from_uuid(Uuid::from_u128(2)).to_string());
        target["content"] = json!("Keep the release workflow deterministic.");
        let link = json!({
            "schema": EXPORT_LINK_SCHEMA_V1,
            "link_id": MemoryLinkId::from_uuid(Uuid::from_u128(3)).to_string(),
            "source_memory_id": records[1]["memory_id"],
            "target_memory_id": target["memory_id"],
            "link_type": "supports",
            "weight": 0.75,
            "created_at": "2026-04-30T00:00:01Z",
            "metadata": {
                "confidence": 0.5, "directed": false, "evidenceCount": 7,
                "lastReinforcedAt": "2026-05-01T00:00:00Z", "source": "agent",
                "createdBy": "release-review", "metadata": {"rationale": "two observed releases"}
            }
        });
        records.insert(3, target);
        records.insert(4, link);
        records[0]["record_count"] = json!(5);
        records[5]["total_records"] = json!(6);
        records[5]["memory_count"] = json!(2);
        records[5]["link_count"] = json!(1);
        Ok(records)
    }

    fn jsonl_values_text(records: &[JsonValue]) -> String {
        records
            .iter()
            .map(JsonValue::to_string)
            .collect::<Vec<_>>()
            .join("\n")
    }

    fn revision_jsonl_values() -> Result<Vec<JsonValue>, String> {
        let mut records = linked_jsonl_values()?;
        let root = records[1]["memory_id"].clone();
        records[1]["logical_id"] = root.clone();
        records[1]["valid_to"] = json!("2026-05-01T00:00:00Z");
        records[3]["logical_id"] = root;
        records[3]["created_at"] = json!("2026-05-02T00:00:00Z");
        records[3]["attempt_family"] = json!({
            "family_id": "fam-import-lineage", "declared_size": 1,
            "attempt_index": 1, "disposition": "selected", "origin": "declared"
        });
        let mut historical = records[1].clone();
        historical["memory_id"] = json!(MemoryId::from_uuid(Uuid::from_u128(4)).to_string());
        historical["content"] = json!("Historical revision retained after tombstoning.");
        historical["created_at"] = json!("2026-05-01T00:00:00Z");
        historical["valid_to"] = json!("2026-05-02T00:00:00Z");
        historical["tombstoned_at"] = json!("2026-05-03T00:00:00Z");
        records.insert(5, historical);
        records[0]["record_count"] = json!(6);
        records[6]["total_records"] = json!(7);
        records[6]["memory_count"] = json!(3);
        Ok(records)
    }

    #[test]
    fn revision_lineage_restores_roots_history_and_family_membership_after_redaction() -> TestResult
    {
        for level in [
            RedactionLevel::None,
            RedactionLevel::Standard,
            RedactionLevel::Paranoid,
        ] {
            let mut records = revision_jsonl_values()?;
            records[0]["redaction_level"] = json!(level.as_str());
            let records = records
                .into_iter()
                .map(|value| {
                    let record = serde_json::from_value::<crate::models::ExportRecord>(value)
                        .map_err(|error| error.to_string())?;
                    serde_json::to_value(crate::output::jsonl_export::redact_record(record, level))
                        .map_err(|error| error.to_string())
                })
                .collect::<Result<Vec<_>, _>>()?;
            let parsed = parse_jsonl_source(&jsonl_values_text(&records));
            let root =
                import_memory_id(&parsed.memories[0], level).map_err(|issue| issue.message)?;
            let head =
                import_memory_id(&parsed.memories[1], level).map_err(|issue| issue.message)?;
            let historical =
                import_memory_id(&parsed.memories[2], level).map_err(|issue| issue.message)?;
            let dir = tempfile::tempdir().map_err(|error| error.to_string())?;
            let options = JsonlImportOptions {
                workspace_path: dir.path().join("workspace"),
                database_path: None,
                source_path: dir.path().join("revisions.jsonl"),
                dry_run: false,
            };
            fs::write(&options.source_path, jsonl_values_text(&records))
                .map_err(|error| error.to_string())?;
            let report = import_jsonl_records(&options).map_err(|error| error.to_string())?;
            ensure(
                report.memories_imported,
                3,
                &format!("{}: {:?}", level.as_str(), report.issues),
            )?;
            let connection = DbConnection::open_file(database_path(&options))
                .map_err(|error| error.to_string())?;
            for id in [&root, &head, &historical] {
                ensure(
                    connection
                        .get_memory_logical_id(id)
                        .map_err(|error| error.to_string())?,
                    Some(root.clone()),
                    "every restored revision shares the restored root",
                )?;
            }
            ensure(
                connection
                    .count_memory_chain(&root)
                    .map_err(|error| error.to_string())?,
                3,
                "chain count",
            )?;
            let head_memory = connection
                .get_memory(&head)
                .map_err(|error| error.to_string())?
                .ok_or("head")?;
            ensure(
                connection
                    .list_live_memory_revisions_for_logical_id(&head_memory.workspace_id, &root)
                    .map_err(|error| error.to_string())?
                    .iter()
                    .map(|memory| memory.id.clone())
                    .collect::<Vec<_>>(),
                vec![head.clone()],
                "exactly one live head",
            )?;
            let family_id = parsed.memories[1]
                .attempt_family
                .as_ref()
                .ok_or("family")?
                .family_id
                .as_str();
            ensure(
                connection
                    .list_attempt_family_membership_logical_ids(
                        &head_memory.workspace_id,
                        family_id,
                    )
                    .map_err(|error| error.to_string())?,
                vec![root.clone()],
                "family ledger keys to the root, not the current revision",
            )?;
            ensure(
                connection
                    .get_memory(&historical)
                    .map_err(|error| error.to_string())?
                    .ok_or("historical")?
                    .tombstoned_at,
                Some("2026-05-03T00:00:00Z".to_owned()),
                "historical tombstone survives",
            )?;
            drop(connection);
            let repeat = import_jsonl_records(&options).map_err(|error| error.to_string())?;
            ensure(
                (
                    repeat.status.as_str(),
                    repeat.memories_imported,
                    repeat.memories_skipped_duplicate,
                ),
                ("completed", 0, 3),
                "idempotent chain replay",
            )?;
            ensure(
                repeat
                    .issues
                    .iter()
                    .any(|issue| issue.code.starts_with("reimport_divergent")),
                false,
                "no false reimport conflict",
            )?;
        }
        Ok(())
    }

    #[test]
    fn invalid_revision_lineage_rejects_before_dry_run_or_storage_creation() -> TestResult {
        let base = revision_jsonl_values()?;
        let dir = tempfile::tempdir().map_err(|error| error.to_string())?;
        for case in ["missing", "blank", "cycle", "workspace", "two_heads"] {
            let mut records = base.clone();
            match case {
                "missing" => {
                    records[3]["logical_id"] =
                        json!(MemoryId::from_uuid(Uuid::from_u128(99)).to_string())
                }
                "blank" => records[3]["logical_id"] = json!(""),
                "cycle" => records[1]["logical_id"] = records[3]["memory_id"].clone(),
                "workspace" => records[3]["workspace_id"] = json!("wsp_other"),
                "two_heads" => records[1]["valid_to"] = JsonValue::Null,
                _ => unreachable!(),
            }
            let source_path = dir.path().join(format!("{case}.jsonl"));
            fs::write(&source_path, jsonl_values_text(&records))
                .map_err(|error| error.to_string())?;
            for dry_run in [true, false] {
                let options = JsonlImportOptions {
                    workspace_path: dir.path().join(format!("{case}-{dry_run}")),
                    database_path: None,
                    source_path: source_path.clone(),
                    dry_run,
                };
                let report = import_jsonl_records(&options).map_err(|error| error.to_string())?;
                ensure(report.status.as_str(), "rejected", case)?;
                ensure(
                    report
                        .issues
                        .iter()
                        .any(|issue| issue.code == "invalid_memory_lineage"),
                    true,
                    case,
                )?;
                ensure(
                    options.workspace_path.exists(),
                    false,
                    "preflight leaves storage absent",
                )?;
            }
        }
        Ok(())
    }

    #[test]
    fn divergent_revision_reimport_preserves_existing_chain_and_new_rows() -> TestResult {
        let dir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let options = JsonlImportOptions {
            workspace_path: dir.path().join("workspace"),
            database_path: None,
            source_path: dir.path().join("revisions.jsonl"),
            dry_run: false,
        };
        let base = revision_jsonl_values()?;
        fs::write(&options.source_path, jsonl_values_text(&base))
            .map_err(|error| error.to_string())?;
        let imported = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(imported.memories_imported, 3, "seed chain")?;
        let connection =
            DbConnection::open_file(database_path(&options)).map_err(|error| error.to_string())?;
        let workspace_id = ensure_workspace(&connection, &options.workspace_path)
            .map_err(|error| error.to_string())?;
        let before = connection
            .list_memories(&workspace_id, None, true)
            .map_err(|error| error.to_string())?;
        let audits_before = connection
            .list_audit_entries(Some(&workspace_id), None)
            .map_err(|error| error.to_string())?;
        drop(connection);
        for case in ["content", "logical_id", "valid_to", "new_head"] {
            let mut records = base.clone();
            match case {
                "content" => {
                    records[1]["content"] =
                        json!("Different source history must not replace the stored chain.")
                }
                "logical_id" => records[3]["logical_id"] = records[3]["memory_id"].clone(),
                "valid_to" => records[1]["valid_to"] = json!("2026-04-30T12:00:00Z"),
                "new_head" => {
                    let new_id = json!(MemoryId::from_uuid(Uuid::from_u128(77)).to_string());
                    records[3]["memory_id"] = new_id.clone();
                    records[4]["target_memory_id"] = new_id;
                }
                _ => unreachable!(),
            }
            fs::write(&options.source_path, jsonl_values_text(&records))
                .map_err(|error| error.to_string())?;
            let report = import_jsonl_records(&options).map_err(|error| error.to_string())?;
            ensure(report.status.as_str(), "rejected", case)?;
            ensure(
                (report.memories_imported, report.links_imported),
                (0, 0),
                case,
            )?;
            ensure(
                report
                    .issues
                    .iter()
                    .any(|issue| issue.code == "reimport_divergent_revision_chain"),
                true,
                case,
            )?;
        }
        let connection =
            DbConnection::open_file(database_path(&options)).map_err(|error| error.to_string())?;
        ensure(
            connection
                .list_memories(&workspace_id, None, true)
                .map_err(|error| error.to_string())?,
            before,
            "every memory unchanged",
        )?;
        ensure(
            connection
                .list_audit_entries(Some(&workspace_id), None)
                .map_err(|error| error.to_string())?,
            audits_before,
            "no audit changes on rejection",
        )
    }

    #[test]
    fn links_round_trip_with_fields_audit_idempotence_and_conflict_preservation() -> TestResult {
        let dir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let options = JsonlImportOptions {
            workspace_path: dir.path().join("workspace"),
            database_path: None,
            source_path: dir.path().join("links.jsonl"),
            dry_run: false,
        };
        let mut records = linked_jsonl_values()?;
        fs::write(&options.source_path, jsonl_values_text(&records))
            .map_err(|error| error.to_string())?;
        let first = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(first.status.as_str(), "completed", "first status")?;
        ensure(first.memories_imported, 2, "two memories")?;
        ensure(first.links_imported, 1, "one link")?;
        ensure(first.ignored_records, 0, "link is no longer ignored")?;
        let connection = DbConnection::open(DatabaseConfig::file(database_path(&options)))
            .map_err(|error| error.to_string())?;
        let id = records[4]["link_id"].as_str().ok_or("link id")?.to_owned();
        let stored = connection
            .get_memory_link(&id)
            .map_err(|error| error.to_string())?
            .ok_or("missing link")?;
        ensure(
            stored.src_memory_id.as_str(),
            records[1]["memory_id"].as_str().ok_or("source id")?,
            "source",
        )?;
        ensure(
            stored.dst_memory_id.as_str(),
            records[3]["memory_id"].as_str().ok_or("target id")?,
            "target",
        )?;
        ensure(stored.relation.as_str(), "supports", "relation")?;
        ensure(
            (
                stored.weight,
                stored.confidence,
                stored.directed,
                stored.evidence_count,
            ),
            (0.75, 0.5, false, 7),
            "link scores and evidence",
        )?;
        ensure(
            stored.created_at.as_str(),
            "2026-04-30T00:00:01Z",
            "original timestamp",
        )?;
        ensure(
            stored.last_reinforced_at.as_deref(),
            Some("2026-05-01T00:00:00Z"),
            "reinforcement time",
        )?;
        ensure(stored.source.as_str(), "agent", "origin")?;
        ensure(
            stored.created_by.as_deref(),
            Some("release-review"),
            "creator",
        )?;
        ensure(
            stored.metadata_json.as_deref(),
            Some(r#"{"rationale":"two observed releases"}"#),
            "metadata",
        )?;
        let audits = connection
            .list_audit_by_target("memory_link", &id, None)
            .map_err(|error| error.to_string())?;
        ensure(audits.len(), 1, "one link audit")?;
        ensure(
            audits[0].action.as_str(),
            crate::db::audit_actions::MEMORY_LINK_CREATE,
            "audit action",
        )?;
        let details: JsonValue =
            serde_json::from_str(audits[0].details.as_deref().ok_or("audit details")?)
                .map_err(|error| error.to_string())?;
        ensure(&details["sourceRecord"], &records[4], "source provenance")?;
        drop(connection);

        let repeated = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(
            (
                repeated.memories_imported,
                repeated.links_imported,
                repeated.links_skipped_duplicate,
            ),
            (0, 0, 1),
            "idempotent repeat",
        )?;
        for field in ["weight", "link_id", "memory_content"] {
            let mut conflicting = records.clone();
            match field {
                "weight" => conflicting[4]["weight"] = json!(0.9),
                "link_id" => {
                    conflicting[4]["link_id"] =
                        json!(MemoryLinkId::from_uuid(Uuid::from_u128(4)).to_string())
                }
                _ => {
                    conflicting[1]["content"] =
                        json!("A divergent local identity must not gain imported edges.")
                }
            }
            fs::write(&options.source_path, jsonl_values_text(&conflicting))
                .map_err(|error| error.to_string())?;
            let report = import_jsonl_records(&options).map_err(|error| error.to_string())?;
            ensure(
                (report.links_imported, report.links_skipped_conflict),
                (0, 1),
                field,
            )?;
        }
        // A conflicting incoming tombstone must not change the local edge either.
        let connection = DbConnection::open(DatabaseConfig::file(database_path(&options)))
            .map_err(|error| error.to_string())?;
        ensure(
            connection
                .get_memory_link(&id)
                .map_err(|error| error.to_string())?,
            Some(stored),
            "conflicts never overwrite the link",
        )?;
        ensure(
            connection
                .list_audit_by_target("memory_link", &id, None)
                .map_err(|error| error.to_string())?
                .len(),
            1,
            "repeat/conflicts add no link audit",
        )?;
        records[1]["tombstoned_at"] = json!("2026-06-01T00:00:00Z");
        drop(connection);
        fs::write(&options.source_path, jsonl_values_text(&records))
            .map_err(|error| error.to_string())?;
        let report = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(
            report.links_skipped_conflict,
            1,
            "conflicting tombstone skips incident link",
        )
    }

    #[test]
    fn invalid_links_reject_the_whole_import_before_creating_storage() -> TestResult {
        let dir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let base = linked_jsonl_values()?;
        let cases = [
            ("/link_type", json!("invented")),
            ("/weight", json!(1.01)),
            ("/weight", json!("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij")),
            ("/metadata/confidence", json!(-0.1)),
            ("/metadata/evidenceCount", json!(-1)),
            ("/metadata/directed", json!("false")),
            ("/metadata/source", json!("invented")),
            ("/metadata/createdBy", json!(" ")),
            ("/metadata/lastReinforcedAt", json!("yesterday")),
            ("/created_at", json!("yesterday")),
            ("/link_id", json!("invalid")),
            (
                "/link_id",
                json!("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij"),
            ),
            (
                "/metadata/source",
                json!("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij"),
            ),
            ("/target_memory_id", base[1]["memory_id"].clone()),
            (
                "/target_memory_id",
                json!(MemoryId::from_uuid(Uuid::from_u128(99)).to_string()),
            ),
            (
                "/metadata/metadata",
                json!({"credential": "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij"}),
            ),
        ];
        for (index, (pointer, value)) in cases.into_iter().enumerate() {
            let mut records = base.clone();
            *records[4].pointer_mut(pointer).ok_or("fixture pointer")? = value;
            let options = JsonlImportOptions {
                workspace_path: dir.path().join(format!("rejected-{index}")),
                database_path: None,
                source_path: dir.path().join(format!("invalid-{index}.jsonl")),
                dry_run: false,
            };
            fs::write(&options.source_path, jsonl_values_text(&records))
                .map_err(|error| error.to_string())?;
            for dry_run in [true, false] {
                let report = import_jsonl_records(&JsonlImportOptions {
                    dry_run,
                    ..options.clone()
                })
                .map_err(|error| error.to_string())?;
                ensure(report.status.as_str(), "rejected", pointer)?;
                ensure(
                    report
                        .data_json()
                        .to_string()
                        .contains("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij"),
                    false,
                    "invalid link diagnostics must not echo secret-shaped field values",
                )?;
                ensure(
                    report.memories_imported + report.links_imported,
                    0,
                    "no partial rows",
                )?;
                ensure(
                    options.workspace_path.exists(),
                    false,
                    "no database, keys or index created",
                )?;
            }
        }
        for same_id in [true, false] {
            let mut records = base.clone();
            let mut duplicate = records[4].clone();
            if !same_id {
                duplicate["link_id"] =
                    json!(MemoryLinkId::from_uuid(Uuid::from_u128(4)).to_string());
            }
            records.insert(5, duplicate);
            let parsed = parse_jsonl_source(&jsonl_values_text(&records));
            ensure(
                prepare_links(&parsed).is_err(),
                true,
                "duplicate link ID/edge rejected",
            )?;
        }
        Ok(())
    }

    #[test]
    fn redacted_links_resolve_to_imported_endpoints_deterministically() -> TestResult {
        for &level in RedactionLevel::all() {
            let mut records = linked_jsonl_values()?;
            records[0]["redaction_level"] = json!(level);
            // Use the production redactor: strict removes link metadata but
            // keeps IDs; standard and paranoid pseudonymize the identifiers.
            let records = records
                .into_iter()
                .map(|value| {
                    let record = serde_json::from_value::<crate::models::ExportRecord>(value)
                        .map_err(|error| error.to_string())?;
                    serde_json::to_value(crate::output::jsonl_export::redact_record(record, level))
                        .map_err(|error| error.to_string())
                })
                .collect::<Result<Vec<_>, _>>()?;
            let dir = tempfile::tempdir().map_err(|error| error.to_string())?;
            let options = JsonlImportOptions {
                workspace_path: dir.path().join("workspace"),
                database_path: None,
                source_path: dir.path().join("redacted.jsonl"),
                dry_run: false,
            };
            fs::write(&options.source_path, jsonl_values_text(&records))
                .map_err(|error| error.to_string())?;
            let first = import_jsonl_records(&options).map_err(|error| error.to_string())?;
            ensure(
                (first.memories_imported, first.links_imported),
                (2, 1),
                &format!("{} rows imported: {:?}", level.as_str(), first.issues),
            )?;
            let connection = DbConnection::open(DatabaseConfig::file(database_path(&options)))
                .map_err(|error| error.to_string())?;
            let links = connection
                .list_all_memory_links(None)
                .map_err(|error| error.to_string())?;
            ensure(links.len(), 1, "one restored link")?;
            for endpoint in [&links[0].src_memory_id, &links[0].dst_memory_id] {
                ensure(
                    first.imported_memory_ids.contains(endpoint),
                    true,
                    "endpoint resolves to restored row",
                )?;
            }
            if matches!(
                level,
                RedactionLevel::None | RedactionLevel::Minimal | RedactionLevel::Strict
            ) {
                ensure(
                    links[0].id.as_str(),
                    records[4]["link_id"].as_str().ok_or("link id")?,
                    "unredacted link ID remains intact",
                )?;
            } else {
                ensure(
                    links[0].id.parse::<MemoryLinkId>().is_ok(),
                    true,
                    "pseudonymized link maps to a valid durable ID",
                )?;
            }
            ensure(
                links[0].source.as_str(),
                if matches!(level, RedactionLevel::Strict | RedactionLevel::Paranoid) {
                    "import"
                } else {
                    "agent"
                },
                "redacted origin uses import default only when metadata was removed",
            )?;
            drop(connection);
            let repeated = import_jsonl_records(&options).map_err(|error| error.to_string())?;
            ensure(
                (repeated.links_imported, repeated.links_skipped_duplicate),
                (0, 1),
                "stable redacted link identity",
            )?;
        }
        Ok(())
    }

    fn import_report_fixture(source_path: &str, source_id: &str) -> JsonlImportReport {
        JsonlImportReport {
            schema: IMPORT_JSONL_SCHEMA_V1,
            workspace_path: "/workspace/project".to_owned(),
            database_path: None,
            source_path: source_path.to_owned(),
            source_id: source_id.to_owned(),
            dry_run: true,
            status: "dry_run".to_owned(),
            header: None,
            footer: None,
            records_total: 0,
            memory_records: 0,
            tag_records: 0,
            link_records: 0,
            ignored_records: 0,
            memories_imported: 0,
            memories_skipped_duplicate: 0,
            tags_imported: 0,
            links_imported: 0,
            links_skipped_duplicate: 0,
            links_skipped_conflict: 0,
            imported_memory_ids: Vec::new(),
            issues: Vec::new(),
        }
    }

    #[test]
    fn import_human_summary_escapes_terminal_control_bytes_in_publication_failures() -> TestResult {
        let mut report = import_report_fixture("/source.jsonl", "source-id");
        report.issues.push(JsonlImportIssue {
            line: None,
            code: "import_index_publish_failed".to_owned(),
            severity: JsonlImportIssueSeverity::Warning,
            message: "publisher said \u{1b}[31mred\nnext".to_owned(),
            repair: Some(format!(
                "ee index rebuild --workspace {}",
                jsonl_import_shell_quote_arg("/tmp/project\nnext")
            )),
        });

        let human = report.human_summary();
        ensure(
            human.contains("publisher said \\u001b[31mred\\nnext")
                && human.contains("Repair: ee index rebuild --workspace $'/tmp/project\\nnext'"),
            true,
            "human output renders terminal controls as visible escapes",
        )?;
        ensure(
            human.contains('\u{1b}')
                || human.contains("red\nnext")
                || human.contains("project\nnext"),
            false,
            "human output never emits raw terminal controls or injected lines",
        )
    }

    #[test]
    fn import_index_job_identity_is_workspace_scoped() -> TestResult {
        let memory_id = "mem_01234567890123456789012345";
        let first = import_search_index_job_id("wsp_first", memory_id);
        let second = import_search_index_job_id("wsp_second", memory_id);

        ensure(
            first == second,
            false,
            "the same imported memory id in two workspaces must not alias one durable job",
        )?;
        ensure(
            first.starts_with("sidx_") && first.len() == 31,
            true,
            "workspace-scoped import job ids preserve the canonical opaque shape",
        )
    }

    #[test]
    fn import_repair_shell_quote_preserves_special_paths_without_raw_controls() -> TestResult {
        ensure(
            jsonl_import_shell_quote_arg("/tmp/a 'quoted' path"),
            "'/tmp/a '\\''quoted'\\'' path'".to_owned(),
            "ordinary special paths retain portable single-quote escaping",
        )?;
        let control_path = jsonl_import_shell_quote_arg("/tmp/a 'quoted'\nline\\tail");
        ensure(
            control_path,
            "$'/tmp/a \\'quoted\\'\\nline\\\\tail'".to_owned(),
            "control-bearing paths use copy/pasteable visible ANSI-C quoting",
        )
    }

    #[test]
    fn parse_jsonl_header_accepts_header_record_only() -> TestResult {
        let header_line = sample_jsonl()
            .lines()
            .next()
            .ok_or_else(|| "sample JSONL must include a header line".to_string())?
            .to_string();
        let header = parse_jsonl_header(&header_line).map_err(|error| error.to_string())?;

        ensure(header.export_id, "exp-001".to_string(), "export id")?;
        ensure(
            parse_jsonl_header(r#"{"schema":"ee.export.memory.v1"}"#),
            Err(JsonlHeaderParseError::WrongSchema {
                schema: "ee.export.memory.v1".to_string(),
            }),
            "wrong schema",
        )
    }

    #[test]
    fn parse_jsonl_header_rejects_blank_required_fields() -> TestResult {
        let header_line = sample_jsonl()
            .lines()
            .next()
            .ok_or_else(|| "sample JSONL must include a header line".to_string())?
            .replace(
                "\"created_at\":\"2026-04-30T00:00:00Z\"",
                "\"created_at\":\"   \"",
            );

        let error = match parse_jsonl_header(&header_line) {
            Ok(_) => return Err("blank created_at must reject header".to_string()),
            Err(error) => error,
        };
        ensure(
            error,
            JsonlHeaderParseError::InvalidHeader {
                message: "header field `created_at` must not be blank".to_string(),
            },
            "blank created_at",
        )
    }

    #[test]
    fn parse_jsonl_source_collects_header_memory_and_tags() -> TestResult {
        let parsed = parse_jsonl_source(&sample_jsonl());

        ensure(parsed.has_errors(), false, "has errors")?;
        ensure(parsed.header.is_some(), true, "header parsed")?;
        ensure(parsed.footer.is_some(), true, "footer parsed")?;
        ensure(parsed.memories.len(), 1, "memory count")?;
        ensure(
            parsed
                .tags_by_memory
                .get("mem_01234567890123456789012345")
                .map(BTreeSet::len),
            Some(1),
            "tag count",
        )
    }

    #[test]
    fn parse_jsonl_source_reports_invalid_blank_header() -> TestResult {
        let input = sample_jsonl()
            .replace("\"ee_version\":\"0.1.0\"", "\"ee_version\":\"\"")
            .replace("\"export_id\":\"exp-001\"", "\"export_id\":\"   \"");
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(parsed.header.is_none(), true, "invalid header omitted")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line == Some(1)
                    && issue.code == "invalid_header"
                    && issue
                        .message
                        .contains("header field `ee_version` must not be blank")
            }),
            true,
            "invalid header issue",
        )
    }

    #[test]
    fn parse_jsonl_source_rejects_missing_header() -> TestResult {
        let parsed = parse_jsonl_source(
            r#"{"schema":"ee.export.memory.v1","memory_id":"mem_01234567890123456789012345","workspace_id":"wsp_01234567890123456789012345","level":"procedural","kind":"rule","content":"content","importance":0.8,"confidence":0.9,"utility":0.7,"created_at":"2026-04-30T00:00:00Z","updated_at":null,"expires_at":null,"source_agent":null,"provenance_uri":null,"superseded_by":null,"supersedes":null,"redacted":false,"redaction_reason":null}"#,
        );

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(
            parsed
                .issues
                .iter()
                .any(|issue| issue.code == "missing_header"),
            true,
            "missing header issue",
        )
    }

    #[test]
    fn parse_jsonl_source_rejects_missing_footer() -> TestResult {
        let input = sample_jsonl()
            .lines()
            .take(3)
            .collect::<Vec<_>>()
            .join("\n");
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(parsed.footer.is_none(), true, "footer absent")?;
        ensure(
            parsed
                .issues
                .iter()
                .any(|issue| issue.code == "missing_footer"),
            true,
            "missing footer issue",
        )
    }

    #[test]
    fn parse_jsonl_source_rejects_blank_footer_required_fields() -> TestResult {
        let input = sample_jsonl().replace(
            "\"completed_at\":\"2026-04-30T00:01:00Z\"",
            "\"completed_at\":\"  \"",
        );
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(parsed.footer.is_none(), true, "invalid footer omitted")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line == Some(4)
                    && issue.code == "invalid_footer"
                    && issue
                        .message
                        .contains("footer field `completed_at` must not be blank")
            }),
            true,
            "invalid footer issue",
        )
    }

    #[test]
    fn parse_jsonl_source_rejects_footer_export_id_mismatch() -> TestResult {
        let input = sample_jsonl().replace(
            "\"schema\":\"ee.export.footer.v1\",\"export_id\":\"exp-001\"",
            "\"schema\":\"ee.export.footer.v1\",\"export_id\":\"exp-other\"",
        );
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(parsed.footer.is_some(), true, "valid footer parsed")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line == Some(4)
                    && issue.code == "footer_export_id_mismatch"
                    && issue.message.contains("exp-other")
                    && issue.message.contains("exp-001")
            }),
            true,
            "footer mismatch issue",
        )
    }

    #[test]
    fn parse_jsonl_source_rejects_records_after_footer() -> TestResult {
        let mut lines = sample_jsonl()
            .lines()
            .map(str::to_owned)
            .collect::<Vec<_>>();
        let footer = lines
            .pop()
            .ok_or_else(|| "sample JSONL must include a footer".to_string())?;
        let trailing_memory = lines
            .get(1)
            .cloned()
            .ok_or_else(|| "sample JSONL must include a memory record".to_string())?
            .replace(
                "mem_01234567890123456789012345",
                "mem_22222222222222222222222222",
            );
        lines.push(footer);
        lines.push(trailing_memory);
        let parsed = parse_jsonl_source(&lines.join("\n"));

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line == Some(5)
                    && issue.code == "footer_not_last"
                    && issue.message.contains("final")
            }),
            true,
            "footer-not-last issue",
        )?;
        ensure(parsed.memories.len(), 1, "trailing memory ignored")
    }

    #[test]
    fn parse_jsonl_source_rejects_orphaned_tag_records() -> TestResult {
        let input = sample_jsonl().replace(
            "\"schema\":\"ee.export.tag.v1\",\"memory_id\":\"mem_01234567890123456789012345\"",
            "\"schema\":\"ee.export.tag.v1\",\"memory_id\":\"mem_99999999999999999999999999\"",
        );
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), true, "has errors")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line == Some(3)
                    && issue.code == "orphaned_tag_record"
                    && issue.message.contains("mem_99999999999999999999999999")
            }),
            true,
            "orphaned tag issue",
        )
    }

    #[test]
    fn import_report_json_redacts_sensitive_source_refs() -> TestResult {
        let report = import_report_fixture(
            "/Users/alice/private/export.jsonl?api_key=redaction-fixture",
            "jsonl:///Users/alice/private/export.jsonl?api_key=redaction-fixture",
        );
        let json = report.data_json();
        let rendered = json.to_string();

        assert!(
            rendered.contains("[REDACTED_PATH]"),
            "source refs should redact path-like values: {rendered}"
        );
        assert!(
            rendered.contains("[REDACTED:"),
            "source refs should redact secret-like values: {rendered}"
        );
        assert!(
            !rendered.contains("/Users/alice") && !rendered.contains("redaction-fixture"),
            "source refs leaked sensitive material: {rendered}"
        );
        ensure(
            report.source_path,
            "/Users/alice/private/export.jsonl?api_key=redaction-fixture".to_owned(),
            "raw report source_path remains available internally",
        )
    }

    #[test]
    fn import_report_json_redacts_windows_source_refs() -> TestResult {
        let report = import_report_fixture(
            r"C:\Users\Alice\private\export.jsonl?api_key=redaction-fixture",
            r"jsonl://C:\Users\Alice\private\export.jsonl?api_key=redaction-fixture",
        );
        let json = report.data_json();
        let rendered = json.to_string();

        assert!(
            rendered.contains("[REDACTED_PATH]"),
            "source refs should redact Windows path-like values: {rendered}"
        );
        assert!(
            rendered.contains("[REDACTED:"),
            "source refs should redact secret-like values: {rendered}"
        );
        assert!(
            !rendered.contains("C:\\Users")
                && !rendered.contains("Alice")
                && !rendered.contains("redaction-fixture"),
            "source refs leaked sensitive Windows material: {rendered}"
        );
        ensure(
            report.source_path,
            r"C:\Users\Alice\private\export.jsonl?api_key=redaction-fixture".to_owned(),
            "raw Windows report source_path remains available internally",
        )
    }

    #[test]
    fn import_report_json_redacts_unc_source_refs() -> TestResult {
        let report = import_report_fixture(
            r"\\fileserver\share\team\export.jsonl",
            r"jsonl://\\fileserver\share\team\export.jsonl",
        );
        let json = report.data_json();
        let rendered = json.to_string();

        assert!(
            rendered.contains("[REDACTED_PATH]"),
            "source refs should redact UNC path-like values: {rendered}"
        );
        assert!(
            !rendered.contains("fileserver") && !rendered.contains("share"),
            "source refs leaked UNC material: {rendered}"
        );
        Ok(())
    }

    #[test]
    fn import_report_json_preserves_safe_source_refs() -> TestResult {
        let report =
            import_report_fixture("fixtures/export.jsonl", "jsonl://fixtures/export.jsonl");
        let json = report.data_json();

        ensure(
            json["sourcePath"].as_str(),
            Some("fixtures/export.jsonl"),
            "safe sourcePath",
        )?;
        ensure(
            json["sourceId"].as_str(),
            Some("jsonl://fixtures/export.jsonl"),
            "safe sourceId",
        )
    }

    #[test]
    fn parse_jsonl_source_warns_on_footer_tag_count_mismatch() -> TestResult {
        let input = sample_jsonl().replace("\"tag_count\":1", "\"tag_count\":2");
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), false, "warning only")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line.is_none()
                    && issue.code == "footer_tag_count_mismatch"
                    && issue.severity == JsonlImportIssueSeverity::Warning
            }),
            true,
            "tag count warning",
        )
    }

    #[test]
    fn parse_jsonl_source_warns_on_footer_artifact_count_mismatch() -> TestResult {
        let artifact_line = r#"{"schema":"ee.export.artifact.v1"}"#;
        let input = sample_jsonl()
            .replace(
                r#"{"schema":"ee.export.footer.v1""#,
                &format!("{artifact_line}\n{{\"schema\":\"ee.export.footer.v1\""),
            )
            .replace("\"total_records\":4", "\"total_records\":5")
            .replace(
                "\"memory_count\":1,\"link_count\"",
                "\"memory_count\":1,\"artifact_count\":2,\"link_count\"",
            );
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), false, "warning only")?;
        ensure(parsed.artifact_records, 1, "raw artifact records")?;
        ensure(
            parsed.ignored_records,
            1,
            "artifact row remains ignored for import",
        )?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line.is_none()
                    && issue.code == "footer_artifact_count_mismatch"
                    && issue.severity == JsonlImportIssueSeverity::Warning
            }),
            true,
            "artifact count warning",
        )
    }

    #[test]
    fn parse_jsonl_source_counts_duplicate_tag_records_separately() -> TestResult {
        let tag_line = r#"{"schema":"ee.export.tag.v1","memory_id":"mem_01234567890123456789012345","tag":"Release","created_at":"2026-04-30T00:00:00Z"}"#;
        let input = sample_jsonl()
            .replace(tag_line, &format!("{tag_line}\n{tag_line}"))
            .replace("\"total_records\":4", "\"total_records\":5")
            .replace("\"tag_count\":1", "\"tag_count\":2");
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), false, "duplicate tag record is valid")?;
        ensure(parsed.tag_records, 2, "raw tag records")?;
        ensure(
            parsed
                .tags_by_memory
                .get("mem_01234567890123456789012345")
                .map(BTreeSet::len),
            Some(1),
            "deduplicated stored tags",
        )?;
        ensure(
            parsed
                .issues
                .iter()
                .any(|issue| issue.code == "footer_tag_count_mismatch"),
            false,
            "footer tag count should compare raw tag records",
        )?;

        let report = report_from_parsed(
            Path::new("/workspace"),
            Path::new("export.jsonl"),
            "jsonl://export.jsonl",
            true,
            &parsed,
        );
        ensure(report.tag_records, 2, "reported tag records")?;

        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );
        ensure(prepared.has_errors(), false, "prepared has no errors")?;
        let memory = prepared
            .memories
            .first()
            .ok_or_else(|| "prepared memory missing".to_owned())?;
        ensure(memory.tag_count, 1, "storage tag count stays deduplicated")
    }

    #[test]
    fn parse_jsonl_source_warns_on_footer_total_records_mismatch() -> TestResult {
        let input = sample_jsonl().replace("\"total_records\":4", "\"total_records\":99");
        let parsed = parse_jsonl_source(&input);

        ensure(parsed.has_errors(), false, "warning only")?;
        ensure(
            parsed.issues.iter().any(|issue| {
                issue.line.is_none()
                    && issue.code == "footer_total_records_mismatch"
                    && issue.severity == JsonlImportIssueSeverity::Warning
                    && issue.message.contains("99")
                    && issue.message.contains("4")
            }),
            true,
            "total record count warning",
        )
    }

    #[test]
    fn prepare_memories_validates_scores() -> TestResult {
        let input = sample_jsonl().replace(r#""confidence":0.9"#, r#""confidence":1.5"#);
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );

        ensure(prepared.has_errors(), true, "prepared has errors")?;
        ensure(
            prepared
                .issues
                .iter()
                .any(|issue| issue.code == "invalid_memory_confidence"),
            true,
            "invalid confidence issue",
        )
    }

    #[test]
    fn prepare_memories_rejects_scores_that_round_into_range_after_narrowing() -> TestResult {
        let input =
            sample_jsonl().replace(r#""confidence":0.9"#, r#""confidence":1.0000000000000002"#);
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );

        ensure(prepared.has_errors(), true, "prepared has errors")?;
        ensure(
            prepared
                .issues
                .iter()
                .any(|issue| issue.code == "invalid_memory_confidence"),
            true,
            "rounded invalid confidence issue",
        )
    }

    #[test]
    fn prepare_memories_preserves_record_trust_metadata() -> TestResult {
        let input = sample_jsonl().replace(
            r#""utility":0.7,"created_at""#,
            r#""utility":0.7,"trust_class":"human_explicit","trust_subclass":"project-rule","created_at""#,
        );
        let parsed = parse_jsonl_source(&input);
        // Record-level human_explicit on a native artifact requires the
        // artifact to authenticate (TC-D14); this test covers preservation,
        // not the gate, so it models the authenticated case.
        let prepared =
            prepare_memories(&parsed, "wsp_01234567890123456789012345", &authenticated());

        ensure(prepared.has_errors(), false, "prepared has no errors")?;
        let memory = prepared
            .memories
            .first()
            .ok_or_else(|| "prepared memory missing".to_string())?;
        ensure(
            memory.input.trust_class.as_str(),
            "human_explicit",
            "record trust_class overrides header",
        )?;
        ensure(
            memory.input.trust_subclass.as_deref(),
            Some("project-rule"),
            "record trust_subclass overrides header",
        )
    }

    #[test]
    fn prepare_memories_preserves_missing_record_trust_subclass() -> TestResult {
        let input = sample_jsonl().replace(
            r#""utility":0.7,"created_at""#,
            r#""utility":0.7,"trust_class":"human_explicit","created_at""#,
        );
        let parsed = parse_jsonl_source(&input);
        let prepared =
            prepare_memories(&parsed, "wsp_01234567890123456789012345", &authenticated());

        ensure(prepared.has_errors(), false, "prepared has no errors")?;
        let memory = prepared
            .memories
            .first()
            .ok_or_else(|| "prepared memory missing".to_string())?;
        ensure(
            memory.input.trust_class.as_str(),
            "human_explicit",
            "record trust_class overrides header",
        )?;
        ensure(
            memory.input.trust_subclass.as_deref(),
            None,
            "missing record trust_subclass stays absent",
        )
    }

    #[test]
    fn prepare_memories_rejects_external_human_explicit_trust_override() -> TestResult {
        let input = sample_jsonl()
            .replace(
                r#""import_source":"native""#,
                r#""import_source":"external_import""#,
            )
            .replace(
                r#""utility":0.7,"created_at""#,
                r#""utility":0.7,"trust_class":"human_explicit","created_at""#,
            );
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );

        ensure(prepared.has_errors(), true, "prepared has errors")?;
        ensure(prepared.memories.len(), 0, "external human memory blocked")?;
        ensure(
            prepared.issues.iter().any(|issue| {
                issue.code == "external_import_human_explicit_trust_class"
                    && issue.message.contains("external_import")
                    && issue.message.contains("agent_assertion")
            }),
            true,
            "external human_explicit issue",
        )
    }

    #[test]
    fn authenticated_jsonl_cannot_mint_peer_human_attested() -> TestResult {
        let input = sample_jsonl().replace(
            r#""utility":0.7,"created_at""#,
            r#""utility":0.7,"trust_class":"peer_human_attested","created_at""#,
        );
        let parsed = parse_jsonl_source(&input);
        let prepared =
            prepare_memories(&parsed, "wsp_01234567890123456789012345", &authenticated());

        ensure(prepared.has_errors(), true, "prepared has errors")?;
        ensure(prepared.memories.len(), 0, "peer attestation row blocked")?;
        ensure(
            prepared.issues.iter().any(|issue| {
                issue.code == PEER_HUMAN_ATTESTED_IMPORT_PATH_REQUIRED_CODE
                    && issue
                        .message
                        .contains("signed active-member admission path")
            }),
            true,
            "peer attestation requires team import issue",
        )
    }

    #[test]
    fn prepare_memories_preserves_lifecycle_metadata() -> TestResult {
        let input = sample_jsonl().replace(
            r#""updated_at":null,"expires_at":null"#,
            r#""updated_at":null,"tombstoned_at":"2026-05-02T00:00:00Z","tombstoned_reason":"superseded by newer release rule","valid_from":"2026-05-01T00:00:00Z","expires_at":"2026-06-01T00:00:00Z""#,
        );
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );
        ensure(prepared.has_errors(), false, "prepared has no errors")?;
        let memory = prepared
            .memories
            .first()
            .ok_or_else(|| "prepared memory missing".to_string())?;

        ensure(
            memory.tombstoned_at.as_deref(),
            Some("2026-05-02T00:00:00Z"),
            "tombstoned_at",
        )?;
        ensure(
            memory.tombstoned_reason.as_deref(),
            Some("superseded by newer release rule"),
            "tombstoned_reason",
        )?;
        ensure(
            memory.input.valid_from.as_deref(),
            Some("2026-05-01T00:00:00Z"),
            "valid_from",
        )?;
        ensure(
            memory.input.valid_to.as_deref(),
            Some("2026-06-01T00:00:00Z"),
            "valid_to fallback from expires_at",
        )
    }

    #[test]
    fn prepare_memories_preserves_export_graph_fields_in_audit_details() -> TestResult {
        let input = sample_jsonl_with_graph_fields();
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );
        ensure(prepared.has_errors(), false, "prepared has no errors")?;
        let memory = prepared
            .memories
            .first()
            .ok_or_else(|| "prepared memory missing".to_string())?;
        ensure(memory.bayes_posterior, Some((2.5, 1.5)), "bayes posterior")?;

        let details: JsonValue =
            serde_json::from_str(&memory.details).map_err(|error| error.to_string())?;
        let graph_fields = details
            .get("sourceGraphFields")
            .ok_or_else(|| format!("missing sourceGraphFields: {details}"))?;
        ensure(
            graph_fields
                .get("pagerank_score")
                .and_then(JsonValue::as_f64),
            Some(0.12),
            "pagerank_score",
        )?;
        ensure(
            graph_fields
                .get("betweenness_score")
                .and_then(JsonValue::as_f64),
            Some(0.34),
            "betweenness_score",
        )?;
        ensure(
            graph_fields
                .get("hits_authority")
                .and_then(JsonValue::as_f64),
            Some(0.56),
            "hits_authority",
        )?;
        ensure(
            graph_fields.get("hits_hub").and_then(JsonValue::as_f64),
            Some(0.78),
            "hits_hub",
        )?;
        ensure(
            graph_fields.get("onion_layer").and_then(JsonValue::as_u64),
            Some(3),
            "onion_layer",
        )?;
        ensure(
            graph_fields.get("k_truss_max").and_then(JsonValue::as_u64),
            Some(4),
            "k_truss_max",
        )?;
        ensure(
            graph_fields
                .get("articulation_point")
                .and_then(JsonValue::as_bool),
            Some(true),
            "articulation_point",
        )
    }

    #[test]
    fn prepare_memories_rejects_partial_bayes_posterior() -> TestResult {
        let input = sample_jsonl().replace(
            r#""utility":0.7,"created_at""#,
            r#""utility":0.7,"bayes_alpha":2.5,"created_at""#,
        );
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(
            &parsed,
            "wsp_01234567890123456789012345",
            &unauthenticated(),
        );

        ensure(prepared.has_errors(), true, "prepared has errors")?;
        ensure(
            prepared
                .issues
                .iter()
                .any(|issue| issue.code == "invalid_memory_bayes_posterior"),
            true,
            "partial bayes posterior issue",
        )
    }

    #[test]
    fn import_jsonl_preserves_memory_chronology_and_provenance() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let mut records = sample_jsonl()
            .lines()
            .map(serde_json::from_str::<JsonValue>)
            .collect::<Result<Vec<_>, _>>()
            .map_err(|error| error.to_string())?;
        let template = records[1].clone();
        let mut footer = records[3].clone();
        records.truncate(1);
        records[0]["record_count"] = json!(5);
        let cases = [
            (
                "2020-01-02T03:04:05.123456789+05:30",
                Some("2020-02-03T04:05:06.987654321-04:00"),
                None,
                None,
            ),
            (
                "2021-01-02T00:00:00Z",
                Some("2021-03-03T00:00:00Z"),
                Some("2021-02-02T00:00:00Z"),
                Some("2019-01-01T00:00:00Z"),
            ),
            ("2022-01-02T00:00:00Z", None, None, None),
            (
                "2023-01-02T00:00:00Z",
                None,
                Some("2023-02-02T00:00:00Z"),
                None,
            ),
        ];
        for (index, &(created, updated, tombstoned, valid_from)) in cases.iter().enumerate() {
            let mut memory = template.clone();
            memory["memory_id"] =
                json!(MemoryId::from_uuid(Uuid::from_u128(index as u128 + 1)).to_string());
            memory["content"] = json!(format!("Preserve historical release rule {index}."));
            memory["created_at"] = json!(created);
            memory["updated_at"] = json!(updated);
            memory["tombstoned_at"] = json!(tombstoned);
            memory["valid_from"] = json!(valid_from);
            memory["bayes_alpha"] = json!(2.5);
            memory["bayes_beta"] = json!(1.5);
            records.push(memory);
        }
        footer["total_records"] = json!(6);
        footer["memory_count"] = json!(4);
        footer["tag_count"] = json!(0);
        records.push(footer);
        let source = tempdir.path().join("chronology.jsonl");
        fs::write(
            &source,
            records
                .iter()
                .map(JsonValue::to_string)
                .collect::<Vec<_>>()
                .join("\n"),
        )
        .map_err(|error| error.to_string())?;
        let options = JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: None,
            source_path: source,
            dry_run: false,
        };
        let report = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(
            report.status.as_str(),
            "completed",
            "chronology import status",
        )?;
        ensure(report.memories_imported, 4, "all chronology cases imported")?;
        let connection = DbConnection::open(DatabaseConfig::file(database_path(&options)))
            .map_err(|error| error.to_string())?;
        for (index, &(created, updated, tombstoned, valid_from)) in cases.iter().enumerate() {
            let id = MemoryId::from_uuid(Uuid::from_u128(index as u128 + 1)).to_string();
            let memory = connection
                .get_memory(&id)
                .map_err(|error| error.to_string())?
                .ok_or("imported chronology memory missing")?;
            ensure(
                memory.created_at.as_str(),
                created,
                "exact original creation time",
            )?;
            ensure(
                memory.updated_at.as_str(),
                updated.or(tombstoned).unwrap_or(created),
                "modification time survives posterior and tombstone restoration",
            )?;
            ensure(
                memory.tombstoned_at.as_deref(),
                tombstoned,
                "original tombstone",
            )?;
            ensure(
                memory.valid_from.as_deref(),
                Some(valid_from.unwrap_or(created)),
                "validity defaults to original creation, not import time",
            )?;
            ensure(
                connection
                    .get_memory_bayes_posterior(&id)
                    .map_err(|error| error.to_string())?,
                Some((2.5, 1.5)),
                "posterior restored even for historical tombstones",
            )?;
            ensure(
                memory.provenance_chain_hash.as_deref(),
                Some(crate::db::compute_memory_provenance_chain_hash(&memory).as_str()),
                "provenance hash covers the stored historical creation time",
            )?;
        }
        Ok(())
    }

    #[test]
    fn import_jsonl_restores_exported_bayes_posterior() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let workspace = tempdir.path().join("workspace");
        fs::create_dir_all(&workspace).map_err(|error| error.to_string())?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, sample_jsonl_with_graph_fields()).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;
        ensure(report.status.as_str(), "completed", "import status")?;
        ensure(report.memories_imported, 1, "memories imported")?;

        let connection =
            DbConnection::open(DatabaseConfig::file(database_path(&JsonlImportOptions {
                workspace_path: workspace,
                database_path: None,
                source_path: PathBuf::new(),
                dry_run: false,
            })))
            .map_err(|error| error.to_string())?;
        let posterior = connection
            .get_memory_bayes_posterior("mem_01234567890123456789012345")
            .map_err(|error| error.to_string())?;
        ensure(posterior, Some((2.5, 1.5)), "restored posterior")
    }

    #[test]
    fn import_jsonl_leaves_index_fresh_and_content_searchable() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let workspace = tempdir.path().join("workspace");
        fs::create_dir_all(&workspace).map_err(|error| error.to_string())?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, sample_jsonl()).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;
        ensure(report.status.as_str(), "completed", "import status")?;
        ensure(report.memories_imported, 1, "memories imported")?;
        ensure(
            report
                .issues
                .iter()
                .any(|issue| issue.code == "import_index_publish_failed"),
            false,
            "import drain publishes without a failure issue",
        )?;

        let status =
            crate::core::index::get_index_status(&crate::core::index::IndexStatusOptions {
                workspace_path: workspace.clone(),
                database_path: None,
                index_dir: None,
            })
            .map_err(|error| format!("index status: {error:?}"))?;
        ensure(
            status.health,
            crate::core::index::IndexHealth::Ready,
            "index ready after import without rebuild",
        )?;
        ensure(
            status.db_generation.is_some(),
            true,
            "db generation present",
        )?;
        ensure(
            status.db_generation == status.index_generation,
            true,
            "import leaves database and index generations equal",
        )?;

        let connection =
            DbConnection::open(DatabaseConfig::file(database_path(&JsonlImportOptions {
                workspace_path: workspace.clone(),
                database_path: None,
                source_path: PathBuf::new(),
                dry_run: false,
            })))
            .map_err(|error| error.to_string())?;
        let workspace_id =
            ensure_workspace(&connection, &workspace).map_err(|error| error.to_string())?;
        let pending = connection
            .list_pending_search_index_jobs(&workspace_id, None)
            .map_err(|error| error.to_string())?;
        ensure(pending.len(), 0, "pending index jobs after import drain")?;

        let search = crate::core::search::run_search_with_filters(
            &crate::core::search::SearchOptions {
                workspace_path: workspace,
                database_path: None,
                index_dir: None,
                query: "cargo fmt release".to_owned(),
                limit: 5,
                speed: crate::search::SpeedMode::Instant,
                explain: false,
                as_of: None,
                include_tombstoned: false,
                include_expired: false,
                include_future: false,
                include_stale: false,
                relevance_floor: Some(0.0),
                dedup_mode: crate::core::search::SearchDedupMode::DocId,
                source_mode: crate::core::search::SearchSourceMode::LexicalOnly,
                strict_source_mode: true,
                memory_scope: crate::models::MemoryScope::Workspace,
                strict_scope: false,
            },
            None,
            &[],
        )
        .map_err(|error| format!("post-import search: {error:?}"))?;
        let actual_ids = search
            .results
            .iter()
            .map(|hit| hit.doc_id.as_str())
            .collect::<BTreeSet<_>>();
        ensure(
            actual_ids,
            BTreeSet::from(["mem_01234567890123456789012345"]),
            "strict lexical search returns exactly the imported memory id",
        )?;
        ensure(
            search
                .degraded
                .iter()
                .any(|entry| entry.code == "search_index_stale"),
            false,
            "no stale advisory after import drain",
        )
    }

    #[cfg(unix)]
    #[test]
    fn import_jsonl_survives_noncompleted_index_publication_with_retryable_job() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let workspace = tempdir.path().join("workspace with spaces");
        fs::create_dir_all(workspace.join(".ee")).map_err(|error| error.to_string())?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, sample_jsonl()).map_err(|error| error.to_string())?;

        let index_dir = workspace
            .join(".ee")
            .join(crate::core::index::DEFAULT_INDEX_SUBDIR);
        let blocked_target = workspace.join(".ee").join("index-publish-blocker");
        fs::create_dir_all(&blocked_target).map_err(|error| error.to_string())?;
        std::os::unix::fs::symlink(&blocked_target, &index_dir)
            .map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source.clone(),
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;
        ensure(
            report.status.as_str(),
            "completed",
            "source import remains completed",
        )?;
        ensure(
            report.memories_imported,
            1,
            "source memory remains imported",
        )?;
        let issue = report
            .issues
            .iter()
            .find(|issue| issue.code == "import_index_publish_failed")
            .ok_or("missing truthful import index publication issue")?;
        let expected_repair = format!(
            "ee index rebuild --workspace {}",
            jsonl_import_shell_quote_arg(workspace.to_string_lossy().as_ref())
        );
        ensure(
            issue
                .message
                .contains("automatic publication of durable search-index jobs did not complete")
                && issue.message.contains("Search may omit imported memories")
                && issue.repair.as_deref() == Some(expected_repair.as_str()),
            true,
            "publication issue carries exact failure truth and shell-safe repair",
        )?;
        ensure(
            report.degraded_json(),
            vec![json!({
                "code": "import_index_publish_failed",
                "severity": "warning",
                "message": issue.message,
                "repair": expected_repair,
            })],
            "publication failure is a response-level degradation",
        )?;
        let data = report.data_json();
        let issue_json = data["issues"]
            .as_array()
            .and_then(|issues| {
                issues.iter().find(|candidate| {
                    candidate["code"].as_str() == Some("import_index_publish_failed")
                })
            })
            .ok_or("machine report omitted publication failure issue")?;
        ensure(
            issue_json["repair"].as_str(),
            Some(expected_repair.as_str()),
            "machine issue carries structured repair",
        )?;
        let human = report.human_summary();
        ensure(
            human.contains("[warning] import_index_publish_failed")
                && human.contains(&format!("Repair: {expected_repair}")),
            true,
            "human output surfaces the publication failure and repair",
        )?;

        let connection = DbConnection::open(DatabaseConfig::file(workspace.join(".ee/ee.db")))
            .map_err(|error| error.to_string())?;
        ensure(
            connection
                .get_memory("mem_01234567890123456789012345")
                .map_err(|error| error.to_string())?
                .is_some(),
            true,
            "source-of-truth imported memory survives publication failure",
        )?;
        let workspace_id =
            ensure_workspace(&connection, &workspace).map_err(|error| error.to_string())?;
        let jobs = connection
            .list_search_index_jobs(&workspace_id, None)
            .map_err(|error| error.to_string())?;
        ensure(
            jobs.iter().any(|job| {
                matches!(
                    job.status_enum(),
                    Some(
                        crate::db::SearchIndexJobStatus::Pending
                            | crate::db::SearchIndexJobStatus::Failed
                    )
                )
            }),
            true,
            "noncompleted publication leaves durable retryable index work",
        )?;

        // Preserve the failed symlink as evidence while making the canonical
        // index path available. An identical reimport must requeue the SAME
        // deterministic failed job even though it inserts no new memory row.
        fs::rename(&index_dir, workspace.join(".ee/index-publish-blocker-link"))
            .map_err(|error| error.to_string())?;
        let retry = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source.clone(),
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;
        ensure(retry.memories_imported, 0, "reimport inserts no duplicate")?;
        ensure(
            retry.memories_skipped_duplicate,
            1,
            "reimport recognizes the durable memory",
        )?;
        ensure(
            retry
                .issues
                .iter()
                .any(|issue| issue.code == "import_index_publish_failed"),
            false,
            "reimport retries and completes prior failed publication",
        )?;
        let retried_job = connection
            .get_search_index_job(&import_search_index_job_id(
                &workspace_id,
                "mem_01234567890123456789012345",
            ))
            .map_err(|error| error.to_string())?
            .ok_or("deterministic import index job disappeared")?;
        ensure(
            retried_job.status_enum(),
            Some(crate::db::SearchIndexJobStatus::Completed),
            "same deterministic job completed after retry",
        )?;
        let status =
            crate::core::index::get_index_status(&crate::core::index::IndexStatusOptions {
                workspace_path: workspace.clone(),
                database_path: None,
                index_dir: None,
            })
            .map_err(|error| format!("index status after retry: {error:?}"))?;
        ensure(
            status.health == crate::core::index::IndexHealth::Ready
                && status.db_generation.is_some()
                && status.db_generation == status.index_generation,
            true,
            "retry converges the index to the committed database generation",
        )?;
        let search = crate::core::search::run_search_with_filters(
            &crate::core::search::SearchOptions {
                workspace_path: workspace.clone(),
                database_path: None,
                index_dir: None,
                query: "cargo fmt release".to_owned(),
                limit: 5,
                speed: crate::search::SpeedMode::Instant,
                explain: false,
                as_of: None,
                include_tombstoned: false,
                include_expired: false,
                include_future: false,
                include_stale: false,
                relevance_floor: Some(0.0),
                dedup_mode: crate::core::search::SearchDedupMode::DocId,
                source_mode: crate::core::search::SearchSourceMode::LexicalOnly,
                strict_source_mode: true,
                memory_scope: crate::models::MemoryScope::Workspace,
                strict_scope: false,
            },
            None,
            &[],
        )
        .map_err(|error| format!("post-retry search: {error:?}"))?;
        ensure(
            search
                .results
                .iter()
                .any(|hit| hit.doc_id == "mem_01234567890123456789012345"),
            true,
            "retried import memory is immediately searchable",
        )?;

        let job_id = import_search_index_job_id(&workspace_id, "mem_01234567890123456789012345");
        let skipped_report = crate::core::index::IndexProcessingJobReport {
            job_id: job_id.clone(),
            job_type: crate::db::SearchIndexJobType::SingleDocument
                .as_str()
                .to_owned(),
            document_source: Some("memory".to_owned()),
            document_id: Some("mem_01234567890123456789012345".to_owned()),
            outcome: "skipped".to_owned(),
            processing_mode: "concurrent_claim".to_owned(),
            fallback_to_full: None,
            documents_total: 1,
            documents_indexed: 0,
            error: Some("another publisher held the claim".to_owned()),
        };
        ensure(
            jsonl_import_publication_failure(
                &connection,
                std::slice::from_ref(&job_id),
                &workspace,
                &workspace.join(".ee/ee.db"),
                Ok(vec![skipped_report]),
            )
            .map_err(|error| error.to_string())?,
            None,
            "authoritative completed job and ready index override a local skipped report",
        )?;

        // A completed job is re-armed only when authoritative index state is
        // missing/stale. Preserve the published directory, then prove the
        // identical import reuses the same logical job to restore it.
        fs::rename(
            &index_dir,
            workspace.join(".ee/index-completed-but-missing"),
        )
        .map_err(|error| error.to_string())?;
        let recovered = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;
        ensure(
            recovered
                .issues
                .iter()
                .any(|issue| issue.code == "import_index_publish_failed"),
            false,
            "completed-job reimport restores a missing derived index",
        )?;
        let import_jobs = connection
            .list_search_index_jobs(&workspace_id, None)
            .map_err(|error| error.to_string())?
            .into_iter()
            .filter(|job| job.id == job_id)
            .collect::<Vec<_>>();
        ensure(
            import_jobs.len(),
            1,
            "missing-index recovery reuses one deterministic logical job",
        )?;
        ensure(
            import_jobs[0].status_enum(),
            Some(crate::db::SearchIndexJobStatus::Completed),
            "re-armed completed job returns to completed",
        )?;
        ensure(
            jsonl_import_index_is_ready(&workspace, &workspace.join(".ee/ee.db")),
            true,
            "completed-job recovery restores ready generation equality",
        )
    }

    fn human_explicit_jsonl() -> String {
        sample_jsonl().replace(
            r#""utility":0.7,"created_at""#,
            r#""utility":0.7,"trust_class":"human_explicit","created_at""#,
        )
    }

    /// Emit a native artifact through the real exporter, authenticated by the
    /// store at `workspace` and bound to `workspace_scope`.
    fn authenticate_sample(
        artifact: &str,
        workspace: &Path,
        workspace_scope: &str,
    ) -> Result<String, String> {
        use crate::models::ExportScope;
        use crate::output::jsonl_export::JsonlExporter;
        use crate::policy::import_auth::authenticate_artifact;

        let root = StoreAuthRoot::open_or_create(workspace_keys_dir(workspace))
            .map_err(|error| error.message())?;
        let mut output = Vec::new();
        let mut exporter = JsonlExporter::new(&mut output, RedactionLevel::None, ExportScope::All);
        let mut footer = None;
        for line in artifact.lines() {
            let value: JsonValue =
                serde_json::from_str(line.trim()).map_err(|error| error.to_string())?;
            match value.get("schema").and_then(JsonValue::as_str) {
                Some(EXPORT_HEADER_SCHEMA_V1) => exporter.write_header(
                    serde_json::from_value(value).map_err(|error| error.to_string())?,
                ),
                Some(EXPORT_MEMORY_SCHEMA_V1) => exporter.write_memory(
                    serde_json::from_value(value).map_err(|error| error.to_string())?,
                ),
                Some(EXPORT_TAG_SCHEMA_V1) => exporter
                    .write_tag(serde_json::from_value(value).map_err(|error| error.to_string())?),
                Some(EXPORT_LINK_SCHEMA_V1) => exporter
                    .write_link(serde_json::from_value(value).map_err(|error| error.to_string())?),
                Some(EXPORT_FOOTER_SCHEMA_V1) => {
                    footer = Some(
                        serde_json::from_value::<ExportFooter>(value)
                            .map_err(|error| error.to_string())?,
                    );
                    Ok(())
                }
                _ => return Err("unsupported authenticated fixture record".to_owned()),
            }
            .map_err(|error| error.to_string())?;
        }
        let (records_root, record_count) = exporter.finalize_records_root();
        let header = authenticate_artifact(
            &root,
            MacDomain::NativeImportRecordsRoot,
            &ArtifactContext {
                artifact_family: EXPORT_ARTIFACT_FAMILY,
                record_encoding_version: EXPORT_RECORD_ENCODING_V1,
                source_key_namespace: STORE_KEY_NAMESPACE_V1,
                workspace_scope,
            },
            &records_root,
            record_count,
        )
        .map_err(|error| error.message())?;
        let mut footer = footer.ok_or("authenticated fixture has no footer")?;
        footer.authentication = Some(header);
        exporter
            .write_footer(footer)
            .map_err(|error| error.to_string())?;
        String::from_utf8(output).map_err(|error| error.to_string())
    }

    /// Workspace fixture for authenticated-import tests: canonical path, a
    /// migrated DB, and the workspace id `ee import jsonl` will resolve.
    fn authenticated_import_workspace(
        tempdir: &tempfile::TempDir,
    ) -> Result<(PathBuf, String), String> {
        let workspace = tempdir.path().join("workspace");
        fs::create_dir_all(workspace.join(crate::config::WORKSPACE_MARKER))
            .map_err(|error| error.to_string())?;
        let workspace = workspace
            .canonicalize()
            .map_err(|error| error.to_string())?;
        let connection = DbConnection::open(DatabaseConfig::file(
            workspace
                .join(crate::config::WORKSPACE_MARKER)
                .join(DEFAULT_DB_FILE),
        ))
        .map_err(|error| error.to_string())?;
        connection.migrate().map_err(|error| error.to_string())?;
        let workspace_id =
            ensure_workspace(&connection, &workspace).map_err(|error| error.to_string())?;
        Ok((workspace, workspace_id))
    }

    #[test]
    fn native_human_explicit_without_authentication_is_refused() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let (workspace, _workspace_id) = authenticated_import_workspace(&tempdir)?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, human_explicit_jsonl()).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "rejected", "import status")?;
        ensure(report.memories_imported, 0, "memories imported")?;
        ensure(
            report
                .issues
                .iter()
                .any(|issue| issue.code == UNAUTHENTICATED_NATIVE_IMPORT_TRUST_CODE),
            true,
            "unauthenticated native trust issue",
        )?;
        let connection = DbConnection::open(DatabaseConfig::file(
            workspace
                .join(crate::config::WORKSPACE_MARKER)
                .join(DEFAULT_DB_FILE),
        ))
        .map_err(|error| error.to_string())?;
        ensure(
            connection
                .get_memory("mem_01234567890123456789012345")
                .map_err(|error| error.to_string())?
                .is_none(),
            true,
            "refused import must leave zero rows",
        )
    }

    #[test]
    fn verified_backup_restore_caps_foreign_human_explicit_trust() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let (workspace, _workspace_id) = authenticated_import_workspace(&tempdir)?;
        let source = tempdir.path().join("verified-backup.jsonl");
        fs::write(&source, human_explicit_jsonl()).map_err(|error| error.to_string())?;

        let report = import_verified_backup_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "completed", "import status")?;
        ensure(report.memories_imported, 1, "memories imported")?;
        ensure(
            report.issues.iter().any(|issue| {
                issue.code == VERIFIED_BACKUP_TRUST_DOWNGRADED_CODE
                    && issue.severity == JsonlImportIssueSeverity::Warning
            }),
            true,
            "verified backup trust downgrade warning",
        )?;
        let connection = DbConnection::open(DatabaseConfig::file(
            workspace
                .join(crate::config::WORKSPACE_MARKER)
                .join(DEFAULT_DB_FILE),
        ))
        .map_err(|error| error.to_string())?;
        let stored = connection
            .get_memory("mem_01234567890123456789012345")
            .map_err(|error| error.to_string())?
            .ok_or("restored memory missing")?;
        ensure(
            stored.trust_class.as_str(),
            "agent_validated",
            "verified backup trust cap",
        )?;
        ensure(
            connection
                .list_search_index_jobs(&stored.workspace_id, None)
                .map_err(|error| error.to_string())?
                .is_empty(),
            true,
            "backup record import leaves the job ledger to authenticated history recovery",
        )
    }

    #[test]
    fn authenticated_native_human_explicit_import_round_trips() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let (workspace, workspace_id) = authenticated_import_workspace(&tempdir)?;
        let artifact = authenticate_sample(&human_explicit_jsonl(), &workspace, &workspace_id)?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, artifact).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "completed", "import status")?;
        ensure(report.memories_imported, 1, "memories imported")?;
        let connection = DbConnection::open(DatabaseConfig::file(
            workspace
                .join(crate::config::WORKSPACE_MARKER)
                .join(DEFAULT_DB_FILE),
        ))
        .map_err(|error| error.to_string())?;
        let stored = connection
            .get_memory("mem_01234567890123456789012345")
            .map_err(|error| error.to_string())?
            .ok_or("imported memory missing")?;
        ensure(
            stored.trust_class.as_str(),
            "human_explicit",
            "authenticated native import preserves human_explicit",
        )
    }

    #[test]
    fn native_authentication_binds_exported_tags_and_links() -> TestResult {
        for mutation in [
            "unchanged",
            "edit_tag",
            "edit_link",
            "remove_tag",
            "remove_link",
            "append_tag",
            "append_link",
            "reorder",
        ] {
            let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
            let (workspace, workspace_id) = authenticated_import_workspace(&tempdir)?;
            let mut records = linked_jsonl_values()?;
            records[1]["trust_class"] = json!("human_explicit");
            records[3]["trust_class"] = json!("human_explicit");
            let artifact =
                authenticate_sample(&jsonl_values_text(&records), &workspace, &workspace_id)?;
            let mut lines = artifact.lines().map(str::to_owned).collect::<Vec<_>>();
            let footer: JsonValue =
                serde_json::from_str(&lines[5]).map_err(|error| error.to_string())?;
            ensure(
                footer["authentication"]["recordCount"].as_u64(),
                Some(4),
                "two memories, one tag, and one link are authenticated",
            )?;

            // Preserve every untouched line byte-for-byte. Reserializing the
            // memory lines here would mask an unauthenticated tag/link defect.
            match mutation {
                "unchanged" => {}
                "edit_tag" | "append_tag" => {
                    let mut tag: JsonValue =
                        serde_json::from_str(&lines[2]).map_err(|error| error.to_string())?;
                    tag["tag"] = json!("forged-routing-tag");
                    if mutation == "edit_tag" {
                        lines[2] = tag.to_string();
                    } else {
                        lines.insert(5, tag.to_string());
                    }
                }
                "edit_link" | "append_link" => {
                    let mut link: JsonValue =
                        serde_json::from_str(&lines[4]).map_err(|error| error.to_string())?;
                    link["weight"] = json!(0.25);
                    if mutation == "edit_link" {
                        lines[4] = link.to_string();
                    } else {
                        link["link_id"] =
                            json!(MemoryLinkId::from_uuid(Uuid::from_u128(4)).to_string());
                        let source = link["source_memory_id"].clone();
                        link["source_memory_id"] = link["target_memory_id"].clone();
                        link["target_memory_id"] = source;
                        lines.insert(5, link.to_string());
                    }
                }
                "remove_tag" => {
                    lines.remove(2);
                }
                "remove_link" => {
                    lines.remove(4);
                }
                "reorder" => lines.swap(2, 4),
                _ => return Err("unknown tamper test case".to_owned()),
            }
            let source_path = tempdir.path().join("source.jsonl");
            fs::write(&source_path, lines.join("\n")).map_err(|error| error.to_string())?;
            let report = import_jsonl_records(&JsonlImportOptions {
                workspace_path: workspace.clone(),
                database_path: None,
                source_path,
                dry_run: false,
            })
            .map_err(|error| error.to_string())?;
            let connection = DbConnection::open(DatabaseConfig::file(
                workspace
                    .join(crate::config::WORKSPACE_MARKER)
                    .join(DEFAULT_DB_FILE),
            ))
            .map_err(|error| error.to_string())?;
            if mutation == "unchanged" {
                ensure(report.status.as_str(), "completed", "valid native artifact")?;
                ensure(
                    report.memories_imported,
                    2,
                    "authenticated memories restored",
                )?;
                ensure(report.tags_imported, 1, "authenticated tag restored")?;
                ensure(report.links_imported, 1, "authenticated link restored")?;
                for record in [&records[1], &records[3]] {
                    let memory = connection
                        .get_memory(record["memory_id"].as_str().ok_or("memory id")?)
                        .map_err(|error| error.to_string())?
                        .ok_or("restored memory")?;
                    ensure(
                        memory.trust_class.as_str(),
                        "human_explicit",
                        "native trust preserved",
                    )?;
                }
            } else {
                ensure(report.status.as_str(), "rejected", mutation)?;
                ensure(
                    report.memories_imported,
                    0,
                    "tampered artifact imports no memories",
                )?;
                ensure(report.tags_imported, 0, "tampered artifact imports no tags")?;
                ensure(
                    report.links_imported,
                    0,
                    "tampered artifact imports no links",
                )?;
                ensure(
                    report
                        .issues
                        .iter()
                        .any(|issue| issue.code == UNAUTHENTICATED_NATIVE_IMPORT_TRUST_CODE),
                    true,
                    mutation,
                )?;
                for record in [&records[1], &records[3]] {
                    ensure(
                        connection
                            .get_memory(record["memory_id"].as_str().ok_or("memory id")?)
                            .map_err(|error| error.to_string())?
                            .is_none(),
                        true,
                        "authentication failure leaves no stored memory",
                    )?;
                }
            }
        }
        Ok(())
    }

    #[test]
    fn tampered_authenticated_artifact_refuses_native_trust() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let (workspace, workspace_id) = authenticated_import_workspace(&tempdir)?;
        let artifact = authenticate_sample(&human_explicit_jsonl(), &workspace, &workspace_id)?
            .replace("Run cargo fmt --check", "Disable all release checks");
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, artifact).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace,
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "rejected", "import status")?;
        ensure(
            report
                .issues
                .iter()
                .any(|issue| issue.code == UNAUTHENTICATED_NATIVE_IMPORT_TRUST_CODE),
            true,
            "tampered artifact must refuse native trust",
        )
    }

    #[test]
    fn foreign_workspace_authentication_refuses_native_trust() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let (workspace, _workspace_id) = authenticated_import_workspace(&tempdir)?;
        // MAC'd by this store, but bound to a different workspace scope.
        let artifact =
            authenticate_sample(&human_explicit_jsonl(), &workspace, "wsp_foreign_scope")?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, artifact).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace,
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "rejected", "import status")?;
        ensure(
            report
                .issues
                .iter()
                .any(|issue| issue.code == UNAUTHENTICATED_NATIVE_IMPORT_TRUST_CODE),
            true,
            "cross-workspace authentication must not admit human_explicit",
        )
    }

    #[test]
    fn store_unavailable_fails_closed_for_native_human_explicit() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let (workspace, _workspace_id) = authenticated_import_workspace(&tempdir)?;
        // A structurally valid authentication block, but this workspace has no
        // initialized key store: fail closed with the store-unavailable code.
        let authentication = format!(
            r#"{{"schema":"{}","keyId":"{}","recordCount":1,"recordsRoot":"{}","mac":"{}"}}"#,
            crate::policy::import_auth::NATIVE_IMPORT_AUTH_SCHEMA,
            "00".repeat(16),
            "11".repeat(32),
            "22".repeat(32),
        );
        let artifact = human_explicit_jsonl().replace(
            r#""error_message":null}"#,
            &format!(r#""error_message":null,"authentication":{authentication}}}"#),
        );
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, artifact).map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: workspace,
            database_path: None,
            source_path: source,
            dry_run: false,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "rejected", "import status")?;
        ensure(
            report
                .issues
                .iter()
                .any(|issue| issue.code == MESH_STORE_AUTHENTICATION_UNAVAILABLE_CODE),
            true,
            "missing key store must fail closed for native human_explicit",
        )
    }

    #[test]
    fn reimport_preserves_existing_row_and_flags_divergence() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let workspace = tempdir.path().join("workspace");
        fs::create_dir_all(&workspace).map_err(|error| error.to_string())?;
        let source = tempdir.path().join("source.jsonl");
        fs::write(&source, sample_jsonl()).map_err(|error| error.to_string())?;
        let options = JsonlImportOptions {
            workspace_path: workspace.clone(),
            database_path: None,
            source_path: source.clone(),
            dry_run: false,
        };
        let first = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(first.memories_imported, 1, "first import")?;

        // Byte-identical reimport: pure no-op, no conflict signal.
        let second = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(second.status.as_str(), "completed", "identical reimport")?;
        ensure(second.memories_skipped_duplicate, 1, "identical skip")?;
        ensure(
            second
                .issues
                .iter()
                .any(|issue| issue.code == "reimport_divergent_existing_row"),
            false,
            "identical reimport must not flag a conflict",
        )?;

        // Divergent reimport: same id, edited content — preserved + flagged.
        fs::write(
            &source,
            sample_jsonl().replace("Run cargo fmt --check", "Never run cargo fmt"),
        )
        .map_err(|error| error.to_string())?;
        let third = import_jsonl_records(&options).map_err(|error| error.to_string())?;
        ensure(third.status.as_str(), "completed", "divergent reimport")?;
        ensure(third.memories_skipped_duplicate, 1, "divergent skip")?;
        ensure(
            third
                .issues
                .iter()
                .any(|issue| issue.code == "reimport_divergent_existing_row"),
            true,
            "divergent reimport must flag the preserved conflict",
        )?;
        let connection = DbConnection::open(DatabaseConfig::file(database_path(&options)))
            .map_err(|error| error.to_string())?;
        let stored = connection
            .get_memory("mem_01234567890123456789012345")
            .map_err(|error| error.to_string())?
            .ok_or("memory missing after reimport")?;
        ensure(
            stored.content.contains("Run cargo fmt --check"),
            true,
            "existing row content must be preserved, never overwritten",
        )?;
        let parsed = parse_jsonl_source(&sample_jsonl());
        let mut prepared = prepare_memories(&parsed, &stored.workspace_id, &unauthenticated());
        let incoming = prepared
            .memories
            .first_mut()
            .ok_or("prepared reimport missing")?;
        incoming.updated_at = "2026-05-01T00:00:00Z".to_owned();
        let issue = reimport_conflict_issue(&stored, incoming)
            .ok_or("modification-time-only divergence must be reported")?;
        ensure(
            issue.message.contains("updated_at"),
            true,
            "updated_at divergence",
        )?;
        incoming.updated_at.clone_from(&stored.updated_at);
        incoming.created_at = "2026-04-29T00:00:00Z".to_owned();
        let issue = reimport_conflict_issue(&stored, incoming)
            .ok_or("creation-time-only divergence must be reported")?;
        ensure(
            issue.message.contains("created_at"),
            true,
            "created_at divergence",
        )
    }

    #[cfg(unix)]
    #[test]
    fn import_rejects_symlinked_database_parent_before_create() -> TestResult {
        use std::os::unix::fs::symlink;

        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let source_path = tempdir.path().join("export.jsonl");
        fs::write(&source_path, sample_jsonl()).map_err(|error| error.to_string())?;

        let real_database_dir = tempdir.path().join("real-db");
        fs::create_dir_all(&real_database_dir).map_err(|error| error.to_string())?;
        let linked_database_dir = tempdir.path().join("linked-db");
        symlink(&real_database_dir, &linked_database_dir).map_err(|error| error.to_string())?;
        let database_path = linked_database_dir.join("ee.db");

        let error = match import_jsonl_records(&JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: Some(database_path),
            source_path,
            dry_run: false,
        }) {
            Ok(report) => {
                return Err(format!(
                    "import should reject symlinked DB path: {report:?}"
                ));
            }
            Err(error) => error,
        };

        assert!(
            error.to_string().contains("symlinked path component"),
            "unexpected error: {error}"
        );
        assert!(
            !real_database_dir.join("ee.db").exists(),
            "import must not create a database through a symlinked parent"
        );
        Ok(())
    }

    #[test]
    fn import_rejects_non_regular_database_path_before_open() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let source_path = tempdir.path().join("export.jsonl");
        fs::write(&source_path, sample_jsonl()).map_err(|error| error.to_string())?;
        let database_path = tempdir.path().join("workspace").join(".ee").join("ee.db");
        fs::create_dir_all(&database_path).map_err(|error| error.to_string())?;

        let error = match import_jsonl_records(&JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: Some(database_path),
            source_path,
            dry_run: false,
        }) {
            Ok(report) => {
                return Err(format!(
                    "import should reject directory DB path: {report:?}"
                ));
            }
            Err(error) => error,
        };

        assert!(
            error.to_string().contains("non-regular database path"),
            "unexpected error: {error}"
        );
        Ok(())
    }

    #[test]
    fn import_accepts_canonical_absolute_source_path() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let source_path = tempdir.path().join("export.jsonl");
        fs::write(&source_path, sample_jsonl()).map_err(|error| error.to_string())?;
        let canonical_source = source_path
            .canonicalize()
            .map_err(|error| error.to_string())?;

        let report = import_jsonl_records(&JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: None,
            source_path: canonical_source,
            dry_run: true,
        })
        .map_err(|error| error.to_string())?;

        ensure(report.status.as_str(), "dry_run", "import status")?;
        ensure(report.memories_imported, 0, "dry run imports no memories")?;
        ensure(
            tempdir.path().join("workspace").exists(),
            false,
            "valid preview creates no workspace",
        )
    }

    #[test]
    fn import_validates_memory_payloads_before_dry_run_or_storage_creation() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let secret = "sk-proj-abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
        let cases = [
            ("memory_id", json!(""), "invalid_memory_id"),
            ("level", json!("unknown"), "invalid_memory_level"),
            ("kind", json!(""), "invalid_memory_kind"),
            ("content", json!(""), "invalid_memory_content"),
            ("content", json!(" \t\n "), "invalid_memory_content"),
            (
                "content",
                json!("a".repeat(65_537)),
                "invalid_memory_content",
            ),
            (
                "content",
                json!(format!("Use {secret}")),
                "memory_contains_secret",
            ),
            ("confidence", json!(1.5), "invalid_memory_confidence"),
            (
                "confidence",
                json!(1.000_000_000_000_000_2),
                "invalid_memory_confidence",
            ),
            ("utility", json!(-0.1), "invalid_memory_utility"),
            ("importance", json!(1.1), "invalid_memory_importance"),
            ("bayes_alpha", json!(2.5), "invalid_memory_bayes_posterior"),
            ("bayes_alpha", json!(0.0), "invalid_memory_bayes_posterior"),
            ("created_at", json!("yesterday"), "invalid_memory_timestamp"),
            ("updated_at", json!(""), "invalid_memory_timestamp"),
            (
                "tombstoned_at",
                json!("2026-13-01T00:00:00Z"),
                "invalid_memory_timestamp",
            ),
            ("valid_from", json!("tomorrow"), "invalid_memory_timestamp"),
            ("valid_to", json!("not-a-date"), "invalid_memory_timestamp"),
            (
                "expires_at",
                json!("2026-01-01"),
                "invalid_memory_timestamp",
            ),
        ];
        for (index, (field, value, expected_code)) in cases.into_iter().enumerate() {
            let mut records = sample_jsonl()
                .lines()
                .map(serde_json::from_str::<JsonValue>)
                .collect::<Result<Vec<_>, _>>()
                .map_err(|error| error.to_string())?;
            records[1][field] = value.clone();
            if field == "memory_id" {
                records[2]["memory_id"] = value;
            }
            if field == "bayes_alpha" && records[1][field] == json!(0.0) {
                records[1]["bayes_beta"] = json!(1.5);
            }
            let source_path = tempdir.path().join(format!("invalid-{index}.jsonl"));
            fs::write(
                &source_path,
                records
                    .iter()
                    .map(JsonValue::to_string)
                    .collect::<Vec<_>>()
                    .join("\n"),
            )
            .map_err(|error| error.to_string())?;
            for dry_run in [true, false] {
                let workspace = tempdir.path().join(format!("workspace-{index}-{dry_run}"));
                let external_database = tempdir.path().join(format!("database-{index}-{dry_run}"));
                let report = import_jsonl_records(&JsonlImportOptions {
                    workspace_path: workspace.clone(),
                    database_path: Some(external_database.join("ee.db")),
                    source_path: source_path.clone(),
                    dry_run,
                })
                .map_err(|error| format!("case {index}/{dry_run}: {error}"))?;
                ensure(
                    report.status.as_str(),
                    "rejected",
                    &format!("case {index}/{field}/{dry_run}: invalid payload rejected"),
                )?;
                ensure(report.memories_imported, 0, "no memories imported")?;
                ensure(report.database_path, None, "destination was never opened")?;
                ensure(
                    report.issues.iter().any(|issue| {
                        issue.code == expected_code
                            && issue.severity == JsonlImportIssueSeverity::Error
                    }),
                    true,
                    &format!(
                        "case {index}/{dry_run}: expected {expected_code}, got {:?}",
                        report.issues
                    ),
                )?;
                ensure(
                    report
                        .issues
                        .iter()
                        .any(|issue| issue.message.contains(secret)),
                    false,
                    "diagnostics must not echo the planted secret",
                )?;
                ensure(
                    workspace.exists(),
                    false,
                    "invalid input creates no workspace",
                )?;
                ensure(
                    external_database.exists(),
                    false,
                    "invalid input creates no database parent",
                )?;
            }
        }
        Ok(())
    }

    #[test]
    fn validated_memory_defaults_still_follow_authenticated_trust() -> TestResult {
        let input = sample_jsonl().replace(
            r#""confidence":0.9"#,
            r#""confidence":null,"trust_class":"human_explicit""#,
        );
        let parsed = parse_jsonl_source(&input);
        let prepared = prepare_memories(&parsed, "workspace", &authenticated());
        ensure(
            prepared.has_errors(),
            false,
            "authenticated memory accepted",
        )?;
        let memory = prepared.memories.first().ok_or("prepared memory missing")?;
        ensure(
            memory.input.confidence,
            TrustClass::HumanExplicit.initial_confidence(),
            "trust default",
        )?;
        ensure(memory.input.utility, 0.7, "explicit utility preserved")?;
        let refused = prepare_memories(&parsed, "workspace", &unauthenticated());
        ensure(
            refused.has_errors(),
            true,
            "validation cannot confer native trust",
        )
    }

    #[test]
    fn validated_memory_accepts_payload_boundaries() -> TestResult {
        let mut parsed = parse_jsonl_source(&sample_jsonl());
        let memory = parsed.memories.first_mut().ok_or("source memory missing")?;
        memory.kind = "unknown".to_owned();
        memory.content = "a".repeat(65_536);
        memory.confidence = Some(0.0);
        memory.utility = Some(1.0);
        memory.importance = None;
        memory.bayes_alpha = Some(0.5);
        memory.bayes_beta = Some(2.5);
        let prepared = prepare_memories(&parsed, "workspace", &unauthenticated());
        ensure(
            prepared.has_errors(),
            false,
            "valid boundary payload accepted",
        )?;
        let memory = prepared.memories.first().ok_or("prepared memory missing")?;
        ensure(
            memory.input.kind.as_str(),
            "unknown",
            "custom kind retained",
        )?;
        ensure(memory.input.content.len(), 65_536, "maximum body retained")?;
        ensure(
            memory.input.confidence,
            0.0,
            "explicit zero confidence retained",
        )?;
        ensure(memory.input.utility, 1.0, "maximum utility retained")?;
        ensure(memory.input.importance, 0.5, "missing importance default")?;
        ensure(
            memory.bayes_posterior,
            Some((0.5, 2.5)),
            "posterior retained",
        )
    }

    #[test]
    fn database_path_safety_accepts_canonical_absolute_missing_tail() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let root = tempdir
            .path()
            .canonicalize()
            .map_err(|error| error.to_string())?;
        let database_path = root.join("workspace").join(".ee").join("ee.db");

        ensure_import_database_path_is_safe_for_write(&database_path)
            .map_err(|error| error.to_string())
    }

    #[cfg(unix)]
    #[test]
    fn import_rejects_symlinked_source_path_components() -> TestResult {
        use std::os::unix::fs::symlink;

        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let real_source_dir = tempdir.path().join("real-source");
        fs::create_dir_all(&real_source_dir).map_err(|error| error.to_string())?;
        let real_source = real_source_dir.join("export.jsonl");
        fs::write(&real_source, sample_jsonl()).map_err(|error| error.to_string())?;

        let linked_source_dir = tempdir.path().join("linked-source");
        symlink(&real_source_dir, &linked_source_dir).map_err(|error| error.to_string())?;
        let parent_error = match import_jsonl_records(&JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: None,
            source_path: linked_source_dir.join("export.jsonl"),
            dry_run: true,
        }) {
            Ok(_) => return Err("import should reject symlinked source parent".to_owned()),
            Err(error) => error,
        };
        assert!(
            parent_error
                .to_string()
                .contains("symlinked path component"),
            "unexpected error: {parent_error}"
        );

        let linked_source_file = tempdir.path().join("linked-export.jsonl");
        symlink(&real_source, &linked_source_file).map_err(|error| error.to_string())?;
        let file_error = match import_jsonl_records(&JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: None,
            source_path: linked_source_file,
            dry_run: true,
        }) {
            Ok(_) => return Err("import should reject symlinked source file".to_owned()),
            Err(error) => error,
        };
        assert!(
            file_error.to_string().contains("symlinked path component"),
            "unexpected error: {file_error}"
        );
        Ok(())
    }

    #[test]
    fn import_rejects_non_regular_source_path_before_read() -> TestResult {
        let tempdir = tempfile::tempdir().map_err(|error| error.to_string())?;
        let source_dir = tempdir.path().join("export.jsonl");
        fs::create_dir_all(&source_dir).map_err(|error| error.to_string())?;

        let error = match import_jsonl_records(&JsonlImportOptions {
            workspace_path: tempdir.path().join("workspace"),
            database_path: None,
            source_path: source_dir,
            dry_run: true,
        }) {
            Ok(_) => return Err("import should reject directory source path".to_owned()),
            Err(error) => error,
        };

        assert!(
            error.to_string().contains("non-regular path"),
            "unexpected error: {error}"
        );
        Ok(())
    }
}