shepherd-registry 6.6.0

/*
    Appellation: registry <module>
    Created At: 2026.08.14
    Contrib: @FL03
*/
//! Typed ownership boundary around one Shepherd SQLite registry.

use std::cell::Cell;
use std::path::{Path, PathBuf};
use std::time::Duration;

use rusqlite::{Connection, OpenFlags, Params, Row, Transaction, TransactionBehavior, types::Type};
use sha2::{Digest, Sha256};

use shepherd_core::dispatch::{
    AgentId, AgentType, DispatchRecord, DispatchState, PendingDispatch, PendingLaunchState,
    ProjectId, ReviewCustody, ReviewCustodyState, Role, RunId, SessionId,
};

use crate::error::{Error, Result};

const CLAIM_SELECT: &str = "SELECT c.project_id, c.run_id, c.role, c.lane_key, c.lane_id, c.agent_id, c.harness, c.agent_type, c.parent_agent_id, c.session_id, c.identity_fingerprint, c.claimed_at, c.resumed_from_agent_id, c.write_scope, c.publication_nonce, p.publication_state, p.record_sha256, p.record_path FROM dispatch_singleton_claims c LEFT JOIN dispatch_singleton_publications p ON p.nonce = c.publication_nonce";
const PUBLICATION_SELECT: &str = "SELECT nonce, project_id, run_id, role, lane_key, record_path, record_sha256, record_json, claim_json, publication_state, prepared_at, published_at, quarantine_reason, updated_at FROM dispatch_singleton_publications";

/// Native input for one logical Engineer-per-run or Conductor-per-lane claim.
///
/// Native agent incarnation fields may change during a cross-harness resume. The
/// stable fingerprint intentionally covers only the logical ownership boundary:
/// project, run, role, lane, parent, and write scope.
#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct DispatchSingletonInput {
    pub project_id: String,
    pub run_id: String,
    pub role: String,
    pub lane_id: Option<String>,
    pub agent_id: String,
    pub harness: String,
    pub agent_type: String,
    pub parent_agent_id: Option<String>,
    pub session_id: String,
    pub write_scope: Vec<String>,
    pub claimed_at: i64,
    pub resumes_agent_id: Option<String>,
}

/// The durable cross-resource state of one singleton publication.
#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantArray,
    strum::VariantNames,
)]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum SingletonPublicationState {
    Preparing,
    Published,
    Quarantined,
}

impl SingletonPublicationState {
    fn as_str(self) -> &'static str {
        match self {
            Self::Preparing => "preparing",
            Self::Published => "published",
            Self::Quarantined => "quarantined",
        }
    }
}

impl TryFrom<String> for SingletonPublicationState {
    type Error = Error;

    fn try_from(value: String) -> Result<Self> {
        match value.as_str() {
            "preparing" => Ok(Self::Preparing),
            "published" => Ok(Self::Published),
            "quarantined" => Ok(Self::Quarantined),
            _ => Err(Error::InvalidSingletonPublication(format!(
                "unknown publication state `{value}`"
            ))),
        }
    }
}

/// The complete SQLite publication intent. `record_json` is the replay
/// payload and `record_sha256` binds it to the filesystem bytes.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DispatchSingletonPublication {
    pub nonce: String,
    pub project_id: String,
    pub run_id: String,
    pub role: String,
    pub lane_key: String,
    pub record_path: String,
    pub record_sha256: String,
    pub record_json: String,
    pub claim: DispatchSingletonInput,
    pub state: SingletonPublicationState,
    pub prepared_at: i64,
    pub published_at: Option<i64>,
    pub quarantine_reason: Option<String>,
    pub updated_at: i64,
}

/// Input used to prepare one nonce-keyed publication before touching the
/// filesystem.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DispatchSingletonPublicationInput {
    pub nonce: String,
    pub claim: DispatchSingletonInput,
    pub record_path: String,
    pub record_sha256: String,
    pub record_json: String,
    pub prepared_at: i64,
}

/// The authoritative current owner of one native singleton key.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DispatchSingletonClaim {
    pub project_id: String,
    pub run_id: String,
    pub role: String,
    pub lane_key: String,
    pub lane_id: Option<String>,
    pub agent_id: String,
    pub harness: String,
    pub agent_type: String,
    pub parent_agent_id: Option<String>,
    pub session_id: String,
    pub identity_fingerprint: String,
    pub claimed_at: i64,
    pub resumed_from_agent_id: Option<String>,
    pub write_scope: Vec<String>,
    pub publication_nonce: Option<String>,
    pub publication_state: Option<SingletonPublicationState>,
    pub record_sha256: Option<String>,
    pub record_path: Option<String>,
}

/// Whether a transaction created a new claim or advanced the exact existing
/// claim through a native resume.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DispatchSingletonClaimOutcome {
    Created(DispatchSingletonClaim),
    Resumed(DispatchSingletonClaim),
}

/// Stable SHA-256 fingerprint for the logical singleton identity.
pub fn dispatch_singleton_fingerprint(input: &DispatchSingletonInput) -> String {
    let fields = [
        input.project_id.as_str(),
        input.run_id.as_str(),
        input.role.as_str(),
        input.lane_id.as_deref().unwrap_or(""),
        input.parent_agent_id.as_deref().unwrap_or(""),
    ];
    let mut scopes = input.write_scope.clone();
    scopes.sort();
    let mut digest = Sha256::new();
    for field in fields {
        update_fingerprint_field(&mut digest, field.as_bytes());
    }
    for scope in scopes {
        update_fingerprint_field(&mut digest, scope.as_bytes());
    }
    let bytes = digest.finalize();
    hex_digest(&bytes)
}

/// The filesystem and mutation posture used when opening a registry.
#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum OpenMode {
    /// Open an existing database and reject every mutation before execution.
    ReadOnly,
    /// Open an existing database for reads and writes, never creating it.
    ReadWrite,
    /// Open a database for reads and writes, creating it when absent.
    ReadWriteCreate,
}

impl OpenMode {
    const fn flags(self) -> OpenFlags {
        let access = match self {
            Self::ReadOnly => OpenFlags::SQLITE_OPEN_READ_ONLY,
            Self::ReadWrite => OpenFlags::SQLITE_OPEN_READ_WRITE,
            Self::ReadWriteCreate => {
                OpenFlags::SQLITE_OPEN_READ_WRITE.union(OpenFlags::SQLITE_OPEN_CREATE)
            }
        };
        access.union(OpenFlags::SQLITE_OPEN_NOFOLLOW)
    }

    const fn can_write(self) -> bool {
        !matches!(self, Self::ReadOnly)
    }
}

/// An opened Shepherd registry with an explicit read/write posture.
#[derive(Debug)]
pub struct Registry {
    connection: Connection,
    mode: OpenMode,
    path: PathBuf,
}

