shepherd-registry 6.7.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::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::time::Duration;

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

use crate::error::{Error, Result};
use crate::migrate::{AppliedMigration, ROLLBACK_COMMAND, RegistryInspection, StructuralFinding};
use crate::path::canonical_path_string;
use shepherd_core::digest::{format_digest, sha256_hex};
use shepherd_core::dispatch::{
    AgentId, AgentType, DispatchRecord, DispatchState, PendingDispatch, PendingLaunchState,
    ProjectId, ReviewCustody, ReviewCustodyState, Role, RunId, SessionId,
};

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";

/// The canonical typed contract for `.shepherd/project.json`.
///
/// `root` is optional only so the explicit bootstrap recovery can recognize a
/// historical rootless document. Runtime and repair callers must still require
/// and validate the exact canonical root before mutation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProjectIdentityDocument {
    id: ProjectId,
    scaffolded_at: i64,
    root: Option<String>,
}

/// A project identity failed deterministic syntax or field validation.
#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
pub enum ProjectIdentityParseError {
    #[error("malformed JSON: {0}")]
    MalformedJson(String),
    #[error("invalid project id: {0}")]
    InvalidId(String),
    #[error("field `scaffolded_at` must be a non-negative integer")]
    InvalidScaffoldedAt,
    #[error("field `root` must be a non-empty string when present")]
    InvalidRoot,
}

#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct ProjectIdentityWire {
    id: String,
    scaffolded_at: i64,
    #[serde(default, deserialize_with = "deserialize_present_project_root")]
    root: Option<String>,
}

fn deserialize_present_project_root<'de, D>(
    deserializer: D,
) -> core::result::Result<Option<String>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    // A missing field is legacy input; a present field must be a string.
    <String as serde::Deserialize>::deserialize(deserializer).map(Some)
}

impl ProjectIdentityDocument {
    /// Parse the exact project identity schema. Unknown fields are rejected so
    /// bootstrap, doctor, repair, and rollback cannot disagree on its meaning.
    pub fn parse(bytes: &[u8]) -> core::result::Result<Self, ProjectIdentityParseError> {
        let wire: ProjectIdentityWire = serde_json::from_slice(bytes)
            .map_err(|error| ProjectIdentityParseError::MalformedJson(error.to_string()))?;
        let id = ProjectId::new(wire.id)
            .map_err(|error| ProjectIdentityParseError::InvalidId(error.to_string()))?;
        if wire.scaffolded_at < 0 {
            return Err(ProjectIdentityParseError::InvalidScaffoldedAt);
        }
        if wire.root.as_deref() == Some("") {
            return Err(ProjectIdentityParseError::InvalidRoot);
        }
        Ok(Self {
            id,
            scaffolded_at: wire.scaffolded_at,
            root: wire.root,
        })
    }

    #[must_use]
    pub fn id(&self) -> &ProjectId {
        &self.id
    }

    #[must_use]
    pub const fn scaffolded_at(&self) -> i64 {
        self.scaffolded_at
    }

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

/// 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());
    }
    format_digest(digest.finalize())
}

/// 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)
    }
}

/// Explicit, project-bound repair request. Repair is never implied by an
/// ordinary read or migration open.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RegistryRepairRequest {
    pub registry_path: PathBuf,
    pub project_root: PathBuf,
    pub snapshot_dir: Option<PathBuf>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RegistryRollbackRequest {
    pub receipt_path: PathBuf,
    pub project_root: PathBuf,
    pub witness_sha256: String,
    pub receipt_sha256: String,
}

