eidetic-engine 0.18.0

//! Extend a previously imported CASS session without rewriting its evidence.
//!
//! Discovery is a snapshot, not proof that an already-known session is finished.
//! Compare original-source commitments with the retained CASS rows,
//! preserve every existing identity/link/admission decision, and capture missing
//! spans with their normal DB screening. Metadata, new evidence, audits and a
//! revision-specific index job commit in one writer-fenced transaction.

use std::collections::{BTreeMap, BTreeSet};

use crate::db::{
    CassSourceCommitment, CreateAuditInput, CreateSessionInput, DbConnection, DbError, DbOperation,
    EvidenceAdmissionTally, SearchIndexJobStatus, StoredEvidenceSpan, StoredSession,
};
use crate::models::AuditId;

use super::{
    CassSessionInfo, CassViewSpanForImport, cass_redaction_audit_input, evidence_input,
    search_index_job_input, session_input, stable_cass_redaction_audit_id, stable_evidence_id,
    stable_search_index_job_id, stable_uuid, with_import_session_transaction,
};

const CHECKPOINT_KEY: &str = "eeCassImportCheckpoint";
const CHECKPOINT_SCHEMA: &str = "ee.cass.session_checkpoint.v1";

#[path = "refresh_batch.rs"]
mod batch;

type SessionMetadata = serde_json::Map<String, serde_json::Value>;

struct MetadataRefresh {
    values: SessionMetadata,
    checkpoint: Option<Checkpoint>,
    fields_changed: bool,
}

#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Checkpoint {
    schema: String,
    snapshot_revision: String,
    index_job_id: String,
    previous_index_job_id: String,
}

pub(super) struct RefreshReport {
    pub changed: bool,
    pub added_lines: Vec<u32>,
    pub index_job_id: Option<String>,
    /// Admission of the added spans, by policy reason.
    pub admission: EvidenceAdmissionTally,
}

/// Recover publication of the last committed snapshot without reading CASS.
/// `include_spans = false` suppresses transcript capture, not recovery of work
/// already committed by a previous import. The original job cannot stand in
/// for a newer checkpoint, even when that original job completed successfully.
pub(super) fn reconcile_session_publication(
    connection: &DbConnection,
    workspace_id: &str,
    session_id: &str,
) -> Result<Option<String>, DbError> {
    with_import_session_transaction(connection, || {
        let stored = connection
            .get_session(session_id)?
            .ok_or_else(|| refusal("cass_refresh_session_missing"))?;
        if stored.workspace_id != workspace_id {
            return Err(refusal("cass_refresh_scope_mismatch"));
        }
        let mut metadata = metadata_object(stored.metadata_json.as_deref())?;
        let checkpoint = take_checkpoint(&mut metadata, workspace_id, session_id)?;
        let job_id = checkpoint.map_or_else(
            || stable_search_index_job_id(workspace_id, session_id),
            |saved| saved.index_job_id,
        );
        // A missing derived queue row is repairable from this durable identity.
        // Never change the evidence, checkpoint, or a live publisher's lease.
        pending_job(connection, workspace_id, session_id, &job_id, true)
    })
}