impl Registry {
    /// SQLite lock contention is bounded, never an unbounded hook or CLI hang.
    pub const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5);

    /// Open `path` with the requested creation and mutation posture.
    pub fn open(path: impl AsRef<Path>, mode: OpenMode) -> Result<Self> {
        let path = path.as_ref().to_path_buf();
        let open_path = safe_open_path(&path)?;
        let connection = Connection::open_with_flags(&open_path, mode.flags())?;
        connection.busy_timeout(Self::DEFAULT_BUSY_TIMEOUT)?;
        connection.execute_batch("PRAGMA foreign_keys = ON; PRAGMA synchronous = FULL;")?;
        if !mode.can_write() {
            connection.execute_batch("PRAGMA query_only = ON;")?;
        }
        Ok(Self {
            connection,
            mode,
            path,
        })
    }

    /// Open a writable registry, create it when absent, and apply every schema migration.
    pub fn open_migrated(path: impl AsRef<Path>) -> Result<Self> {
        let registry = Self::open(path, OpenMode::ReadWriteCreate)?;
        registry.apply_migrations()?;
        Ok(registry)
    }

    /// The exact path passed to [`Self::open`].
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// The immutable open posture for this handle.
    pub const fn mode(&self) -> OpenMode {
        self.mode
    }

    /// Apply every embedded migration and return the resulting schema version.
    pub fn apply_migrations(&self) -> Result<u32> {
        self.require_write()?;
        crate::migrate::apply_all(&self.connection)
    }

    /// Return the greatest recorded schema version.
    pub fn schema_version(&self) -> Result<u32> {
        let version: i64 = self.connection.query_row(
            "SELECT COALESCE(MAX(version), 0) FROM schema_versions",
            [],
            |row| row.get(0),
        )?;
        u32::try_from(version)
            .map_err(|_| Error::unknown(format!("schema_versions.version out of range: {version}")))
    }

    /// Execute one parameterized mutating statement.
    pub fn execute<P>(&self, sql: &str, params: P) -> Result<usize>
    where
        P: Params,
    {
        self.require_write()?;
        Ok(self.connection.execute(sql, params)?)
    }

    /// Decode every row returned by a parameterized query.
    pub fn query<T, P, F>(&self, sql: &str, params: P, mut decode: F) -> Result<Vec<T>>
    where
        P: Params,
        F: FnMut(&Row<'_>) -> rusqlite::Result<T>,
    {
        let mut statement = self.connection.prepare(sql)?;
        let rows = statement.query_map(params, |row| decode(row))?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(decode_query_error)
    }

    /// Decode exactly one row returned by a parameterized query.
    pub fn query_one<T, P, F>(&self, sql: &str, params: P, decode: F) -> Result<T>
    where
        P: Params,
        F: FnOnce(&Row<'_>) -> rusqlite::Result<T>,
    {
        Ok(self.connection.query_row(sql, params, decode)?)
    }

    /// Run `body` inside one explicit transaction.
    pub fn transaction<T, F>(&mut self, body: F) -> Result<T>
    where
        F: FnOnce(&RegistryTransaction<'_>) -> Result<T>,
    {
        self.require_write()?;
        let transaction = self.connection.transaction()?;
        let wrapped = RegistryTransaction {
            transaction: &transaction,
            commit_on_error: Cell::new(false),
        };
        let result = body(&wrapped);

        match result {
            Ok(value) => {
                transaction.commit()?;
                Ok(value)
            }
            Err(cause) if wrapped.commit_on_error.get() => {
                transaction.commit()?;
                Err(cause)
            }
            Err(cause) => match transaction.rollback() {
                Ok(()) => Err(cause),
                Err(rollback) => Err(Error::TransactionRollback {
                    cause: cause.to_string(),
                    rollback: rollback.to_string(),
                }),
            },
        }
    }

    /// Run `body` under SQLite's immediate writer lock.
    ///
    /// Singleton claims use this boundary so two native processes cannot both
    /// observe an empty key and proceed. The generic error lets callers roll
    /// back the registry transaction when their filesystem publication fails.
    pub fn transaction_immediate<T, E, F>(&mut self, body: F) -> core::result::Result<T, E>
    where
        E: From<Error> + core::fmt::Display,
        F: FnOnce(&RegistryTransaction<'_>) -> core::result::Result<T, E>,
    {
        self.require_write().map_err(E::from)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(Error::from)
            .map_err(E::from)?;
        let wrapped = RegistryTransaction {
            transaction: &transaction,
            commit_on_error: Cell::new(false),
        };
        let result = body(&wrapped);
        let commit_on_error = wrapped.commit_on_error.get();

        match result {
            Ok(value) => transaction
                .commit()
                .map(|()| value)
                .map_err(Error::from)
                .map_err(E::from),
            Err(cause) if commit_on_error => transaction
                .commit()
                .map_err(Error::from)
                .map_err(E::from)
                .and(Err(cause)),
            Err(cause) => match transaction.rollback() {
                Ok(()) => Err(cause),
                Err(rollback) => Err(E::from(Error::TransactionRollback {
                    cause: cause.to_string(),
                    rollback: rollback.to_string(),
                })),
            },
        }
    }

    /// Load the current claim for one authoritative singleton key.
    pub fn load_dispatch_singleton(
        &self,
        project_id: &str,
        run_id: &str,
        role: &str,
        lane_key: &str,
    ) -> Result<Option<DispatchSingletonClaim>> {
        let rows = self.query(
            &format!("{CLAIM_SELECT} WHERE c.project_id = ?1 AND c.run_id = ?2 AND c.role = ?3 AND c.lane_key = ?4"),
            (project_id, run_id, role, lane_key),
            decode_claim,
        )?;
        Ok(rows.into_iter().next())
    }

    /// Load one nonce-keyed publication intent for replay reconciliation.
    pub fn load_dispatch_publication(
        &self,
        nonce: &str,
    ) -> Result<Option<DispatchSingletonPublication>> {
        let rows = self.query(
            &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
            [nonce],
            decode_publication,
        )?;
        Ok(rows.into_iter().next())
    }

    /// Load every publication intent. The filesystem adapter owns the replay
    /// walk; the registry owns the durable list and state transitions.
    pub fn list_dispatch_publications(&self) -> Result<Vec<DispatchSingletonPublication>> {
        self.query(
            &format!("{PUBLICATION_SELECT} ORDER BY prepared_at, nonce"),
            (),
            decode_publication,
        )
    }

    fn require_write(&self) -> Result<()> {
        if self.mode.can_write() {
            Ok(())
        } else {
            Err(Error::ReadOnly)
        }
    }
}

fn safe_open_path(path: &Path) -> Result<PathBuf> {
    let file_name = path.file_name().ok_or_else(|| {
        Error::UnsafePath(format!(
            "registry path has no file name: {}",
            path.display()
        ))
    })?;
    let absolute = if path.is_absolute() {
        path.to_path_buf()
    } else {
        std::env::current_dir()
            .map_err(|source| Error::UnsafePath(format!("cannot resolve registry cwd: {source}")))?
            .join(path)
    };
    let parent = absolute
        .parent()
        .filter(|parent| !parent.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("/"));
    reject_symlink_ancestors(parent)?;
    let resolved = parent.join(file_name);
    match std::fs::symlink_metadata(&resolved) {
        Ok(metadata) if metadata.file_type().is_symlink() => Err(Error::UnsafePath(format!(
            "symbolic-link database target {}",
            path.display()
        ))),
        Ok(metadata) if metadata.file_type().is_file() => Ok(resolved),
        Ok(_) => Err(Error::UnsafePath(format!(
            "registry target is not a regular file: {}",
            path.display()
        ))),
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(resolved),
        Err(source) => Err(Error::UnsafePath(format!(
            "cannot inspect registry target {}: {source}",
            path.display()
        ))),
    }
}

fn reject_symlink_ancestors(path: &Path) -> Result<()> {
    let mut walked = PathBuf::new();
    for component in path.components() {
        walked.push(component.as_os_str());
        if matches!(
            component,
            std::path::Component::Prefix(_) | std::path::Component::RootDir
        ) || walked.parent().is_none()
        {
            continue;
        }
        let metadata = std::fs::symlink_metadata(&walked).map_err(|source| {
            Error::UnsafePath(format!(
                "cannot inspect registry ancestor {}: {source}",
                walked.display()
            ))
        })?;
        if metadata.file_type().is_symlink() || !metadata.is_dir() {
            return Err(Error::UnsafePath(format!(
                "registry ancestor is not a regular directory: {}",
                walked.display()
            )));
        }
    }
    Ok(())
}

/// The bounded query and mutation surface available inside a transaction.
#[derive(Debug)]
pub struct RegistryTransaction<'connection> {
    transaction: &'connection Transaction<'connection>,
    commit_on_error: Cell<bool>,
}

impl RegistryTransaction<'_> {
    /// Execute one parameterized statement in this transaction.
    pub fn execute<P>(&self, sql: &str, params: P) -> Result<usize>
    where
        P: Params,
    {
        Ok(self.transaction.execute(sql, params)?)
    }

    /// Decode every row returned by a parameterized query in this transaction.
    pub fn query<T, P, F>(&self, sql: &str, params: P, mut decode: F) -> Result<Vec<T>>
    where
        P: Params,
        F: FnMut(&Row<'_>) -> rusqlite::Result<T>,
    {
        let mut statement = self.transaction.prepare(sql)?;
        let rows = statement.query_map(params, |row| decode(row))?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(decode_query_error)
    }

    /// Decode exactly one row returned by a parameterized query in this transaction.
    pub fn query_one<T, P, F>(&self, sql: &str, params: P, decode: F) -> Result<T>
    where
        P: Params,
        F: FnOnce(&Row<'_>) -> rusqlite::Result<T>,
    {
        Ok(self.transaction.query_row(sql, params, decode)?)
    }

    fn commit_quarantine_on_error(&self) {
        self.commit_on_error.set(true);
    }

    /// Persist a nonce-keyed publication intent and its current singleton
    /// pointer in one immediate SQLite transaction.
    pub fn prepare_dispatch_singleton(
        &self,
        input: &DispatchSingletonPublicationInput,
    ) -> Result<DispatchSingletonPublication> {
        validate_publication_input(input)?;
        let lane_key = singleton_lane_key(&input.claim)?;
        let fingerprint = dispatch_singleton_fingerprint(&input.claim);
        let current = self
            .query(
                &format!("{CLAIM_SELECT} WHERE c.project_id = ?1 AND c.run_id = ?2 AND c.role = ?3 AND c.lane_key = ?4"),
                (
                    &input.claim.project_id,
                    &input.claim.run_id,
                    &input.claim.role,
                    &lane_key,
                ),
                decode_claim,
            )?
            .into_iter()
            .next();
        if let Some(existing) = self
            .query(
                &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                [&input.nonce],
                decode_publication,
            )?
            .into_iter()
            .next()
        {
            if publication_differs(&existing, input, &lane_key) {
                self.commit_quarantine_on_error();
                quarantine_existing_publication(
                    self,
                    &existing,
                    "nonce was reused for different identity or bytes",
                    input.prepared_at.max(existing.updated_at),
                )?;
                return Err(Error::SingletonPublicationConflict {
                    nonce: input.nonce.clone(),
                    reason:
                        "nonce was reused for different identity or bytes; publication quarantined"
                            .into(),
                });
            }
            if existing.state == SingletonPublicationState::Quarantined {
                return Err(Error::SingletonPublicationConflict {
                    nonce: input.nonce.clone(),
                    reason: "quarantined publication nonce cannot be replayed".into(),
                });
            }
            return Ok(existing);
        }
        validate_current_claim(current.as_ref(), input, &fingerprint)?;
        self.insert_dispatch_publication(input, lane_key, fingerprint, true)
    }

    /// Transfer a singleton only while Native holds the run lock and has
    /// rechecked the exact review-replace custody and pending contract. This
    /// is a new incarnation, not a resume of a malignant agent. The old
    /// publication remains published terminal history; no claim is released.
    pub fn prepare_review_replacement_singleton(
        &self,
        input: &DispatchSingletonPublicationInput,
        subject: &DispatchRecord,
        pending: &PendingDispatch,
        custody: &ReviewCustody,
    ) -> Result<DispatchSingletonPublication> {
        let (lane_key, fingerprint) =
            self.validate_review_replacement_singleton(input, subject, pending, custody, false)?;
        // The old claim stays occupied while this intent is merely preparing.
        self.insert_dispatch_publication(input, lane_key, fingerprint, false)
    }

    /// Native calls this under its run lock only after the replacement's
    /// exact active record and pending activation are durable. Compare the
    /// previous terminal pointer and bytes again, then advance it and publish
    /// this intent in the same immediate transaction.
    pub fn publish_review_replacement_singleton(
        &self,
        publication: &DispatchSingletonPublication,
        subject: &DispatchRecord,
        pending: &PendingDispatch,
        custody: &ReviewCustody,
        published_at: i64,
    ) -> Result<()> {
        let input = DispatchSingletonPublicationInput {
            nonce: publication.nonce.clone(),
            claim: publication.claim.clone(),
            record_path: publication.record_path.clone(),
            record_sha256: publication.record_sha256.clone(),
            record_json: publication.record_json.clone(),
            prepared_at: publication.prepared_at,
        };
        if publication.state != SingletonPublicationState::Preparing {
            return Err(invalid_review_publication(
                "replacement intent must still be preparing",
            ));
        }
        let (lane_key, fingerprint) =
            self.validate_review_replacement_singleton(&input, subject, pending, custody, true)?;
        update_or_insert_claim_transaction(
            self,
            &input.claim,
            lane_key,
            fingerprint,
            Some(&input.nonce),
        )?;
        self.mark_dispatch_singleton_published(&input.nonce, published_at)
    }

    fn validate_review_replacement_singleton(
        &self,
        input: &DispatchSingletonPublicationInput,
        subject: &DispatchRecord,
        pending: &PendingDispatch,
        custody: &ReviewCustody,
        prepared: bool,
    ) -> Result<(String, String)> {
        validate_publication_input(input)?;
        validate_terminal_review_subject(subject, pending, custody)?;
        let replacement: DispatchRecord = serde_json::from_str(&input.record_json)
            .map_err(|error| invalid_review_publication(error.to_string()))?;
        replacement
            .validate_loaded()
            .map_err(|error| invalid_review_publication(error.to_string()))?;
        if custody.state != ReviewCustodyState::Replaced
            || custody.replacement_agent_id.as_ref() != Some(&replacement.agent_id)
            || replacement.state != DispatchState::Active
            || replacement.agent_id == subject.agent_id
            || replacement.session_id == subject.session_id
            || replacement.project_id != subject.project_id
            || replacement.run != subject.run
            || replacement.root_session_id != subject.root_session_id
            || replacement.run_incarnation != subject.run_incarnation
            || replacement.harness != subject.harness
            || replacement.role != subject.role
            || replacement.lane != subject.lane
            || replacement.parent_agent_id != subject.parent_agent_id
            || replacement.write_scope != subject.write_scope
            || replacement.resumes_agent_id.is_some()
            || replacement.started_at < custody.updated_at
            || !claim_matches_record(&input.claim, &replacement)
            || input.record_json != canonical_dispatch_json(&replacement)?
        {
            return Err(invalid_review_publication(
                "replacement does not match the terminal review authorization",
            ));
        }
        let lane_key = singleton_lane_key(&input.claim)?;
        let current = self.query(
            &format!("{CLAIM_SELECT} WHERE c.project_id = ?1 AND c.run_id = ?2 AND c.role = ?3 AND c.lane_key = ?4"),
            (&input.claim.project_id, &input.claim.run_id, &input.claim.role, &lane_key), decode_claim,
        )?.into_iter().next().ok_or_else(|| invalid_review_publication("malignant singleton claim is absent"))?;
        let nonce = current
            .publication_nonce
            .as_ref()
            .ok_or_else(|| invalid_review_publication("malignant singleton has no publication"))?;
        let old = self
            .query(
                &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                [nonce],
                decode_publication,
            )?
            .into_iter()
            .next()
            .ok_or_else(|| invalid_review_publication("malignant publication is absent"))?;
        let source_json = canonical_dispatch_json(subject)?;
        let existing = self
            .query(
                &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                [&input.nonce],
                decode_publication,
            )?
            .into_iter()
            .next();
        let intent_matches = if prepared {
            existing.as_ref().is_some_and(|intent| {
                intent.state == SingletonPublicationState::Preparing
                    && intent.prepared_at == input.prepared_at
                    && !publication_differs(intent, input, &lane_key)
            })
        } else {
            existing.is_none()
        };
        if old.state != SingletonPublicationState::Published
            || current.agent_id != subject.agent_id.as_str()
            || current.publication_state != Some(SingletonPublicationState::Published)
            || current.identity_fingerprint != dispatch_singleton_fingerprint(&old.claim)
            || current.record_path.as_deref() != Some(old.record_path.as_str())
            || current.record_sha256.as_deref() != Some(old.record_sha256.as_str())
            || !current_claim_matches_input(&current, &old.claim)
            || !claim_matches_record(&old.claim, subject)
            || old.record_json != source_json
            || old.record_sha256 != hex_digest(&Sha256::digest(source_json.as_bytes()))
            || old.record_path != format!("{}/dispatch/{}.json", subject.run, subject.agent_id)
            || old.project_id != input.claim.project_id
            || old.run_id != input.claim.run_id
            || old.role != input.claim.role
            || old.lane_key != lane_key
            || dispatch_singleton_fingerprint(&input.claim) != current.identity_fingerprint
            || !intent_matches
        {
            return Err(invalid_review_publication(
                "malignant singleton claim or publication changed before replacement",
            ));
        }
        let fingerprint = dispatch_singleton_fingerprint(&input.claim);
        Ok((lane_key, fingerprint))
    }

    /// Complete only the exact Native fourth-rejection transition after a
    /// crash between filesystem quarantine and SQLite receipt refresh. The
    /// caller holds the same run lock used by activation and custody writes.
    pub fn refresh_review_terminal_singleton(
        &self,
        expected: &DispatchSingletonPublication,
        record_json: &str,
        pending: &PendingDispatch,
        custody: &ReviewCustody,
    ) -> Result<()> {
        let terminal: DispatchRecord = serde_json::from_str(record_json)
            .map_err(|error| invalid_review_publication(error.to_string()))?;
        validate_terminal_review_subject(&terminal, pending, custody)?;
        let mut prior: DispatchRecord = serde_json::from_str(&expected.record_json)
            .map_err(|error| invalid_review_publication(error.to_string()))?;
        prior
            .validate_loaded()
            .map_err(|error| invalid_review_publication(error.to_string()))?;
        if !claim_matches_record(&expected.claim, &prior) {
            return Err(invalid_review_publication(
                "prior publication does not match its singleton claim",
            ));
        }
        prior
            .quarantine_malignant(
                custody
                    .stopped_at
                    .ok_or_else(|| invalid_review_publication("missing stop time"))?,
            )
            .map_err(|error| invalid_review_publication(error.to_string()))?;
        let current = self
            .query(
                &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                [&expected.nonce],
                decode_publication,
            )?
            .into_iter()
            .next()
            .ok_or_else(|| invalid_review_publication("terminal publication disappeared"))?;
        if prior != terminal
            || canonical_dispatch_json(&terminal)? != record_json
            || current.record_path
                != format!("{}/dispatch/{}.json", terminal.run, terminal.agent_id)
            || current.record_sha256 != hex_digest(&Sha256::digest(current.record_json.as_bytes()))
        {
            return Err(invalid_review_publication(
                "terminal publication is not the exact Native quarantine transition",
            ));
        }
        if &current != expected {
            let mut already_refreshed = expected.clone();
            already_refreshed.record_json = record_json.into();
            already_refreshed.record_sha256 = hex_digest(&Sha256::digest(record_json.as_bytes()));
            already_refreshed.updated_at = current.updated_at;
            if current == already_refreshed
                && current.state == SingletonPublicationState::Published
                && current.updated_at >= expected.updated_at.max(custody.updated_at)
            {
                return Ok(());
            }
            return Err(Error::SingletonPublicationConflict {
                nonce: expected.nonce.clone(),
                reason: "publication changed during terminal recovery".into(),
            });
        }
        if current.state != SingletonPublicationState::Published {
            return Err(invalid_review_publication(
                "terminal recovery requires a published source",
            ));
        }
        self.refresh_dispatch_singleton_record(
            &current.nonce,
            record_json,
            &hex_digest(&Sha256::digest(record_json.as_bytes())),
            current.updated_at.max(custody.updated_at),
        )
    }

    fn insert_dispatch_publication(
        &self,
        input: &DispatchSingletonPublicationInput,
        lane_key: String,
        fingerprint: String,
        claim_on_prepare: bool,
    ) -> Result<DispatchSingletonPublication> {
        let publication = publication_from_input(input, lane_key.clone());
        let claim_json = encode_claim(&publication.claim)?;
        self.execute(
            "INSERT INTO dispatch_singleton_publications (nonce, project_id, run_id, role, lane_key, record_path, record_sha256, record_json, claim_json, publication_state, prepared_at, published_at, quarantine_reason, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, NULL, NULL, ?11)",
            (
                &publication.nonce,
                &publication.project_id,
                &publication.run_id,
                &publication.role,
                &publication.lane_key,
                &publication.record_path,
                &publication.record_sha256,
                &publication.record_json,
                &claim_json,
                publication.state.as_str(),
                publication.prepared_at,
            ),
        )?;
        if claim_on_prepare {
            update_or_insert_claim_transaction(
                self,
                &input.claim,
                lane_key,
                fingerprint,
                Some(&input.nonce),
            )?;
        }
        Ok(publication)
    }

    /// Mark a prepared nonce published after its final filesystem name is
    /// durable. Replaying this call is idempotent.
    pub fn mark_dispatch_singleton_published(&self, nonce: &str, published_at: i64) -> Result<()> {
        if published_at < 0 {
            return Err(Error::InvalidSingletonPublication(
                "published_at must be non-negative".into(),
            ));
        }
        let Some(publication) = self
            .query(
                &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                [nonce],
                decode_publication,
            )?
            .into_iter()
            .next()
        else {
            return Err(Error::SingletonPublicationConflict {
                nonce: nonce.into(),
                reason: "publication intent is absent".into(),
            });
        };
        match publication.state {
            SingletonPublicationState::Published => return Ok(()),
            SingletonPublicationState::Quarantined => {
                return Err(Error::SingletonPublicationConflict {
                    nonce: nonce.into(),
                    reason: "quarantined publication cannot be published".into(),
                });
            }
            SingletonPublicationState::Preparing => {}
        }
        if published_at < publication.prepared_at {
            return Err(Error::InvalidSingletonPublication(
                "published_at must not precede prepared_at".into(),
            ));
        }
        if self.execute(
            "UPDATE dispatch_singleton_publications SET publication_state = 'published', published_at = ?1, updated_at = ?1, quarantine_reason = NULL WHERE nonce = ?2 AND publication_state = 'preparing'",
            (published_at, nonce),
        )? == 0
        {
            return Err(Error::SingletonPublicationConflict {
                nonce: nonce.into(),
                reason: "publication changed during publish".into(),
            });
        }
        Ok(())
    }

    /// Refresh the durable payload hash after an in-place terminal stop
    /// rewrites the same canonical dispatch path. Ownership and nonce stay
    /// unchanged; only the persisted record bytes advance.
    pub fn refresh_dispatch_singleton_record(
        &self,
        nonce: &str,
        record_json: &str,
        record_sha256: &str,
        updated_at: i64,
    ) -> Result<()> {
        if updated_at < 0 {
            return Err(Error::InvalidSingletonPublication(
                "terminal record refresh has invalid timestamp".into(),
            ));
        }
        validate_record_json(record_json)?;
        validate_record_hash(record_json, record_sha256)?;
        let Some(publication) = self
            .query(
                &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                [nonce],
                decode_publication,
            )?
            .into_iter()
            .next()
        else {
            return Err(Error::SingletonPublicationConflict {
                nonce: nonce.into(),
                reason: "published publication intent is absent".into(),
            });
        };
        if publication.state != SingletonPublicationState::Published {
            return Err(Error::SingletonPublicationConflict {
                nonce: nonce.into(),
                reason: "only a published publication can refresh its record".into(),
            });
        }
        if updated_at < publication.updated_at {
            return Err(Error::InvalidSingletonPublication(
                "terminal record refresh moves updated_at backwards".into(),
            ));
        }
        if self.execute(
            "UPDATE dispatch_singleton_publications SET record_json = ?1, record_sha256 = ?2, updated_at = ?3 WHERE nonce = ?4 AND publication_state = 'published'",
            (record_json, record_sha256, updated_at, nonce),
        )? == 0
        {
            return Err(Error::SingletonPublicationConflict {
                nonce: nonce.into(),
                reason: "published publication changed during refresh".into(),
            });
        }
        Ok(())
    }

    /// Quarantine a corrupt or interrupted publication and release its
    /// current logical singleton pointer. The row remains as audit history.
    pub fn quarantine_dispatch_singleton(
        &self,
        nonce: &str,
        reason: &str,
        quarantined_at: i64,
    ) -> Result<()> {
        if reason.is_empty() || reason.len() > 512 || reason.chars().any(char::is_control) {
            return Err(Error::InvalidSingletonPublication(
                "quarantine reason is empty, oversized, or contains control text".into(),
            ));
        }
        if quarantined_at < 0 {
            return Err(Error::InvalidSingletonPublication(
                "quarantined_at must be non-negative".into(),
            ));
        }
        let changed = self.execute(
            "UPDATE dispatch_singleton_publications SET publication_state = 'quarantined', quarantine_reason = ?1, published_at = NULL, updated_at = ?2 WHERE nonce = ?3 AND publication_state <> 'quarantined'",
            (reason, quarantined_at, nonce),
        )?;
        if changed == 0 {
            let Some(publication) = self
                .query(
                    &format!("{PUBLICATION_SELECT} WHERE nonce = ?1"),
                    [nonce],
                    decode_publication,
                )?
                .into_iter()
                .next()
            else {
                return Err(Error::SingletonPublicationConflict {
                    nonce: nonce.into(),
                    reason: "publication intent is absent".into(),
                });
            };
            if publication.state != SingletonPublicationState::Quarantined {
                return Err(Error::SingletonPublicationConflict {
                    nonce: nonce.into(),
                    reason: "publication changed during quarantine".into(),
                });
            }
        }
        self.execute(
            "DELETE FROM dispatch_singleton_claims WHERE publication_nonce = ?1",
            [nonce],
        )?;
        Ok(())
    }

    /// Atomically create or resume one Engineer/Conductor singleton claim.
    pub fn claim_dispatch_singleton(
        &self,
        input: &DispatchSingletonInput,
    ) -> Result<DispatchSingletonClaimOutcome> {
        validate_claim_input(input)?;
        let lane_key = singleton_lane_key(input)?;
        let fingerprint = dispatch_singleton_fingerprint(input);
        let existing = self
            .query(
                &format!("{CLAIM_SELECT} WHERE c.project_id = ?1 AND c.run_id = ?2 AND c.role = ?3 AND c.lane_key = ?4"),
                (&input.project_id, &input.run_id, &input.role, &lane_key),
                decode_claim,
            )?
            .into_iter()
            .next();

        let Some(existing) = existing else {
            if input.resumes_agent_id.is_some() {
                return Err(Error::InvalidDispatchClaim(
                    "resume source has no authoritative singleton claim".into(),
                ));
            }
            let claim = claim_from_input(input, lane_key, fingerprint);
            let write_scope = encode_scope(&claim.write_scope)?;
            self.execute(
                "INSERT INTO dispatch_singleton_claims (project_id, run_id, role, lane_key, lane_id, agent_id, harness, agent_type, parent_agent_id, session_id, identity_fingerprint, write_scope, claimed_at, resumed_from_agent_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
                (
                    &claim.project_id,
                    &claim.run_id,
                    &claim.role,
                    &claim.lane_key,
                    &claim.lane_id,
                    &claim.agent_id,
                    &claim.harness,
                    &claim.agent_type,
                    &claim.parent_agent_id,
                    &claim.session_id,
                    &claim.identity_fingerprint,
                    &write_scope,
                    claim.claimed_at,
                    &claim.resumed_from_agent_id,
                ),
            )?;
            return Ok(DispatchSingletonClaimOutcome::Created(claim));
        };

        if existing.publication_state == Some(SingletonPublicationState::Preparing)
            || input.resumes_agent_id.as_deref() != Some(existing.agent_id.as_str())
            || existing.identity_fingerprint != fingerprint
            || input.agent_id == existing.agent_id
        {
            return Err(Error::DispatchClaimConflict {
                project_id: existing.project_id,
                run_id: existing.run_id,
                role: existing.role,
                lane_key: existing.lane_key,
                agent_id: existing.agent_id,
            });
        }

        let current = claim_from_input(input, lane_key, fingerprint);
        let write_scope = encode_scope(&current.write_scope)?;
        self.execute(
            "UPDATE dispatch_singleton_claims SET lane_id = ?1, agent_id = ?2, harness = ?3, agent_type = ?4, parent_agent_id = ?5, session_id = ?6, identity_fingerprint = ?7, write_scope = ?8, claimed_at = ?9, resumed_from_agent_id = ?10, publication_nonce = NULL WHERE project_id = ?11 AND run_id = ?12 AND role = ?13 AND lane_key = ?14 AND agent_id = ?15",
            (
                &current.lane_id,
                &current.agent_id,
                &current.harness,
                &current.agent_type,
                &current.parent_agent_id,
                &current.session_id,
                &current.identity_fingerprint,
                &write_scope,
                current.claimed_at,
                &current.resumed_from_agent_id,
                &current.project_id,
                &current.run_id,
                &current.role,
                &current.lane_key,
                &existing.agent_id,
            ),
        )?;
        Ok(DispatchSingletonClaimOutcome::Resumed(current))
    }
}

fn validate_publication_input(input: &DispatchSingletonPublicationInput) -> Result<()> {
    validate_claim_input(&input.claim)?;
    validate_nonce(&input.nonce)?;
    if input.prepared_at < 0 {
        return Err(Error::InvalidSingletonPublication(
            "prepared_at must be non-negative".into(),
        ));
    }
    validate_record_path(
        &input.record_path,
        &input.claim.run_id,
        &input.claim.agent_id,
    )?;
    validate_record_json(&input.record_json)?;
    validate_record_hash(&input.record_json, &input.record_sha256)?;
    Ok(())
}

fn invalid_review_publication(reason: impl Into<String>) -> Error {
    Error::InvalidSingletonPublication(reason.into())
}

fn canonical_dispatch_json(record: &DispatchRecord) -> Result<String> {
    let mut json = serde_json::to_string(record)
        .map_err(|error| invalid_review_publication(error.to_string()))?;
    json.push('\n');
    Ok(json)
}

fn claim_matches_record(claim: &DispatchSingletonInput, record: &DispatchRecord) -> bool {
    claim.project_id == record.project_id.as_str()
        && claim.run_id == record.run.as_str()
        && claim.role == record.role.as_str()
        && claim.agent_id == record.agent_id.as_str()
        && claim.lane_id.as_deref() == record.lane.as_ref().map(|lane| lane.as_str())
        && claim.harness == record.harness.to_string()
        && claim.agent_type == record.agent_type.as_str()
        && claim.parent_agent_id.as_deref() == record.parent_agent_id.as_ref().map(AgentId::as_str)
        && claim.session_id == record.session_id.as_str()
        && claim.write_scope == record.write_scope
        && claim.claimed_at == record.started_at
        && claim.resumes_agent_id.as_deref()
            == record.resumes_agent_id.as_ref().map(AgentId::as_str)
}

fn current_claim_matches_input(
    current: &DispatchSingletonClaim,
    input: &DispatchSingletonInput,
) -> bool {
    current.project_id == input.project_id
        && current.run_id == input.run_id
        && current.role == input.role
        && current.lane_id == input.lane_id
        && current.agent_id == input.agent_id
        && current.harness == input.harness
        && current.agent_type == input.agent_type
        && current.parent_agent_id == input.parent_agent_id
        && current.session_id == input.session_id
        && current.write_scope == input.write_scope
        && current.claimed_at == input.claimed_at
        && current.resumed_from_agent_id == input.resumes_agent_id
}

fn validate_terminal_review_subject(
    subject: &DispatchRecord,
    pending: &PendingDispatch,
    custody: &ReviewCustody,
) -> Result<()> {
    subject
        .validate_loaded()
        .map_err(|error| invalid_review_publication(error.to_string()))?;
    pending
        .validate()
        .map_err(|error| invalid_review_publication(error.to_string()))?;
    custody
        .validate()
        .map_err(|error| invalid_review_publication(error.to_string()))?;
    if subject.state != DispatchState::Malignant
        || custody.state == ReviewCustodyState::Active
        || !matches!(subject.role, Role::Engineer | Role::Conductor)
        || custody.project_id != subject.project_id
        || custody.run != subject.run
        || custody.root_session_id != subject.root_session_id
        || custody.subject_agent_id != subject.agent_id
        || custody.subject_session_id != subject.session_id
        || custody.subject_role != subject.role
        || custody.lane != subject.lane
        || custody.stopped_at != subject.stopped_at
        || custody.pending_launch_id_hash != pending.launch_id_hash
        || custody.task_sha256 != pending.task_sha256
        || pending.launch_state != PendingLaunchState::Quarantined
        || pending.project_id != subject.project_id
        || pending.run != subject.run
        || pending.root_session_id != subject.root_session_id
        || pending.role != subject.role
        || pending.lane != subject.lane
        || pending.expected_attachment.agent_id != subject.agent_id
        || pending.expected_child_session_id != subject.session_id
        || pending.expected_attachment.target != subject.harness
        || pending
            .parent_dispatch_id
            .as_ref()
            .map(|parent| parent.as_str())
            != subject.parent_agent_id.as_ref().map(AgentId::as_str)
        || pending
            .write_scope
            .iter()
            .map(|path| path.as_str())
            .collect::<Vec<_>>()
            != subject
                .write_scope
                .iter()
                .map(String::as_str)
                .collect::<Vec<_>>()
    {
        return Err(invalid_review_publication(
            "review custody is not the exact terminal singleton subject",
        ));
    }
    Ok(())
}

fn validate_nonce(nonce: &str) -> Result<()> {
    if nonce.len() < 8
        || nonce.len() > 128
        || nonce
            .chars()
            .any(|value| !value.is_ascii_lowercase() && !value.is_ascii_digit() && value != '-')
    {
        return Err(Error::InvalidSingletonPublication(
            "nonce must be 8..=128 lowercase ASCII characters, digits, or hyphens".into(),
        ));
    }
    Ok(())
}

fn validate_record_path(path: &str, run_id: &str, agent_id: &str) -> Result<()> {
    let expected = format!("{run_id}/dispatch/{agent_id}.json");
    if path != expected {
        return Err(Error::InvalidSingletonPublication(format!(
            "record_path must be the canonical `{expected}` path"
        )));
    }
    Ok(())
}

fn validate_record_json(record_json: &str) -> Result<()> {
    if record_json.trim().is_empty() {
        return Err(Error::InvalidSingletonPublication(
            "record_json must be non-empty".into(),
        ));
    }
    let value: serde_json::Value = serde_json::from_str(record_json).map_err(|error| {
        Error::InvalidSingletonPublication(format!("record_json is not valid JSON: {error}"))
    })?;
    if !value.is_object() {
        return Err(Error::InvalidSingletonPublication(
            "record_json must be a JSON object".into(),
        ));
    }
    Ok(())
}

fn validate_record_hash(record_json: &str, record_sha256: &str) -> Result<()> {
    if record_sha256 != hex_digest(&Sha256::digest(record_json.as_bytes())) {
        return Err(Error::InvalidSingletonPublication(
            "record_json does not match record_sha256".into(),
        ));
    }
    if record_sha256.len() != 64
        || !record_sha256
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    {
        return Err(Error::InvalidSingletonPublication(
            "record_sha256 must be lowercase hexadecimal SHA-256".into(),
        ));
    }
    Ok(())
}

fn publication_from_input(
    input: &DispatchSingletonPublicationInput,
    lane_key: String,
) -> DispatchSingletonPublication {
    DispatchSingletonPublication {
        nonce: input.nonce.clone(),
        project_id: input.claim.project_id.clone(),
        run_id: input.claim.run_id.clone(),
        role: input.claim.role.clone(),
        lane_key,
        record_path: input.record_path.clone(),
        record_sha256: input.record_sha256.clone(),
        record_json: input.record_json.clone(),
        claim: input.claim.clone(),
        state: SingletonPublicationState::Preparing,
        prepared_at: input.prepared_at,
        published_at: None,
        quarantine_reason: None,
        updated_at: input.prepared_at,
    }
}

fn publication_differs(
    existing: &DispatchSingletonPublication,
    input: &DispatchSingletonPublicationInput,
    lane_key: &str,
) -> bool {
    existing.project_id != input.claim.project_id
        || existing.run_id != input.claim.run_id
        || existing.role != input.claim.role
        || existing.lane_key != lane_key
        || existing.claim != input.claim
        || existing.record_path != input.record_path
        || existing.record_sha256 != input.record_sha256
        || existing.record_json != input.record_json
}

fn validate_current_claim(
    current: Option<&DispatchSingletonClaim>,
    input: &DispatchSingletonPublicationInput,
    fingerprint: &str,
) -> Result<()> {
    let Some(existing) = current else {
        if input.claim.resumes_agent_id.is_some() {
            return Err(Error::InvalidDispatchClaim(
                "resume source has no authoritative singleton claim".into(),
            ));
        }
        return Ok(());
    };
    if existing.publication_state == Some(SingletonPublicationState::Quarantined) {
        return Ok(());
    }
    let resumable = input.claim.resumes_agent_id.as_deref() == Some(existing.agent_id.as_str())
        && existing.identity_fingerprint == fingerprint
        && input.claim.agent_id != existing.agent_id
        && matches!(
            existing.publication_state,
            Some(SingletonPublicationState::Published) | None
        );
    if resumable {
        Ok(())
    } else {
        Err(Error::DispatchClaimConflict {
            project_id: existing.project_id.clone(),
            run_id: existing.run_id.clone(),
            role: existing.role.clone(),
            lane_key: existing.lane_key.clone(),
            agent_id: existing.agent_id.clone(),
        })
    }
}

fn update_or_insert_claim_transaction(
    transaction: &RegistryTransaction<'_>,
    input: &DispatchSingletonInput,
    lane_key: String,
    fingerprint: String,
    publication_nonce: Option<&str>,
) -> Result<()> {
    let claim = claim_from_input(input, lane_key.clone(), fingerprint);
    let existing = transaction.query(
        &format!("{CLAIM_SELECT} WHERE c.project_id = ?1 AND c.run_id = ?2 AND c.role = ?3 AND c.lane_key = ?4"),
        (&input.project_id, &input.run_id, &input.role, &lane_key),
        decode_claim,
    )?.into_iter().next();
    if existing.is_some() {
        let write_scope = encode_scope(&claim.write_scope)?;
        transaction.execute(
            "UPDATE dispatch_singleton_claims SET lane_id = ?1, agent_id = ?2, harness = ?3, agent_type = ?4, parent_agent_id = ?5, session_id = ?6, identity_fingerprint = ?7, write_scope = ?8, claimed_at = ?9, resumed_from_agent_id = ?10, publication_nonce = ?11 WHERE project_id = ?12 AND run_id = ?13 AND role = ?14 AND lane_key = ?15",
            (
                &claim.lane_id,
                &claim.agent_id,
                &claim.harness,
                &claim.agent_type,
                &claim.parent_agent_id,
                &claim.session_id,
                &claim.identity_fingerprint,
                &write_scope,
                claim.claimed_at,
                &claim.resumed_from_agent_id,
                publication_nonce,
                &claim.project_id,
                &claim.run_id,
                &claim.role,
                &claim.lane_key,
            ),
        )?;
    } else {
        let write_scope = encode_scope(&claim.write_scope)?;
        transaction.execute(
            "INSERT INTO dispatch_singleton_claims (project_id, run_id, role, lane_key, lane_id, agent_id, harness, agent_type, parent_agent_id, session_id, identity_fingerprint, write_scope, claimed_at, resumed_from_agent_id, publication_nonce) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
            (
                &claim.project_id,
                &claim.run_id,
                &claim.role,
                &claim.lane_key,
                &claim.lane_id,
                &claim.agent_id,
                &claim.harness,
                &claim.agent_type,
                &claim.parent_agent_id,
                &claim.session_id,
                &claim.identity_fingerprint,
                &write_scope,
                claim.claimed_at,
                &claim.resumed_from_agent_id,
                publication_nonce,
            ),
        )?;
    }
    Ok(())
}

fn singleton_lane_key(input: &DispatchSingletonInput) -> Result<String> {
    match input.role.as_str() {
        "engineer" => Ok("__run__".into()),
        "conductor" => input
            .lane_id
            .as_deref()
            .filter(|lane| !lane.is_empty() && lane != &"__run__")
            .map(ToOwned::to_owned)
            .ok_or_else(|| {
                Error::InvalidDispatchClaim("Conductor claims require a non-run lane".into())
            }),
        role => Err(Error::InvalidDispatchClaim(format!(
            "singleton claims do not support role `{role}`"
        ))),
    }
}

fn validate_claim_input(input: &DispatchSingletonInput) -> Result<()> {
    ProjectId::new(input.project_id.clone()).map_err(|error| invalid_claim("project_id", error))?;
    RunId::new(input.run_id.clone()).map_err(|error| invalid_claim("run_id", error))?;
    let role = Role::from_name(&input.role).map_err(|error| invalid_claim("role", error))?;
    let agent_id =
        AgentId::new(input.agent_id.clone()).map_err(|error| invalid_claim("agent_id", error))?;
    let agent_type = AgentType::new(input.agent_type.clone())
        .map_err(|error| invalid_claim("agent_type", error))?;
    SessionId::new(input.session_id.clone()).map_err(|error| invalid_claim("session_id", error))?;
    if !matches!(
        input.harness.as_str(),
        "claude" | "codex" | "pi" | "prime_agent"
    ) {
        return Err(Error::InvalidDispatchClaim(format!(
            "harness `{}` is not canonical",
            input.harness
        )));
    }
    singleton_lane_key(input)?;
    match role {
        Role::Conductor if input.lane_id.is_none() => {
            return Err(Error::InvalidDispatchClaim(
                "Conductor singleton claims require a lane id".into(),
            ));
        }
        _ => {}
    }
    if input.harness == "claude"
        && agent_type.as_str() != role.as_str()
        && agent_type.as_str() != role.carrier()
    {
        return Err(Error::InvalidDispatchClaim(format!(
            "Claude agent type `{}` disagrees with role `{role}`",
            agent_type.as_str()
        )));
    }
    if let Some(parent) = &input.parent_agent_id {
        let parent = AgentId::new(parent.clone())
            .map_err(|error| invalid_claim("parent_agent_id", error))?;
        if parent == agent_id {
            return Err(Error::InvalidDispatchClaim(
                "parent agent id must differ from agent id".into(),
            ));
        }
    }
    if let Some(source) = &input.resumes_agent_id {
        AgentId::new(source.clone()).map_err(|error| invalid_claim("resumes_agent_id", error))?;
        if source == &input.agent_id {
            return Err(Error::InvalidDispatchClaim(
                "resume source must differ from the new agent id".into(),
            ));
        }
    }
    if input.claimed_at < 0 {
        return Err(Error::InvalidDispatchClaim(
            "claimed_at must be non-negative".into(),
        ));
    }
    let mut scopes = input.write_scope.clone();
    scopes.sort();
    if scopes.windows(2).any(|pair| pair[0] == pair[1]) {
        return Err(Error::InvalidDispatchClaim(
            "write_scope entries must be unique".into(),
        ));
    }
    for scope in &input.write_scope {
        shepherd_core::dispatch::validate_write_scope_pattern(scope)
            .map_err(|error| invalid_claim("write_scope", error))?;
    }
    Ok(())
}

fn invalid_claim(field: &str, error: impl core::fmt::Display) -> Error {
    Error::InvalidDispatchClaim(format!("{field} is not canonical: {error}"))
}

fn claim_from_input(
    input: &DispatchSingletonInput,
    lane_key: String,
    fingerprint: String,
) -> DispatchSingletonClaim {
    DispatchSingletonClaim {
        project_id: input.project_id.clone(),
        run_id: input.run_id.clone(),
        role: input.role.clone(),
        lane_key,
        lane_id: input.lane_id.clone(),
        agent_id: input.agent_id.clone(),
        harness: input.harness.clone(),
        agent_type: input.agent_type.clone(),
        parent_agent_id: input.parent_agent_id.clone(),
        session_id: input.session_id.clone(),
        identity_fingerprint: fingerprint,
        claimed_at: input.claimed_at,
        resumed_from_agent_id: input.resumes_agent_id.clone(),
        write_scope: input.write_scope.clone(),
        publication_nonce: None,
        publication_state: None,
        record_sha256: None,
        record_path: None,
    }
}

fn decode_claim(row: &Row<'_>) -> rusqlite::Result<DispatchSingletonClaim> {
    let claim = DispatchSingletonClaim {
        project_id: row.get(0)?,
        run_id: row.get(1)?,
        role: row.get(2)?,
        lane_key: row.get(3)?,
        lane_id: row.get(4)?,
        agent_id: row.get(5)?,
        harness: row.get(6)?,
        agent_type: row.get(7)?,
        parent_agent_id: row.get(8)?,
        session_id: row.get(9)?,
        identity_fingerprint: row.get(10)?,
        claimed_at: row.get(11)?,
        resumed_from_agent_id: row.get(12)?,
        write_scope: decode_scope(&row.get::<_, String>(13)?)
            .map_err(|error| row_error(13, error))?,
        publication_nonce: row.get(14)?,
        publication_state: row
            .get::<_, Option<String>>(15)?
            .map(SingletonPublicationState::try_from)
            .transpose()
            .map_err(|error| row_error(15, error))?,
        record_sha256: row.get(16)?,
        record_path: row.get(17)?,
    };
    validate_loaded_claim(&claim).map_err(|error| row_error(0, error))?;
    Ok(claim)
}

fn decode_publication(row: &Row<'_>) -> rusqlite::Result<DispatchSingletonPublication> {
    let state = SingletonPublicationState::try_from(row.get::<_, String>(9)?)
        .map_err(|error| row_error(9, error))?;
    let claim = serde_json::from_str::<DispatchSingletonInput>(&row.get::<_, String>(8)?).map_err(
        |error| {
            row_error(
                8,
                Error::InvalidDispatchClaim(format!("claim_json is invalid: {error}")),
            )
        },
    )?;
    let publication = DispatchSingletonPublication {
        nonce: row.get(0)?,
        project_id: row.get(1)?,
        run_id: row.get(2)?,
        role: row.get(3)?,
        lane_key: row.get(4)?,
        record_path: row.get(5)?,
        record_sha256: row.get(6)?,
        record_json: row.get(7)?,
        claim,
        state,
        prepared_at: row.get(10)?,
        published_at: row.get(11)?,
        quarantine_reason: row.get(12)?,
        updated_at: row.get(13)?,
    };
    validate_loaded_publication(&publication).map_err(|error| row_error(0, error))?;
    Ok(publication)
}

fn row_error(index: usize, error: Error) -> rusqlite::Error {
    rusqlite::Error::FromSqlConversionFailure(index, Type::Text, Box::new(error))
}

fn decode_query_error(error: rusqlite::Error) -> Error {
    match error {
        rusqlite::Error::FromSqlConversionFailure(index, kind, source) => {
            match source.downcast::<Error>() {
                Ok(error) => *error,
                Err(source) => Error::Sqlite(rusqlite::Error::FromSqlConversionFailure(
                    index, kind, source,
                )),
            }
        }
        error => Error::Sqlite(error),
    }
}

fn claim_input_from_loaded(claim: &DispatchSingletonClaim) -> DispatchSingletonInput {
    DispatchSingletonInput {
        project_id: claim.project_id.clone(),
        run_id: claim.run_id.clone(),
        role: claim.role.clone(),
        lane_id: claim.lane_id.clone(),
        agent_id: claim.agent_id.clone(),
        harness: claim.harness.clone(),
        agent_type: claim.agent_type.clone(),
        parent_agent_id: claim.parent_agent_id.clone(),
        session_id: claim.session_id.clone(),
        write_scope: claim.write_scope.clone(),
        claimed_at: claim.claimed_at,
        resumes_agent_id: claim.resumed_from_agent_id.clone(),
    }
}

fn validate_loaded_claim(claim: &DispatchSingletonClaim) -> Result<()> {
    let input = claim_input_from_loaded(claim);
    validate_claim_input(&input)?;
    let expected_lane_key = singleton_lane_key(&input)?;
    if claim.lane_key != expected_lane_key
        || claim.identity_fingerprint != dispatch_singleton_fingerprint(&input)
    {
        return Err(Error::InvalidDispatchClaim(
            "loaded claim lane key or identity fingerprint does not match its fields".into(),
        ));
    }
    match (&claim.publication_nonce, claim.publication_state) {
        (None, None) => {
            if claim.record_sha256.is_some() || claim.record_path.is_some() {
                return Err(Error::InvalidDispatchClaim(
                    "unpublished claim carries publication facts".into(),
                ));
            }
        }
        (Some(nonce), Some(state)) => {
            validate_nonce(nonce).map_err(|error| {
                Error::InvalidDispatchClaim(format!("publication nonce is invalid: {error}"))
            })?;
            if state == SingletonPublicationState::Quarantined
                || claim.record_sha256.is_none()
                || claim.record_path.is_none()
            {
                return Err(Error::InvalidDispatchClaim(
                    "live claim points at a missing or quarantined publication".into(),
                ));
            }
            let path = claim.record_path.as_deref().unwrap_or_default();
            validate_record_path(path, &claim.run_id, &claim.agent_id).map_err(|error| {
                Error::InvalidDispatchClaim(format!("publication path is invalid: {error}"))
            })?;
            let hash = claim.record_sha256.as_deref().unwrap_or_default();
            if hash.len() != 64
                || !hash
                    .bytes()
                    .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
            {
                return Err(Error::InvalidDispatchClaim(
                    "publication hash is not lowercase SHA-256".into(),
                ));
            }
        }
        _ => {
            return Err(Error::InvalidDispatchClaim(
                "publication nonce and state must be present together".into(),
            ));
        }
    }
    Ok(())
}

fn validate_loaded_publication(publication: &DispatchSingletonPublication) -> Result<()> {
    validate_nonce(&publication.nonce)?;
    validate_claim_input(&publication.claim)?;
    let expected_lane_key = singleton_lane_key(&publication.claim)?;
    if publication.project_id != publication.claim.project_id
        || publication.run_id != publication.claim.run_id
        || publication.role != publication.claim.role
        || publication.lane_key != expected_lane_key
    {
        return Err(Error::InvalidSingletonPublication(
            "publication identity does not match its immutable claim snapshot".into(),
        ));
    }
    validate_record_path(
        &publication.record_path,
        &publication.claim.run_id,
        &publication.claim.agent_id,
    )?;
    validate_record_json(&publication.record_json)?;
    validate_record_hash(&publication.record_json, &publication.record_sha256)?;
    if publication.prepared_at < 0 || publication.updated_at < publication.prepared_at {
        return Err(Error::InvalidSingletonPublication(
            "publication timestamps are not monotonic".into(),
        ));
    }
    if publication
        .published_at
        .is_some_and(|at| at < publication.prepared_at)
    {
        return Err(Error::InvalidSingletonPublication(
            "published_at precedes prepared_at".into(),
        ));
    }
    match publication.state {
        SingletonPublicationState::Preparing
            if publication.published_at.is_none() && publication.quarantine_reason.is_none() => {}
        SingletonPublicationState::Published
            if publication.published_at.is_some() && publication.quarantine_reason.is_none() => {}
        SingletonPublicationState::Quarantined
            if publication.published_at.is_none()
                && publication
                    .quarantine_reason
                    .as_deref()
                    .is_some_and(|reason| {
                        !reason.is_empty()
                            && reason.len() <= 512
                            && !reason.chars().any(char::is_control)
                    }) => {}
        _ => {
            return Err(Error::InvalidSingletonPublication(
                "publication state does not match its timestamps and reason".into(),
            ));
        }
    }
    Ok(())
}

fn encode_scope(scope: &[String]) -> Result<String> {
    serde_json::to_string(scope)
        .map_err(|error| Error::InvalidDispatchClaim(format!("cannot encode write_scope: {error}")))
}

fn decode_scope(value: &str) -> Result<Vec<String>> {
    serde_json::from_str(value).map_err(|error| {
        Error::InvalidDispatchClaim(format!("write_scope is invalid JSON: {error}"))
    })
}

fn encode_claim(claim: &DispatchSingletonInput) -> Result<String> {
    serde_json::to_string(claim).map_err(|error| {
        Error::InvalidSingletonPublication(format!("cannot encode claim: {error}"))
    })
}

fn quarantine_existing_publication(
    transaction: &RegistryTransaction<'_>,
    publication: &DispatchSingletonPublication,
    reason: &str,
    quarantined_at: i64,
) -> Result<()> {
    if quarantined_at < publication.prepared_at {
        return Err(Error::InvalidSingletonPublication(
            "quarantine timestamp precedes preparation".into(),
        ));
    }
    if publication.state != SingletonPublicationState::Quarantined {
        transaction.execute(
            "UPDATE dispatch_singleton_publications SET publication_state = 'quarantined', published_at = NULL, quarantine_reason = ?1, updated_at = ?2 WHERE nonce = ?3 AND publication_state <> 'quarantined'",
            (reason, quarantined_at, &publication.nonce),
        )?;
    }
    transaction.execute(
        "DELETE FROM dispatch_singleton_claims WHERE publication_nonce = ?1",
        [&publication.nonce],
    )?;
    Ok(())
}

fn update_fingerprint_field(digest: &mut Sha256, value: &[u8]) {
    digest.update((value.len() as u64).to_be_bytes());
    digest.update(value);
}

fn hex_digest(bytes: &[u8]) -> String {
    use std::fmt::Write as _;

    let mut result = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        write!(result, "{byte:02x}").expect("writing to a String cannot fail");
    }
    result
}