/// Immutable filesystem identity captured before and after a repair.
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RegistryFileIdentity {
    pub path: String,
    pub sha256: String,
    pub length: u64,
    pub mode: u32,
    #[cfg(unix)]
    pub device: u64,
    #[cfg(unix)]
    pub inode: u64,
    #[cfg(unix)]
    pub links: u64,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RegistrySidecar {
    pub suffix: String,
    pub source_path: String,
    pub present: bool,
    pub length: u64,
    pub mode: u32,
    pub sha256: Option<String>,
    pub snapshot_path: Option<String>,
    pub disposition: String,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RegistryRepairReceipt {
    pub schema: String,
    pub status: String,
    pub project_id: String,
    pub primary_root: String,
    pub source: RegistryFileIdentity,
    pub sidecars: Vec<RegistrySidecar>,
    pub structural_findings: Vec<StructuralFinding>,
    pub applied: Vec<AppliedMigration>,
    pub after_applied: Option<Vec<AppliedMigration>>,
    pub repair_plan: Vec<u32>,
    pub advanced_migrations: Vec<u32>,
    pub snapshot_path: String,
    pub snapshot_sha256: String,
    pub snapshot_mode: u32,
    pub snapshot_receipt_path: String,
    pub plan_sha256: String,
    pub receipt_path: String,
    pub created_at: i64,
    pub after: Option<RegistryFileIdentity>,
    pub rollback_command: String,
    pub failure: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
struct RegistryRepairPlan {
    schema: String,
    project_id: String,
    primary_root: String,
    source: RegistryFileIdentity,
    sidecars: Vec<RegistrySidecar>,
    structural_findings: Vec<StructuralFinding>,
    applied: Vec<AppliedMigration>,
    repair_plan: Vec<u32>,
    snapshot_path: String,
    snapshot_sha256: String,
    snapshot_mode: u32,
    receipt_path: String,
    created_at: i64,
    rollback_command: String,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
pub struct RegistryRepairReport {
    pub receipt: RegistryRepairReceipt,
    pub snapshot_path: PathBuf,
    pub receipt_path: PathBuf,
    pub receipt_sha256: String,
    pub rollback_command: String,
}

/// 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 = open_connection(&open_path, mode)?;
        let registry = Self {
            connection,
            mode,
            path,
        };
        if schema_versions_exists(&registry.connection)? {
            let inspection = crate::migrate::inspect(&registry.connection)?;
            if !inspection.ok() {
                return Err(inspection
                    .first_error()
                    .unwrap_or_else(|| Error::unknown("registry structural inspection failed")));
            }
        } else if matches!(mode, OpenMode::ReadOnly | OpenMode::ReadWrite) {
            return Err(Error::MigrationPostcondition {
                version: 1,
                object: "table:schema_versions".into(),
            });
        }
        Ok(registry)
    }

    /// Open a writable registry, create it when absent, and apply every schema migration.
    pub fn open_migrated(path: impl AsRef<Path>) -> Result<Self> {
        // Migration compatibility is intentionally the one unchecked posture:
        // apply_all owns the legacy empty-ledger backfill and then this handle
        // receives the same structural inspection as every ordinary open.
        let path = path.as_ref().to_path_buf();
        let open_path = safe_open_path(&path)?;
        let registry = Self {
            connection: open_connection(&open_path, OpenMode::ReadWriteCreate)?,
            mode: OpenMode::ReadWriteCreate,
            path,
        };
        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()?;
        let version = crate::migrate::apply_all(&self.connection)?;
        let inspection = crate::migrate::inspect(&self.connection)?;
        if !inspection.ok() {
            return Err(inspection
                .first_error()
                .unwrap_or_else(|| Error::unknown("registry structural inspection failed")));
        }
        Ok(version)
    }

    /// Read-only, no-follow, query-only structural inspection. This bypasses
    /// [`Registry::open`] validation only so doctor can report every finding
    /// instead of receiving the first typed error and stopping.
    pub fn inspect_path(path: impl AsRef<Path>) -> Result<RegistryInspection> {
        let path = path.as_ref().to_path_buf();
        let open_path = safe_open_path(&path)?;
        let connection = open_connection(&open_path, OpenMode::ReadOnly)?;
        crate::migrate::inspect(&connection)
    }

    /// Execute one explicit, snapshot-backed repair selected by the runner's
    /// structural inventory. Every refusal happens before mutation.
    pub fn repair(request: &RegistryRepairRequest) -> Result<RegistryRepairReport> {
        repair_registry(request)
    }

    /// Restore a repaired registry from the immutable snapshot recorded in a
    /// receipt. The current source must still be the exact repaired identity.
    pub fn rollback(request: &RegistryRollbackRequest) -> Result<RegistryRepairReport> {
        rollback_registry(request)
    }

    /// 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 open_connection(path: &Path, mode: OpenMode) -> Result<Connection> {
    let connection = Connection::open_with_flags(path, mode.flags())?;
    connection.busy_timeout(Registry::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(connection)
}

fn schema_versions_exists(conn: &Connection) -> Result<bool> {
    Ok(conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'schema_versions')",
        [],
        |row| row.get(0),
    )?)
}

const RECEIPT_SCHEMA: &str = "shepherd.registry-repair-receipt/1";
const MAX_RECEIPT_BYTES: u64 = 2 * 1024 * 1024;
const SIDECAR_SUFFIXES: &[&str] = &["-wal", "-shm", "-journal"];

fn repair_registry(request: &RegistryRepairRequest) -> Result<RegistryRepairReport> {
    let project_root = canonical_directory(&request.project_root, "project root")?;
    let registry_path = canonical_regular_file(&request.registry_path, "registry")?;
    let expected_registry = project_root.join(".shepherd").join("shepherd.db");
    if registry_path != expected_registry {
        return Err(Error::RepairRefused(format!(
            "registry path {} is not the canonical project registry {}",
            registry_path.display(),
            expected_registry.display()
        )));
    }
    let project_identity = read_and_validate_project_identity(&project_root)?;
    let project_id = project_identity.id().as_str().to_owned();
    let snapshot_parent = match &request.snapshot_dir {
        Some(path) => ensure_snapshot_directory(path)?,
        None => ensure_snapshot_directory(&project_root.join(".shepherd/registry-repairs"))?,
    };
    let mut sidecars = inspect_sidecars(&registry_path)?;
    reject_ambiguous_sidecars(&sidecars)?;
    let source_before = file_identity(&registry_path)?;

    let connection = open_connection(&registry_path, OpenMode::ReadWrite)?;
    connection.execute_batch("BEGIN IMMEDIATE;")?;
    let locked_sidecars = inspect_sidecars(&registry_path)?;
    if locked_sidecars != sidecars {
        sidecars = locked_sidecars;
    }
    if let Err(error) = reject_locked_sidecars(&mut sidecars) {
        let _ = connection.execute_batch("ROLLBACK;");
        return Err(error);
    }
    let inspection = match crate::migrate::inspect(&connection) {
        Ok(inspection) => inspection,
        Err(error) => {
            let _ = connection.execute_batch("ROLLBACK;");
            return Err(error);
        }
    };
    let project_metadata = match verify_project_row(&connection, &project_identity, &project_root) {
        Ok(state) => state,
        Err(error) => {
            let _ = connection.execute_batch("ROLLBACK;");
            return Err(error);
        }
    };
    let version = match crate::migrate::exact_repair_version(&inspection) {
        Ok(version) => version,
        Err(error) => {
            let _ = connection.execute_batch("ROLLBACK;");
            return Err(error);
        }
    };

    let artifact_dir = create_artifact_directory(&snapshot_parent, &source_before.sha256)?;
    let snapshot_path = artifact_dir.join("shepherd.db");
    let snapshot = copy_file_create_new(&registry_path, &snapshot_path, 0o600)?;
    let snapshot_sha256 = snapshot.sha256.clone();
    for sidecar in &mut sidecars {
        if !sidecar.present {
            continue;
        }
        let source = PathBuf::from(&sidecar.source_path);
        let destination = artifact_dir.join(
            source
                .file_name()
                .ok_or_else(|| Error::UnsafePath("sidecar has no file name".into()))?,
        );
        let captured = copy_file_create_new(&source, &destination, sidecar.mode)?;
        sidecar.snapshot_path = Some(destination.display().to_string());
        sidecar.sha256 = Some(captured.sha256);
        sidecar.length = captured.length;
    }
    let source_after_snapshot = file_identity(&registry_path)?;
    if source_after_snapshot != source_before {
        let _ = connection.execute_batch("ROLLBACK;");
        return Err(Error::RepairRefused(
            "registry changed while creating the immutable snapshot".into(),
        ));
    }

    let now = now_seconds();
    let receipt_path = artifact_dir.join("repair-receipt.json");
    let plan_path = artifact_dir.join("repair-plan.json");
    let plan = RegistryRepairPlan {
        schema: "shepherd.registry-repair-plan/1".into(),
        project_id: project_id.clone(),
        primary_root: project_root.display().to_string(),
        source: source_before.clone(),
        sidecars: sidecars.clone(),
        structural_findings: inspection.findings.clone(),
        applied: inspection.applied.clone(),
        repair_plan: vec![version],
        snapshot_path: snapshot_path.display().to_string(),
        snapshot_sha256: snapshot_sha256.clone(),
        snapshot_mode: snapshot.mode,
        receipt_path: receipt_path.display().to_string(),
        created_at: now,
        rollback_command: ROLLBACK_COMMAND.into(),
    };
    let plan_bytes = json_bytes(&plan)?;
    let plan_sha256 = format_digest(Sha256::digest(&plan_bytes));
    write_bytes_create_new(&plan_path, &plan_bytes, 0o600)?;
    let mut receipt = RegistryRepairReceipt {
        schema: RECEIPT_SCHEMA.into(),
        status: "planned".into(),
        project_id: project_id.clone(),
        primary_root: project_root.display().to_string(),
        source: source_before.clone(),
        sidecars: sidecars.clone(),
        structural_findings: inspection.findings.clone(),
        applied: inspection.applied.clone(),
        after_applied: None,
        repair_plan: vec![version],
        advanced_migrations: Vec::new(),
        snapshot_path: snapshot_path.display().to_string(),
        snapshot_sha256: snapshot_sha256.clone(),
        snapshot_mode: snapshot.mode,
        snapshot_receipt_path: plan_path.display().to_string(),
        plan_sha256: plan_sha256.clone(),
        receipt_path: receipt_path.display().to_string(),
        created_at: now,
        after: None,
        rollback_command: ROLLBACK_COMMAND.into(),
        failure: None,
    };
    let mutation = (|| -> Result<(RegistryInspection, Vec<u32>)> {
        if project_metadata == ProjectMetadataState::LegacyMissing {
            let metadata =
                serde_json::json!({"root": project_root.display().to_string()}).to_string();
            let changed = connection.execute(
                "UPDATE projects SET metadata = ?1, updated_at = ?2 WHERE id = ?3 AND metadata IS NULL",
                rusqlite::params![metadata, now, &project_id],
            )?;
            if changed != 1 {
                return Err(Error::RepairRefused(
                    "legacy project metadata changed before repair could bind it".into(),
                ));
            }
        }
        crate::migrate::apply_missing_in_transaction(&connection, version)?;
        let advanced = crate::migrate::apply_pending_in_transaction(&connection)?;
        let after = crate::migrate::inspect(&connection)?;
        if !after.ok() {
            return Err(after.first_error().unwrap_or_else(|| {
                Error::RepairRefused("repair left structural findings".into())
            }));
        }
        Ok((after, advanced))
    })();
    let after_inspection = match mutation {
        Ok((after, advanced)) => {
            connection.execute_batch("COMMIT;")?;
            receipt.advanced_migrations = advanced;
            after
        }
        Err(error) => {
            let _ = connection.execute_batch("ROLLBACK;");
            receipt.status = "failed".into();
            receipt.failure = Some(error.to_string());
            write_json_create_new(&receipt_path, &receipt)?;
            return Err(Error::RepairRefused(format!(
                "repair rolled back; receipt: {} ({error})",
                receipt_path.display()
            )));
        }
    };
    drop(connection);

    let after = match file_identity(&registry_path) {
        Ok(after) => after,
        Err(error) => {
            receipt.status = "failed".into();
            receipt.failure = Some(error.to_string());
            write_json_create_new(&receipt_path, &receipt)?;
            return Err(error);
        }
    };
    receipt.status = "success".into();
    receipt.after = Some(after);
    receipt.after_applied = Some(after_inspection.applied);
    let receipt_sha256 = write_json_create_new_hashed(&receipt_path, &receipt)?;
    let rollback_command =
        rendered_rollback_command(&receipt_path, &receipt.plan_sha256, &receipt_sha256);
    Ok(RegistryRepairReport {
        receipt,
        snapshot_path,
        receipt_path,
        receipt_sha256,
        rollback_command,
    })
}

fn rollback_registry(request: &RegistryRollbackRequest) -> Result<RegistryRepairReport> {
    let receipt_path = canonical_regular_file(&request.receipt_path, "repair receipt")?;
    let receipt_bytes = read_receipt_bytes(&receipt_path)?;
    let receipt_sha256 = format_digest(Sha256::digest(&receipt_bytes));
    if receipt_sha256 != request.receipt_sha256 {
        return Err(Error::InvalidReceipt(
            "rollback receipt witness does not match exact receipt bytes".into(),
        ));
    }
    let receipt: RegistryRepairReceipt = serde_json::from_slice(&receipt_bytes)
        .map_err(|error| Error::InvalidReceipt(error.to_string()))?;
    if receipt.schema != RECEIPT_SCHEMA || receipt.status != "success" {
        return Err(Error::InvalidReceipt(
            "rollback requires a successful repair receipt".into(),
        ));
    }
    let plan_path = canonical_regular_file(
        Path::new(&receipt.snapshot_receipt_path),
        "repair plan witness",
    )?;
    let plan_bytes = fs::read(&plan_path)?;
    let plan_sha256 = format_digest(Sha256::digest(&plan_bytes));
    if plan_sha256 != request.witness_sha256 || plan_sha256 != receipt.plan_sha256 {
        return Err(Error::InvalidReceipt(
            "rollback witness does not authenticate the immutable repair plan".into(),
        ));
    }
    let plan: RegistryRepairPlan = serde_json::from_slice(&plan_bytes)
        .map_err(|error| Error::InvalidReceipt(format!("repair plan is invalid: {error}")))?;
    validate_plan_against_receipt(&plan, &receipt)?;
    let project_root = canonical_directory(&request.project_root, "project root")?;
    if project_root.display().to_string() != receipt.primary_root {
        return Err(Error::InvalidReceipt(
            "receipt project root does not match the requested canonical root".into(),
        ));
    }
    let project_identity = read_and_validate_project_identity(&project_root)?;
    let project_id = project_identity.id().as_str().to_owned();
    if project_id != receipt.project_id {
        return Err(Error::InvalidReceipt(
            "receipt project identity does not match project.json".into(),
        ));
    }
    let registry_path =
        canonical_regular_file(&project_root.join(".shepherd/shepherd.db"), "registry")?;
    if registry_path.display().to_string() != receipt.source.path {
        return Err(Error::InvalidReceipt(
            "receipt source path does not match the canonical project registry".into(),
        ));
    }
    let current = file_identity(&registry_path)?;
    let expected_after = receipt.after.as_ref().ok_or_else(|| {
        Error::InvalidReceipt("successful receipt has no repaired source identity".into())
    })?;
    if &current != expected_after {
        return Err(Error::InvalidReceipt(
            "registry changed after repair; refusing rollback".into(),
        ));
    }
    let snapshot_path = canonical_regular_file(Path::new(&receipt.snapshot_path), "snapshot")?;
    let snapshot = file_identity(&snapshot_path)?;
    if snapshot.sha256 != receipt.snapshot_sha256 || snapshot.mode != receipt.snapshot_mode {
        return Err(Error::InvalidReceipt(
            "snapshot hash or mode does not match the receipt".into(),
        ));
    }
    preflight_sidecar_snapshots(&receipt.sidecars)?;
    let sidecars = inspect_sidecars(&registry_path)?;
    reject_rollback_sidecars(&sidecars, &receipt.sidecars)?;

    let artifact_dir = receipt_path
        .parent()
        .ok_or_else(|| Error::InvalidReceipt("receipt has no parent directory".into()))?;
    // Reserve every fallible evidence destination before the first live rename.
    // Receipt exhaustion must therefore leave the repaired source untouched.
    let rollback_path = unique_artifact_path(artifact_dir, "rollback-receipt", "json")?;
    let guard_dir = create_artifact_directory(artifact_dir, &current.sha256)?;
    let guard_main = guard_dir.join("repaired.db");
    copy_file_create_new(&registry_path, &guard_main, current.mode)?;
    let mut guard_sidecars = Vec::new();
    for sidecar in &sidecars {
        let target = PathBuf::from(format!("{}{}", registry_path.display(), sidecar.suffix));
        let guard_path = if sidecar.present {
            let path = guard_dir.join(format!("repaired{}", sidecar.suffix));
            copy_file_create_new(&target, &path, sidecar.mode)?;
            Some(path)
        } else {
            None
        };
        guard_sidecars.push((sidecar.suffix.clone(), guard_path));
    }

    let restore_main = guard_dir.join("restore.db");
    copy_file_create_new(&snapshot_path, &restore_main, receipt.source.mode)?;
    let mut restore_sidecars = Vec::new();
    for sidecar in &receipt.sidecars {
        let staged = if let Some(snapshot_path) = &sidecar.snapshot_path {
            let snapshot = canonical_regular_file(Path::new(snapshot_path), "sidecar snapshot")?;
            let path = guard_dir.join(format!("restore{}", sidecar.suffix));
            copy_file_create_new(&snapshot, &path, sidecar.mode)?;
            Some(path)
        } else {
            None
        };
        restore_sidecars.push((sidecar.suffix.clone(), staged));
    }

    let replacement = replace_live_from_staged(&registry_path, &restore_main, &restore_sidecars);
    let displaced = match replacement {
        Ok(displaced) => displaced,
        Err(error) => {
            return Err(rollback_failure(
                error,
                &registry_path,
                &guard_main,
                &guard_sidecars,
                &guard_dir,
                &[],
            ));
        }
    };
    let post_swap: Result<RegistryRepairReport> = (|| {
        let restored_sidecars = inspect_sidecars(&registry_path)?;
        let restored = file_identity(&registry_path)?;
        if restored.sha256 != receipt.snapshot_sha256 {
            return Err(Error::RepairRefused(
                "restored registry hash does not match the immutable snapshot".into(),
            ));
        }
        let restored_inspection = Registry::inspect_path(&registry_path)?;
        if restored_inspection.findings != receipt.structural_findings {
            return Err(Error::RepairRefused(
                "restored registry structural findings differ from the receipt".into(),
            ));
        }
        let rollback_receipt = RegistryRepairReceipt {
            schema: RECEIPT_SCHEMA.into(),
            status: "rolled_back".into(),
            project_id,
            primary_root: project_root.display().to_string(),
            source: current,
            sidecars: restored_sidecars,
            structural_findings: restored_inspection.findings,
            applied: receipt.applied,
            after_applied: Some(restored_inspection.applied),
            repair_plan: receipt.repair_plan,
            advanced_migrations: receipt.advanced_migrations,
            snapshot_path: receipt.snapshot_path,
            snapshot_sha256: receipt.snapshot_sha256,
            snapshot_mode: receipt.snapshot_mode,
            snapshot_receipt_path: receipt.snapshot_receipt_path,
            plan_sha256: receipt.plan_sha256,
            receipt_path: rollback_path.display().to_string(),
            created_at: now_seconds(),
            after: Some(restored),
            rollback_command: receipt.rollback_command,
            failure: None,
        };
        let receipt_sha256 = write_json_create_new_hashed(&rollback_path, &rollback_receipt)?;
        let rollback_command = rendered_rollback_command(
            &rollback_path,
            &rollback_receipt.plan_sha256,
            &receipt_sha256,
        );
        Ok(RegistryRepairReport {
            receipt: rollback_receipt,
            snapshot_path,
            receipt_path: rollback_path,
            receipt_sha256,
            rollback_command,
        })
    })();
    match post_swap {
        Ok(report) => {
            if let Err(error) = cleanup_rollback_artifacts(&guard_dir, &displaced) {
                return Err(Error::RepairRefused(format!(
                    "rollback receipt is durable but temporary guard cleanup failed: {error}; guard: {}",
                    guard_dir.display()
                )));
            }
            Ok(report)
        }
        Err(error) => Err(rollback_failure(
            error,
            &registry_path,
            &guard_main,
            &guard_sidecars,
            &guard_dir,
            &displaced,
        )),
    }
}

fn cleanup_rollback_artifacts(guard_dir: &Path, displaced: &[PathBuf]) -> Result<()> {
    cleanup_displaced_paths(displaced)?;
    fs::remove_dir_all(guard_dir)?;
    sync_parent_directory(guard_dir)
}

fn cleanup_displaced_paths(displaced: &[PathBuf]) -> Result<()> {
    for path in displaced {
        match fs::symlink_metadata(path) {
            Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
                return Err(Error::UnsafePath(format!(
                    "displaced cleanup target is not a regular file: {}",
                    path.display()
                )));
            }
            Ok(_) => fs::remove_file(path)?,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => {
                return Err(Error::UnsafePath(format!(
                    "inspect displaced cleanup target {}: {error}",
                    path.display()
                )));
            }
        }
    }
    Ok(())
}