/// The caller obtained a complete bounded `cass view` snapshot before entering
/// here. A failure never partially updates this session; earlier successfully
/// imported sessions remain committed, as in the ordinary import path.
pub(super) fn refresh_session(
    connection: &DbConnection,
    workspace_id: &str,
    session_id: &str,
    discovered: &CassSessionInfo,
    spans: &[CassViewSpanForImport],
) -> Result<RefreshReport, DbError> {
    with_import_session_transaction(connection, || {
        let stored = connection
            .get_session(session_id)?
            .ok_or_else(|| refusal("cass_refresh_session_missing"))?;
        if stored.workspace_id != workspace_id || stored.cass_session_id != discovered.source_path {
            return Err(refusal("cass_refresh_scope_mismatch"));
        }

        // Keep the raw-reference order for the existing checkpoint contract and
        // stable evidence IDs. The live DB boundary stores BLAKE3(raw reference)
        // in cass_span_id, so retained-row lookup needs a separate projection.
        // Hash incoming raw references exactly once, never already-stored keys.
        let mut incoming = BTreeMap::new();
        let mut by_stored_reference = BTreeMap::new();
        let mut lines = BTreeSet::new();
        for span in spans {
            // Reconciliation is a persistence boundary too. Do not let an
            // inconsistent source locator or payload create an unrefreshable
            // checkpoint, even when a future caller bypasses the view parser.
            if span.start_line == 0
                || span.source_bytes == 0
                || span.source_bytes > super::CASS_STDOUT_LINE_MAX_BYTES
                || span.end_line != span.start_line
                || span.cass_span_id != format!("{}:{}", discovered.source_path, span.start_line)
                || span.content_hash
                    != format!("blake3:{}", blake3::hash(span.excerpt.as_bytes()).to_hex())
            {
                return Err(refusal("cass_refresh_invalid_span"));
            }
            let stored_reference = format!(
                "blake3:{}",
                blake3::hash(span.cass_span_id.as_bytes()).to_hex()
            );
            if incoming.insert(span.cass_span_id.as_str(), span).is_some()
                || by_stored_reference.insert(stored_reference, span).is_some()
                || !lines.insert(span.start_line)
            {
                return Err(refusal("cass_refresh_duplicate_span"));
            }
        }
        let previous = connection.list_evidence_spans_for_session(session_id)?;
        let mut retained = BTreeSet::new();
        let mut reconciliations = Vec::new();
        for row in &previous {
            if row.workspace_id != workspace_id || row.session_id != session_id {
                return Err(refusal("cass_refresh_scope_mismatch"));
            }
            // Existing databases may retain the original locator rather than
            // its privacy projection. Match either representation to the raw
            // incoming identity without rewriting or re-admitting the row.
            let observed = by_stored_reference
                .get(row.cass_span_id.as_str())
                .or_else(|| incoming.get(row.cass_span_id.as_str()));
            // A derived incident card summarizes this transcript; it is not a
            // line of it (ADR 0091). Its derived reference can never name an
            // upstream line, so it neither blocks nor joins reconciliation.
            if row.is_derived_incident_card() && observed.is_none() {
                continue;
            }
            if !matches!(row.producer_kind.as_str(), "cass_import" | "legacy_unknown") {
                // Other producers may attach their own evidence to a session.
                // They cannot confer authority on a second interpretation of
                // an occupied CASS source slot, even under a different row ID.
                if observed.is_some() {
                    return Err(refusal("cass_refresh_producer_conflict"));
                }
                continue;
            }
            let Some(span) = observed else {
                return Err(refusal("cass_refresh_history_missing"));
            };
            // Retention membership is still in the raw incoming identity space,
            // just like additions and stable_evidence_id below. No stored row is
            // rewritten or re-admitted merely because its source is recognized.
            if !retained.insert(span.cass_span_id.as_str())
                || row.start_line != span.start_line
                || row.end_line != span.end_line
                || row.span_kind != span.span_kind.as_str()
                || row.role.as_deref() != span.role.map(|role| role.as_str())
            {
                return Err(refusal("cass_refresh_history_changed"));
            }
            if let Some(proof) = retained_source_proof(row, span)? {
                let (audit_id, audit) =
                    reconciliation_audit(workspace_id, session_id, row, span, proof);
                if let Some(existing) = connection.get_audit(&audit_id)? {
                    if existing.workspace_id != audit.workspace_id
                        || existing.actor != audit.actor
                        || existing.action != audit.action
                        || existing.target_type != audit.target_type
                        || existing.target_id != audit.target_id
                        || existing.details != audit.details
                    {
                        return Err(refusal("cass_refresh_reconciliation_audit_conflict"));
                    }
                } else {
                    reconciliations.push((audit_id, audit));
                }
            }
        }

        let mut input = session_input(workspace_id, discovered);
        let merged = merged_metadata(&stored, &input, workspace_id, session_id)?;
        let metadata_changed = !same_metadata(&stored, &input) || merged.fields_changed;
        let checkpoint = merged.checkpoint;
        let mut metadata = merged.values;
        let additions: Vec<_> = incoming
            .values()
            .copied()
            .filter(|span| !retained.contains(span.cass_span_id.as_str()))
            .collect();
        let revision = snapshot_revision(workspace_id, session_id, &input, &incoming);
        let changed = metadata_changed
            || !additions.is_empty()
            || !reconciliations.is_empty()
            || checkpoint
                .as_ref()
                .is_some_and(|saved| saved.snapshot_revision != revision);
        let previous_job = checkpoint.as_ref().map_or_else(
            || stable_search_index_job_id(workspace_id, session_id),
            |saved| saved.index_job_id.clone(),
        );

        if !changed {
            // The checkpoint survives a failed/cancelled publication. Missing
            // durable work can be reconstructed, as for the original import.
            let index_job_id =
                pending_job(connection, workspace_id, session_id, &previous_job, true)?;
            return Ok(RefreshReport {
                changed: false,
                added_lines: Vec::new(),
                index_job_id,
                admission: EvidenceAdmissionTally::default(),
            });
        }

        // Bind each transition to its predecessor, not only the new payload.
        // A legitimate A -> B -> A -> B history must get new publication work
        // rather than reusing the completed job from the first B snapshot.
        let job_id = refresh_job_id(workspace_id, session_id, &previous_job, &revision);
        if connection.get_search_index_job(&job_id)?.is_some() {
            return Err(refusal("cass_refresh_revision_already_exists"));
        }
        let checkpoint = Checkpoint {
            schema: CHECKPOINT_SCHEMA.to_owned(),
            snapshot_revision: revision.clone(),
            index_job_id: job_id.clone(),
            previous_index_job_id: previous_job,
        };
        metadata.insert(
            CHECKPOINT_KEY.to_owned(),
            serde_json::to_value(&checkpoint)
                .map_err(|_| refusal("cass_refresh_checkpoint_invalid"))?,
        );
        input.metadata_json = Some(serde_json::Value::Object(metadata).to_string());

        let (mut added_lines, admission) =
            batch::insert_additions(connection, workspace_id, &stored, &additions)?;
        for (audit_id, audit) in &reconciliations {
            connection.insert_audit(audit_id, audit)?;
        }
        // Persist the checkpoint even when discovery fields did not change.
        // Its job and every captured row belong to this same transaction.
        update_metadata(connection, workspace_id, session_id, &input)?;
        connection
            .insert_search_index_job(&job_id, &search_index_job_input(workspace_id, session_id))?;
        // Audit identity follows the transition too: revisiting the same
        // payload is another recorded refresh, not the old audit row.
        let audit_id = AuditId::from_uuid(stable_uuid(&format!(
            "audit:cass-refresh:{session_id}:{job_id}"
        )))
        .to_string();
        connection.insert_audit(
            &audit_id,
            &CreateAuditInput {
                workspace_id: Some(workspace_id.to_owned()),
                actor: Some("ee import cass".to_owned()),
                action: "cass.session.refreshed".to_owned(),
                target_type: Some("session".to_owned()),
                target_id: Some(session_id.to_owned()),
                details: Some(
                    serde_json::json!({
                        "schema": "ee.cass.refresh_audit.v1",
                        "sessionId": session_id,
                        "snapshotRevision": revision,
                        "addedSpanCount": added_lines.len(),
                        "retainedCassSpanCount": retained.len(),
                        "reconciledScreeningCount": reconciliations.len(),
                        "metadataChanged": metadata_changed,
                    })
                    .to_string(),
                ),
            },
        )?;
        added_lines.sort_unstable();
        Ok(RefreshReport {
            changed: true,
            added_lines,
            index_job_id: Some(job_id),
            admission,
        })
    })
}

/// Return a proof label only when the screened representation changed. Never
/// re-admit or replace a historical row to make a comparison succeed.
///
/// New rows commit to the full original line, including bytes removed by
/// redaction and bounding. Old imports did not keep that commitment. Only an
/// intact clean excerpt or a strictly recognized importer withholding digest
/// can prove retained source evidence. A complete excerpt cannot attest to any
/// serialization normalization done before it was stored; never relabel it as
/// an import-time raw commitment. A lossy projection cannot prove source identity.
///
/// The one exception is a canonical-JSON legacy excerpt whose current source
/// line is byte-for-byte the retained excerpt (#70): there are no unretained
/// bytes left to authenticate, so the row is kept rather than freezing the
/// session. A canonical excerpt whose current source differs stays
/// unverifiable.
fn retained_source_proof(
    row: &StoredEvidenceSpan,
    span: &CassViewSpanForImport,
) -> Result<Option<&'static str>, DbError> {
    let retained_hash = format!("blake3:{}", blake3::hash(row.excerpt.as_bytes()).to_hex());
    if row.content_hash != retained_hash
        || row
            .canonical_excerpt_hash
            .as_ref()
            .is_some_and(|hash| hash != &retained_hash)
    {
        return Err(refusal("cass_refresh_history_changed"));
    }
    let observed_hash = format!("blake3:{}", span.source_hash.to_hex());
    let proof = if let Some(source) =
        CassSourceCommitment::from_metadata(row.metadata_json.as_deref())
            .map_err(|_| refusal("cass_refresh_history_changed"))?
    {
        let metadata: serde_json::Value =
            serde_json::from_str(row.metadata_json.as_deref().unwrap_or(""))
                .map_err(|_| refusal("cass_refresh_history_changed"))?;
        if row.producer_kind != "cass_import"
            || metadata["schema"] != crate::db::EVIDENCE_SECURITY_METADATA_SCHEMA_V1
            || metadata["producerKind"] != row.producer_kind
            || metadata["canonicalExcerptHash"] != retained_hash
        {
            return Err(refusal("cass_refresh_history_changed"));
        }
        if source.content_hash != observed_hash || source.byte_length != span.source_bytes as u64 {
            return Err(refusal("cass_refresh_history_changed"));
        }
        "captured_source_commitment"
    } else if intact_clean_legacy_excerpt(row) {
        let identical = retained_hash == observed_hash && row.excerpt.len() == span.source_bytes;
        if !legacy_excerpt_may_be_normalized_json(&row.excerpt) {
            if !identical {
                return Err(refusal("cass_refresh_history_changed"));
            }
            "retained_complete_excerpt"
        } else if identical {
            // The record may be a v0.17 normalization of a padded original, so
            // it proves nothing about an older source. But today's source line
            // is exactly these bytes, so every byte the row stands for is
            // retained: keep it instead of refusing every v0.17-imported
            // compact JSON transcript line (#70).
            "retained_identical_canonical_excerpt"
        } else {
            return Err(refusal("cass_refresh_history_unverifiable"));
        }
    } else if let Some(digest) = legacy_withheld_source_digest(row) {
        if digest != observed_hash {
            return Err(refusal("cass_refresh_history_changed"));
        }
        "retained_withholding_digest"
    } else {
        // Neither a session's path/mtime/size hash nor replaying the old lossy
        // screener can authenticate discarded bytes. Do not manufacture a
        // historical source digest from the new observation.
        return Err(refusal("cass_refresh_history_unverifiable"));
    };
    Ok((row.excerpt != span.excerpt || row.content_hash != span.content_hash).then_some(proof))
}

fn intact_clean_legacy_excerpt(row: &StoredEvidenceSpan) -> bool {
    row.secret_redaction_status == "clean"
        && serde_json::from_str::<Vec<String>>(&row.redaction_classes_json)
            .is_ok_and(|classes| classes.is_empty())
        // UTF-8 truncation can retreat up to three bytes from the old cap.
        // Every length in that interval can be an unmarked lossy prefix.
        && row.excerpt.len() < super::ingestion::MAX_EXCERPT_BYTES - 3
        && !row.excerpt.contains("[REDACTED:")
        && !row.excerpt.contains("[TRUNCATED]")
}

fn legacy_excerpt_may_be_normalized_json(excerpt: &str) -> bool {
    // v0.17 could turn an oversized, whitespace-padded JSON object into a
    // short canonical record without marking it as truncated or redacted;
    // later producers also normalized object streams joined by newlines.
    // Those records retained semantic content but did not retain source bytes.
    // An already-canonical original is indistinguishable, so neither may
    // confer a complete-source proof of an older source without an independent
    // commitment; only a byte-identical current source can keep it (#70).
    if !excerpt.starts_with('{') {
        return false;
    }
    excerpt.split('\n').all(|line| {
        let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
            return false;
        };
        value.is_object() && serde_json::to_string(&value).is_ok_and(|canonical| canonical == line)
    })
}

fn legacy_withheld_source_digest(row: &StoredEvidenceSpan) -> Option<String> {
    if row.producer_kind != "cass_import" || row.secret_redaction_status != "redacted" {
        return None;
    }
    let classes: Vec<String> = serde_json::from_str(&row.redaction_classes_json).ok()?;
    let value: serde_json::Value = serde_json::from_str(&row.excerpt).ok()?;
    let reason = value.get("reason")?.as_str()?;
    if !matches!(
        reason,
        "external_ingestion_oversized"
            | "external_ingestion_encoded_json_unreadable"
            | "external_ingestion_encoded_json_redaction_invalid"
    ) || !classes.iter().any(|class| class == reason)
    {
        return None;
    }
    let digest = value.get("sourceDigest")?.as_str()?;
    // Byte-for-byte comparison also rejects extra/duplicate keys, altered
    // classifications, and arbitrary source objects that only resemble a marker.
    let expected = serde_json::json!({
        "type": "external_ingestion_withheld",
        "reason": reason,
        "sourceDigest": digest,
        "redaction": format!("[REDACTED:{reason}]"),
    })
    .to_string();
    (row.excerpt == expected).then(|| digest.to_owned())
}

fn reconciliation_audit(
    workspace_id: &str,
    session_id: &str,
    row: &StoredEvidenceSpan,
    span: &CassViewSpanForImport,
    proof: &str,
) -> (String, CreateAuditInput) {
    let details = serde_json::json!({
        "schema": "ee.cass.screening_reconciliation.v1",
        "sessionId": session_id,
        "evidenceId": row.id,
        "sourceProof": proof,
        "sourceContentHash": format!("blake3:{}", span.source_hash.to_hex()),
        "observedSourceByteLength": span.source_bytes,
        "retainedExcerptHash": row.content_hash,
        "observedExcerptHash": span.content_hash,
        "observedExcerptProducerVersion": super::CASS_EXCERPT_PRODUCER_VERSION,
        "historicalEvidenceRewritten": false,
    })
    .to_string();
    let id = AuditId::from_uuid(stable_uuid(&format!(
        "audit:cass-screening-reconciliation:{}:{}",
        row.id,
        blake3::hash(details.as_bytes()).to_hex()
    )))
    .to_string();
    (
        id,
        CreateAuditInput {
            workspace_id: Some(workspace_id.to_owned()),
            actor: Some("ee import cass".to_owned()),
            action: "cass.evidence.screening_reconciled".to_owned(),
            target_type: Some("evidence_span".to_owned()),
            target_id: Some(row.id.clone()),
            details: Some(details),
        },
    )
}

fn same_metadata(stored: &StoredSession, input: &CreateSessionInput) -> bool {
    stored.agent_name == input.agent_name
        && stored.started_at == input.started_at
        && stored.ended_at == input.ended_at
        && stored.message_count == input.message_count
        && stored.token_count == input.token_count
        && stored.content_hash == input.content_hash
}

fn refresh_job_id(
    workspace_id: &str,
    session_id: &str,
    previous_job: &str,
    revision: &str,
) -> String {
    stable_search_index_job_id(
        workspace_id,
        &format!("refresh:{session_id}:{previous_job}:{revision}"),
    )
}

/// Only the importer's discovery fields are refreshed. Preserve unrelated
/// session annotations, and keep the resumable checkpoint out of the observed
/// metadata comparison so an unchanged retry does not create another write.
fn merged_metadata(
    stored: &StoredSession,
    input: &CreateSessionInput,
    workspace_id: &str,
    session_id: &str,
) -> Result<MetadataRefresh, DbError> {
    let mut stored_metadata = metadata_object(stored.metadata_json.as_deref())?;
    let checkpoint = take_checkpoint(&mut stored_metadata, workspace_id, session_id)?;
    let incoming = metadata_object(input.metadata_json.as_deref())?;
    let changed = incoming
        .iter()
        .any(|(key, value)| stored_metadata.get(key) != Some(value));
    stored_metadata.extend(incoming);
    Ok(MetadataRefresh {
        values: stored_metadata,
        checkpoint,
        fields_changed: changed,
    })
}

fn metadata_object(value: Option<&str>) -> Result<SessionMetadata, DbError> {
    match value {
        None => Ok(serde_json::Map::new()),
        Some(value) => serde_json::from_str::<serde_json::Value>(value)
            .ok()
            .and_then(|value| value.as_object().cloned())
            .ok_or_else(|| refusal("cass_refresh_metadata_invalid")),
    }
}

/// Full refresh and metadata-only retry must interpret the same checkpoint.
/// Reject corruption before consulting any completed job or creating new work.
fn take_checkpoint(
    metadata: &mut SessionMetadata,
    workspace_id: &str,
    session_id: &str,
) -> Result<Option<Checkpoint>, DbError> {
    let checkpoint: Option<Checkpoint> = metadata
        .remove(CHECKPOINT_KEY)
        .map(serde_json::from_value)
        .transpose()
        .map_err(|_| refusal("cass_refresh_checkpoint_invalid"))?;
    if let Some(saved) = &checkpoint {
        let valid_revision = saved
            .snapshot_revision
            .strip_prefix("blake3:")
            .is_some_and(|hash| {
                hash.len() == 64
                    && hash
                        .bytes()
                        .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
            });
        if saved.schema != CHECKPOINT_SCHEMA
            || !valid_revision
            || saved.index_job_id == saved.previous_index_job_id
            || saved.index_job_id
                != refresh_job_id(
                    workspace_id,
                    session_id,
                    &saved.previous_index_job_id,
                    &saved.snapshot_revision,
                )
        {
            return Err(refusal("cass_refresh_checkpoint_invalid"));
        }
    }
    Ok(checkpoint)
}