fn rendered_rollback_command(
    receipt_path: &Path,
    plan_sha256: &str,
    receipt_sha256: &str,
) -> String {
    format!(
        "shepherd registry rollback --receipt {} --witness-sha256 {} --receipt-sha256 {} --confirm",
        receipt_path.display(),
        plan_sha256,
        receipt_sha256
    )
}

fn validate_plan_against_receipt(
    plan: &RegistryRepairPlan,
    receipt: &RegistryRepairReceipt,
) -> Result<()> {
    if plan.schema != "shepherd.registry-repair-plan/1"
        || plan.project_id != receipt.project_id
        || plan.primary_root != receipt.primary_root
        || plan.source != receipt.source
        || plan.sidecars != receipt.sidecars
        || plan.structural_findings != receipt.structural_findings
        || plan.applied != receipt.applied
        || plan.repair_plan != receipt.repair_plan
        || plan.snapshot_path != receipt.snapshot_path
        || plan.snapshot_sha256 != receipt.snapshot_sha256
        || plan.snapshot_mode != receipt.snapshot_mode
        || plan.receipt_path != receipt.receipt_path
        || plan.created_at != receipt.created_at
        || plan.rollback_command != receipt.rollback_command
    {
        return Err(Error::InvalidReceipt(
            "successful receipt fields disagree with its immutable repair plan".into(),
        ));
    }
    Ok(())
}