/// Model attribution and locator authority are not supplied by discovery and
/// must not be erased on refresh. V097 invalidates material session updates.
fn update_metadata(
    connection: &DbConnection,
    workspace_id: &str,
    session_id: &str,
    input: &CreateSessionInput,
) -> Result<(), DbError> {
    // This public DbConnection surface accepts complete SQL, not parameters.
    // Encode text as UTF-8 hex literals: even quotes, NUL and delimiter-bearing
    // metadata cannot become SQL. Do not propagate SQL-bearing driver errors.
    let query = format!(
        "UPDATE sessions SET agent_name = {}, started_at = {}, ended_at = {}, message_count = {}, token_count = {}, content_hash = {}, metadata_json = {} WHERE id = {} AND workspace_id = {}",
        optional_text(input.agent_name.as_deref()),
        optional_text(input.started_at.as_deref()),
        optional_text(input.ended_at.as_deref()),
        input.message_count,
        input
            .token_count
            .map_or_else(|| "NULL".to_owned(), |count| count.to_string()),
        sql_text(&input.content_hash),
        optional_text(input.metadata_json.as_deref()),
        sql_text(session_id),
        sql_text(workspace_id),
    );
    connection
        .execute_raw(&query)
        .map(|_| ())
        .map_err(|_| refusal("cass_refresh_metadata_write_failed"))
}

fn optional_text(value: Option<&str>) -> String {
    value.map_or_else(|| "NULL".to_owned(), sql_text)
}

fn sql_text(value: &str) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut hex = String::with_capacity(value.len().saturating_mul(2));
    for byte in value.bytes() {
        hex.push(char::from(HEX[usize::from(byte >> 4)]));
        hex.push(char::from(HEX[usize::from(byte & 15)]));
    }
    format!("CAST(X'{hex}' AS TEXT)")
}

fn snapshot_revision(
    workspace_id: &str,
    session_id: &str,
    input: &CreateSessionInput,
    spans: &BTreeMap<&str, &CassViewSpanForImport>,
) -> String {
    let mut hash = blake3::Hasher::new();
    hash.update(b"ee.cass.refresh_snapshot.v2\0");
    let metadata = serde_json::json!([
        workspace_id,
        session_id,
        input.agent_name,
        input.started_at,
        input.ended_at,
        input.message_count,
        input.token_count,
        input.content_hash,
        input.metadata_json
    ])
    .to_string();
    hash.update(&(metadata.len() as u64).to_le_bytes());
    hash.update(metadata.as_bytes());
    for (reference, span) in spans {
        let identity = serde_json::json!([
            reference,
            span.start_line,
            span.end_line,
            span.span_kind.as_str(),
            span.role.map(|role| role.as_str()),
            span.content_hash,
            format!("blake3:{}", span.source_hash.to_hex()),
            span.source_bytes,
            span.redacted,
            span.redacted_reasons
        ])
        .to_string();
        hash.update(&(identity.len() as u64).to_le_bytes());
        hash.update(identity.as_bytes());
    }
    format!("blake3:{}", hash.finalize().to_hex())
}

fn pending_job(
    connection: &DbConnection,
    workspace_id: &str,
    session_id: &str,
    job_id: &str,
    create_missing: bool,
) -> Result<Option<String>, DbError> {
    match connection.get_search_index_job(job_id)? {
        Some(job) => {
            if job.workspace_id != workspace_id
                || job.document_source.as_deref() != Some("session")
                || job.document_id.as_deref() != Some(session_id)
            {
                return Err(refusal("cass_refresh_index_job_scope_mismatch"));
            }
            // A single-document intake job may publish a full or coalesced
            // generation. The publisher updates documents_total to that corpus
            // size; it is progress, not part of the immutable job identity.
            if job.job_type_enum() != Some(crate::db::SearchIndexJobType::SingleDocument) {
                return Err(refusal("cass_refresh_index_job_invalid"));
            }
            match job.status_enum() {
                Some(SearchIndexJobStatus::Completed) => Ok(None),
                Some(_) => Ok(Some(job_id.to_owned())),
                None => Err(refusal("cass_refresh_index_job_invalid")),
            }
        }
        None if create_missing => {
            connection.insert_search_index_job(
                job_id,
                &search_index_job_input(workspace_id, session_id),
            )?;
            Ok(Some(job_id.to_owned()))
        }
        None => Err(refusal("cass_refresh_index_job_missing")),
    }
}

fn refusal(code: &str) -> DbError {
    DbError::MalformedRow {
        operation: DbOperation::Execute,
        message: format!(
            "{code}: CASS session refresh was not committed; retry a complete stable transcript, or inspect the original session before replacing retained history"
        ),
    }
}

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

#[cfg(test)]
mod canonical_reference_tests {
    #![allow(clippy::unwrap_used, clippy::expect_used)]

    use super::*;
    use crate::cass::CassAgent;
    use crate::db::CreateWorkspaceInput;
    use crate::models::WorkspaceId;
    use serde_json::json;

    fn span(session: &CassSessionInfo, line: u32) -> CassViewSpanForImport {
        super::super::parse_view_line_value(
            &json!({"line": line, "content": format!("Build observation {line}. 調査完了")}),
            &session.source_path,
        )
        .unwrap()
    }

    fn fixture(source: &str, count: u32) -> (DbConnection, String, String, CassSessionInfo) {
        let db = DbConnection::open_memory().unwrap();
        db.migrate().unwrap();
        let workspace = WorkspaceId::from_uuid(uuid::Uuid::from_u128(8711)).to_string();
        db.insert_workspace(
            &workspace,
            &CreateWorkspaceInput {
                path: "/canonical-refresh-workspace".to_owned(),
                name: None,
            },
        )
        .unwrap();
        let session = CassSessionInfo::new(source).with_agent(CassAgent::Codex);
        let spans: Vec<_> = (1..=count).map(|line| span(&session, line)).collect();
        let result =
            super::super::persist_session_import_if_absent(&db, &workspace, &session, &spans)
                .unwrap();
        let super::super::SessionImportPersistResult::Imported { session_id, .. } = result else {
            panic!("new fixture must import");
        };
        (db, workspace, session_id, session)
    }