fn absolute_path(path: &Path) -> Result<PathBuf> {
    if path.is_absolute() {
        Ok(path.to_path_buf())
    } else {
        Ok(std::env::current_dir()
            .map_err(|error| {
                Error::UnsafePath(format!("resolve path {}: {error}", path.display()))
            })?
            .join(path))
    }
}

fn canonical_directory(path: &Path, label: &str) -> Result<PathBuf> {
    let absolute = absolute_path(path)?;
    reject_symlink_ancestors(&absolute)?;
    let metadata = fs::symlink_metadata(&absolute).map_err(|error| {
        Error::UnsafePath(format!("inspect {label} {}: {error}", absolute.display()))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_dir() {
        return Err(Error::UnsafePath(format!(
            "{label} is not a canonical regular directory: {}",
            absolute.display()
        )));
    }
    let canonical = fs::canonicalize(&absolute).map_err(|error| {
        Error::UnsafePath(format!(
            "canonicalize {label} {}: {error}",
            absolute.display()
        ))
    })?;
    if canonical_path_string(&canonical) != canonical_path_string(&absolute) {
        return Err(Error::UnsafePath(format!(
            "{label} is not already canonical: {}",
            absolute.display()
        )));
    }
    Ok(canonical)
}

fn canonical_regular_file(path: &Path, label: &str) -> Result<PathBuf> {
    let absolute = absolute_path(path)?;
    let parent = absolute
        .parent()
        .ok_or_else(|| Error::UnsafePath(format!("{label} has no parent")))?;
    reject_symlink_ancestors(parent)?;
    let metadata = fs::symlink_metadata(&absolute).map_err(|error| {
        Error::UnsafePath(format!("inspect {label} {}: {error}", absolute.display()))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_file() {
        return Err(Error::UnsafePath(format!(
            "{label} is not a regular no-follow file: {}",
            absolute.display()
        )));
    }
    reject_hard_link(&metadata, label, &absolute)?;
    let canonical_parent = fs::canonicalize(parent).map_err(|error| {
        Error::UnsafePath(format!(
            "canonicalize {label} parent {}: {error}",
            parent.display()
        ))
    })?;
    let canonical = canonical_parent.join(
        absolute
            .file_name()
            .ok_or_else(|| Error::UnsafePath(format!("{label} has no file name")))?,
    );
    if canonical_path_string(&canonical) != canonical_path_string(&absolute) {
        return Err(Error::UnsafePath(format!(
            "{label} is not already canonical: {}",
            absolute.display()
        )));
    }
    Ok(canonical)
}

fn ensure_snapshot_directory(path: &Path) -> Result<PathBuf> {
    let absolute = absolute_path(path)?;
    if absolute.exists() {
        return canonical_directory(&absolute, "snapshot directory");
    }
    let mut missing = Vec::new();
    let mut cursor = absolute.clone();
    while !cursor.exists() {
        let name = cursor
            .file_name()
            .ok_or_else(|| Error::UnsafePath("snapshot directory has no name".into()))?;
        missing.push(name.to_owned());
        cursor = cursor
            .parent()
            .ok_or_else(|| Error::UnsafePath("snapshot directory has no existing ancestor".into()))?
            .to_path_buf();
    }
    let existing = canonical_directory(&cursor, "snapshot directory ancestor")?;
    let mut created = existing;
    for name in missing.iter().rev() {
        created.push(name);
        #[cfg(unix)]
        let mut builder = fs::DirBuilder::new();
        #[cfg(not(unix))]
        let builder = fs::DirBuilder::new();
        #[cfg(unix)]
        {
            use std::os::unix::fs::DirBuilderExt;
            builder.mode(0o700);
        }
        builder.create(&created).map_err(|error| {
            Error::UnsafePath(format!(
                "create snapshot directory {}: {error}",
                created.display()
            ))
        })?;
    }
    canonical_directory(&absolute, "snapshot directory")
}

fn create_artifact_directory(parent: &Path, source_sha256: &str) -> Result<PathBuf> {
    let stamp = now_seconds();
    let prefix = source_sha256.get(..16).unwrap_or(source_sha256);
    for suffix in 0..1000_u32 {
        let name = if suffix == 0 {
            format!("repair-{stamp}-{prefix}")
        } else {
            format!("repair-{stamp}-{prefix}-{suffix}")
        };
        let candidate = parent.join(name);
        #[cfg(unix)]
        let mut builder = fs::DirBuilder::new();
        #[cfg(not(unix))]
        let builder = fs::DirBuilder::new();
        #[cfg(unix)]
        {
            use std::os::unix::fs::DirBuilderExt;
            builder.mode(0o700);
        }
        match builder.create(&candidate) {
            Ok(()) => return canonical_directory(&candidate, "repair artifact directory"),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(error) => {
                return Err(Error::UnsafePath(format!(
                    "create repair artifact directory {}: {error}",
                    candidate.display()
                )));
            }
        }
    }
    Err(Error::UnsafePath(
        "could not allocate a unique repair artifact directory".into(),
    ))
}

fn unique_artifact_path(parent: &Path, stem: &str, extension: &str) -> Result<PathBuf> {
    for suffix in 0..1000_u32 {
        let name = if suffix == 0 {
            format!("{stem}.{extension}")
        } else {
            format!("{stem}-{suffix}.{extension}")
        };
        let candidate = parent.join(name);
        if !candidate.exists() {
            return Ok(candidate);
        }
        let metadata = fs::symlink_metadata(&candidate).map_err(|error| {
            Error::UnsafePath(format!(
                "inspect artifact candidate {}: {error}",
                candidate.display()
            ))
        })?;
        if metadata.file_type().is_symlink() || !metadata.is_file() {
            return Err(Error::UnsafePath(format!(
                "artifact candidate is unsafe: {}",
                candidate.display()
            )));
        }
    }
    Err(Error::UnsafePath(
        "could not allocate a unique repair receipt path".into(),
    ))
}

fn unique_sibling(path: &Path, stem: &str) -> Result<PathBuf> {
    let parent = path
        .parent()
        .ok_or_else(|| Error::UnsafePath("registry has no parent".into()))?;
    let base = path
        .file_name()
        .ok_or_else(|| Error::UnsafePath("registry has no file name".into()))?
        .to_string_lossy();
    for suffix in 0..1000_u32 {
        let name = if suffix == 0 {
            format!(".{base}.{stem}.tmp")
        } else {
            format!(".{base}.{stem}-{suffix}.tmp")
        };
        let candidate = parent.join(name);
        if !candidate.exists() {
            return Ok(candidate);
        }
    }
    Err(Error::UnsafePath(
        "could not allocate an atomic restore temporary".into(),
    ))
}

fn reject_hard_link(metadata: &fs::Metadata, label: &str, path: &Path) -> Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt;
        if metadata.nlink() > 1 {
            return Err(Error::UnsafePath(format!(
                "{label} has hard-link ambiguity: {}",
                path.display()
            )));
        }
    }
    let _ = metadata;
    let _ = label;
    let _ = path;
    Ok(())
}

fn file_identity(path: &Path) -> Result<RegistryFileIdentity> {
    let canonical = canonical_regular_file(path, "registry file")?;
    let metadata = fs::symlink_metadata(&canonical).map_err(|error| {
        Error::UnsafePath(format!(
            "inspect registry file {}: {error}",
            canonical.display()
        ))
    })?;
    let sha256 = hash_file(&canonical)?;
    Ok(RegistryFileIdentity {
        path: canonical.display().to_string(),
        sha256,
        length: metadata.len(),
        mode: file_mode(&metadata),
        #[cfg(unix)]
        device: {
            use std::os::unix::fs::MetadataExt;
            metadata.dev()
        },
        #[cfg(unix)]
        inode: {
            use std::os::unix::fs::MetadataExt;
            metadata.ino()
        },
        #[cfg(unix)]
        links: {
            use std::os::unix::fs::MetadataExt;
            metadata.nlink()
        },
    })
}

fn file_mode(metadata: &fs::Metadata) -> u32 {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        metadata.permissions().mode() & 0o777
    }
    #[cfg(not(unix))]
    {
        let _ = metadata;
        0o600
    }
}

fn hash_file(path: &Path) -> Result<String> {
    let mut file = File::open(path)?;
    let mut digest = Sha256::new();
    let mut buffer = [0_u8; 64 * 1024];
    loop {
        let read = file.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        digest.update(&buffer[..read]);
    }
    Ok(format_digest(digest.finalize()))
}

fn copy_file_create_new(
    source: &Path,
    destination: &Path,
    mode: u32,
) -> Result<RegistryFileIdentity> {
    #[cfg(not(unix))]
    let _ = mode;
    let source = canonical_regular_file(source, "snapshot source")?;
    let source_metadata = fs::symlink_metadata(&source)?;
    reject_hard_link(&source_metadata, "snapshot source", &source)?;
    let mut input = File::open(&source)?;
    let mut options = OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(mode);
    }
    let mut output = options.open(destination)?;
    let mut digest = Sha256::new();
    let mut length = 0_u64;
    let mut buffer = [0_u8; 64 * 1024];
    loop {
        let read = input.read(&mut buffer)?;
        if read == 0 {
            break;
        }
        output.write_all(&buffer[..read])?;
        digest.update(&buffer[..read]);
        length = length.saturating_add(u64::try_from(read).unwrap_or(u64::MAX));
    }
    output.sync_all()?;
    sync_parent_directory(destination)?;
    let _ = (digest, length);
    file_identity(destination)
}

fn write_json_create_new<T: serde::Serialize>(path: &Path, value: &T) -> Result<()> {
    let bytes = json_bytes(value)?;
    write_bytes_create_new(path, &bytes, 0o600)
}

fn write_json_create_new_hashed<T: serde::Serialize>(path: &Path, value: &T) -> Result<String> {
    let bytes = json_bytes(value)?;
    write_bytes_create_new(path, &bytes, 0o600)?;
    Ok(format_digest(Sha256::digest(&bytes)))
}

fn json_bytes<T: serde::Serialize>(value: &T) -> Result<Vec<u8>> {
    let mut bytes =
        serde_json::to_vec_pretty(value).map_err(|error| Error::unknown(error.to_string()))?;
    bytes.push(b'\n');
    Ok(bytes)
}

fn write_bytes_create_new(path: &Path, bytes: &[u8], mode: u32) -> Result<()> {
    #[cfg(not(unix))]
    let _ = mode;
    let mut options = OpenOptions::new();
    options.write(true).create_new(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.mode(mode);
    }
    let mut file = options.open(path)?;
    file.write_all(bytes)?;
    file.sync_all()?;
    sync_parent_directory(path)?;
    Ok(())
}

fn sync_parent_directory(path: &Path) -> Result<()> {
    #[cfg(unix)]
    {
        let parent = path
            .parent()
            .ok_or_else(|| Error::UnsafePath("artifact has no parent directory".into()))?;
        File::open(parent)?.sync_all()?;
    }
    #[cfg(not(unix))]
    let _ = path;
    Ok(())
}