    #[test]
    fn live_storage_projection_survives_retries_and_repeated_growth() {
        let (db, workspace, id, session) = fixture("/private/canonical-session.jsonl", 1);
        let first = span(&session, 1);
        let first_id = stable_evidence_id(&id, &first.cass_span_id);
        let before = db.get_evidence_span(&first_id).unwrap().unwrap();
        let digest = format!(
            "blake3:{}",
            blake3::hash(first.cass_span_id.as_bytes()).to_hex()
        );
        assert_eq!(before.cass_span_id, digest);
        assert_ne!(before.cass_span_id, first.cass_span_id);
        assert_eq!(before.upstream_ref_hash.as_deref(), Some(digest.as_str()));
        let original_job = stable_search_index_job_id(&workspace, &id);
        let retry = refresh_session(&db, &workspace, &id, &session, &[first]).unwrap();
        assert!(!retry.changed);
        assert_eq!(retry.index_job_id.as_deref(), Some(original_job.as_str()));

        for count in [2, 12] {
            let mut spans: Vec<_> = (1..=count).map(|line| span(&session, line)).collect();
            let report = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
            assert!(report.changed);
            let first_added = if count == 2 { 2 } else { 3 };
            assert_eq!(
                report.added_lines,
                (first_added..=count).collect::<Vec<_>>()
            );
            let job = report.index_job_id.unwrap();
            assert_ne!(job, original_job);
            let stored = db.get_session(&id).unwrap().unwrap();
            let metadata: serde_json::Value =
                serde_json::from_str(stored.metadata_json.as_deref().unwrap()).unwrap();
            let raw: BTreeMap<_, _> = spans
                .iter()
                .map(|span| (span.cass_span_id.as_str(), span))
                .collect();
            assert_eq!(
                metadata[CHECKPOINT_KEY]["snapshotRevision"],
                snapshot_revision(&workspace, &id, &session_input(&workspace, &session), &raw)
            );
            spans.reverse();
            let retry = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
            assert!(!retry.changed);
            assert!(retry.added_lines.is_empty());
            assert_eq!(retry.index_job_id.as_deref(), Some(job.as_str()));
            assert_eq!(db.get_session(&id).unwrap().unwrap(), stored);
            assert_eq!(db.get_evidence_span(&first_id).unwrap().unwrap(), before);
            assert_eq!(
                db.list_evidence_spans_for_session(&id).unwrap().len(),
                count as usize
            );
            for incoming in &spans {
                let expected_id = stable_evidence_id(&id, &incoming.cass_span_id);
                let row = db.get_evidence_span(&expected_id).unwrap().unwrap();
                assert_eq!(row.excerpt, incoming.excerpt);
            }
        }
    }

    #[test]
    fn backfilled_rows_are_retained_on_the_next_refresh() {
        let (db, workspace, id, session) = fixture("/private/backfill-session.jsonl", 0);
        for count in [2, 3] {
            let spans: Vec<_> = (1..=count).map(|line| span(&session, line)).collect();
            let report = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
            assert_eq!(
                report.added_lines,
                if count == 2 { vec![1, 2] } else { vec![3] }
            );
            let rows = db.list_evidence_spans_for_session(&id).unwrap();
            assert_eq!(rows.len(), count as usize);
            let retry = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
            assert!(!retry.changed);
            assert_eq!(db.list_evidence_spans_for_session(&id).unwrap(), rows);
        }
    }

    #[test]
    fn unicode_and_digest_looking_upstream_references_are_hashed_exactly_once() {
        for source in [
            "/private/資料:session.jsonl".to_owned(),
            format!("blake3:{}", "a".repeat(64)),
        ] {
            let (db, workspace, id, session) = fixture(&source, 1);
            let incoming = span(&session, 1);
            let expected_id = stable_evidence_id(&id, &incoming.cass_span_id);
            let before = db.get_evidence_span(&expected_id).unwrap().unwrap();
            let report = refresh_session(&db, &workspace, &id, &session, &[incoming]).unwrap();
            assert!(!report.changed);
            assert_eq!(db.get_evidence_span(&expected_id).unwrap().unwrap(), before);
        }
    }

    #[test]
    fn canonical_lookup_preserves_denial_but_refuses_substituted_reference() {
        let (db, workspace, id, session) = fixture("/private/denied-session.jsonl", 1);
        let first = span(&session, 1);
        let first_id = stable_evidence_id(&id, &first.cass_span_id);
        db.execute_raw(
            "UPDATE evidence_spans SET search_eligibility = 'denied', pack_eligibility = 'denied'",
        )
        .unwrap();
        let denied = db.get_evidence_span(&first_id).unwrap().unwrap();
        let incoming = vec![first, span(&session, 2)];
        let grown = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert_eq!(grown.added_lines, vec![2]);
        assert_eq!(db.get_evidence_span(&first_id).unwrap().unwrap(), denied);
        assert!(
            db.get_search_admitted_evidence_span(&first_id, &workspace)
                .unwrap()
                .is_none()
        );

        let substituted = format!("blake3:{}", blake3::hash(b"other source:1").to_hex());
        db.execute_raw(&format!(
            "UPDATE evidence_spans SET cass_span_id = {} WHERE id = {}",
            sql_text(&substituted),
            sql_text(&first_id)
        ))
        .unwrap();
        let before = db.get_session(&id).unwrap().unwrap();
        let mut newer = incoming;
        newer.push(span(&session, 3));
        let error = refresh_session(&db, &workspace, &id, &session, &newer)
            .err()
            .unwrap();
        assert!(error.to_string().contains("cass_refresh_history_missing"));
        assert!(!error.to_string().contains(&session.source_path));
        assert_eq!(db.get_session(&id).unwrap().unwrap(), before);
        assert_eq!(db.list_evidence_spans_for_session(&id).unwrap().len(), 2);
    }

    fn row_counts(db: &DbConnection) -> [i64; 4] {
        [
            "sessions",
            "evidence_spans",
            "search_index_jobs",
            "audit_log",
        ]
        .map(|table| db.count_table_rows(table).unwrap())
    }

    fn assert_refresh_refused_without_writes(
        db: &DbConnection,
        workspace: &str,
        id: &str,
        session: &CassSessionInfo,
        spans: &[CassViewSpanForImport],
        code: &str,
    ) {
        let before_counts = row_counts(db);
        let before_session = db.get_session(id).unwrap();
        let before_evidence = db.list_evidence_spans_for_session(id).unwrap();
        let error = refresh_session(db, workspace, id, session, spans)
            .err()
            .expect("invalid retained history must refuse refresh")
            .to_string();
        assert!(error.contains(code), "{error}");
        assert!(!error.contains(&session.source_path), "{error}");
        assert!(!error.contains("PRIVATE_PAYLOAD_SENTINEL"), "{error}");
        assert_eq!(row_counts(db), before_counts);
        assert_eq!(db.get_session(id).unwrap(), before_session);
        assert_eq!(
            db.list_evidence_spans_for_session(id).unwrap(),
            before_evidence
        );
    }