fn read_receipt_bytes(path: &Path) -> Result<Vec<u8>> {
    let metadata = fs::symlink_metadata(path)?;
    if metadata.len() > MAX_RECEIPT_BYTES
        || metadata.file_type().is_symlink()
        || !metadata.is_file()
    {
        return Err(Error::InvalidReceipt(
            "receipt is not a bounded regular file".into(),
        ));
    }
    Ok(fs::read(path)?)
}

fn inspect_sidecars(registry_path: &Path) -> Result<Vec<RegistrySidecar>> {
    let mut result = Vec::new();
    for suffix in SIDECAR_SUFFIXES {
        let path = PathBuf::from(format!("{}{}", registry_path.display(), suffix));
        match fs::symlink_metadata(&path) {
            Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
                return Err(Error::UnsafePath(format!(
                    "registry sidecar is not a regular no-follow file: {}",
                    path.display()
                )));
            }
            Ok(metadata) => {
                reject_hard_link(&metadata, "registry sidecar", &path)?;
                let sha256 = hash_file(&path)?;
                result.push(RegistrySidecar {
                    suffix: (*suffix).into(),
                    source_path: path.display().to_string(),
                    present: true,
                    length: metadata.len(),
                    mode: file_mode(&metadata),
                    sha256: Some(sha256),
                    snapshot_path: None,
                    disposition: if metadata.len() == 0 {
                        "empty-sidecar-captured".into()
                    } else {
                        "nonempty-sidecar-rejected".into()
                    },
                });
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                result.push(RegistrySidecar {
                    suffix: (*suffix).into(),
                    source_path: path.display().to_string(),
                    present: false,
                    length: 0,
                    mode: 0,
                    sha256: None,
                    snapshot_path: None,
                    disposition: "absent".into(),
                })
            }
            Err(error) => {
                return Err(Error::UnsafePath(format!(
                    "inspect registry sidecar {}: {error}",
                    path.display()
                )));
            }
        }
    }
    Ok(result)
}

fn reject_ambiguous_sidecars(sidecars: &[RegistrySidecar]) -> Result<()> {
    if let Some(sidecar) = sidecars
        .iter()
        .find(|sidecar| sidecar.present && sidecar.length > 0)
    {
        return Err(Error::RepairRefused(format!(
            "{} sidecar is nonempty or live; checkpoint it before repair",
            sidecar.source_path
        )));
    }
    Ok(())
}

fn reject_locked_sidecars(sidecars: &mut [RegistrySidecar]) -> Result<()> {
    if let Some(sidecar) = sidecars
        .iter_mut()
        .find(|sidecar| sidecar.present && sidecar.length > 0 && sidecar.suffix == "-wal")
    {
        return Err(Error::RepairRefused(format!(
            "{} WAL sidecar remained nonempty after writer lock; refusing ambiguous snapshot",
            sidecar.source_path
        )));
    }
    for sidecar in sidecars.iter_mut() {
        if sidecar.present && sidecar.length > 0 && sidecar.suffix == "-shm" {
            sidecar.disposition = "created-or-retained-under-repair-lock".into();
        }
    }
    Ok(())
}

fn reject_rollback_sidecars(
    sidecars: &[RegistrySidecar],
    recorded: &[RegistrySidecar],
) -> Result<()> {
    for sidecar in sidecars
        .iter()
        .filter(|sidecar| sidecar.present && sidecar.length > 0)
    {
        let recorded_sidecar = recorded
            .iter()
            .find(|candidate| candidate.suffix == sidecar.suffix);
        let owned_shm = sidecar.suffix == "-shm"
            && recorded_sidecar.is_some_and(|candidate| {
                candidate.disposition == "created-or-retained-under-repair-lock"
                    && candidate.sha256 == sidecar.sha256
            });
        if !owned_shm {
            return Err(Error::RepairRefused(format!(
                "{} sidecar is nonempty or changed after repair; refusing rollback",
                sidecar.source_path
            )));
        }
    }
    Ok(())
}

fn preflight_sidecar_snapshots(recorded: &[RegistrySidecar]) -> Result<()> {
    for sidecar in recorded {
        let Some(snapshot_path) = &sidecar.snapshot_path else {
            continue;
        };
        let snapshot = file_identity(Path::new(snapshot_path))?;
        if sidecar.sha256.as_deref() != Some(snapshot.sha256.as_str())
            || snapshot.length != sidecar.length
            || snapshot.mode != sidecar.mode
        {
            return Err(Error::InvalidReceipt(format!(
                "sidecar snapshot {} does not match the authenticated receipt",
                snapshot_path
            )));
        }
    }
    Ok(())
}

fn replace_live_from_staged(
    registry_path: &Path,
    staged_main: &Path,
    staged_sidecars: &[(String, Option<PathBuf>)],
) -> Result<Vec<PathBuf>> {
    let mut displaced = Vec::new();
    if let Some(path) = install_staged_over_live(staged_main, registry_path)? {
        displaced.push(path);
    }
    for (suffix, staged) in staged_sidecars {
        let target = PathBuf::from(format!("{}{}", registry_path.display(), suffix));
        match staged {
            Some(staged) => {
                if let Some(path) = install_staged_over_live(staged, &target)? {
                    displaced.push(path);
                }
            }
            None => match fs::symlink_metadata(&target) {
                Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
                    return Err(Error::UnsafePath(format!(
                        "sidecar restore target is not a regular file: {}",
                        target.display()
                    )));
                }
                Ok(_) => {
                    let displaced_path = unique_sibling(&target, "displaced")?;
                    fs::rename(&target, &displaced_path)?;
                    displaced.push(displaced_path);
                    sync_parent_directory(&target)?;
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => {
                    return Err(Error::UnsafePath(format!(
                        "inspect sidecar restore target {}: {error}",
                        target.display()
                    )));
                }
            },
        }
    }
    Ok(displaced)
}

fn install_staged_over_live(staged: &Path, target: &Path) -> Result<Option<PathBuf>> {
    let displaced = match fs::symlink_metadata(target) {
        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
            return Err(Error::UnsafePath(format!(
                "live replacement target is not a regular file: {}",
                target.display()
            )));
        }
        Ok(_) => {
            let displaced = unique_sibling(target, "displaced")?;
            fs::rename(target, &displaced)?;
            Some(displaced)
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
        Err(error) => {
            return Err(Error::UnsafePath(format!(
                "inspect live replacement target {}: {error}",
                target.display()
            )));
        }
    };
    if let Err(error) = fs::rename(staged, target) {
        if let Some(displaced) = displaced
            && let Err(restore) = fs::rename(&displaced, target)
        {
            return Err(Error::RollbackCompensation {
                cause: error.to_string(),
                compensation: restore.to_string(),
            });
        }
        return Err(Error::RepairRefused(format!(
            "install staged replacement {}: {error}",
            target.display()
        )));
    }
    sync_parent_directory(target)?;
    Ok(displaced)
}

fn rollback_failure(
    cause: Error,
    registry_path: &Path,
    guard_main: &Path,
    guard_sidecars: &[(String, Option<PathBuf>)],
    guard_dir: &Path,
    displaced: &[PathBuf],
) -> Error {
    match compensate_live_from_guard(registry_path, guard_main, guard_sidecars) {
        Ok(()) => match cleanup_displaced_paths(displaced)
            .and_then(|()| cleanup_rollback_artifacts(guard_dir, &[]))
        {
            Ok(()) => Error::RepairRefused(cause.to_string()),
            Err(cleanup) => Error::RollbackCompensation {
                cause: cause.to_string(),
                compensation: format!(
                    "cleanup failed: {cleanup}; guard: {}; displaced: {:?}",
                    guard_dir.display(),
                    displaced
                ),
            },
        },
        Err(compensation) => Error::RollbackCompensation {
            cause: cause.to_string(),
            compensation: format!(
                "{}; guard: {}; displaced: {:?}",
                compensation,
                guard_dir.display(),
                displaced
            ),
        },
    }
}