    /// Model the metadata that existed before complete-source commitments.
    /// Keep the old producer metadata hash too, rather than leaving a hidden
    /// hash of a modern `cassSource` claim in this historical fixture.
    fn use_legacy_source_metadata(db: &DbConnection, evidence_id: &str) {
        let row = db.get_evidence_span(evidence_id).unwrap().unwrap();
        let mut metadata: serde_json::Value =
            serde_json::from_str(row.metadata_json.as_deref().unwrap()).unwrap();
        assert!(
            metadata
                .as_object_mut()
                .unwrap()
                .remove("cassSource")
                .is_some()
        );
        let classes: Vec<String> = serde_json::from_str(&row.redaction_classes_json).unwrap();
        let producer_metadata = json!({
            "schema": crate::models::CASS_EVIDENCE_SPAN_SCHEMA_V1,
            "redactionStatus": row.secret_redaction_status,
            "redactionClasses": classes,
        })
        .to_string();
        metadata["sourceMetadataHash"] = json!(format!(
            "blake3:{}",
            blake3::hash(producer_metadata.as_bytes()).to_hex()
        ));
        db.execute_raw(&format!(
            "UPDATE evidence_spans SET metadata_json = {} WHERE id = {}",
            sql_text(&metadata.to_string()),
            sql_text(evidence_id),
        ))
        .unwrap();
        let stored = db.get_evidence_span(evidence_id).unwrap().unwrap();
        assert!(
            CassSourceCommitment::from_metadata(stored.metadata_json.as_deref())
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn migrated_history_can_grow_without_regaining_admission_or_changing_ids() {
        for producer in ["cass_import", "legacy_unknown"] {
            for raw_reference in [false, true] {
                let (db, workspace, id, session) = fixture("/private/migrated.jsonl", 2);
                let first = span(&session, 1);
                let first_id = stable_evidence_id(&id, &first.cass_span_id);
                db.execute_raw(&format!(
                    "UPDATE evidence_spans SET producer_kind = {}, search_eligibility = 'denied', pack_eligibility = 'denied' WHERE id = {}",
                    sql_text(producer),
                    sql_text(&first_id),
                ))
                .unwrap();
                if producer == "legacy_unknown" {
                    use_legacy_source_metadata(&db, &first_id);
                }
                if raw_reference {
                    db.execute_raw(&format!(
                        "UPDATE evidence_spans SET cass_span_id = {} WHERE id = {}",
                        sql_text(&first.cass_span_id),
                        sql_text(&first_id),
                    ))
                    .unwrap();
                }
                let retained = db.list_evidence_spans_for_session(&id).unwrap();
                db.execute_raw("UPDATE search_index_jobs SET status = 'completed'")
                    .unwrap();
                let old_job = stable_search_index_job_id(&workspace, &id);
                let mut incoming: Vec<_> = (1..=4).map(|line| span(&session, line)).collect();
                let grown = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
                assert!(grown.changed);
                assert_eq!(grown.added_lines, vec![3, 4]);
                let job = grown.index_job_id.unwrap();
                assert_ne!(job, old_job);
                assert_eq!(
                    db.get_search_index_job(&job)
                        .unwrap()
                        .unwrap()
                        .status_enum(),
                    Some(SearchIndexJobStatus::Pending)
                );
                for row in &retained {
                    assert_eq!(db.get_evidence_span(&row.id).unwrap().as_ref(), Some(row));
                }
                assert!(
                    db.get_search_admitted_evidence_span(&first_id, &workspace)
                        .unwrap()
                        .is_none()
                );
                for fresh in &incoming[2..] {
                    let fresh_id = stable_evidence_id(&id, &fresh.cass_span_id);
                    let row = db
                        .get_search_admitted_evidence_span(&fresh_id, &workspace)
                        .unwrap()
                        .expect("only newly captured evidence receives normal admission");
                    assert_eq!(row.excerpt, fresh.excerpt);
                }
                let before_retry = row_counts(&db);
                let saved = db.get_session(&id).unwrap();
                incoming.reverse();
                let retry = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
                assert!(!retry.changed);
                assert!(retry.added_lines.is_empty());
                assert_eq!(retry.index_job_id.as_deref(), Some(job.as_str()));
                assert_eq!(row_counts(&db), before_retry);
                assert_eq!(db.get_session(&id).unwrap(), saved);
                db.execute_raw("UPDATE search_index_jobs SET status = 'completed'")
                    .unwrap();
                let completed = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
                assert!(!completed.changed);
                assert!(completed.index_job_id.is_none());
                assert_eq!(row_counts(&db), before_retry);
                for row in retained {
                    assert_eq!(db.get_evidence_span(&row.id).unwrap(), Some(row));
                }
            }
        }
    }

    #[test]
    fn migrated_history_cannot_be_truncated_or_rewritten_during_growth() {
        let (db, workspace, id, session) = fixture("/private/migrated-history.jsonl", 2);
        db.execute_raw(
            "UPDATE evidence_spans SET producer_kind = 'legacy_unknown', search_eligibility = 'denied', pack_eligibility = 'denied'",
        )
        .unwrap();
        for row in db.list_evidence_spans_for_session(&id).unwrap() {
            use_legacy_source_metadata(&db, &row.id);
        }
        let mut incoming: Vec<_> = (1..=3).map(|line| span(&session, line)).collect();
        let mut missing = incoming.clone();
        missing.remove(0);
        assert_refresh_refused_without_writes(
            &db,
            &workspace,
            &id,
            &session,
            &missing,
            "cass_refresh_history_missing",
        );
        incoming[0] = super::super::parse_view_line_value(
            &json!({"line": 1, "content": "PRIVATE_PAYLOAD_SENTINEL changed history"}),
            &session.source_path,
        )
        .unwrap();
        assert_refresh_refused_without_writes(
            &db,
            &workspace,
            &id,
            &session,
            &incoming,
            "cass_refresh_history_changed",
        );
    }

    #[test]
    fn legacy_unknown_source_commitment_cannot_authorize_refresh() {
        let (db, workspace, id, session) = fixture("/private/legacy-source-claim.jsonl", 1);
        let first = span(&session, 1);
        let evidence_id = stable_evidence_id(&id, &first.cass_span_id);
        let stored = db.get_evidence_span(&evidence_id).unwrap().unwrap();
        assert!(
            CassSourceCommitment::from_metadata(stored.metadata_json.as_deref())
                .unwrap()
                .is_some()
        );
        db.execute_raw(
            "UPDATE evidence_spans SET producer_kind = 'legacy_unknown', search_eligibility = 'denied', pack_eligibility = 'denied'",
        )
        .unwrap();
        let incoming = [first, span(&session, 2)];
        for _ in 0..2 {
            assert_refresh_refused_without_writes(
                &db,
                &workspace,
                &id,
                &session,
                &incoming,
                "cass_refresh_history_changed",
            );
        }
    }

    #[test]
    fn other_producer_cannot_be_reimported_under_a_fresh_cass_identity() {
        let (db, workspace, id, session) = fixture("/private/producer-conflict.jsonl", 1);
        let other_id = stable_evidence_id(&id, "other-producer-identity");
        db.execute_raw(&format!(
            "UPDATE evidence_spans SET id = {}, producer_kind = 'journal_distill', search_eligibility = 'denied', pack_eligibility = 'denied'",
            sql_text(&other_id),
        ))
        .unwrap();
        let incoming = [span(&session, 1), span(&session, 2)];
        assert_refresh_refused_without_writes(
            &db,
            &workspace,
            &id,
            &session,
            &incoming,
            "cass_refresh_producer_conflict",
        );
    }

    #[test]
    fn migrated_evidence_keeps_its_historical_identity_during_growth() {
        let (db, workspace, id, session) = fixture("/private/historical-identity.jsonl", 1);
        let original = span(&session, 1);
        let current_id = stable_evidence_id(&id, &original.cass_span_id);
        let historical_id = stable_evidence_id(&id, "historical-evidence-identity");
        db.execute_raw(&format!(
            "UPDATE evidence_spans SET id = {}, producer_kind = 'legacy_unknown', cass_span_id = {}, search_eligibility = 'denied', pack_eligibility = 'denied'",
            sql_text(&historical_id),
            sql_text(&original.cass_span_id),
        ))
        .unwrap();
        use_legacy_source_metadata(&db, &historical_id);
        let retained = db.get_evidence_span(&historical_id).unwrap().unwrap();
        let incoming = [original, span(&session, 2)];
        let report = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert_eq!(report.added_lines, vec![2]);
        assert_eq!(
            db.get_evidence_span(&historical_id).unwrap(),
            Some(retained)
        );
        assert!(db.get_evidence_span(&current_id).unwrap().is_none());
        assert_eq!(db.list_evidence_spans_for_session(&id).unwrap().len(), 2);
        let again = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert!(!again.changed);
        assert!(again.added_lines.is_empty());
        assert_eq!(again.index_job_id, report.index_job_id);
    }

    #[test]
    fn unrelated_producer_evidence_is_preserved_without_blocking_cass_growth() {
        let (db, workspace, id, session) = fixture("/private/mixed-producers.jsonl", 1);
        let other_id = stable_evidence_id(&id, "journal-entry-identity");
        db.execute_raw(&format!(
            "UPDATE evidence_spans SET id = {}, producer_kind = 'journal_distill', cass_span_id = 'journal:unrelated-entry', search_eligibility = 'denied', pack_eligibility = 'denied'",
            sql_text(&other_id),
        ))
        .unwrap();
        let retained = db.get_evidence_span(&other_id).unwrap().unwrap();
        let incoming = [span(&session, 1), span(&session, 2)];
        let report = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert_eq!(report.added_lines, vec![1, 2]);
        assert_eq!(db.get_evidence_span(&other_id).unwrap(), Some(retained));
        assert_eq!(db.list_evidence_spans_for_session(&id).unwrap().len(), 3);
        let again = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert!(!again.changed);
        assert!(again.added_lines.is_empty());
    }

    #[test]
    fn derived_incident_cards_neither_block_nor_join_transcript_refresh() {
        let (db, workspace, id, session) = fixture("/private/incident-card.jsonl", 1);
        let card_id = crate::core::incident_card::incident_card_id(&workspace, "ev_failure");
        let excerpt = format!(
            "{}1-1): `cargo test` failed, then passed after a fix.\nSymptom: cargo: test failed\nFix: Added the missing derive to the widget type.\nVerified: `cargo test` succeeded afterwards.",
            crate::core::incident_card::INCIDENT_CARD_PREFIX
        );
        db.insert_evidence_span(
            &card_id,
            &crate::db::CreateEvidenceSpanInput {
                workspace_id: workspace.clone(),
                session_id: id.clone(),
                memory_id: None,
                producer_kind: crate::db::EvidenceProducerKind::CassImport,
                cass_span_id: format!("ee-incident-card:test:{card_id}"),
                span_kind: "summary".to_owned(),
                start_line: 1,
                end_line: 1,
                start_byte: None,
                end_byte: None,
                role: None,
                content_hash: format!("blake3:{}", blake3::hash(excerpt.as_bytes()).to_hex()),
                excerpt,
                metadata_json: None,
                inherited_redaction_classes: Vec::new(),
            },
        )
        .unwrap();
        let card = db.get_evidence_span(&card_id).unwrap().unwrap();
        assert!(card.is_derived_incident_card());
        assert_eq!(card.search_eligibility, "admitted");

        let incoming = [span(&session, 1), span(&session, 2)];
        let report = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert_eq!(report.added_lines, vec![2]);
        assert_eq!(db.get_evidence_span(&card_id).unwrap(), Some(card));
        let again = refresh_session(&db, &workspace, &id, &session, &incoming).unwrap();
        assert!(!again.changed);
        assert!(again.added_lines.is_empty());
    }

    #[test]
    fn invalid_new_evidence_cannot_poison_a_durable_refresh_checkpoint() {
        let (db, workspace, id, session) = fixture("/private/invalid-refresh.jsonl", 1);
        for malformed in 0..5 {
            let mut incoming = vec![span(&session, 1), span(&session, 2)];
            match malformed {
                0 => incoming[1].start_line = 0,
                1 => incoming[1].end_line = 3,
                2 => incoming[1].cass_span_id = "/private/other-session.jsonl:2".to_owned(),
                3 => incoming[1].content_hash = "not-the-excerpt-digest".to_owned(),
                _ => incoming[1].excerpt = "PRIVATE_PAYLOAD_SENTINEL substituted".to_owned(),
            }
            assert_refresh_refused_without_writes(
                &db,
                &workspace,
                &id,
                &session,
                &incoming,
                "cass_refresh_invalid_span",
            );
        }
    }

    #[test]
    fn publication_retry_follows_latest_checkpoint_without_touching_evidence() {
        let (db, workspace, id, session) = fixture("/private/publication-retry.jsonl", 1);
        db.execute_raw(
            "UPDATE evidence_spans SET search_eligibility = 'denied', pack_eligibility = 'denied'",
        )
        .unwrap();
        db.execute_raw("UPDATE search_index_jobs SET status = 'completed'")
            .unwrap();
        let spans = [span(&session, 1), span(&session, 2)];
        let grown = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
        let latest = grown.index_job_id.unwrap();
        assert_ne!(latest, stable_search_index_job_id(&workspace, &id));
        let before_session = db.get_session(&id).unwrap();
        let before_evidence = db.list_evidence_spans_for_session(&id).unwrap();
        let before_counts = row_counts(&db);
        for status in ["pending", "failed", "cancelled", "running", "completed"] {
            db.execute_raw(&format!(
                "UPDATE search_index_jobs SET status = {}, documents_total = 17 WHERE id = {}",
                sql_text(status),
                sql_text(&latest),
            ))
            .unwrap();
            let before_job = db.get_search_index_job(&latest).unwrap().unwrap();
            let recovered = reconcile_session_publication(&db, &workspace, &id).unwrap();
            assert_eq!(recovered, (status != "completed").then(|| latest.clone()));
            assert_eq!(db.get_session(&id).unwrap(), before_session);
            assert_eq!(
                db.list_evidence_spans_for_session(&id).unwrap(),
                before_evidence
            );
            assert_eq!(row_counts(&db), before_counts);
            let after_job = db.get_search_index_job(&latest).unwrap().unwrap();
            assert_eq!(
                after_job.status, before_job.status,
                "do not steal live work"
            );
            assert_eq!(after_job.documents_total, 17, "retain publisher progress");
        }
    }

    #[test]
    fn publication_retry_recreates_missing_latest_job_not_its_completed_predecessor() {
        let (db, workspace, id, session) = fixture("/private/recover-latest.jsonl", 1);
        db.execute_raw("UPDATE search_index_jobs SET status = 'completed'")
            .unwrap();
        let spans = [span(&session, 1), span(&session, 2)];
        let latest = refresh_session(&db, &workspace, &id, &session, &spans)
            .unwrap()
            .index_job_id
            .unwrap();
        let stored = db.get_session(&id).unwrap();
        let evidence = db.list_evidence_spans_for_session(&id).unwrap();
        // Only this test's derived queue row is removed; the checkpoint stays.
        db.execute_raw(&format!(
            "DELETE FROM search_index_jobs WHERE id = {}",
            sql_text(&latest)
        ))
        .unwrap();
        let before = row_counts(&db);
        for _ in 0..2 {
            assert_eq!(
                reconcile_session_publication(&db, &workspace, &id).unwrap(),
                Some(latest.clone())
            );
            let work = db.get_search_index_job(&latest).unwrap().unwrap();
            assert_eq!(work.status_enum(), Some(SearchIndexJobStatus::Pending));
            assert_eq!(work.document_id.as_deref(), Some(id.as_str()));
            assert_eq!(db.get_session(&id).unwrap(), stored);
            assert_eq!(db.list_evidence_spans_for_session(&id).unwrap(), evidence);
            let after = row_counts(&db);
            assert_eq!(after, [before[0], before[1], before[2] + 1, before[3]]);
        }
        let original = stable_search_index_job_id(&workspace, &id);
        assert_eq!(
            db.get_search_index_job(&original)
                .unwrap()
                .unwrap()
                .status_enum(),
            Some(SearchIndexJobStatus::Completed)
        );
    }

    #[test]
    fn publication_retry_preserves_original_job_recovery_before_any_refresh() {
        let (db, workspace, id, _) = fixture("/private/recover-original.jsonl", 0);
        let original = stable_search_index_job_id(&workspace, &id);
        let stored = db.get_session(&id).unwrap();
        assert_eq!(
            reconcile_session_publication(&db, &workspace, &id).unwrap(),
            Some(original.clone())
        );
        db.execute_raw("UPDATE search_index_jobs SET status = 'completed'")
            .unwrap();
        assert!(
            reconcile_session_publication(&db, &workspace, &id)
                .unwrap()
                .is_none()
        );
        db.execute_raw(&format!(
            "DELETE FROM search_index_jobs WHERE id = {}",
            sql_text(&original)
        ))
        .unwrap();
        assert_eq!(
            reconcile_session_publication(&db, &workspace, &id).unwrap(),
            Some(original)
        );
        assert_eq!(db.get_session(&id).unwrap(), stored);
        assert_eq!(db.count_table_rows("search_index_jobs").unwrap(), 1);
    }

    #[test]
    fn publication_retry_refuses_corrupt_checkpoint_before_consulting_completed_job() {
        let (db, workspace, id, _) = fixture("/private/corrupt-checkpoint.jsonl", 1);
        db.execute_raw("UPDATE search_index_jobs SET status = 'completed'")
            .unwrap();
        let original = stable_search_index_job_id(&workspace, &id);
        let invalid_json = "not-json PRIVATE_PAYLOAD_SENTINEL";
        let before_invalid = db.get_session(&id).unwrap();
        assert!(metadata_object(Some(invalid_json)).is_err());
        // The live schema rejects invalid JSON itself. Exercise that guard
        // rather than disabling it to manufacture an impossible stored row.
        assert!(
            db.execute_raw(&format!(
                "UPDATE sessions SET metadata_json = {} WHERE id = {}",
                sql_text(invalid_json),
                sql_text(&id),
            ))
            .is_err()
        );
        assert_eq!(db.get_session(&id).unwrap(), before_invalid);
        for metadata in [
            "[]".to_owned(),
            json!({CHECKPOINT_KEY: null}).to_string(),
            json!({CHECKPOINT_KEY: {"schema": CHECKPOINT_SCHEMA}}).to_string(),
            json!({CHECKPOINT_KEY: {
                "schema": CHECKPOINT_SCHEMA,
                "snapshotRevision": format!("blake3:{}", "a".repeat(64)),
                "indexJobId": original,
                "previousIndexJobId": original,
            }})
            .to_string(),
        ] {
            db.execute_raw(&format!(
                "UPDATE sessions SET metadata_json = {} WHERE id = {}",
                sql_text(&metadata),
                sql_text(&id)
            ))
            .unwrap();
            let stored = db.get_session(&id).unwrap();
            let before = row_counts(&db);
            let error = reconcile_session_publication(&db, &workspace, &id)
                .unwrap_err()
                .to_string();
            assert!(error.contains("cass_refresh_"));
            assert!(!error.contains("PRIVATE_PAYLOAD_SENTINEL"));
            assert!(!error.contains("/private/"));
            assert_eq!(db.get_session(&id).unwrap(), stored);
            assert_eq!(row_counts(&db), before);
        }
    }

    #[test]
    fn publication_retry_refuses_mismatched_completed_work() {
        let (db, workspace, id, _) = fixture("/private/mismatched-job.jsonl", 1);
        let original = stable_search_index_job_id(&workspace, &id);
        db.execute_raw(
            "UPDATE search_index_jobs SET status = 'completed', document_source = 'memory'",
        )
        .unwrap();
        let before = row_counts(&db);
        let error = reconcile_session_publication(&db, &workspace, &id)
            .unwrap_err()
            .to_string();
        assert!(error.contains("cass_refresh_index_job_scope_mismatch"));
        assert_eq!(row_counts(&db), before);
        assert_eq!(
            db.get_search_index_job(&original)
                .unwrap()
                .unwrap()
                .document_source
                .as_deref(),
            Some("memory")
        );
        db.execute_raw(
            "UPDATE search_index_jobs SET document_source = 'session', job_type = 'full_rebuild'",
        )
        .unwrap();
        let error = reconcile_session_publication(&db, &workspace, &id)
            .unwrap_err()
            .to_string();
        assert!(error.contains("cass_refresh_index_job_invalid"));
        assert_eq!(row_counts(&db), before);
    }

    #[test]
    fn publication_retry_cannot_create_work_for_a_missing_or_foreign_session() {
        let (db, workspace, id, _) = fixture("/private/session-scope.jsonl", 1);
        let before = row_counts(&db);
        for (scope, target, code) in [
            (
                workspace.as_str(),
                "sess_missing",
                "cass_refresh_session_missing",
            ),
            ("wsp_foreign", id.as_str(), "cass_refresh_scope_mismatch"),
        ] {
            let error = reconcile_session_publication(&db, scope, target)
                .unwrap_err()
                .to_string();
            assert!(error.contains(code));
            assert_eq!(row_counts(&db), before);
        }
    }

    #[test]
    fn real_index_publication_round_trips_through_metadata_only_recovery() {
        let (db, workspace, id, session) = fixture("/private/real-publication.jsonl", 1);
        let index = tempfile::tempdir().unwrap();
        let original = stable_search_index_job_id(&workspace, &id);
        let published =
            crate::core::index::process_index_job_for_connection(&db, &original, index.path())
                .unwrap();
        assert_eq!(published.outcome, "completed", "{published:?}");
        assert!(
            published.documents_total > 1,
            "session and evidence must be indexed"
        );
        assert!(
            reconcile_session_publication(&db, &workspace, &id)
                .unwrap()
                .is_none()
        );

        let spans = [span(&session, 1), span(&session, 2), span(&session, 3)];
        let growth = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
        let latest = growth.index_job_id.unwrap();
        assert_ne!(latest, original);
        assert_eq!(
            reconcile_session_publication(&db, &workspace, &id).unwrap(),
            Some(latest.clone())
        );
        let before_session = db.get_session(&id).unwrap();
        let before_evidence = db.list_evidence_spans_for_session(&id).unwrap();
        let published =
            crate::core::index::process_index_job_for_connection(&db, &latest, index.path())
                .unwrap();
        assert_eq!(published.outcome, "completed", "{published:?}");
        assert!(published.documents_total > 1);
        let work = db.get_search_index_job(&latest).unwrap().unwrap();
        assert_eq!(work.status_enum(), Some(SearchIndexJobStatus::Completed));
        assert_eq!(work.documents_total, published.documents_total);
        assert!(
            reconcile_session_publication(&db, &workspace, &id)
                .unwrap()
                .is_none()
        );
        let retry = refresh_session(&db, &workspace, &id, &session, &spans).unwrap();
        assert!(!retry.changed);
        assert!(retry.index_job_id.is_none());
        assert_eq!(db.get_session(&id).unwrap(), before_session);
        assert_eq!(
            db.list_evidence_spans_for_session(&id).unwrap(),
            before_evidence
        );
    }
}