fn compensate_live_from_guard(
    registry_path: &Path,
    guard_main: &Path,
    guard_sidecars: &[(String, Option<PathBuf>)],
) -> Result<()> {
    let mut displaced = Vec::new();
    let mut staged = Vec::new();
    let result = (|| -> Result<()> {
        let main_restore = unique_sibling(registry_path, "compensate")?;
        copy_file_create_new(guard_main, &main_restore, file_identity(guard_main)?.mode).map_err(
            |error| {
                Error::RepairRefused(format!(
                    "stage compensation main {}: {error}",
                    main_restore.display()
                ))
            },
        )?;
        staged.push(main_restore.clone());
        if let Some(path) = install_staged_over_live(&main_restore, registry_path)? {
            displaced.push(path);
        }
        for (suffix, guard) in guard_sidecars {
            let target = PathBuf::from(format!("{}{}", registry_path.display(), suffix));
            match guard {
                Some(guard) => {
                    let stage = unique_sibling(&target, "compensate")?;
                    copy_file_create_new(guard, &stage, file_identity(guard)?.mode).map_err(
                        |error| {
                            Error::RepairRefused(format!(
                                "stage compensation sidecar {}: {error}",
                                stage.display()
                            ))
                        },
                    )?;
                    staged.push(stage.clone());
                    if let Some(path) = install_staged_over_live(&stage, &target)? {
                        displaced.push(path);
                    }
                }
                None => match fs::symlink_metadata(&target) {
                    Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
                        return Err(Error::UnsafePath(format!(
                            "compensation target is not a regular file: {}",
                            target.display()
                        )));
                    }
                    Ok(_) => {
                        let displaced_path = unique_sibling(&target, "compensate-displaced")?;
                        fs::rename(&target, &displaced_path)?;
                        sync_parent_directory(&target)?;
                        displaced.push(displaced_path);
                    }
                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                    Err(error) => {
                        return Err(Error::UnsafePath(format!(
                            "inspect compensation target: {error}"
                        )));
                    }
                },
            }
        }
        cleanup_displaced_paths(&displaced)?;
        cleanup_staged_paths(&staged)?;
        Ok(())
    })();
    match result {
        Ok(()) => Ok(()),
        Err(error) => Err(Error::RepairRefused(format!(
            "{error}; compensation custody retained; staged: {:?}; displaced: {:?}",
            staged, displaced
        ))),
    }
}

fn cleanup_staged_paths(staged: &[PathBuf]) -> Result<()> {
    for path in staged {
        match fs::symlink_metadata(path) {
            Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
                return Err(Error::UnsafePath(format!(
                    "staged cleanup target is not a regular file: {}",
                    path.display()
                )));
            }
            Ok(_) => fs::remove_file(path)?,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => {
                return Err(Error::UnsafePath(format!(
                    "inspect staged cleanup target {}: {error}",
                    path.display()
                )));
            }
        }
    }
    Ok(())
}

fn read_and_validate_project_identity(project_root: &Path) -> Result<ProjectIdentityDocument> {
    let identity = project_root.join(".shepherd/project.json");
    let metadata = fs::symlink_metadata(&identity).map_err(|error| {
        Error::RepairRefused(format!(
            "inspect project identity {}: {error}",
            identity.display()
        ))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_file() {
        return Err(Error::RepairRefused(
            "project identity is not a regular no-follow file".into(),
        ));
    }
    reject_hard_link(&metadata, "project identity", &identity)?;
    let bytes = fs::read(&identity)?;
    if bytes.len() > 65_536 {
        return Err(Error::RepairRefused(
            "project identity exceeds the bounded reader".into(),
        ));
    }
    let document = ProjectIdentityDocument::parse(&bytes)
        .map_err(|error| Error::RepairRefused(format!("project identity is malformed: {error}")))?;
    let root = document.root().ok_or_else(|| {
        Error::RepairRefused("project identity root is absent or malformed".into())
    })?;
    if canonical_path_string(Path::new(root)) != canonical_path_string(project_root) {
        return Err(Error::RepairRefused(
            "project identity root does not match the canonical primary root".into(),
        ));
    }
    Ok(document)
}

#[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")]
enum ProjectMetadataState {
    Bound,
    LegacyMissing,
}

fn verify_project_row(
    conn: &Connection,
    identity: &ProjectIdentityDocument,
    project_root: &Path,
) -> Result<ProjectMetadataState> {
    let rows = conn
        .prepare("SELECT id, metadata, created_at FROM projects ORDER BY id")?
        .query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, Option<String>>(1)?,
                row.get::<_, i64>(2)?,
            ))
        })?
        .collect::<rusqlite::Result<Vec<_>>>()?;
    if rows.len() != 1 || rows[0].0 != identity.id().as_str() {
        return Err(Error::RepairRefused(format!(
            "project identity must bind exactly one projects row and no copied namespace, found {} row(s)",
            rows.len()
        )));
    }
    if rows[0].2 != identity.scaffolded_at() {
        return Err(Error::RepairRefused(
            "project identity scaffolded_at disagrees with the matching projects row".into(),
        ));
    }
    let (_, metadata, _) = &rows[0];
    let Some(metadata) = metadata.as_deref() else {
        return Ok(ProjectMetadataState::LegacyMissing);
    };
    let value: serde_json::Value = serde_json::from_str(metadata).map_err(|error| {
        Error::RepairRefused(format!("matching projects metadata is malformed: {error}"))
    })?;
    let root = value
        .get("root")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| Error::RepairRefused("matching projects metadata has no root".into()))?;
    if canonical_path_string(Path::new(root)) != canonical_path_string(project_root) {
        return Err(Error::RepairRefused(
            "matching projects row disagrees with the canonical primary root".into(),
        ));
    }
    Ok(ProjectMetadataState::Bound)
}

fn now_seconds() -> i64 {
    i64::try_from(
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs(),
    )
    .unwrap_or(i64::MAX)
}

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 != sha256_hex(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 != sha256_hex(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 = sha256_hex(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,
            &sha256_hex(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 != sha256_hex(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);
}