holochain_state 0.8.0-dev.6

//! Read operations on the per-DNA DHT store.
//!
//! Methods on [`DhtStoreRead`] expose domain-meaningful reads that delegate to
//! `holochain_data`'s `DbRead<Dht>` primitives and return values in terms of
//! the project's existing domain types. The parent module holds the
//! corresponding write operations.

use super::DhtStore;
use crate::prelude::ActionSequenceAndHash;
use crate::query::{StateQueryError, StateQueryResult};
use crate::scratch::SyncScratch;
use holo_hash::{
    ActionHash, AgentPubKey, AnyLinkableHash, DhtOpHash, EntryHash, ExternalHash, HasHash,
};
use holochain_data::kind::Dht;
use holochain_data::DbRead;
use holochain_state_types::{SourceChainCursor, SourceChainDump, SourceChainDumpRecord};
use holochain_types::op::DhtOpHashed;
use holochain_types::prelude::{
    ActionHashedContainer, AgentActivity, AgentActivityResponse, ChainItems, ChainItemsSource,
    MustGetAgentActivityResponse, ScheduledFn, Timestamp,
};
use holochain_types::warrant::WarrantOp;
use holochain_zome_types::prelude::{
    Action, ActionData, CapAccess, CapSecret, ChainFilter, ChainFork, ChainHead, ChainQueryFilter,
    ChainStatus, EntryType, EntryVisibility, GrantConstraintType, HighestObserved, LimitConditions,
    LinkTag, LinkTypeFilter, Record, RecordEntry, RecordValidity, SignedActionHashed,
    SignedWarrant, ValidationReceiptSet,
};
use holochain_zome_types::validate::ValidationStatus;
use std::collections::{HashMap, HashSet};
use std::time::Duration;

impl DhtStore<DbRead<Dht>> {
    /// Returns `true` if `hash` appears in any op-bearing DHT table
    /// (`ChainOp`, `LimboChainOp`, `WarrantOp`, `LimboWarrantOp`).
    pub async fn op_exists(&self, hash: &DhtOpHash) -> StateQueryResult<bool> {
        Ok(self.db().op_exists(hash).await?)
    }

    /// Count integrated ops (chain ops plus warrants) held in the DHT store.
    pub async fn count_integrated_ops(&self) -> StateQueryResult<u64> {
        Ok(self.db().count_integrated_ops().await?)
    }

    /// Total bytes occupied on disk by this DNA's DhtStore, including the
    /// unused (free) bytes within each page.
    pub async fn size_on_disk(&self) -> StateQueryResult<u64> {
        Ok(self.db().get_size_on_disk().await?)
    }

    /// Bytes actually in use by this DNA's DhtStore, excluding the free
    /// space within pages.
    pub async fn used_size(&self) -> StateQueryResult<u64> {
        Ok(self.db().get_used_size().await?)
    }

    /// Returns `true` if a scheduled-function row exists for `author` and
    /// `scheduled_fn`, regardless of liveness — i.e. whether the current time
    /// falls within the row's `[start_at, end_at]` window.
    ///
    /// This is a membership check on the scheduled-functions table: it stays
    /// `true` for a persisted cron function between firings, and becomes
    /// `false` once an ephemeral function is consumed.
    pub async fn is_function_scheduled(
        &self,
        author: &AgentPubKey,
        scheduled_fn: &ScheduledFn,
    ) -> StateQueryResult<bool> {
        Ok(self
            .db()
            .is_function_scheduled(
                author,
                scheduled_fn.zome_name().0.as_ref(),
                scheduled_fn.fn_name().0.as_str(),
            )
            .await?)
    }

    /// Count integrated, locally-validated chain ops that passed validation
    /// (rejected and GET-cached ops excluded).
    #[cfg(any(test, feature = "inspection"))]
    pub async fn count_valid_integrated_ops(&self) -> StateQueryResult<u64> {
        Ok(self.db().count_valid_integrated_ops().await?)
    }

    /// Count chain ops that passed both sys- and app-validation but are not yet
    /// integrated.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn count_valid_not_integrated_ops(&self) -> StateQueryResult<u64> {
        Ok(self.db().count_valid_not_integrated_ops().await?)
    }

    /// Count chain ops authored by `author` that are not yet integrated.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn count_pending_ops_for_author(
        &self,
        author: &AgentPubKey,
    ) -> StateQueryResult<u64> {
        Ok(self.db().count_pending_ops_for_author(author).await?)
    }

    /// Hashes of integrated chain ops that were rejected (GET-cached copies
    /// excluded).
    #[cfg(any(test, feature = "inspection"))]
    pub async fn rejected_integrated_op_hashes(&self) -> StateQueryResult<Vec<DhtOpHash>> {
        Ok(self.db().rejected_integrated_op_hashes().await?)
    }

    /// `(op_hash, basis, storage_center_loc)` for every integrated chain op.
    /// Used to verify DHT-location consistency across conductors.
    pub async fn integrated_op_locations(
        &self,
    ) -> StateQueryResult<Vec<(DhtOpHash, holo_hash::AnyLinkableHash, u32)>> {
        Ok(self
            .db()
            .integrated_op_locations()
            .await?
            .into_iter()
            .map(|r| {
                let basis: AnyLinkableHash = ExternalHash::from_raw_36(r.basis_hash).into();
                (
                    DhtOpHash::from_raw_36(r.hash),
                    basis,
                    r.storage_center_loc as u32,
                )
            })
            .collect())
    }

    /// Op hashes of the integrated chain ops for `action_hash`.
    pub async fn chain_op_hashes_for_action(
        &self,
        action_hash: holo_hash::ActionHash,
    ) -> StateQueryResult<Vec<DhtOpHash>> {
        Ok(self
            .db()
            .get_chain_ops_for_action(action_hash)
            .await?
            .into_iter()
            .map(|r| DhtOpHash::from_raw_36(r.hash))
            .collect())
    }

    /// Total count of every op (integrated and limbo) held in the DHT store.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn count_all_ops(&self) -> StateQueryResult<u64> {
        Ok(self.db().count_all_ops().await?)
    }

    /// Whether the integrated chain op `op_hash` requires a validation receipt.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn op_requires_receipt(&self, op_hash: &DhtOpHash) -> StateQueryResult<bool> {
        Ok(self.db().op_requires_receipt(op_hash).await?)
    }

    /// Whether `op_hash` is present in the limbo (not-yet-integrated) chain ops.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn limbo_op_exists(&self, op_hash: &DhtOpHash) -> StateQueryResult<bool> {
        Ok(self.db().limbo_op_exists(op_hash).await?)
    }

    /// Hashes of limbo chain ops flagged as requiring a validation receipt.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn limbo_op_hashes_requiring_receipt(&self) -> StateQueryResult<Vec<DhtOpHash>> {
        Ok(self.db().limbo_op_hashes_requiring_receipt().await?)
    }

    /// Hashes of integrated chain ops with the given DHT `basis`.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn get_ops_at_basis(
        &self,
        basis: &AnyLinkableHash,
    ) -> StateQueryResult<Vec<DhtOpHash>> {
        Ok(self.db().get_ops_at_basis(basis).await?)
    }

    /// Count of rows in the public `Entry` table.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn count_entries(&self) -> StateQueryResult<u64> {
        Ok(self.db().count_entries().await?)
    }

    /// Count of public `Entry` rows that belong to a private-visibility action.
    /// This is always zero — private entries live only in the separate
    /// `PrivateEntry` table and are never placed in the shared `Entry` table.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn count_private_entries_in_public_table(&self) -> StateQueryResult<u64> {
        Ok(self.db().count_private_entries_in_public_table().await?)
    }

    /// All validation receipts stored for the op `op_hash`, decoded from their
    /// blobs. Used by tests that verify receipt delivery and signatures.
    #[cfg(any(test, feature = "inspection"))]
    pub async fn validation_receipts_for_op(
        &self,
        op_hash: &DhtOpHash,
    ) -> StateQueryResult<Vec<holochain_types::prelude::SignedValidationReceipt>> {
        self.db()
            .get_validation_receipts(op_hash.clone())
            .await?
            .into_iter()
            .map(|row| {
                holochain_serialized_bytes::decode(&row.blob).map_err(|e| {
                    crate::query::StateQueryError::Other(format!("decode receipt: {e}"))
                })
            })
            .collect()
    }

    /// Integrated warrants authored by `author` (the warrant issuer),
    /// reconstructed as `WarrantOp`s.
    pub async fn warrants_by_author(
        &self,
        author: AgentPubKey,
    ) -> StateQueryResult<Vec<holochain_types::warrant::WarrantOp>> {
        self.db()
            .get_warrants_by_author(author)
            .await?
            .into_iter()
            .map(warrant_row_to_signed_warrant)
            .map(|r| r.map(holochain_types::warrant::WarrantOp::from))
            .collect()
    }

    /// Terminal validation status of an integrated warrant op, or `None` if no
    /// such warrant op exists.
    pub async fn warrant_validation_status(
        &self,
        op_hash: &DhtOpHash,
    ) -> StateQueryResult<Option<ValidationStatus>> {
        Ok(self
            .db()
            .warrant_op_validation_status(op_hash)
            .await?
            .map(|s| {
                if s == 2 {
                    ValidationStatus::Rejected
                } else {
                    ValidationStatus::Valid
                }
            }))
    }

    /// Drop any op whose hash is already recorded in the DHT store.
    /// Input order is preserved for surviving ops.
    ///
    /// The bool coupled with each op indicates whether to set the validation
    /// receipt required flag. This flag must be preserved through the
    /// filtering.
    pub async fn filter_existing_ops(
        &self,
        ops: Vec<(DhtOpHashed, bool)>,
    ) -> StateQueryResult<Vec<(DhtOpHashed, bool)>> {
        let hashes: Vec<DhtOpHash> = ops.iter().map(|(o, _)| o.as_hash().clone()).collect();
        let present = self.db().op_hashes_present(&hashes).await?;
        Ok(ops
            .into_iter()
            .zip(present)
            .filter_map(|(op, exists)| if exists { None } else { Some(op) })
            .collect())
    }

    /// Ops authored by `author` that are ready to be published to the network,
    /// each paired with its DHT basis.
    ///
    /// `min_publish_interval` bounds how recently an op may have been published
    /// and still be skipped: the cutoff is `now - min_publish_interval`, and an
    /// op qualifies when it has never been published or was last published at
    /// or before the cutoff. Private `CreateEntry` ops are never returned — the
    /// query is the guard that keeps private entries off the network.
    ///
    /// The basis is reconstructed as an [`ExternalHash`]: the publish path
    /// routes purely by the basis location, so the hash-type wrapper is
    /// irrelevant and the stored 36-byte basis is sufficient.
    pub async fn get_ops_to_publish(
        &self,
        author: &AgentPubKey,
        min_publish_interval: Duration,
    ) -> StateQueryResult<Vec<(AnyLinkableHash, DhtOpHash)>> {
        let recency = Timestamp::now()
            .as_micros()
            .saturating_sub(min_publish_interval.as_micros() as i64);
        let rows = self.db().get_ops_to_publish(author, recency).await?;
        Ok(rows
            .into_iter()
            .map(|r| {
                let basis: AnyLinkableHash = ExternalHash::from_raw_36(r.basis_hash).into();
                (basis, DhtOpHash::from_raw_36(r.dht_hash))
            })
            .collect())
    }

    /// Count ops authored by `author` that may still need publishing in the
    /// future (ignoring the recency window). The publish workflow uses this to
    /// decide whether to keep looping.
    pub async fn num_still_needing_publish(&self, author: &AgentPubKey) -> StateQueryResult<usize> {
        Ok(self.db().num_still_needing_publish(author).await? as usize)
    }

    /// Whether `author`'s genesis records are present.
    ///
    /// Genesis writes three chain actions (`Dna`, `AgentValidationPkg`,
    /// `AgentId`); their presence in the store marks the cell as genesised.
    /// The count is capped at three, so a long chain does not affect the cost.
    pub async fn has_genesis(&self, author: &AgentPubKey) -> StateQueryResult<bool> {
        Ok(self.db().count_author_actions_capped(author, 3).await? >= 3)
    }

    /// The author's source-chain head, or `None` pre-genesis.
    pub async fn chain_head_for_author(
        &self,
        author: &AgentPubKey,
    ) -> StateQueryResult<Option<crate::source_chain::HeadInfo>> {
        Ok(self
            .db()
            .chain_head_for_author(author)
            .await?
            .map(|(action, seq, timestamp)| crate::source_chain::HeadInfo {
                action,
                seq,
                timestamp,
            }))
    }

    /// Get the source-chain lock for `author`, if one exists.
    ///
    /// Returns *any* lock row for the author, including one that has already
    /// expired. Callers are responsible for checking `ChainLock::is_expired_at`
    /// when expiry matters; several countersigning call sites must respect an
    /// expired lock
    /// (a lock that exists at all means a session is mid-flight and the chain
    /// must not be re-used until it is cleaned up).
    ///
    /// This is why a `Timestamp::MIN` sentinel is passed to the underlying
    /// `holochain_data` reader: that reader filters out locks whose
    /// `expires_at <= now`, so `MIN` makes the expiry filter a no-op and every
    /// existing row is returned.
    pub async fn get_chain_lock(
        &self,
        author: AgentPubKey,
    ) -> StateQueryResult<Option<crate::chain_lock::ChainLock>> {
        Ok(self
            .db()
            .get_chain_lock(author, Timestamp::MIN)
            .await?
            .map(|row| {
                crate::chain_lock::ChainLock::from_parts(
                    row.subject,
                    Timestamp::from_micros(row.expires_at_timestamp),
                )
            }))
    }

    /// Whether `author`'s source-chain head is a committed countersigning
    /// session, and if so the session record, its entry hash, and the decoded
    /// [`CounterSigningSessionData`](holochain_zome_types::prelude::CounterSigningSessionData).
    ///
    /// This reads the chain head and head record from the store. `author` is
    /// threaded through to [`retrieve_record`](Self::retrieve_record) so the
    /// author's own (potentially private) countersigning entry is visible.
    ///
    /// Returns `None` when the chain is empty (pre-genesis), the head record is
    /// unavailable, or the head action does not carry a `CounterSign` entry. A
    /// `Some` result guarantees the `CounterSign` entry blob is present in the
    /// store — the match requires it — so callers need not re-check the entry.
    pub async fn current_countersigning_session(
        &self,
        author: &AgentPubKey,
    ) -> StateQueryResult<
        Option<(
            Record,
            holo_hash::EntryHash,
            holochain_zome_types::prelude::CounterSigningSessionData,
        )>,
    > {
        use holochain_types::prelude::Entry;

        let head = match self.chain_head_for_author(author).await? {
            Some(h) => h,
            None => return Ok(None),
        };
        let record = match self.retrieve_record(&head.action, Some(author)).await? {
            Some(r) => r,
            None => return Ok(None),
        };
        let entry_hash = record.action().data.entry_hash().cloned();
        let entry = record.entry.clone().into_option();
        Ok(match (entry_hash, entry) {
            (Some(entry_hash), Some(Entry::CounterSign(cs, _))) => Some((record, entry_hash, *cs)),
            _ => None,
        })
    }

    /// Validation receipts for every chain op of `action_hash`, grouped into a
    /// [`ValidationReceiptSet`] per op.
    ///
    /// Each stored receipt is the full serialized `SignedValidationReceipt`, so
    /// the validator-reported status and validator set are returned exactly as
    /// received. `receipts_complete` reflects the op's `ChainOpPublish` row.
    pub async fn validation_receipts_for_action(
        &self,
        action_hash: holo_hash::ActionHash,
    ) -> StateQueryResult<Vec<ValidationReceiptSet>> {
        use holochain_zome_types::prelude::{
            ChainOpType, ValidationReceiptInfo, ValidationReceiptSet,
        };

        let rows = self
            .db()
            .validation_receipts_for_action(action_hash)
            .await?;
        let mut sets: HashMap<DhtOpHash, ValidationReceiptSet> = HashMap::new();
        for row in rows {
            let op_hash = DhtOpHash::from_raw_36(row.op_hash);
            let op_type = ChainOpType::try_from(row.op_type).map_err(|e| {
                crate::query::StateQueryError::Other(format!(
                    "invalid op_type {}: {e}",
                    row.op_type
                ))
            })?;
            let receipts_complete = row.receipts_complete.map(|v| v != 0).unwrap_or(false);
            let receipt: holochain_types::prelude::SignedValidationReceipt =
                holochain_serialized_bytes::decode(&row.receipt_blob).map_err(|e| {
                    crate::query::StateQueryError::Other(format!("decode receipt: {e}"))
                })?;
            sets.entry(op_hash.clone())
                .or_insert_with(|| ValidationReceiptSet {
                    op_hash,
                    op_type: op_type.to_string(),
                    receipts_complete,
                    receipts: Vec::new(),
                })
                .receipts
                .push(ValidationReceiptInfo {
                    validation_status: receipt.receipt.validation_status,
                    validators: receipt.receipt.validators,
                });
        }
        Ok(sets.into_values().collect())
    }

    /// Whether `action_hash` (authored by `action_author`) has any
    /// pending-or-valid `ChainIntegrity` warrant declaring it an invalid chain
    /// op.
    ///
    /// Rejected and abandoned warrants are ignored — a disproven accusation
    /// must not mark an action invalid. Warrants still pending validation are
    /// included, since one may be validated imminently.
    pub async fn is_action_warranted_as_invalid(
        &self,
        action_hash: &holo_hash::ActionHash,
        action_author: &AgentPubKey,
    ) -> StateQueryResult<bool> {
        use holochain_zome_types::warrant::{ChainIntegrityWarrant, WarrantProof};

        let proofs = self
            .db()
            .pending_or_valid_warrant_proofs_by_warrantee(action_author)
            .await?;
        for proof_blob in proofs {
            let proof: WarrantProof =
                holochain_serialized_bytes::decode(&proof_blob).map_err(|e| {
                    crate::query::StateQueryError::Other(format!("decode warrant proof: {e}"))
                })?;
            if let WarrantProof::ChainIntegrity(ChainIntegrityWarrant::InvalidChainOp {
                action,
                ..
            }) = &proof
            {
                if &action.0 == action_hash {
                    return Ok(true);
                }
            }
        }
        Ok(false)
    }

    /// Terminal validation status of the chain op for `(action_hash, op_type)`,
    /// or `None` if it has no validation outcome yet (still pending, cache-only,
    /// or absent). The outcome is read from the integrated `ChainOp` or, if the
    /// op has been validated but not yet integrated, from `LimboChainOp`.
    ///
    /// Used by the sys-validation warrant-dependency readiness check: a warrant
    /// can only be evaluated once the op it warrants has its own validation
    /// outcome.
    pub async fn op_validation_status(
        &self,
        action_hash: &holo_hash::ActionHash,
        op_type: holochain_zome_types::op::ChainOpType,
    ) -> StateQueryResult<Option<ValidationStatus>> {
        let raw = self
            .db()
            .op_validation_outcome(action_hash, i64::from(op_type))
            .await?;
        // `validation_status` is a `RecordValidity` discriminant: 1 = Accepted
        // (Valid), 2 = Rejected.
        Ok(raw.map(|v| {
            if v == 2 {
                ValidationStatus::Rejected
            } else {
                ValidationStatus::Valid
            }
        }))
    }

    /// Find an existing action that shares `prev_action` with the given
    /// `action` but has a different hash. Used by sys-validation to detect
    /// chain forks.
    ///
    /// Returns `Ok(None)` if `action` has no `prev_action`, or if no
    /// sibling exists. Returns `Err` with a descriptive message if a
    /// sibling action by a different author is found (cross-author
    /// `prev_action` collision).
    ///
    /// The returned tuple is `(hash, signature)` for the sibling action —
    /// the exact shape the sys-validation workflow needs to author a
    /// `ChainFork` warrant.
    pub async fn find_fork_for_action(
        &self,
        action: &Action,
    ) -> StateQueryResult<Option<(holo_hash::ActionHash, holochain_types::prelude::Signature)>>
    {
        let Some(prev) = action.header.prev_action.as_ref() else {
            return Ok(None);
        };
        let incoming_hash = holo_hash::ActionHash::with_data_sync(action);

        let siblings = self
            .db()
            .get_actions_by_prev_hash(prev, &incoming_hash)
            .await?;

        let incoming_author = &action.header.author;
        if let Some(sibling) = siblings.into_iter().next() {
            let existing_author = &sibling.hashed.content.header.author;
            if existing_author != incoming_author {
                return Err(crate::query::StateQueryError::Other(format!(
                    "Cross-author prev_action collision: incoming author {incoming_author} \
                     differs from existing author {existing_author} for prev_action {prev:?}"
                )));
            }
            let hash = sibling.as_hash().clone();
            let signature = sibling.signature().clone();
            return Ok(Some((hash, signature)));
        }
        Ok(None)
    }

    /// Retrieve the entry for `hash` if present. `author = Some` includes that
    /// agent's private entry; `None` returns public entries only.
    pub async fn retrieve_entry(
        &self,
        hash: &holo_hash::EntryHash,
        author: Option<&holo_hash::AgentPubKey>,
    ) -> StateQueryResult<Option<holochain_types::prelude::Entry>> {
        Ok(self.db().get_entry(hash.clone(), author).await?)
    }

    /// Retrieve the complete record (action + entry, if any) for `hash`.
    ///
    /// If the action references an entry and that entry is not available
    /// locally, returns `None` (matching the cascade's `retrieve_public_record`
    /// contract). `author = Some` allows that agent's private entry.
    pub async fn retrieve_record(
        &self,
        hash: &holo_hash::ActionHash,
        author: Option<&holo_hash::AgentPubKey>,
    ) -> StateQueryResult<Option<Record>> {
        let Some(sah) = self.db().get_action(hash.clone()).await? else {
            return Ok(None);
        };
        let action = &sah.hashed.content;
        let entry = match action.data.entry_hash().cloned() {
            Some(entry_hash) if private_entry_visible_to(action, author) => {
                match self.db().get_entry(entry_hash.clone(), author).await? {
                    Some(entry) => Some(entry),
                    // A public entry referenced but unavailable means "no
                    // record"; an absent private entry is simply `Hidden`.
                    None if action_entry_is_private(action) => None,
                    None => return Ok(None),
                }
            }
            // A private entry is `Hidden` unless the caller is the author —
            // never attach another agent's private entry, even if the caller
            // happens to hold a same-hash private entry of their own.
            Some(_) => None,
            None => None,
        };
        let record_entry =
            RecordEntry::new(action_entry_type(action).map(|et| et.visibility()), entry);
        Ok(Some(Record::new(sah, record_entry)))
    }

    /// Retrieve the record for `hash` only if it is still live — i.e. its
    /// action exists and no integrated `Delete` targets it. Returns `None` if
    /// the action is absent, deleted, or (per `retrieve_record`) references an
    /// entry that is unavailable. `author = Some` allows that agent's private
    /// entry.
    pub async fn get_live_record(
        &self,
        hash: &holo_hash::ActionHash,
        author: Option<&holo_hash::AgentPubKey>,
    ) -> StateQueryResult<Option<Record>> {
        if !self
            .db()
            .get_deleted_records(hash.clone())
            .await?
            .is_empty()
        {
            return Ok(None);
        }
        self.retrieve_record(hash, author).await
    }

    /// The live `Record` for an entry: among the entry's valid, integrated,
    /// undeleted `CreateEntry` creates (visible to `author`), prefer the one
    /// authored by `author`, else the first by integration order. Returns
    /// `None` if there are no live creates. `author = Some` includes that
    /// agent's private entry.
    pub async fn get_live_entry(
        &self,
        entry_hash: &holo_hash::EntryHash,
        author: Option<&holo_hash::AgentPubKey>,
    ) -> StateQueryResult<Option<Record>> {
        let creates = self.db().get_live_entry_creates(entry_hash, author).await?;
        let chosen = match author {
            Some(a) => {
                let authored = creates
                    .iter()
                    .find(|sah| &sah.hashed.content.header.author == a)
                    .cloned();
                authored.or_else(|| creates.into_iter().next())
            }
            None => creates.into_iter().next(),
        };
        let Some(sah) = chosen else {
            return Ok(None);
        };
        let entry = self.db().get_entry(entry_hash.clone(), author).await?;
        let record_entry = RecordEntry::new(
            action_entry_type(&sah.hashed.content).map(|et| et.visibility()),
            entry,
        );
        Ok(Some(Record::new(sah, record_entry)))
    }

    /// Assemble the [`EntryDetails`](holochain_zome_types::metadata::EntryDetails) for `entry_hash`: the entry, its valid
    /// create actions, rejected create actions, the deletes on it, the updates
    /// from it, and its Live/Dead status. Returns `None` if the entry is not
    /// available. `author = Some` allows that agent's private entry.
    pub async fn get_entry_details(
        &self,
        entry_hash: &holo_hash::EntryHash,
        author: Option<&holo_hash::AgentPubKey>,
    ) -> StateQueryResult<Option<holochain_zome_types::metadata::EntryDetails>> {
        let Some(entry) = self.db().get_entry(entry_hash.clone(), author).await? else {
            return Ok(None);
        };
        let actions = self
            .db()
            .get_create_actions_for_entry(entry_hash, author, RecordValidity::Accepted)
            .await?;
        let rejected_actions = self
            .db()
            .get_create_actions_for_entry(entry_hash, author, RecordValidity::Rejected)
            .await?;
        let deletes = self.db().get_delete_actions_for_entry(entry_hash).await?;
        let updates = self.db().get_update_actions_for_entry(entry_hash).await?;
        let entry_dht_status = if self
            .db()
            .get_live_entry_creates(entry_hash, author)
            .await?
            .is_empty()
        {
            holochain_zome_types::metadata::EntryDhtStatus::Dead
        } else {
            holochain_zome_types::metadata::EntryDhtStatus::Live
        };
        Ok(Some(holochain_zome_types::metadata::EntryDetails {
            entry,
            actions,
            rejected_actions,
            deletes,
            updates,
            entry_dht_status,
        }))
    }

    /// Retrieve the signed action for `hash` if present, without CRUD
    /// resolution.
    pub async fn retrieve_action(
        &self,
        hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Option<SignedActionHashed>> {
        Ok(self.db().get_action(hash.clone()).await?)
    }

    /// The signed action carried by the integrated chain op `op_hash`, if the
    /// op is held in the store. Used by the receipt-receive path to inspect the
    /// action's entry type.
    pub async fn action_for_op(
        &self,
        op_hash: &DhtOpHash,
    ) -> StateQueryResult<Option<SignedActionHashed>> {
        let Some(row) = self.db().get_chain_op(op_hash.clone()).await? else {
            return Ok(None);
        };
        let action_hash = holo_hash::ActionHash::from_raw_36(row.action_hash);
        self.retrieve_action(&action_hash).await
    }

    /// Retrieve the signed action for `hash`, checking the store first and
    /// falling back to the in-memory scratch on a miss.
    ///
    /// Use this on the **requester** path only. Authority handlers must never
    /// see scratch data — use [`retrieve_action`](Self::retrieve_action) there.
    pub async fn retrieve_action_with_scratch(
        &self,
        hash: &holo_hash::ActionHash,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<SignedActionHashed>> {
        if let Some(sah) = self.retrieve_action(hash).await? {
            return Ok(Some(sah));
        }
        scratch_action(scratch, hash)
    }

    /// Retrieve the entry for `hash`, checking the store first and falling back
    /// to the in-memory scratch on a miss.
    ///
    /// `author` visibility follows the same rule as
    /// [`retrieve_entry`](Self::retrieve_entry): `None` returns public entries
    /// only from the store; entries in the scratch are always returned because
    /// the scratch is by definition this agent's own authored data.
    ///
    /// Use this on the **requester** path only.
    pub async fn retrieve_entry_with_scratch(
        &self,
        hash: &holo_hash::EntryHash,
        author: Option<&holo_hash::AgentPubKey>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<holochain_types::prelude::Entry>> {
        if let Some(entry) = self.retrieve_entry(hash, author).await? {
            return Ok(Some(entry));
        }
        scratch_entry(scratch, hash)
    }

    /// Retrieve the complete record (action + entry, if any) for `hash`,
    /// resolving both the action and the entry from store-or-scratch.
    ///
    /// Preserves the [`retrieve_record`](Self::retrieve_record) contract: if
    /// the action references an entry that is unavailable in both the store
    /// *and* the scratch, returns `None`.
    ///
    /// Use this on the **requester** path only.
    pub async fn retrieve_record_with_scratch(
        &self,
        hash: &holo_hash::ActionHash,
        author: Option<&holo_hash::AgentPubKey>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<Record>> {
        // Resolve the action from store, then scratch.
        let action = match self.retrieve_action(hash).await? {
            Some(sah) => sah,
            None => match scratch_action(scratch, hash)? {
                Some(sah) => sah,
                None => return Ok(None),
            },
        };
        let action_content = &action.hashed.content;

        // Resolve the entry (if the action references one). A private entry is
        // only attached for its own author — see `private_entry_visible_to`.
        let entry = match action_content.data.entry_hash().cloned() {
            Some(entry_hash) if private_entry_visible_to(action_content, author) => {
                // Try the store first, then the scratch (scratch entries are
                // this agent's own data; the visibility guard above already
                // ensures we only reach here for the author of a private entry).
                match self.retrieve_entry(&entry_hash, author).await? {
                    Some(e) => Some(e),
                    None => match scratch_entry(scratch, &entry_hash)? {
                        Some(e) => Some(e),
                        // Unavailable everywhere: a public entry means "no
                        // record"; an absent private entry is `Hidden`.
                        None if action_entry_is_private(action_content) => None,
                        None => return Ok(None),
                    },
                }
            }
            // Private entry, caller is not the author -> Hidden.
            Some(_) => None,
            None => None,
        };
        let record_entry = RecordEntry::new(
            action_entry_type(action_content).map(|et| et.visibility()),
            entry,
        );

        Ok(Some(Record::new(action, record_entry)))
    }

    /// Retrieve the record for `hash` only if it is still live, checking the
    /// store **and** the in-memory scratch for delete tombstones.
    ///
    /// Returns `None` when:
    /// - the store holds an integrated `Delete` targeting `hash`, **or**
    /// - the scratch holds a `Delete` action whose `deletes_address == hash`.
    ///
    /// Otherwise delegates to [`retrieve_record_with_scratch`](Self::retrieve_record_with_scratch)
    /// to resolve the action and entry from store-or-scratch.
    ///
    /// Use this on the **requester** path only. Authority handlers must never
    /// see scratch data — use [`get_live_record`](Self::get_live_record) there.
    pub async fn get_live_record_with_scratch(
        &self,
        hash: &holo_hash::ActionHash,
        author: Option<&holo_hash::AgentPubKey>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<Record>> {
        // An integrated delete in the store is decisive — record is dead.
        if !self
            .db()
            .get_deleted_records(hash.clone())
            .await?
            .is_empty()
        {
            return Ok(None);
        }
        // A pending delete in the scratch also kills the record.
        if scratch_delete_targets(scratch)?.contains(hash) {
            return Ok(None);
        }
        self.retrieve_record_with_scratch(hash, author, scratch)
            .await
    }

    /// The live `Record` for an entry, consulting both the store and the
    /// in-memory scratch for create candidates and delete tombstones.
    ///
    /// Live creates are the union of:
    /// - the store's validated, integrated, undeleted `CreateEntry` creates
    ///   (from [`get_live_entry_creates`](holochain_data::DbRead::get_live_entry_creates)), and
    /// - scratch `Create`/`Update` actions whose `entry_hash() == entry_hash`
    ///   and that are not targeted by a scratch `Delete`.
    ///
    /// Among these, the authored-or-first preference from
    /// [`get_live_entry`](Self::get_live_entry) is preserved: if `author` is
    /// `Some`, the first action by that author wins; otherwise the first store
    /// create (by integration order) is chosen, then the first scratch create.
    ///
    /// The entry is resolved via store-or-scratch (same as
    /// [`retrieve_entry_with_scratch`](Self::retrieve_entry_with_scratch)).
    ///
    /// Use this on the **requester** path only.
    pub async fn get_live_entry_with_scratch(
        &self,
        entry_hash: &holo_hash::EntryHash,
        author: Option<&holo_hash::AgentPubKey>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<Record>> {
        // Collect validated, integrated store creates, then drop any tombstoned
        // by a pending scratch `Delete` (the store read only excludes integrated
        // store deletes, so the scratch-delete exclusion is applied here).
        let scratch_deleted = scratch_delete_targets(scratch)?;
        let store_creates: Vec<_> = self
            .db()
            .get_live_entry_creates(entry_hash, author)
            .await?
            .into_iter()
            .filter(|sah| !scratch_deleted.contains(sah.as_hash()))
            .collect();

        // Collect live scratch creates/updates for this entry.
        let scratch_creates: Vec<SignedActionHashed> =
            scratch_live_entry_creates(scratch, entry_hash)?;

        // Author-preference: prefer a create by `author`; failing that, first
        // store create (integration order), then first scratch create.
        let chosen = match author {
            Some(a) => {
                // Search store creates first, then scratch creates.
                let authored = store_creates
                    .iter()
                    .find(|sah| &sah.hashed.content.header.author == a)
                    .cloned();
                let authored = authored.or_else(|| {
                    scratch_creates
                        .iter()
                        .find(|sah| &sah.hashed.content.header.author == a)
                        .cloned()
                });
                authored
                    .or_else(|| store_creates.into_iter().next())
                    .or_else(|| scratch_creates.into_iter().next())
            }
            None => store_creates
                .into_iter()
                .next()
                .or_else(|| scratch_creates.into_iter().next()),
        };

        let Some(action) = chosen else {
            return Ok(None);
        };

        // Resolve the entry from store-or-scratch.
        let entry = self
            .retrieve_entry_with_scratch(entry_hash, author, scratch)
            .await?;

        let record_entry = RecordEntry::new(
            action_entry_type(&action.hashed.content).map(|et| et.visibility()),
            entry,
        );

        Ok(Some(Record::new(action, record_entry)))
    }

    /// Assemble the [`RecordDetails`](holochain_zome_types::metadata::RecordDetails) for `hash`: the record, its validation
    /// status (from its integrated `CreateRecord` op), the deletes targeting it,
    /// and the updates of it. Returns `None` if there is no integrated
    /// `CreateRecord` op for `hash`. `author = Some` allows that agent's private
    /// entry.
    pub async fn get_record_details(
        &self,
        hash: &holo_hash::ActionHash,
        author: Option<&holo_hash::AgentPubKey>,
    ) -> StateQueryResult<Option<holochain_zome_types::metadata::RecordDetails>> {
        use holochain_zome_types::op::ChainOpType;
        let ops = self.db().get_chain_ops_for_action(hash.clone()).await?;
        let Some(store_op) = ops
            .iter()
            .find(|r| ChainOpType::try_from(r.op_type) == Ok(ChainOpType::CreateRecord))
        else {
            return Ok(None);
        };
        let validation_status = match RecordValidity::try_from(store_op.validation_status) {
            Ok(RecordValidity::Accepted) => holochain_zome_types::validate::ValidationStatus::Valid,
            Ok(RecordValidity::Rejected) => {
                holochain_zome_types::validate::ValidationStatus::Rejected
            }
            Err(v) => {
                return Err(crate::query::StateQueryError::Other(format!(
                    "invalid validation_status {v} on CreateRecord op for {hash:?}"
                )))
            }
        };
        let Some(record) = self.retrieve_record(hash, author).await? else {
            return Ok(None);
        };
        let deletes = self.db().get_delete_actions_for_record(hash).await?;
        let updates = self.db().get_update_actions_for_record(hash).await?;
        Ok(Some(holochain_zome_types::metadata::RecordDetails {
            record,
            validation_status,
            deletes,
            updates,
        }))
    }

    /// Assemble the [`RecordDetails`](holochain_zome_types::metadata::RecordDetails) for `hash`, overlaying the in-memory
    /// scratch for deletes and updates.
    ///
    /// # Contract
    ///
    /// Validation status is derived from the integrated `CreateRecord` op when one
    /// exists. A scratch-only record (authored locally but not yet committed and
    /// published) has no such op; it is returned with `Valid` status, since the
    /// author's own uncommitted record is valid from their perspective. This
    /// lets host functions resolve a just-authored record mid-call — e.g. an
    /// `update` reading the `create` it depends on within one zome call. An action
    /// reachable only via a non-record op (no `CreateRecord`, not in the scratch)
    /// still returns `None`.
    ///
    /// Deletes and updates are the **union** of:
    /// - store-integrated `DeleteRecord` / `UpdateRecord` actions, and
    /// - scratch `Delete`/`Update` actions targeting `hash`.
    ///
    /// Use this on the **requester** path only. Authority handlers must never
    /// see scratch data — use [`get_record_details`](Self::get_record_details) there.
    pub async fn get_record_details_with_scratch(
        &self,
        hash: &holo_hash::ActionHash,
        author: Option<&holo_hash::AgentPubKey>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<holochain_zome_types::metadata::RecordDetails>> {
        use holochain_zome_types::op::ChainOpType;

        // Resolve the record via store-or-scratch first, so a locally-authored,
        // not-yet-committed record in the scratch is visible to its author.
        let Some(record) = self
            .retrieve_record_with_scratch(hash, author, scratch)
            .await?
        else {
            return Ok(None);
        };

        // Validation status comes from the integrated CreateRecord op when one
        // exists. A scratch-only record (authored locally, not yet committed)
        // has no such op and is Valid from the author's perspective. A store
        // action reachable only via a non-record op (no CreateRecord op, not in
        // the scratch) has no record details.
        let ops = self.db().get_chain_ops_for_action(hash.clone()).await?;
        let store_op = ops
            .iter()
            .find(|r| ChainOpType::try_from(r.op_type) == Ok(ChainOpType::CreateRecord));
        let validation_status = match store_op {
            Some(store_op) => match RecordValidity::try_from(store_op.validation_status) {
                Ok(RecordValidity::Accepted) => ValidationStatus::Valid,
                Ok(RecordValidity::Rejected) => ValidationStatus::Rejected,
                Err(v) => {
                    return Err(crate::query::StateQueryError::Other(format!(
                        "invalid validation_status {v} on CreateRecord op for {hash:?}"
                    )))
                }
            },
            None if scratch_action(scratch, hash)?.is_some() => ValidationStatus::Valid,
            None => return Ok(None),
        };

        // Store deletes + scratch deletes targeting this action hash.
        let mut deletes: Vec<SignedActionHashed> =
            self.db().get_delete_actions_for_record(hash).await?;
        deletes.extend(scratch_deletes_for_record(scratch, hash)?);

        // Store updates + scratch updates targeting this action hash.
        let mut updates: Vec<SignedActionHashed> =
            self.db().get_update_actions_for_record(hash).await?;
        updates.extend(scratch_updates_for_record(scratch, hash)?);

        Ok(Some(holochain_zome_types::metadata::RecordDetails {
            record,
            validation_status,
            deletes,
            updates,
        }))
    }

    /// Assemble the [`EntryDetails`](holochain_zome_types::metadata::EntryDetails) for `entry_hash`, overlaying the in-memory
    /// scratch for creates, deletes, updates, and Live/Dead status.
    ///
    /// Returns `None` when the entry is absent from both the store and the scratch.
    ///
    /// The returned fields are assembled as follows:
    /// - **`actions`** (accepted creates): store accepted creates ∪ scratch
    ///   `Create`/`Update` actions whose new-entry side equals `entry_hash`.
    ///   Scratch creates carry no validation status, so they are treated as
    ///   unconditionally accepted.
    /// - **`rejected_actions`**: store only (the scratch holds no validation status).
    /// - **`deletes`**: store deletes ∪ scratch `Delete` actions whose
    ///   `deletes_entry_address == entry_hash`.
    /// - **`updates`**: store updates ∪ scratch `Update` actions whose
    ///   `original_entry_address == entry_hash`.
    /// - **`entry_dht_status`**: Live when there is at least one live create in the
    ///   union of store live creates and scratch live creates (those not targeted by
    ///   a scratch `Delete`).
    ///
    /// Use this on the **requester** path only.
    pub async fn get_entry_details_with_scratch(
        &self,
        entry_hash: &holo_hash::EntryHash,
        author: Option<&holo_hash::AgentPubKey>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Option<holochain_zome_types::metadata::EntryDetails>> {
        // Resolve the entry from store-or-scratch; return None if absent in both.
        let entry = match self.retrieve_entry(entry_hash, author).await? {
            Some(e) => e,
            None => match scratch_entry(scratch, entry_hash)? {
                Some(e) => e,
                None => return Ok(None),
            },
        };

        // Accepted creates: store accepted creates + all scratch creates for
        // this entry.
        let mut actions: Vec<SignedActionHashed> = self
            .db()
            .get_create_actions_for_entry(entry_hash, author, RecordValidity::Accepted)
            .await?;
        actions.extend(scratch_creates_for_entry(scratch, entry_hash)?);

        // Rejected creates: store only (scratch has no validation status).
        let rejected_actions = self
            .db()
            .get_create_actions_for_entry(entry_hash, author, RecordValidity::Rejected)
            .await?;

        // Deletes: store + scratch.
        let mut deletes: Vec<SignedActionHashed> =
            self.db().get_delete_actions_for_entry(entry_hash).await?;
        deletes.extend(scratch_deletes_for_entry(scratch, entry_hash)?);

        // Updates: store + scratch.
        let mut updates: Vec<SignedActionHashed> =
            self.db().get_update_actions_for_entry(entry_hash).await?;
        updates.extend(scratch_updates_for_entry(scratch, entry_hash)?);

        // Live/Dead: Live if there is any undeleted create in the store-or-scratch
        // union. Use get_live_entry_with_scratch as the authoritative liveness check.
        let entry_dht_status = if self
            .get_live_entry_with_scratch(entry_hash, author, scratch)
            .await?
            .is_some()
        {
            holochain_zome_types::metadata::EntryDhtStatus::Live
        } else {
            holochain_zome_types::metadata::EntryDhtStatus::Dead
        };

        Ok(Some(holochain_zome_types::metadata::EntryDetails {
            entry,
            actions,
            rejected_actions,
            deletes,
            updates,
            entry_dht_status,
        }))
    }

    /// For `base`, every `CreateLink` (live and tombstoned) matching
    /// `type_query`/`tag`, paired with its `DeleteLink`s.
    pub async fn get_link_details(
        &self,
        base: &holo_hash::AnyLinkableHash,
        type_query: &LinkTypeFilter,
        tag: Option<&LinkTag>,
    ) -> StateQueryResult<Vec<(SignedActionHashed, Vec<SignedActionHashed>)>> {
        let creates = self.db().get_link_create_actions(base).await?;
        let mut out = Vec::with_capacity(creates.len());
        for create in creates {
            let ActionData::CreateLink(d) = &create.hashed.content.data else {
                continue;
            };
            if !type_query.contains(&d.zome_index, &d.link_type) {
                continue;
            }
            if let Some(t) = tag {
                if !d.tag.0.starts_with(&t.0) {
                    continue;
                }
            }
            let deletes = self.db().get_delete_link_actions(create.as_hash()).await?;
            out.push((create, deletes));
        }
        Ok(out)
    }

    /// The live links on `base` (CreateLink minus DeleteLink tombstones),
    /// filtered by link type, tag prefix, author, and time. Builds each
    /// [`holochain_zome_types::link::Link`] from its `CreateLink` action.
    ///
    /// Time bounds: `after` is inclusive (`timestamp >= after`) and `before` is
    /// inclusive (`timestamp <= before`).
    pub async fn get_links(
        &self,
        base: &holo_hash::AnyLinkableHash,
        type_query: &LinkTypeFilter,
        tag: Option<&LinkTag>,
        filter: &crate::query::link::GetLinksFilter,
    ) -> StateQueryResult<Vec<holochain_zome_types::link::Link>> {
        let actions = self.db().get_live_link_actions(base).await?;
        let mut links = Vec::with_capacity(actions.len());
        for sah in actions {
            let header = &sah.hashed.content.header;
            let ActionData::CreateLink(d) = &sah.hashed.content.data else {
                continue;
            };
            if !type_query.contains(&d.zome_index, &d.link_type) {
                continue;
            }
            if let Some(t) = tag {
                if !d.tag.0.starts_with(&t.0) {
                    continue;
                }
            }
            if let Some(author) = &filter.author {
                if &header.author != author {
                    continue;
                }
            }
            if let Some(before) = filter.before {
                if header.timestamp > before {
                    continue;
                }
            }
            if let Some(after) = filter.after {
                if header.timestamp < after {
                    continue;
                }
            }
            links.push(holochain_zome_types::link::Link {
                author: header.author.clone(),
                base: d.base_address.clone(),
                target: d.target_address.clone(),
                timestamp: header.timestamp,
                zome_index: d.zome_index,
                link_type: d.link_type,
                tag: d.tag.clone(),
                create_link_hash: sah.as_hash().clone(),
            });
        }
        // Return links in creation (timestamp) order.
        links.sort_by_key(|l| l.timestamp);
        Ok(links)
    }

    /// The live links on `base`, overlaid with the in-memory scratch.
    ///
    /// Extends [`get_links`](Self::get_links) with two scratch-layer adjustments:
    ///
    /// - **Scratch `DeleteLink` tombstones**: any store live link whose
    ///   `create_link_hash` is targeted by a scratch `DeleteLink` is removed.
    ///   (`get_links` only excludes *store* delete-link tombstones, so pending
    ///   scratch deletes must be excluded here.)
    /// - **Scratch `CreateLink` additions**: `CreateLink` actions in the scratch
    ///   whose `base_address == base` are included if they are not tombstoned by
    ///   a store `DeleteLink` OR a scratch `DeleteLink`, and they pass the same
    ///   type/tag/author/time filter.
    ///
    /// Use this on the **requester** path only. Authority handlers must never
    /// see scratch data — use [`get_links`](Self::get_links) there.
    pub async fn get_links_with_scratch(
        &self,
        base: &holo_hash::AnyLinkableHash,
        type_query: &LinkTypeFilter,
        tag: Option<&LinkTag>,
        filter: &crate::query::link::GetLinksFilter,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Vec<holochain_zome_types::link::Link>> {
        // Start with store live links, already filtered by store delete tombstones.
        let mut store_links = self.get_links(base, type_query, tag, filter).await?;

        // Collect scratch delete-link targets (link_add_address of each scratch
        // DeleteLink) so we can exclude any store link they tombstone.
        let scratch_dl_targets = scratch_delete_link_targets(scratch)?;

        // Exclude store links tombstoned by a scratch DeleteLink.
        store_links.retain(|l| !scratch_dl_targets.contains(&l.create_link_hash));

        // Add scratch CreateLinks for this base that pass all filters.
        let scratch_creates = scratch_create_links_for_base(scratch, base)?;
        for sah in scratch_creates {
            let action = sah.action();
            let ActionData::CreateLink(cl) = &action.data else {
                continue;
            };
            let create_hash = sah.as_hash();

            // Exclude if tombstoned by a store DeleteLink.
            if !self
                .db()
                .get_delete_link_actions(create_hash)
                .await?
                .is_empty()
            {
                continue;
            }
            // Exclude if tombstoned by a scratch DeleteLink.
            if scratch_dl_targets.contains(create_hash) {
                continue;
            }
            // Apply the same type/tag/author/time filter.
            if !type_query.contains(&cl.zome_index, &cl.link_type) {
                continue;
            }
            if let Some(t) = tag {
                if !cl.tag.0.starts_with(&t.0) {
                    continue;
                }
            }
            if let Some(author) = &filter.author {
                if action.header.author != *author {
                    continue;
                }
            }
            if let Some(before) = filter.before {
                if action.header.timestamp > before {
                    continue;
                }
            }
            if let Some(after) = filter.after {
                if action.header.timestamp < after {
                    continue;
                }
            }
            store_links.push(holochain_zome_types::link::Link {
                author: action.header.author.clone(),
                base: cl.base_address.clone(),
                target: cl.target_address.clone(),
                timestamp: action.header.timestamp,
                zome_index: cl.zome_index,
                link_type: cl.link_type,
                tag: cl.tag.clone(),
                create_link_hash: create_hash.clone(),
            });
        }

        // Return links (store + scratch) in creation (timestamp) order.
        store_links.sort_by_key(|l| l.timestamp);
        Ok(store_links)
    }

    /// For `base`, every `CreateLink` (live and tombstoned) matching
    /// `type_query`/`tag`, paired with its `DeleteLink`s — overlaid with the
    /// in-memory scratch.
    ///
    /// Extends [`get_link_details`](Self::get_link_details) with two
    /// scratch-layer adjustments:
    ///
    /// - **Scratch `DeleteLink` augmentations**: each store create's delete list
    ///   is extended with scratch `DeleteLink`s targeting that create (i.e.,
    ///   `link_add_address == create_hash`).
    /// - **Scratch `CreateLink` additions**: `CreateLink` actions in the scratch
    ///   whose `base_address == base` are included (ALL — tombstoned or not,
    ///   mirroring `get_link_details` showing every create and its deletes), each
    ///   paired with store deletes + scratch deletes targeting it, filtered by
    ///   type/tag.
    ///
    /// Use this on the **requester** path only.
    pub async fn get_link_details_with_scratch(
        &self,
        base: &holo_hash::AnyLinkableHash,
        type_query: &LinkTypeFilter,
        tag: Option<&LinkTag>,
        scratch: &SyncScratch,
    ) -> StateQueryResult<Vec<(SignedActionHashed, Vec<SignedActionHashed>)>> {
        // Start from store details (creates + their store deletes).
        let mut store_details = self.get_link_details(base, type_query, tag).await?;

        // Collect scratch DeleteLinks indexed by the create they tombstone.
        let scratch_dl_by_create = scratch_delete_links_by_create(scratch)?;

        // Augment each store create's delete list with scratch DeleteLinks
        // targeting that create.
        for (create_sah, deletes) in &mut store_details {
            let create_hash = create_sah.as_hash();
            if let Some(scratch_deletes) = scratch_dl_by_create.get(create_hash) {
                deletes.extend(scratch_deletes.iter().cloned());
            }
        }

        // Add scratch CreateLinks for this base, each paired with their deletes.
        let scratch_creates = scratch_create_links_for_base(scratch, base)?;
        for sah in scratch_creates {
            let action = sah.action();
            let ActionData::CreateLink(cl) = &action.data else {
                continue;
            };
            let create_hash = sah.as_hash();
            // Apply type/tag filter.
            if !type_query.contains(&cl.zome_index, &cl.link_type) {
                continue;
            }
            if let Some(t) = tag {
                if !cl.tag.0.starts_with(&t.0) {
                    continue;
                }
            }
            // Collect deletes: store deletes + scratch deletes targeting this
            // create.
            let mut deletes: Vec<SignedActionHashed> =
                self.db().get_delete_link_actions(create_hash).await?;
            if let Some(scratch_deletes) = scratch_dl_by_create.get(create_hash) {
                deletes.extend(scratch_deletes.iter().cloned());
            }
            store_details.push((sah.clone(), deletes));
        }

        Ok(store_details)
    }

    /// Agent activity for `author`: the integrated `AgentActivity`
    /// actions classified into valid/rejected, with chain status, highest
    /// observed, and warrants. Store-only (no scratch).
    ///
    /// The `ChainItems::Full` case builds
    /// [`Record`]s and the hashes case
    /// the `HoloHashed<Action>` shape, both of which the shared
    /// `build_agent_activity_response` machinery accepts.
    pub async fn get_agent_activity(
        &self,
        author: &holo_hash::AgentPubKey,
        filter: &ChainQueryFilter,
        options: &crate::dht_store::GetAgentActivityOptions,
    ) -> StateQueryResult<AgentActivityResponse> {
        let items = self
            .db()
            .get_agent_activity(author.clone(), options.include_full_records)
            .await?;

        let warrants = if options.include_warrants {
            self.db()
                .get_warrants_by_warrantee(author.clone())
                .await?
                .into_iter()
                .map(warrant_row_to_signed_warrant)
                .collect::<StateQueryResult<Vec<_>>>()?
        } else {
            Vec::new()
        };

        if options.include_full_records {
            let mut valid = Vec::new();
            let mut rejected = Vec::new();
            for item in items {
                let record = record_from_parts(item.action, item.entry);
                match item.validation_status {
                    RecordValidity::Accepted => valid.push(record),
                    RecordValidity::Rejected => rejected.push(record),
                }
            }
            Ok(build_agent_activity_response(
                author.clone(),
                valid,
                rejected,
                warrants,
                filter,
                options,
            ))
        } else {
            let mut valid = Vec::new();
            let mut rejected = Vec::new();
            for item in items {
                let action_hashed = item.action.hashed;
                match item.validation_status {
                    RecordValidity::Accepted => valid.push(action_hashed),
                    RecordValidity::Rejected => rejected.push(action_hashed),
                }
            }
            Ok(build_agent_activity_response(
                author.clone(),
                valid,
                rejected,
                warrants,
                filter,
                options,
            ))
        }
    }

    /// Agent activity for `author`, overlaid with the in-memory scratch.
    ///
    /// Extends [`get_agent_activity`](Self::get_agent_activity) with scratch-authored
    /// actions and warrants:
    ///
    /// - **Scratch activity**: every action in the scratch authored by `author` is
    ///   treated as valid (no rejection can exist for uncommitted actions) and merged
    ///   into the valid set. Scratch activity extends `highest_observed` and shifts
    ///   the chain status accordingly.
    /// - **Scratch warrants**: warrants in the scratch targeting `author` are merged
    ///   into the warrant list when `options.include_warrants` is true.
    ///
    /// The store-only rejected list is passed through unchanged — the scratch holds
    /// no rejected activity.
    ///
    /// Use this on the **requester** path only. Authority handlers must never see
    /// scratch data — use [`get_agent_activity`](Self::get_agent_activity) there.
    pub async fn get_agent_activity_with_scratch(
        &self,
        author: &holo_hash::AgentPubKey,
        filter: &ChainQueryFilter,
        options: &crate::dht_store::GetAgentActivityOptions,
        scratch: &SyncScratch,
    ) -> StateQueryResult<AgentActivityResponse> {
        let items = self
            .db()
            .get_agent_activity(author.clone(), options.include_full_records)
            .await?;

        // Collect store warrants (gated by option).
        let store_warrants: Vec<SignedWarrant> = if options.include_warrants {
            self.db()
                .get_warrants_by_warrantee(author.clone())
                .await?
                .into_iter()
                .map(warrant_row_to_signed_warrant)
                .collect::<StateQueryResult<Vec<_>>>()?
        } else {
            Vec::new()
        };

        // Collect scratch activity + warrants in one lock.
        let (scratch_valid, scratch_warrant_ops) = scratch.apply_and_then(
            |s| -> StateQueryResult<(
                Vec<AgentActivity>,
                Vec<holochain_types::warrant::WarrantOp>,
            )> {
                let activity = agent_activity_from_scratch(s, author, None, None);
                let warrants = if options.include_warrants {
                    warrants_for_agent_from_scratch(s, author)
                } else {
                    Vec::new()
                };
                Ok((activity, warrants))
            },
        )?;

        // Convert store warrants to WarrantOps and merge with scratch warrants.
        let store_warrant_ops: Vec<holochain_types::warrant::WarrantOp> = store_warrants
            .into_iter()
            .map(holochain_types::warrant::WarrantOp::from)
            .collect();
        let merged_warrant_ops = merge_warrants(vec![store_warrant_ops, scratch_warrant_ops]);
        // Convert back to SignedWarrant for AgentActivityResponse (WarrantOp derefs to it).
        let warrants: Vec<SignedWarrant> = merged_warrant_ops
            .into_iter()
            .map(|op| (*op).clone())
            .collect();

        if options.include_full_records {
            let mut store_valid_activity = Vec::new();
            let mut rejected = Vec::new();
            for item in items {
                match item.validation_status {
                    RecordValidity::Accepted => store_valid_activity.push(AgentActivity {
                        action: item.action,
                        cached_entry: None,
                    }),
                    // The rejected list is Records, and keeps the store row's entry.
                    RecordValidity::Rejected => {
                        rejected.push(record_from_parts(item.action, item.entry))
                    }
                }
            }
            // Merge store and scratch valid activity, dedup by action hash.
            let merged_activity = merge_agent_activity(vec![store_valid_activity, scratch_valid]);
            // Valid records carry no entry here, mirroring the cascade's
            // `cached_entry: None` convention on the requester path (the entry is
            // filled separately by the cascade); rejected records above keep theirs.
            let merged_valid: Vec<Record> = merged_activity
                .into_iter()
                .map(|a| record_from_parts(a.action, None))
                .collect();
            Ok(build_agent_activity_response(
                author.clone(),
                merged_valid,
                rejected,
                warrants,
                filter,
                options,
            ))
        } else {
            let mut store_valid_activity = Vec::new();
            let mut rejected_hashed = Vec::new();
            for item in items {
                match item.validation_status {
                    RecordValidity::Accepted => store_valid_activity.push(AgentActivity {
                        action: item.action,
                        cached_entry: None,
                    }),
                    RecordValidity::Rejected => rejected_hashed.push(item.action.hashed),
                }
            }
            // Merge store and scratch valid activity, dedup by action hash.
            let merged_activity = merge_agent_activity(vec![store_valid_activity, scratch_valid]);
            let merged_valid: Vec<holo_hash::HoloHashed<Action>> = merged_activity
                .into_iter()
                .map(|a| a.action.hashed)
                .collect();
            Ok(build_agent_activity_response(
                author.clone(),
                merged_valid,
                rejected_hashed,
                warrants,
                filter,
                options,
            ))
        }
    }

    /// Store-only `must_get_agent_activity`: resolve `filter.chain_top`, scan the
    /// bounded `AgentActivity` range, exclude forked actions, apply the
    /// timestamp/take filters, and decide completeness.
    /// No scratch, no network, no cross-source merge (the requester layers those
    /// on).
    pub async fn must_get_agent_activity(
        &self,
        author: &holo_hash::AgentPubKey,
        filter: &ChainFilter,
    ) -> StateQueryResult<MustGetAgentActivityResponse> {
        // A take of zero is a degenerate filter.
        if filter.get_take() == Some(0) {
            return Err(crate::query::StateQueryError::InvalidInput(
                "ChainFilter take must be greater than 0".to_string(),
            ));
        }

        // Resolve the chain top; if absent we cannot answer.
        let Some((chain_top_seq, chain_top_timestamp)) = self
            .db()
            .get_action_seq_and_timestamp(author.clone(), filter.chain_top.clone())
            .await?
        else {
            return Ok(MustGetAgentActivityResponse::ChainTopNotFound(
                filter.chain_top.clone(),
            ));
        };

        // An until_timestamp after the chain top is unsatisfiable.
        if let Some(until_timestamp) = filter.get_until_timestamp() {
            if until_timestamp > chain_top_timestamp {
                return Ok(
                    MustGetAgentActivityResponse::UntilTimestampGreaterThanChainHead(
                        until_timestamp,
                    ),
                );
            }
        }

        // Resolve the until_hash lower bound, if any.
        let mut resolved_until_seq = None;
        if let Some(until_hash) = filter.get_until_hash() {
            resolved_until_seq = self
                .db()
                .get_action_seq_and_timestamp(author.clone(), until_hash.clone())
                .await?
                .map(|(seq, _)| seq);
            if let Some(until_seq) = resolved_until_seq {
                if until_seq > chain_top_seq {
                    return Ok(MustGetAgentActivityResponse::UntilHashAfterChainHead(
                        until_hash.clone(),
                    ));
                }
            }
        }

        // Scan the bounded range (already ordered seq DESC, hash DESC).
        let mut activity: Vec<AgentActivity> = self
            .db()
            .get_filtered_agent_activity(author.clone(), chain_top_seq, resolved_until_seq)
            .await?
            .into_iter()
            .map(|action| AgentActivity {
                action,
                cached_entry: None,
            })
            .collect();

        // Keep only the canonical chain reachable from the chain top.
        exclude_forked_activity(&mut activity, &filter.chain_top);

        // Apply the until_timestamp filter; the bool records whether the kept set
        // has a deterministic lower-bound witness.
        let canonical_chain_precedes_until_timestamp =
            apply_timestamp_filter(&mut activity, filter.get_until_timestamp());

        // Apply the take filter.
        if let Some(take) = filter.get_take() {
            activity.truncate(take as usize);
        }

        let completeness = check_agent_activity_completeness(
            &activity,
            filter,
            canonical_chain_precedes_until_timestamp,
        );

        Ok(match completeness {
            MustGetAgentActivityCompleteness::Complete => {
                let warrants = self
                    .db()
                    .get_warrants_by_warrantee(author.clone())
                    .await?
                    .into_iter()
                    .map(warrant_row_to_signed_warrant)
                    .collect::<StateQueryResult<Vec<_>>>()?
                    .into_iter()
                    .map(WarrantOp::from)
                    .collect();
                MustGetAgentActivityResponse::Activity { activity, warrants }
            }
            MustGetAgentActivityCompleteness::IncompleteChain => {
                MustGetAgentActivityResponse::IncompleteChain
            }
            MustGetAgentActivityCompleteness::UntilHashMissing(hash) => {
                MustGetAgentActivityResponse::UntilHashMissing(hash)
            }
            MustGetAgentActivityCompleteness::UntilTimestampIndeterminate(timestamp) => {
                MustGetAgentActivityResponse::UntilTimestampIndeterminate(timestamp)
            }
        })
    }

    /// Store-and-scratch `must_get_agent_activity`: resolve `filter.chain_top`,
    /// scan the bounded `AgentActivity` range from both the store and the
    /// in-memory scratch, merge, exclude forked actions, apply timestamp/take
    /// filters, and decide completeness.
    ///
    /// The scratch is consulted as a fallback for chain-top and until-hash
    /// resolution (store first, then scratch), and its activity is merged into
    /// the store-scanned range via `merge_agent_activity`. Warrants are
    /// attached (from store + scratch) only when the response is `Complete`.
    ///
    /// Use this on the **requester** path only. Authority handlers must never
    /// see scratch data — use [`must_get_agent_activity`](Self::must_get_agent_activity) there.
    pub async fn must_get_agent_activity_with_scratch(
        &self,
        author: &holo_hash::AgentPubKey,
        filter: &ChainFilter,
        scratch: &SyncScratch,
    ) -> StateQueryResult<MustGetAgentActivityResponse> {
        // A take of zero is a degenerate filter.
        if filter.get_take() == Some(0) {
            return Err(crate::query::StateQueryError::InvalidInput(
                "ChainFilter take must be greater than 0".to_string(),
            ));
        }

        // Resolve the chain top: store first, scratch fallback.
        let maybe_chain_top = self
            .db()
            .get_action_seq_and_timestamp(author.clone(), filter.chain_top.clone())
            .await?;
        let (chain_top_seq, chain_top_timestamp) = match maybe_chain_top {
            Some(pair) => pair,
            None => {
                // Store miss — try the scratch.
                let from_scratch =
                    scratch.apply_and_then(|s| {
                        Ok::<_, crate::query::StateQueryError>(
                            action_seq_and_timestamp_from_scratch(s, author, &filter.chain_top),
                        )
                    })?;
                match from_scratch {
                    Some(pair) => pair,
                    None => {
                        return Ok(MustGetAgentActivityResponse::ChainTopNotFound(
                            filter.chain_top.clone(),
                        ));
                    }
                }
            }
        };

        // An until_timestamp after the chain top is unsatisfiable.
        if let Some(until_timestamp) = filter.get_until_timestamp() {
            if until_timestamp > chain_top_timestamp {
                return Ok(
                    MustGetAgentActivityResponse::UntilTimestampGreaterThanChainHead(
                        until_timestamp,
                    ),
                );
            }
        }

        // Resolve the until_hash lower bound (store first, scratch fallback).
        let mut resolved_until_seq = None;
        if let Some(until_hash) = filter.get_until_hash() {
            // Try the store.
            resolved_until_seq = self
                .db()
                .get_action_seq_and_timestamp(author.clone(), until_hash.clone())
                .await?
                .map(|(seq, _)| seq);
            // Fall back to scratch on a store miss.
            if resolved_until_seq.is_none() {
                resolved_until_seq = scratch.apply_and_then(|s| {
                    Ok::<_, crate::query::StateQueryError>(
                        action_seq_and_timestamp_from_scratch(s, author, until_hash)
                            .map(|(seq, _)| seq),
                    )
                })?;
            }
            if let Some(until_seq) = resolved_until_seq {
                if until_seq > chain_top_seq {
                    return Ok(MustGetAgentActivityResponse::UntilHashAfterChainHead(
                        until_hash.clone(),
                    ));
                }
            }
        }

        // Scan the store bounded range.
        let store_activity: Vec<AgentActivity> = self
            .db()
            .get_filtered_agent_activity(author.clone(), chain_top_seq, resolved_until_seq)
            .await?
            .into_iter()
            .map(|action| AgentActivity {
                action,
                cached_entry: None,
            })
            .collect();

        // Scan the scratch bounded range.
        let scratch_activity: Vec<AgentActivity> = scratch.apply_and_then(|s| {
            Ok::<_, crate::query::StateQueryError>(agent_activity_from_scratch(
                s,
                author,
                Some(chain_top_seq),
                resolved_until_seq,
            ))
        })?;

        // Merge and deduplicate store + scratch activity.
        let mut activity = merge_agent_activity(vec![store_activity, scratch_activity]);

        // Keep only the canonical chain reachable from the chain top.
        exclude_forked_activity(&mut activity, &filter.chain_top);

        // Apply the until_timestamp filter; the bool records whether the kept set
        // has a deterministic lower-bound witness.
        let canonical_chain_precedes_until_timestamp =
            apply_timestamp_filter(&mut activity, filter.get_until_timestamp());

        // Apply the take filter.
        if let Some(take) = filter.get_take() {
            activity.truncate(take as usize);
        }

        let completeness = check_agent_activity_completeness(
            &activity,
            filter,
            canonical_chain_precedes_until_timestamp,
        );

        Ok(match completeness {
            MustGetAgentActivityCompleteness::Complete => {
                // Store warrants.
                let store_warrant_ops: Vec<WarrantOp> = self
                    .db()
                    .get_warrants_by_warrantee(author.clone())
                    .await?
                    .into_iter()
                    .map(warrant_row_to_signed_warrant)
                    .collect::<StateQueryResult<Vec<_>>>()?
                    .into_iter()
                    .map(WarrantOp::from)
                    .collect();
                // Scratch warrants.
                let scratch_warrant_ops: Vec<WarrantOp> = scratch.apply_and_then(|s| {
                    Ok::<_, crate::query::StateQueryError>(warrants_for_agent_from_scratch(
                        s, author,
                    ))
                })?;
                let warrants = merge_warrants(vec![store_warrant_ops, scratch_warrant_ops]);
                MustGetAgentActivityResponse::Activity { activity, warrants }
            }
            MustGetAgentActivityCompleteness::IncompleteChain => {
                MustGetAgentActivityResponse::IncompleteChain
            }
            MustGetAgentActivityCompleteness::UntilHashMissing(hash) => {
                MustGetAgentActivityResponse::UntilHashMissing(hash)
            }
            MustGetAgentActivityCompleteness::UntilTimestampIndeterminate(timestamp) => {
                MustGetAgentActivityResponse::UntilTimestampIndeterminate(timestamp)
            }
        })
    }

    /// Return chain ops that have passed system validation and are awaiting
    /// app validation. Warrants have no app-validation stage, so they are not
    /// included.
    pub async fn ops_pending_app_validation(
        &self,
        limit: u32,
    ) -> StateQueryResult<Vec<DhtOpHashed>> {
        let db = self.db();
        let rows = db
            .limbo_chain_ops_pending_app_validation_with_action(limit)
            .await?;
        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            out.push(chain_op_from_joined_row(&row)?);
        }
        Ok(out)
    }

    /// Return pending validation receipts: one entry per integrated, validated
    /// op that still has `require_receipt = 1`. Each tuple is
    /// `(ValidationReceipt, action_author)` so the caller can group by author.
    ///
    /// The `validators` field on each
    /// [`ValidationReceipt`](holochain_types::prelude::ValidationReceipt) is
    /// populated from the supplied `validators` argument (the locally-running
    /// agents for this DNA).
    pub async fn pending_validation_receipts(
        &self,
        validators: Vec<holo_hash::AgentPubKey>,
    ) -> StateQueryResult<
        Vec<(
            holochain_types::prelude::ValidationReceipt,
            holo_hash::AgentPubKey,
        )>,
    > {
        let rows = self.db().pending_validation_receipts().await?;
        rows.into_iter()
            .map(|r| {
                let dht_op_hash = holo_hash::DhtOpHash::from_raw_36(r.op_hash);
                let author = holo_hash::AgentPubKey::from_raw_36(r.action_author);
                // Map from RecordValidity (Accepted=1, Rejected=2) to
                // ValidationStatus (Valid=0, Rejected=1).
                let record_validity =
                    RecordValidity::try_from(r.validation_status).map_err(|v| {
                        crate::query::StateQueryError::Other(format!(
                            "invalid validation_status {v} in ChainOp row"
                        ))
                    })?;
                let validation_status = match record_validity {
                    RecordValidity::Accepted => {
                        holochain_zome_types::validate::ValidationStatus::Valid
                    }
                    RecordValidity::Rejected => {
                        holochain_zome_types::validate::ValidationStatus::Rejected
                    }
                };
                let when_integrated =
                    holochain_types::prelude::Timestamp::from_micros(r.when_integrated);
                Ok((
                    holochain_types::prelude::ValidationReceipt {
                        dht_op_hash,
                        validation_status,
                        validators: validators.clone(),
                        when_integrated,
                    },
                    author,
                ))
            })
            .collect()
    }

    /// Authority-serving create-link actions for `base`: locally-validated only,
    /// each paired with its validation status.
    pub async fn get_authority_link_creates(
        &self,
        base: &holo_hash::AnyLinkableHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_link_creates(base)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Authority-serving delete-link actions targeting `base`'s links:
    /// locally-validated only, each paired with its validation status.
    pub async fn get_authority_delete_links(
        &self,
        base: &holo_hash::AnyLinkableHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_delete_links(base)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Authority-serving `CreateRecord` action for `action_hash` (locally-validated
    /// only), paired with its validation status.
    pub async fn get_authority_store_record(
        &self,
        action_hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Option<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_store_record(action_hash)
            .await?
            .map(|(sah, validity)| (sah, record_validity_to_status(validity))))
    }

    /// Authority-serving deletes targeting record `action_hash` (locally-validated
    /// only), each paired with its validation status.
    pub async fn get_authority_deletes_for_record(
        &self,
        action_hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_deletes_for_record(action_hash)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Authority-serving updates targeting record `action_hash` (locally-validated
    /// only), each paired with its validation status.
    pub async fn get_authority_updates_for_record(
        &self,
        action_hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_updates_for_record(action_hash)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Authority-serving create actions for entry `entry_hash` (locally-validated
    /// only), each paired with its validation status.
    pub async fn get_authority_entry_creates(
        &self,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_entry_creates(entry_hash)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Authority-serving deletes targeting entry `entry_hash` (locally-validated
    /// only), each paired with its validation status.
    pub async fn get_authority_deletes_for_entry(
        &self,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_deletes_for_entry(entry_hash)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Authority-serving updates targeting entry `entry_hash` (locally-validated
    /// only), each paired with its validation status.
    pub async fn get_authority_updates_for_entry(
        &self,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<(SignedActionHashed, ValidationStatus)>> {
        Ok(self
            .db()
            .get_authority_updates_for_entry(entry_hash)
            .await?
            .into_iter()
            .map(|(sah, validity)| (sah, record_validity_to_status(validity)))
            .collect())
    }

    /// Warrants held against `warrantee`'s chain, as signed warrants.
    ///
    /// Used by the get authorities to pair a warrant with any `Rejected`
    /// record they serve, so the receiver can verify the rejection without
    /// being forced into pointless validation work.
    pub async fn get_warrants_by_warrantee(
        &self,
        warrantee: holo_hash::AgentPubKey,
    ) -> StateQueryResult<Vec<SignedWarrant>> {
        self.db()
            .get_warrants_by_warrantee(warrantee)
            .await?
            .into_iter()
            .map(warrant_row_to_signed_warrant)
            .collect()
    }

    /// Return ops awaiting system validation, sorted across chain ops and
    /// warrants by `(sys_validation_attempts, when_received)`, up to `limit`
    /// rows total.
    pub async fn ops_pending_sys_validation(
        &self,
        limit: u32,
    ) -> StateQueryResult<Vec<DhtOpHashed>> {
        let db = self.db();
        let chain_rows = db
            .limbo_chain_ops_pending_sys_validation_with_action(limit)
            .await?;
        let warrant_rows = db.limbo_warrants_pending_sys_validation(limit).await?;

        let mut out: Vec<(i64, i64, DhtOpHashed)> =
            Vec::with_capacity(chain_rows.len() + warrant_rows.len());

        for row in chain_rows {
            let attempts = row.sys_validation_attempts;
            let when_received = row.when_received;
            let op = chain_op_from_joined_row(&row)?;
            out.push((attempts, when_received, op));
        }
        for row in warrant_rows {
            let attempts = row.sys_validation_attempts;
            let when_received = row.when_received;
            let op = warrant_from_limbo_row(&row)?;
            out.push((attempts, when_received, op));
        }

        out.sort_by_key(|(attempts, when_received, _)| (*attempts, *when_received));
        out.truncate(limit as usize);
        Ok(out.into_iter().map(|(_, _, op)| op).collect())
    }

    /// Dump the author's source chain as a `SourceChainDump`.
    ///
    /// Reads all actions authored by `author` from the DhtStore in
    /// chain-sequence order, resolves each action's entry from both the public
    /// `Entry` table and the author's private `PrivateEntry` table, and counts
    /// the ops that have been published at least once.
    ///
    /// This is the production path behind the admin `DumpState` / `DumpFullState`
    /// APIs. It is **not** gated behind `inspection`.
    pub async fn dump_source_chain(
        &self,
        author: &AgentPubKey,
    ) -> StateQueryResult<SourceChainDump> {
        self.dump_source_chain_paginated(author, None, None).await
    }

    /// Dump one exclusive page of the author's source chain.
    ///
    /// # Errors
    ///
    /// Returns an error for a zero limit or an action-hash cursor that does not
    /// identify an accepted action authored by `author`.
    pub async fn dump_source_chain_paginated(
        &self,
        author: &AgentPubKey,
        cursor: Option<&SourceChainCursor>,
        limit: Option<u32>,
    ) -> StateQueryResult<SourceChainDump> {
        if limit == Some(0) {
            return Err(StateQueryError::InvalidInput(
                "dump limit must be greater than zero".to_string(),
            ));
        }

        let after_sequence = match cursor {
            Some(SourceChainCursor::Sequence(sequence)) => Some(*sequence),
            Some(SourceChainCursor::ActionHash(action_hash)) => Some(
                self.db()
                    .get_accepted_action_sequence(author, action_hash)
                    .await?
                    .ok_or_else(|| {
                        StateQueryError::InvalidInput(format!(
                            "source-chain cursor {action_hash} is unknown or belongs to another author"
                        ))
                    })?,
            ),
            None => None,
        };

        let actions = self
            .db()
            .get_actions_by_author_paginated(author, after_sequence, limit)
            .await?;

        let mut records = Vec::with_capacity(actions.len());
        for sah in &actions {
            let action_address = sah.as_hash().clone();
            let signature = sah.signature().clone();
            let action = sah.action().clone();

            // Resolve the entry (public OR private) when the action references one.
            let entry = match action.entry_hash() {
                Some(entry_hash) => {
                    self.db()
                        .get_entry(entry_hash.clone(), Some(author))
                        .await?
                }
                None => None,
            };

            records.push(SourceChainDumpRecord {
                signature,
                action_address,
                action,
                entry,
            });
        }

        let published_ops_count = self.db().count_published_ops_for_author(author).await? as usize;

        Ok(SourceChainDump {
            records,
            published_ops_count,
        })
    }

    /// Read the author's committed source-chain records.
    ///
    /// Returns every action authored by `author`, in ascending chain-sequence
    /// order, as a [`Record`].
    /// This is the committed-record source behind
    /// [`SourceChain::query`](crate::source_chain::SourceChain::query): no
    /// [`ChainQueryFilter`] filtering or fork disambiguation is applied here.
    /// The caller overlays the scratch and then applies
    /// [`ChainQueryFilter::filter_records`], which is the authoritative filter.
    ///
    /// An entry is attached only when `include_entries` is set and the entry is
    /// either public or `public_only` is `false`. A private entry is therefore
    /// redacted (`None`) under `public_only`. The private entry itself is
    /// resolved via `get_entry(.., Some(author))`, so an author always sees
    /// their own private entries and never another agent's.
    pub async fn source_chain_records(
        &self,
        author: &AgentPubKey,
        include_entries: bool,
        public_only: bool,
    ) -> StateQueryResult<Vec<Record>> {
        let sahs: Vec<SignedActionHashed> = self.db().get_actions_by_author(author.clone()).await?;

        // Decide which entries to attach. Collect the entry hashes that
        // actually need fetching so they can be read in a single batch rather
        // than one-at-a-time (an N+1 over the chain length).
        let mut wanted_entry_hashes: Vec<EntryHash> = Vec::new();
        // For each action, `Some(hash)` means "attach this entry if present";
        // `None` means "no entry" (condition not met, or no entry hash).
        let attach: Vec<Option<EntryHash>> = sahs
            .iter()
            .map(|sah| {
                let action = &sah.hashed.content;
                let private_entry = action_entry_is_private(action);

                if include_entries && (!private_entry || !public_only) {
                    if let Some(entry_hash) = action.data.entry_hash() {
                        wanted_entry_hashes.push(entry_hash.clone());
                        return Some(entry_hash.clone());
                    }
                }
                None
            })
            .collect();

        // Batch-fetch the needed entries. The author context resolves the
        // author's own private entries, exactly as the per-action
        // `get_entry(.., Some(author))` did.
        let entries = self
            .db()
            .get_entries_by_hashes(&wanted_entry_hashes, Some(author))
            .await?;

        // Assemble records in chain order. An absent hash maps to `None`,
        // matching the previous `get_entry` miss behaviour.
        let records = sahs
            .into_iter()
            .zip(attach)
            .map(|(sah, want)| {
                let entry = want.and_then(|h| entries.get(&h).cloned());
                let record_entry = RecordEntry::new(
                    action_entry_type(&sah.hashed.content).map(|et| et.visibility()),
                    entry,
                );
                Record::new(sah, record_entry)
            })
            .collect();

        Ok(records)
    }

    /// The author's valid agent-key `Create` action.
    ///
    /// Returns the author's `Create` action whose entry is the `AgentPubKey`
    /// entry (`entry_hash == author.into()`) and that has NOT been updated or
    /// deleted **by the author**. Returns `None` when no such action exists or
    /// it has been modified, so the caller maps that to
    /// [`SourceChainError::InvalidAgentKey`](crate::source_chain::SourceChainError::InvalidAgentKey).
    ///
    /// The "not updated/deleted" exclusion is applied via
    /// `entry_updated_or_deleted_by_author`.
    pub async fn valid_create_agent_key_action(
        &self,
        author: &AgentPubKey,
    ) -> StateQueryResult<Option<Action>> {
        let agent_key_entry_hash: EntryHash = author.clone().into();

        // The agent-key `Create` is written to the author's chain at genesis;
        // find it among the author's actions. The `Action` table carries
        // `action_type` but not `entry_type`, so filter on `Create` + the entry
        // hash with an `AgentPubKey` entry-type guard.
        let actions = self.db().get_actions_by_author(author.clone()).await?;
        let create = actions.iter().find_map(|sah| {
            let action = sah.hashed.content.clone();
            let is_agent_key_create = match &action.data {
                ActionData::Create(d) => {
                    d.entry_type == EntryType::AgentPubKey && d.entry_hash == agent_key_entry_hash
                }
                _ => false,
            };
            is_agent_key_create.then_some(action)
        });

        let Some(create) = create else {
            return Ok(None);
        };

        // Agent key must not have been updated or deleted by the author.
        if self
            .entry_updated_or_deleted_by_author(&agent_key_entry_hash, author)
            .await?
        {
            return Ok(None);
        }

        Ok(Some(create))
    }

    /// The author's candidate capability grants for the validity loop in
    /// [`SourceChain::valid_cap_grant`](crate::source_chain::SourceChain::valid_cap_grant).
    ///
    /// - When `check_secret` is `Some`, only grants that carry a secret
    ///   (`Transferable` / `Assigned`) are considered. The exact-secret match is
    ///   left to the caller's [`CapAccess::is_valid_for_zome_call`], which is
    ///   equivalent because it requires `secret == given` for both secret-bearing
    ///   variants (a secret-bearing check therefore never matches an
    ///   `Unrestricted` grant).
    /// - When `check_secret` is `None`, only `Unrestricted` grants are
    ///   considered.
    ///
    /// Candidates are restricted to grants authored by `author` (the `CapGrant`
    /// index join filters on `Action.author`) and dropped when their entry has
    /// been updated or deleted by the author.
    ///
    /// The granted [`Capability`](holochain_zome_types::prelude::Capability) is not
    /// part of the index, so candidates of every capability kind are returned;
    /// matching the requested capability is the caller's job.
    pub async fn valid_cap_grants(
        &self,
        author: &AgentPubKey,
        check_secret: Option<&CapSecret>,
    ) -> StateQueryResult<Vec<CapAccess>> {
        // A secret-bearing check looks only at the grants that carry a secret; an
        // unrestricted check only at unrestricted grants.
        let constraint_types: &[GrantConstraintType] = if check_secret.is_some() {
            &[
                GrantConstraintType::Transferable,
                GrantConstraintType::Assigned,
            ]
        } else {
            &[GrantConstraintType::Unrestricted]
        };

        let mut grants = Vec::new();
        for &constraint_type in constraint_types {
            let rows = self
                .db()
                .get_cap_grants_by_access(author.clone(), constraint_type.into())
                .await?;

            // Resolve each grant's entry hash (from its create/update action)
            // and drop grants whose entry was updated or deleted by the author.
            // The per-grant `get_action` / modification checks stay one-by-one
            // (cap-grant counts are small), but the grant entries are then read
            // in a single batch rather than one query per grant.
            let mut candidate_hashes: Vec<EntryHash> = Vec::new();
            for row in rows {
                let action_hash = ActionHash::from_raw_36(row.action_hash);
                let Some(sah) = self.db().get_action(action_hash).await? else {
                    continue;
                };
                let Some(entry_hash) = sah.hashed.content.data.entry_hash().cloned() else {
                    continue;
                };
                // Skip grants whose entry was updated or deleted by the author.
                if self
                    .entry_updated_or_deleted_by_author(&entry_hash, author)
                    .await?
                {
                    continue;
                }
                candidate_hashes.push(entry_hash);
            }

            // Cap-grant entries are private; resolve them from the author's
            // `PrivateEntry` store in one batch and deserialize. Iterating
            // `candidate_hashes` (which preserves row order and any duplicates)
            // keeps the produced grants identical to the per-grant lookup.
            let entries = self
                .db()
                .get_entries_by_hashes(&candidate_hashes, Some(author))
                .await?;
            for entry_hash in candidate_hashes {
                let Some(entry) = entries.get(&entry_hash) else {
                    continue;
                };
                if let Some(grant) = entry.as_cap_access() {
                    grants.push(grant);
                }
            }
        }
        Ok(grants)
    }

    /// `true` if `entry_hash` has been updated or deleted by `author` — the
    /// "not updated/deleted" exclusion shared by the agent-key and cap-grant
    /// validity reads.
    ///
    /// The `UpdatedRecord` / `DeletedRecord` indexes are keyed on the original /
    /// deleted *entry* hash (see
    /// [`get_update_actions_for_entry`](holochain_data::DbRead::get_update_actions_for_entry)
    /// / [`get_delete_actions_for_entry`](holochain_data::DbRead::get_delete_actions_for_entry)),
    /// and the author filter restricts to the author's own modifications. The
    /// author's own `Update`/`Delete` actions populate these indexes at
    /// source-chain flush time.
    async fn entry_updated_or_deleted_by_author(
        &self,
        entry_hash: &EntryHash,
        author: &AgentPubKey,
    ) -> StateQueryResult<bool> {
        let updates = self.db().get_update_actions_for_entry(entry_hash).await?;
        if updates
            .iter()
            .any(|sah| &sah.hashed.content.header.author == author)
        {
            return Ok(true);
        }
        let deletes = self.db().get_delete_actions_for_entry(entry_hash).await?;
        Ok(deletes
            .iter()
            .any(|sah| &sah.hashed.content.header.author == author))
    }
}

/// The action's declared entry type, if it creates or updates an entry.
fn action_entry_type(action: &Action) -> Option<&EntryType> {
    match &action.data {
        ActionData::Create(d) => Some(&d.entry_type),
        ActionData::Update(d) => Some(&d.entry_type),
        _ => None,
    }
}

/// Build a [`Record`](Record) from a signed
/// action and an optional entry, deriving the [`RecordEntry`] slot from the
/// action's entry visibility.
fn record_from_parts(
    sah: SignedActionHashed,
    entry: Option<holochain_types::prelude::Entry>,
) -> Record {
    let visibility = action_entry_type(&sah.hashed.content).map(|et| et.visibility());
    let record_entry = RecordEntry::new(visibility, entry);
    Record::new(sah, record_entry)
}

/// Whether `action`'s entry (if any) is declared private.
fn action_entry_is_private(action: &Action) -> bool {
    action_entry_type(action).map(|et| et.visibility()) == Some(&EntryVisibility::Private)
}

/// Whether the requesting `author` may have `action`'s entry attached to the
/// assembled record. Always true for public entries; for a **private** entry
/// only when the caller is the action's author. A private entry must never be
/// served to a different author — not even when the caller holds a *same-hash*
/// private entry of their own (the entry hash is shared by content, but the
/// privacy is per author). This provides the entry-visibility hiding that the
/// by-hash `get_entry` lookup alone does not.
fn private_entry_visible_to(action: &Action, author: Option<&holo_hash::AgentPubKey>) -> bool {
    !action_entry_is_private(action) || author == Some(&action.header.author)
}

/// Map the record validity to the validation status served on the wire.
fn record_validity_to_status(v: RecordValidity) -> ValidationStatus {
    match v {
        RecordValidity::Accepted => ValidationStatus::Valid,
        RecordValidity::Rejected => ValidationStatus::Rejected,
    }
}

// Groups every helper that reads directly out of the scratch
// (`crate::scratch::Scratch`, see `scratch.rs`).
mod scratch_reads {
    use super::{StateQueryResult, SyncScratch};
    use holochain_zome_types::prelude::{ActionData, SignedActionHashed};

    /// Look up a [`SignedActionHashed`] by `hash` in the scratch space.
    ///
    /// Returns `Ok(None)` when the action is not present. Maps
    /// [`SyncScratchError`](crate::scratch::SyncScratchError) into
    /// [`StateQueryError`](crate::query::StateQueryError) via the existing `From` impl.
    pub(super) fn scratch_action(
        scratch: &SyncScratch,
        hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Option<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            Ok::<_, crate::query::StateQueryError>(
                s.actions()
                    .find(|sah| sah.action_address() == hash)
                    .cloned(),
            )
        })
    }

    /// Look up an [`Entry`] by `hash` in the scratch space.
    ///
    /// Entries in the scratch are authored by this agent and are therefore always
    /// visible, regardless of the `author` parameter.
    pub(super) fn scratch_entry(
        scratch: &SyncScratch,
        hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Option<holochain_types::prelude::Entry>> {
        // Delegate to the `Store` impl on `Scratch` (which likewise returns the
        // entry regardless of author — scratch data is this agent's own).
        use crate::query::Store;
        scratch.apply_and_then(|s| s.get_entry(hash))
    }

    /// The action hashes tombstoned by a `Delete` in this (locked) scratch — the
    /// `deletes_address` of every scratch `Delete`. Operates on a `&Scratch` so it
    /// can be reused inside an existing `apply`/`apply_and_then` closure without
    /// re-locking.
    fn delete_targets_in(
        s: &crate::scratch::Scratch,
    ) -> std::collections::HashSet<holo_hash::ActionHash> {
        s.actions()
            .filter_map(|sah| match &sah.action().data {
                ActionData::Delete(d) => Some(d.deletes_address.clone()),
                _ => None,
            })
            .collect()
    }

    /// The action hashes tombstoned by a pending scratch `Delete`.
    pub(super) fn scratch_delete_targets(
        scratch: &SyncScratch,
    ) -> StateQueryResult<std::collections::HashSet<holo_hash::ActionHash>> {
        scratch.apply_and_then(|s| Ok::<_, crate::query::StateQueryError>(delete_targets_in(s)))
    }

    /// Return all `Create`/`Update` actions in the scratch whose `entry_hash()`
    /// equals `entry_hash` and that are not targeted by a scratch `Delete`.
    ///
    /// This is the scratch-side analogue of
    /// [`get_live_entry_creates`](holochain_data::DbRead::get_live_entry_creates):
    /// it collects candidates and filters out any that have a pending delete
    /// tombstone in the same scratch.
    pub(super) fn scratch_live_entry_creates(
        scratch: &SyncScratch,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            // Action hashes of all scratch Deletes, so tombstoned creates are excluded.
            let deleted_addresses = delete_targets_in(s);

            let creates: Vec<SignedActionHashed> = s
                .actions()
                .filter(|sah| {
                    // Must be a Create or Update referencing this entry hash.
                    let references_entry = sah.action().entry_hash() == Some(entry_hash);
                    if !references_entry {
                        return false;
                    }
                    matches!(
                        sah.action().data,
                        ActionData::Create(_) | ActionData::Update(_)
                    )
                })
                .filter(|sah| {
                    // Exclude if a Delete in the scratch targets this action.
                    !deleted_addresses.contains(sah.action_address())
                })
                .cloned()
                .collect();

            Ok::<Vec<_>, crate::query::StateQueryError>(creates)
        })
    }

    /// Return all `Delete` scratch actions whose `deletes_address` equals `hash`
    /// (record-level tombstones for `get_record_details`).
    pub(super) fn scratch_deletes_for_record(
        scratch: &SyncScratch,
        hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter(|sah| match &sah.action().data {
                    ActionData::Delete(d) => &d.deletes_address == hash,
                    _ => false,
                })
                .cloned()
                .collect();
            Ok::<Vec<_>, crate::query::StateQueryError>(out)
        })
    }

    /// Return all `Update` scratch actions whose `original_action_address` equals
    /// `hash` (record-level updates for `get_record_details`).
    pub(super) fn scratch_updates_for_record(
        scratch: &SyncScratch,
        hash: &holo_hash::ActionHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter(|sah| match &sah.action().data {
                    ActionData::Update(u) => &u.original_action_address == hash,
                    _ => false,
                })
                .cloned()
                .collect();
            Ok::<Vec<_>, crate::query::StateQueryError>(out)
        })
    }

    /// Return all `Delete` scratch actions whose `deletes_entry_address` equals
    /// `entry_hash` (entry-level tombstones for `get_entry_details`).
    pub(super) fn scratch_deletes_for_entry(
        scratch: &SyncScratch,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter(|sah| match &sah.action().data {
                    ActionData::Delete(d) => &d.deletes_entry_address == entry_hash,
                    _ => false,
                })
                .cloned()
                .collect();
            Ok::<Vec<_>, crate::query::StateQueryError>(out)
        })
    }

    /// Return all `Update` scratch actions whose `original_entry_address` equals
    /// `entry_hash` (entry-level updates for `get_entry_details`).
    pub(super) fn scratch_updates_for_entry(
        scratch: &SyncScratch,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter(|sah| match &sah.action().data {
                    ActionData::Update(u) => &u.original_entry_address == entry_hash,
                    _ => false,
                })
                .cloned()
                .collect();
            Ok::<Vec<_>, crate::query::StateQueryError>(out)
        })
    }

    /// Return all `Create`/`Update` scratch actions whose new-entry side
    /// (`entry_hash()`) equals `entry_hash`. This is the accepted-creates analogue
    /// for `get_entry_details` — the new entry being introduced, not the one being
    /// replaced.
    ///
    /// Unlike [`scratch_live_entry_creates`], this does **not** exclude tombstoned
    /// creates: `EntryDetails.actions` lists every create regardless of deletes
    /// (liveness is reported separately via `entry_dht_status`).
    pub(super) fn scratch_creates_for_entry(
        scratch: &SyncScratch,
        entry_hash: &holo_hash::EntryHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter(|sah| {
                    matches!(
                        sah.action().data,
                        ActionData::Create(_) | ActionData::Update(_)
                    ) && sah.action().entry_hash() == Some(entry_hash)
                })
                .cloned()
                .collect();
            Ok::<Vec<_>, crate::query::StateQueryError>(out)
        })
    }

    /// Return all scratch `CreateLink` actions whose `base_address` equals `base`.
    ///
    /// Used by the requester path to collect pending authored links for a base.
    pub(super) fn scratch_create_links_for_base(
        scratch: &SyncScratch,
        base: &holo_hash::AnyLinkableHash,
    ) -> StateQueryResult<Vec<SignedActionHashed>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter(|sah| match &sah.action().data {
                    ActionData::CreateLink(cl) => &cl.base_address == base,
                    _ => false,
                })
                .cloned()
                .collect();
            Ok::<Vec<_>, crate::query::StateQueryError>(out)
        })
    }

    /// Return the `link_add_address` of every `DeleteLink` in the scratch — the
    /// `ActionHash`es of the `CreateLink` actions they tombstone.
    ///
    /// Used by `get_links_with_scratch` to exclude store links tombstoned by a
    /// pending scratch `DeleteLink`.
    pub(super) fn scratch_delete_link_targets(
        scratch: &SyncScratch,
    ) -> StateQueryResult<std::collections::HashSet<holo_hash::ActionHash>> {
        scratch.apply_and_then(|s| {
            let out = s
                .actions()
                .filter_map(|sah| match &sah.action().data {
                    ActionData::DeleteLink(dl) => Some(dl.link_add_address.clone()),
                    _ => None,
                })
                .collect();
            Ok::<_, crate::query::StateQueryError>(out)
        })
    }

    /// Build a map from `CreateLink` action hash → scratch `DeleteLink` actions
    /// that tombstone it. Used by `get_link_details_with_scratch` to augment each
    /// create's delete list with pending scratch deletes.
    pub(super) fn scratch_delete_links_by_create(
        scratch: &SyncScratch,
    ) -> StateQueryResult<std::collections::HashMap<holo_hash::ActionHash, Vec<SignedActionHashed>>>
    {
        scratch.apply_and_then(|s| {
            let mut map: std::collections::HashMap<holo_hash::ActionHash, Vec<SignedActionHashed>> =
                std::collections::HashMap::new();
            for sah in s.actions() {
                if let ActionData::DeleteLink(dl) = &sah.action().data {
                    map.entry(dl.link_add_address.clone())
                        .or_default()
                        .push(sah.clone());
                }
            }
            Ok::<_, crate::query::StateQueryError>(map)
        })
    }
}
use scratch_reads::{
    scratch_action, scratch_create_links_for_base, scratch_creates_for_entry,
    scratch_delete_link_targets, scratch_delete_links_by_create, scratch_delete_targets,
    scratch_deletes_for_entry, scratch_deletes_for_record, scratch_entry,
    scratch_live_entry_creates, scratch_updates_for_entry, scratch_updates_for_record,
};

// ---- scratch helpers for agent activity ----

/// Collect all actions authored by `author` from the scratch, optionally
/// bounded by `chain_top_seq` (inclusive upper bound) and `until_seq`
/// (inclusive lower bound). Used inside a single `apply_and_then` closure so
/// the `SyncScratch` mutex is held for only one lock.
fn agent_activity_from_scratch(
    s: &crate::scratch::Scratch,
    author: &holo_hash::AgentPubKey,
    // `None` = no upper bound (the whole authored chain); `Some(seq)` bounds to
    // `action_seq <= seq` for the `must_get` chain-top window.
    chain_top_seq: Option<u32>,
    until_seq: Option<u32>,
) -> Vec<AgentActivity> {
    s.actions()
        .filter(|sah| {
            let action = sah.action();
            if action.author() != author {
                return false;
            }
            let seq = action.action_seq();
            if let Some(top) = chain_top_seq {
                if seq > top {
                    return false;
                }
            }
            if let Some(until) = until_seq {
                if seq < until {
                    return false;
                }
            }
            true
        })
        .map(|sah| AgentActivity {
            action: sah.clone(),
            // Entries are not cached in scratch activity (mirrors cascade TODO).
            cached_entry: None,
        })
        .collect()
}

/// Look up the action sequence and timestamp for `action_hash` from the scratch,
/// filtered to `author`. Returns `None` when the action is absent.
fn action_seq_and_timestamp_from_scratch(
    s: &crate::scratch::Scratch,
    author: &holo_hash::AgentPubKey,
    action_hash: &holo_hash::ActionHash,
) -> Option<(u32, holochain_types::prelude::Timestamp)> {
    s.actions()
        .find(|sah| sah.action().author() == author && &sah.hashed.hash == action_hash)
        .map(|sah| (sah.action().action_seq(), sah.action().timestamp()))
}

/// Collect all scratch warrants whose subject (`InvalidChainOp.action_author` or
/// `ChainFork.chain_author`) matches `agent`.
fn warrants_for_agent_from_scratch(
    s: &crate::scratch::Scratch,
    agent: &holo_hash::AgentPubKey,
) -> Vec<holochain_types::warrant::WarrantOp> {
    use holochain_zome_types::warrant::{ChainIntegrityWarrant, WarrantProof};
    s.warrants()
        .filter(|sw| {
            let WarrantProof::ChainIntegrity(ref w) = sw.proof;
            match w {
                ChainIntegrityWarrant::InvalidChainOp {
                    ref action_author, ..
                } => action_author == agent,
                ChainIntegrityWarrant::ChainFork {
                    ref chain_author, ..
                } => chain_author == agent,
            }
        })
        .map(|sw| holochain_types::warrant::WarrantOp::from(sw.clone()))
        .collect()
}

/// Flatten, sort and deduplicate `AgentActivity` lists.
///
/// Sort key: `(Reverse(seq), Reverse(hash))` — newest action first, then by
/// hash descending for stability.
fn merge_agent_activity(lists: Vec<Vec<AgentActivity>>) -> Vec<AgentActivity> {
    use std::cmp::Reverse;
    let total: usize = lists.iter().map(|l| l.len()).sum();
    let mut merged = Vec::with_capacity(total);
    for list in lists {
        merged.extend(list);
    }
    merged.sort_unstable_by_key(|a| {
        (
            Reverse(a.action.action().action_seq()),
            Reverse(a.action.hashed.hash.clone()),
        )
    });
    merged.dedup_by_key(|a| a.action.hashed.hash.clone());
    merged
}

/// Flatten, sort and deduplicate `WarrantOp` lists.
///
/// Sort key: `to_hash()` — stable ordering by op hash.
fn merge_warrants(
    lists: Vec<Vec<holochain_types::warrant::WarrantOp>>,
) -> Vec<holochain_types::warrant::WarrantOp> {
    use holo_hash::HashableContentExtSync;
    let total: usize = lists.iter().map(|l| l.len()).sum();
    let mut merged = Vec::with_capacity(total);
    for list in lists {
        merged.extend(list);
    }
    merged.sort_unstable_by_key(|w| w.to_hash());
    merged.dedup_by_key(|w| w.to_hash());
    merged
}

/// Reconstruct a [`DhtOpHashed`] (`ChainOp` variant) from a
/// [`LimboChainOpJoinedRow`] without any additional database round-trips.
/// The action and entry blobs are decoded in-process from the joined columns.
fn chain_op_from_joined_row(
    row: &holochain_data::dht::LimboChainOpJoinedRow,
) -> StateQueryResult<DhtOpHashed> {
    use holo_hash::{ActionHash, AgentPubKey};
    use holochain_types::op::{ChainOp, DhtOp, OpEntry};
    use holochain_zome_types::prelude::{
        Action, ActionData, ActionHeader, ChainOpType, Entry, EntryType, EntryVisibility,
        Signature, SignedAction,
    };

    let op_type = ChainOpType::try_from(row.op_type).map_err(|n| {
        crate::query::StateQueryError::Other(format!("invalid op_type {n} in LimboChainOp row"))
    })?;

    // Decode the Action from the joined columns.
    let action_data: ActionData = holochain_serialized_bytes::decode(&row.action_data)
        .map_err(|e| crate::query::StateQueryError::Other(format!("decode ActionData: {e}")))?;
    let action = Action {
        header: ActionHeader {
            author: AgentPubKey::from_raw_36(row.action_author.clone()),
            timestamp: holochain_types::prelude::Timestamp::from_micros(row.action_timestamp),
            action_seq: row.action_seq as u32,
            prev_action: row
                .action_prev_hash
                .as_ref()
                .map(|h| ActionHash::from_raw_36(h.clone())),
        },
        data: action_data,
    };
    let sig_bytes: [u8; 64] = row.action_signature.as_slice().try_into().map_err(|_| {
        crate::query::StateQueryError::Other(format!(
            "signature column has {} bytes, expected 64",
            row.action_signature.len()
        ))
    })?;
    let signed_action = SignedAction::new(action.clone(), Signature(sig_bytes));

    // Decode the optional entry blob from the LEFT JOIN.
    let decoded_entry: Option<Entry> = row
        .entry_blob
        .as_ref()
        .map(|blob| {
            holochain_serialized_bytes::decode(blob)
                .map_err(|e| crate::query::StateQueryError::Other(format!("decode Entry: {e}")))
        })
        .transpose()?;

    // Build the `OpEntry` an entry-bearing op variant carries.
    //
    // `OpEntry` has only `Present`/`Hidden`/`ActionOnly`. An action with no
    // entry-bearing data at all (`entry_type: None`) maps to `ActionOnly`, as
    // does an entry-bearing *public* action whose entry blob isn't locally
    // present: there is no distinction between "this action has no entry" and
    // "the entry exists but isn't fetched yet", both just mean "no entry
    // payload right now". A *private* action's absent entry maps to `Hidden`,
    // recording that the entry is deliberately withheld, not merely unfetched.
    let op_entry_for = |entry_type: Option<&EntryType>| -> OpEntry {
        let Some(entry_type) = entry_type else {
            return OpEntry::ActionOnly;
        };
        match &decoded_entry {
            Some(entry) => OpEntry::Present(entry.clone()),
            None if entry_type.visibility() == &EntryVisibility::Private => OpEntry::Hidden,
            None => OpEntry::ActionOnly,
        }
    };
    let entry_type = match &action.data {
        ActionData::Create(d) => Some(&d.entry_type),
        ActionData::Update(d) => Some(&d.entry_type),
        _ => None,
    };

    let chain_op = match op_type {
        ChainOpType::CreateRecord => ChainOp::CreateRecord(signed_action, op_entry_for(entry_type)),
        ChainOpType::CreateEntry => {
            if entry_type.is_none() {
                return Err(crate::query::StateQueryError::Other(
                    "CreateEntry action is not a Create/Update".into(),
                ));
            }
            let entry = decoded_entry.clone().ok_or_else(|| {
                crate::query::StateQueryError::Other("Entry for CreateEntry not found".into())
            })?;
            ChainOp::CreateEntry(signed_action, OpEntry::Present(entry))
        }
        ChainOpType::AgentActivity => ChainOp::AgentActivity(signed_action),
        ChainOpType::UpdateEntry => {
            if !matches!(action.data, ActionData::Update(_)) {
                return Err(crate::query::StateQueryError::Other(
                    "UpdateEntry action is not Update".into(),
                ));
            }
            ChainOp::UpdateEntry(signed_action, op_entry_for(entry_type))
        }
        ChainOpType::UpdateRecord => {
            if !matches!(action.data, ActionData::Update(_)) {
                return Err(crate::query::StateQueryError::Other(
                    "UpdateRecord action is not Update".into(),
                ));
            }
            ChainOp::UpdateRecord(signed_action, op_entry_for(entry_type))
        }
        ChainOpType::DeleteEntry => {
            if !matches!(action.data, ActionData::Delete(_)) {
                return Err(crate::query::StateQueryError::Other(
                    "DeleteEntry action is not Delete".into(),
                ));
            }
            ChainOp::DeleteEntry(signed_action)
        }
        ChainOpType::DeleteRecord => {
            if !matches!(action.data, ActionData::Delete(_)) {
                return Err(crate::query::StateQueryError::Other(
                    "DeleteRecord action is not Delete".into(),
                ));
            }
            ChainOp::DeleteRecord(signed_action)
        }
        ChainOpType::CreateLink => {
            if !matches!(action.data, ActionData::CreateLink(_)) {
                return Err(crate::query::StateQueryError::Other(
                    "CreateLink op action is not a CreateLink action".into(),
                ));
            }
            ChainOp::CreateLink(signed_action)
        }
        ChainOpType::DeleteLink => {
            if !matches!(action.data, ActionData::DeleteLink(_)) {
                return Err(crate::query::StateQueryError::Other(
                    "DeleteLink op action is not a DeleteLink action".into(),
                ));
            }
            ChainOp::DeleteLink(signed_action)
        }
    };

    let op = DhtOp::ChainOp(Box::new(chain_op));
    let op_hash = holo_hash::DhtOpHash::from_raw_36(row.hash.clone());
    Ok(DhtOpHashed::with_pre_hashed(op, op_hash))
}

/// Reconstruct a [`DhtOpHashed`] (`WarrantOp` variant) from a `LimboWarrant` row.
fn warrant_from_limbo_row(
    row: &holochain_data::models::dht::LimboWarrantRow,
) -> StateQueryResult<DhtOpHashed> {
    use holochain_types::op::DhtOp;
    use holochain_types::prelude::Signature;
    use holochain_types::warrant::WarrantOp;
    use holochain_zome_types::warrant::{SignedWarrant, Warrant, WarrantProof};

    let proof: WarrantProof = holochain_serialized_bytes::decode(&row.proof)?;
    let author = holo_hash::AgentPubKey::from_raw_36(row.author.clone());
    let warrantee = holo_hash::AgentPubKey::from_raw_36(row.warrantee.clone());
    let timestamp = holochain_types::prelude::Timestamp::from_micros(row.timestamp);

    let warrant = Warrant::new(proof, author, timestamp, warrantee);
    // The signature is not stored in the limbo row — warrants are self-proving
    // via their proof content.  Use a zeroed signature as a placeholder, the
    // same approach used elsewhere in the codebase when reconstructing
    // WarrantOps from storage without a separate signature column.
    let signature = Signature::from([0u8; 64]);
    let signed_warrant = SignedWarrant::new(warrant, signature);
    let op = DhtOp::WarrantOp(Box::new(WarrantOp::from(signed_warrant)));
    let op_hash = holo_hash::DhtOpHash::from_raw_36(row.hash.clone());
    Ok(DhtOpHashed::with_pre_hashed(op, op_hash))
}

/// Reconstruct a [`SignedWarrant`] from an integrated `WarrantRow`.
fn warrant_row_to_signed_warrant(
    row: holochain_data::models::dht::WarrantRow,
) -> StateQueryResult<SignedWarrant> {
    use holochain_types::prelude::{Signature, Timestamp};
    use holochain_zome_types::warrant::{SignedWarrant, Warrant, WarrantProof};

    let proof: WarrantProof = holochain_serialized_bytes::decode(&row.proof)?;
    let author = holo_hash::AgentPubKey::from_raw_36(row.author);
    let warrantee = holo_hash::AgentPubKey::from_raw_36(row.warrantee);
    let timestamp = Timestamp::from_micros(row.timestamp);
    let warrant = Warrant::new(proof, author, timestamp, warrantee);
    let sig: [u8; 64] = row.signature.as_slice().try_into().map_err(|_| {
        crate::query::StateQueryError::Other(format!(
            "warrant signature column has {} bytes, expected 64",
            row.signature.len()
        ))
    })?;
    Ok(SignedWarrant::new(warrant, Signature::from(sig)))
}

/// Highest observed sequence number across the valid and rejected lists, each
/// assumed sorted ascending by sequence. The hash list holds the last valid
/// and/or last rejected action, which coincide only when both share the top
/// sequence.
fn compute_highest_observed<T: ActionSequenceAndHash>(
    valid: &[T],
    rejected: &[T],
) -> Option<HighestObserved> {
    let mut highest_observed: Option<u32> = None;
    let mut hashes = Vec::new();
    let mut check_highest = |seq: u32, hash: &holo_hash::ActionHash| {
        if let Some(last) = highest_observed.as_mut() {
            match seq.cmp(last) {
                std::cmp::Ordering::Less => {}
                std::cmp::Ordering::Equal => hashes.push(hash.clone()),
                std::cmp::Ordering::Greater => {
                    hashes.clear();
                    hashes.push(hash.clone());
                    *last = seq;
                }
            }
        } else {
            highest_observed = Some(seq);
            hashes.push(hash.clone());
        }
    };
    if let Some(v) = valid.last() {
        check_highest(v.action_seq(), v.address());
    }
    if let Some(r) = rejected.last() {
        check_highest(r.action_seq(), r.address());
    }
    highest_observed.map(|action_seq| HighestObserved {
        action_seq,
        hash: hashes,
    })
}

/// Compute the [`ChainStatus`] from the valid and rejected lists, returning the
/// (sorted) lists alongside. A fork is two valid actions at the same sequence.
fn compute_chain_status<T: ActionSequenceAndHash>(
    valid: impl Iterator<Item = T>,
    rejected: impl Iterator<Item = T>,
) -> (ChainStatus, Vec<T>, Vec<T>) {
    let mut valid: Vec<_> = valid.collect();
    let mut rejected: Vec<_> = rejected.collect();
    valid.sort_unstable_by_key(|a| a.action_seq());
    rejected.sort_unstable_by_key(|a| a.action_seq());

    let mut valid_out: Vec<T> = Vec::with_capacity(valid.len());
    let mut status = None;
    for current in valid {
        if status.is_none() {
            let fork = valid_out.last().and_then(|v: &T| {
                if current.action_seq() == v.action_seq() {
                    Some(v)
                } else {
                    None
                }
            });
            if let Some(fork) = fork {
                status = Some(ChainStatus::Forked(ChainFork {
                    fork_seq: current.action_seq(),
                    first_action: current.address().clone(),
                    second_action: fork.address().clone(),
                }));
            }
        }
        valid_out.push(current);
    }

    let status = status.unwrap_or_else(|| match (valid_out.last(), rejected.first()) {
        (None, None) => ChainStatus::Empty,
        (Some(v), None) => ChainStatus::Valid(ChainHead {
            action_seq: v.action_seq(),
            hash: v.address().clone(),
        }),
        (None, Some(r)) | (Some(_), Some(r)) => ChainStatus::Invalid(ChainHead {
            action_seq: r.action_seq(),
            hash: r.address().clone(),
        }),
    });

    (status, valid_out, rejected)
}

/// An intermediary type describing the completeness of a `must_get` result.
enum MustGetAgentActivityCompleteness {
    Complete,
    IncompleteChain,
    UntilHashMissing(holo_hash::ActionHash),
    UntilTimestampIndeterminate(holochain_types::prelude::Timestamp),
}

/// Remove forked actions by walking the chain backwards from `chain_top`.
/// Input must already be sorted by action seq descending.
fn exclude_forked_activity(activity: &mut Vec<AgentActivity>, chain_top: &holo_hash::ActionHash) {
    if activity.is_empty() {
        return;
    }
    let chain_hashes = collect_canonical_chain_hashes(activity, chain_top);
    activity.retain(|a| chain_hashes.contains(&a.action.hashed.hash));
}

/// Walk the chain from `chain_top` backwards (via `prev_action`), collecting the
/// reachable action hashes. Input must already be sorted by action seq descending.
fn collect_canonical_chain_hashes(
    activity: &[AgentActivity],
    chain_top: &holo_hash::ActionHash,
) -> HashSet<holo_hash::ActionHash> {
    let index_by_hash: HashMap<holo_hash::ActionHash, usize> = activity
        .iter()
        .enumerate()
        .map(|(i, a)| (a.action.hashed.hash.clone(), i))
        .collect();

    let mut chain_hashes: HashSet<holo_hash::ActionHash> = HashSet::new();

    let Some(&walk_index) = index_by_hash.get(chain_top) else {
        return chain_hashes;
    };

    let mut walk_index = walk_index;
    for _ in 0..activity.len() {
        let current = &activity[walk_index];
        let current_hash = current.action.hashed.hash.clone();
        if !chain_hashes.insert(current_hash) {
            break;
        }
        let Some(prev_hash) = current.action.action().prev_action() else {
            break;
        };
        let Some(&prev_index) = index_by_hash.get(prev_hash) else {
            break;
        };
        walk_index = prev_index;
    }

    chain_hashes
}

/// Apply the `until_timestamp` filter. Returns `true` when the kept set has a
/// deterministic lower-bound witness (an action with timestamp < `until_timestamp`).
fn apply_timestamp_filter(
    activity: &mut Vec<AgentActivity>,
    until_timestamp: Option<holochain_types::prelude::Timestamp>,
) -> bool {
    match until_timestamp {
        None => false,
        Some(until_ts) => {
            let precedes_boundary = activity
                .last()
                .map(|a| a.action.action().timestamp() < until_ts)
                .unwrap_or(false);
            activity.retain(|a| a.action.action().timestamp() >= until_ts);
            precedes_boundary
        }
    }
}

/// Decide whether `activity` is a complete response for `filter`.
fn check_agent_activity_completeness(
    activity: &[AgentActivity],
    filter: &ChainFilter,
    canonical_chain_precedes_until_timestamp: bool,
) -> MustGetAgentActivityCompleteness {
    let has_gap = activity
        .windows(2)
        .any(|w| w[0].action.action().action_seq() != w[1].action.action().action_seq() + 1);
    let reaches_genesis = activity
        .last()
        .map(|last| last.action.action().action_seq() == 0)
        .unwrap_or(false);

    match &filter.limit_conditions {
        LimitConditions::ToGenesis => {
            if has_gap || !reaches_genesis {
                MustGetAgentActivityCompleteness::IncompleteChain
            } else {
                MustGetAgentActivityCompleteness::Complete
            }
        }
        LimitConditions::UntilHash(until_hash) => {
            if !activity.iter().any(|a| &a.action.hashed.hash == until_hash) {
                MustGetAgentActivityCompleteness::UntilHashMissing(until_hash.clone())
            } else if has_gap {
                MustGetAgentActivityCompleteness::IncompleteChain
            } else {
                MustGetAgentActivityCompleteness::Complete
            }
        }
        LimitConditions::UntilTimestamp(until_timestamp) => {
            let any_satisfies_timestamp = activity
                .iter()
                .any(|a| a.action.action().timestamp() >= *until_timestamp);
            if !any_satisfies_timestamp
                || (!reaches_genesis && !canonical_chain_precedes_until_timestamp)
            {
                MustGetAgentActivityCompleteness::UntilTimestampIndeterminate(*until_timestamp)
            } else if has_gap {
                MustGetAgentActivityCompleteness::IncompleteChain
            } else {
                MustGetAgentActivityCompleteness::Complete
            }
        }
        LimitConditions::Take(take) => {
            let take = *take as usize;
            if activity.len() >= take {
                if has_gap {
                    MustGetAgentActivityCompleteness::IncompleteChain
                } else {
                    MustGetAgentActivityCompleteness::Complete
                }
            } else if has_gap || !reaches_genesis {
                MustGetAgentActivityCompleteness::IncompleteChain
            } else {
                MustGetAgentActivityCompleteness::Complete
            }
        }
    }
}

/// Assemble the [`AgentActivityResponse`] from the classified lists.
///
/// Generic over the item shape via [`ActionHashedContainer`]: the
/// `include_full_records` case passes
/// [`Record`](Record)s (whose `Vec` renders as
/// [`ChainItems::Full`]) and the hashes case passes the `HoloHashed<Action>`
/// shape (rendering as [`ChainItems::Hashes`]).
fn build_agent_activity_response<T>(
    agent: holo_hash::AgentPubKey,
    valid: Vec<T>,
    rejected: Vec<T>,
    warrants: Vec<SignedWarrant>,
    filter: &ChainQueryFilter,
    options: &crate::dht_store::GetAgentActivityOptions,
) -> AgentActivityResponse
where
    T: ActionHashedContainer + Clone,
    Vec<T>: ChainItemsSource,
{
    // `compute_chain_status` returns the lists sorted ascending by sequence, so
    // computing `highest_observed` from its output makes the "sorted input"
    // precondition structurally guaranteed rather than relying on the caller's
    // SQL ordering.
    let (status, valid, rejected) = compute_chain_status(valid.into_iter(), rejected.into_iter());

    // A chain whose head is a `CloseChain` action is reported as `Closed`. This
    // only upgrades a `Valid` status; `compute_chain_status` returns `Forked`
    // and `Invalid` ahead of `Valid`, so higher-priority states are preserved.
    let status = match status {
        ChainStatus::Valid(head)
            if matches!(
                valid.last().map(|v| &v.action().data),
                Some(ActionData::CloseChain(_))
            ) =>
        {
            ChainStatus::Closed(head)
        }
        other => other,
    };

    let highest_observed = compute_highest_observed(&valid, &rejected);

    let valid_activity = if options.include_valid_activity {
        filter.filter_actions(valid).to_chain_items()
    } else {
        ChainItems::NotRequested
    };
    let rejected_activity = if options.include_rejected_activity {
        filter.filter_actions(rejected).to_chain_items()
    } else {
        ChainItems::NotRequested
    };
    // Warrant gating is owned entirely by the caller, which passes an empty
    // `warrants` when `include_warrants` is false; this fn passes it through.
    AgentActivityResponse {
        agent,
        valid_activity,
        rejected_activity,
        warrants,
        status,
        highest_observed,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dht_store::{AppOutcome, GetAgentActivityOptions, SysOutcome};
    use holo_hash::{ActionHash, AgentPubKey, DhtOpHash, EntryHash, HoloHashed};
    use holochain_data::kind::Dht;
    use holochain_data::DbWrite;
    use holochain_types::chain::MustGetAgentActivityResponse;
    use holochain_types::op::{ChainOp, DhtOp, DhtOpHashed, OpEntry};
    use holochain_zome_types::prelude::{
        build_action, Action, ActionData, ActionHeader, AppEntryDef, ChainFilter, ChainTopOrdering,
        CloseChainData, CreateData, CreateLinkData, DeleteData, DeleteLinkData, DnaData,
        EntryVisibility, Signature, SignedAction, UpdateData,
    };
    use std::sync::Arc;
    use EntryType;

    #[test]
    fn invalid_status_carries_the_rejected_actions_own_head_not_the_valid_tip() {
        use holochain_zome_types::prelude::{ChainHead, ChainStatus};

        let valid_tip_hash = ActionHash::from_raw_36(vec![3; 36]);
        let rejected_hash = ActionHash::from_raw_36(vec![4; 36]);
        let valid = vec![
            (0u32, ActionHash::from_raw_36(vec![0; 36])),
            (3, valid_tip_hash.clone()),
        ];
        let rejected = vec![(4u32, rejected_hash.clone())];

        let (status, valid_out, rejected_out) =
            compute_chain_status(valid.into_iter(), rejected.into_iter());

        //  The chain head inside `Invalid` should be the first rejected action and not the valid
        //  chain head.
        assert_eq!(
            status,
            ChainStatus::Invalid(ChainHead {
                action_seq: 4,
                hash: rejected_hash,
            })
        );
        assert_eq!(valid_out.last().unwrap().1, valid_tip_hash);
        assert_eq!(rejected_out.len(), 1);
    }

    fn make_fork_op(author: &AgentPubKey, prev: &ActionHash, seq: u32, seed: u8) -> DhtOpHashed {
        let action = Action {
            header: ActionHeader {
                author: author.clone(),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: seq,
                prev_action: Some(prev.clone()),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]),
            }),
        };
        let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
        let chain_op = ChainOp::AgentActivity(signed_action);
        DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)))
    }

    fn dht_id() -> Dht {
        Dht::new(Arc::new(holo_hash::DnaHash::from_raw_36(vec![0u8; 36])))
    }

    fn make_chain_op(seed: u8, require_validation_receipt: bool) -> (DhtOpHashed, bool) {
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let action = Action {
            header: ActionHeader {
                author,
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(200); 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]),
            }),
        };
        let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
        let chain_op = ChainOp::AgentActivity(signed_action);
        (
            DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op))),
            require_validation_receipt,
        )
    }

    /// Build a synthetic `DhtOpHashed` with the given pre-computed hash.
    fn make_chain_op_with_hash(
        seed: u8,
        hash: DhtOpHash,
        require_validation_receipt: bool,
    ) -> (DhtOpHashed, bool) {
        let op = make_chain_op(seed, require_validation_receipt);
        (
            HoloHashed::with_pre_hashed(op.0.into_inner().0, hash),
            require_validation_receipt,
        )
    }

    /// Regression: a **private** entry must be `Hidden` from a non-author, even
    /// when the entry is retrievable by hash (here it sits in the public `Entry`
    /// table — the same exposure a same-hash private entry of the caller's own
    /// produces). The action itself stays visible.
    #[tokio::test]
    async fn retrieve_record_hides_private_entry_from_non_author() {
        use holochain_types::prelude::{AppEntryBytes, Entry, RecordEntry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let alice = AgentPubKey::from_raw_36(vec![1u8; 36]);
        let bobbo = AgentPubKey::from_raw_36(vec![2u8; 36]);

        let entry = Entry::App(AppEntryBytes(
            holochain_serialized_bytes::SerializedBytes::from(
                holochain_serialized_bytes::UnsafeBytes::from(vec![9u8; 8]),
            ),
        ));
        let entry_hash = EntryHash::with_data_sync(&entry);
        let action = Action {
            header: ActionHeader {
                author: alice.clone(),
                timestamp: Timestamp::from_micros(1000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![3u8; 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Private,
                )),
                entry_hash: entry_hash.clone(),
            }),
        };
        let action_hash = ActionHash::with_data_sync(&action);

        // Recording the CreateRecord op with the entry present lands the entry in
        // the public `Entry` table — the leak surface.
        let signed_action = SignedAction::new(action, Signature::from([7u8; 64]));
        let chain_op = ChainOp::CreateRecord(signed_action, OpEntry::Present(entry.clone()));
        let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)));
        store.record_incoming_ops(vec![(op, false)]).await.unwrap();

        // A non-author gets the record, but the private entry is Hidden.
        let record = store
            .as_read()
            .retrieve_record(&action_hash, Some(&bobbo))
            .await
            .unwrap()
            .expect("the action is public, so a record is returned");
        assert_eq!(
            *record.entry(),
            RecordEntry::Hidden,
            "a private entry must be Hidden from a non-author"
        );

        // The author sees their own private entry.
        let record = store
            .as_read()
            .retrieve_record(&action_hash, Some(&alice))
            .await
            .unwrap()
            .expect("author's record");
        assert!(
            matches!(*record.entry(), RecordEntry::Present(_)),
            "the author sees their own private entry"
        );
    }

    /// Builds the `(EntryHash, Entry, Action)` for a committed countersigning
    /// session at `seq` on `author`'s chain. The session is the minimal valid
    /// shape: two signing agents (`author` at index 0, `other` at index 1).
    fn countersigning_head(
        author: &AgentPubKey,
        other: &AgentPubKey,
        seq: u32,
        visibility: EntryVisibility,
    ) -> (
        EntryHash,
        holochain_types::prelude::Entry,
        Action,
        holochain_zome_types::prelude::PreflightRequest,
    ) {
        use holochain_types::prelude::{AppEntryBytes, Entry};
        use holochain_zome_types::prelude::{
            ActionBase, CounterSigningAgentState, CounterSigningSessionData,
            CounterSigningSessionTimes, CreateBase, PreflightBytes, PreflightRequest,
        };

        let app_entry_type = EntryType::App(AppEntryDef::new(0.into(), 0.into(), visibility));
        let session_times = CounterSigningSessionTimes::try_new(
            Timestamp::from_micros(1_000),
            Timestamp::from_micros(1_000_000_000),
        )
        .unwrap();
        let preflight_request = PreflightRequest::try_new(
            EntryHash::from_raw_36(vec![5u8; 36]),
            vec![(author.clone(), vec![]), (other.clone(), vec![])],
            vec![],
            0,
            false,
            session_times,
            ActionBase::Create(CreateBase::new(app_entry_type.clone())),
            PreflightBytes(vec![]),
        )
        .unwrap();
        let session_data = CounterSigningSessionData::try_new(
            preflight_request.clone(),
            vec![
                (
                    CounterSigningAgentState::new(0, ActionHash::from_raw_36(vec![10u8; 36]), 2),
                    Signature::from([0u8; 64]),
                ),
                (
                    CounterSigningAgentState::new(1, ActionHash::from_raw_36(vec![11u8; 36]), 2),
                    Signature::from([1u8; 64]),
                ),
            ],
            vec![],
        )
        .unwrap();

        let entry = Entry::CounterSign(
            Box::new(session_data),
            AppEntryBytes(holochain_serialized_bytes::SerializedBytes::from(
                holochain_serialized_bytes::UnsafeBytes::from(vec![1u8; 8]),
            )),
        );
        let entry_hash = EntryHash::with_data_sync(&entry);
        let action = Action {
            header: ActionHeader {
                author: author.clone(),
                timestamp: Timestamp::from_micros(2_000),
                action_seq: seq,
                prev_action: Some(ActionHash::from_raw_36(vec![3u8; 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: app_entry_type,
                entry_hash: entry_hash.clone(),
            }),
        };
        (entry_hash, entry, action, preflight_request)
    }

    /// Insert `action` and its `entry` as a committed self-authored chain head:
    /// an integrated op whose `Action` is written with `record_validity =
    /// Accepted` (so `chain_head_for_author`, which reads that state, sees the
    /// head) plus the entry in the store.
    ///
    /// This mirrors the source-chain flush, which writes each self-authored
    /// action as `Accepted`. `record_incoming_ops` instead inserts actions as
    /// pending (`NULL`), which are deliberately excluded from the chain head —
    /// so it is not a faithful stand-in for a committed self-authored head here.
    async fn insert_integrated_head(
        store: &DhtStore<DbWrite<Dht>>,
        action: Action,
        entry_hash: &EntryHash,
        entry: &holochain_types::prelude::Entry,
        private_author: Option<&AgentPubKey>,
    ) {
        let signed_action = SignedAction::new(action, Signature::from([7u8; 64]));
        let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(ChainOp::AgentActivity(
            signed_action,
        ))));
        store
            .test_insert_authored_chain_op(op, None, None, None, None)
            .await
            .unwrap();
        store
            .test_insert_entry(entry_hash, entry, private_author)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn current_countersigning_session_returns_session_at_chain_head() {
        use holochain_zome_types::prelude::EntryVisibility;

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let alice = AgentPubKey::from_raw_36(vec![1u8; 36]);
        let bob = AgentPubKey::from_raw_36(vec![2u8; 36]);

        let (cs_entry_hash, cs_entry, head_action, preflight_request) =
            countersigning_head(&alice, &bob, 3, EntryVisibility::Public);
        let head_hash = ActionHash::with_data_sync(&head_action);

        // Insert the head as a committed self-authored op: an integrated
        // `AgentActivity` op (so it is the chain head) plus the
        // `CounterSign` entry in the store. This is the shape a real
        // source-chain flush produces.
        insert_integrated_head(&store, head_action, &cs_entry_hash, &cs_entry, None).await;

        let (record, entry_hash, returned_session) = store
            .as_read()
            .current_countersigning_session(&alice)
            .await
            .unwrap()
            .expect("the chain head is a countersigning session");

        assert_eq!(entry_hash, cs_entry_hash);
        assert_eq!(record.action_address(), &head_hash);
        assert_eq!(
            returned_session.preflight_request().fingerprint().unwrap(),
            preflight_request.fingerprint().unwrap(),
        );
    }

    #[tokio::test]
    async fn current_countersigning_session_none_when_head_is_normal_entry() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let alice = AgentPubKey::from_raw_36(vec![4u8; 36]);

        // A normal app entry at the chain head (seq 1), inserted as an
        // integrated self-authored head so the head is genuinely found and the
        // `None` result comes from the entry being an ordinary App entry rather
        // than a `CounterSign`.
        let entry = holochain_types::prelude::Entry::App(holochain_types::prelude::AppEntryBytes(
            holochain_serialized_bytes::SerializedBytes::from(
                holochain_serialized_bytes::UnsafeBytes::from(vec![9u8; 8]),
            ),
        ));
        let entry_hash = EntryHash::with_data_sync(&entry);
        let action = Action {
            header: ActionHeader {
                author: alice.clone(),
                timestamp: Timestamp::from_micros(2_000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![3u8; 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: entry_hash.clone(),
            }),
        };
        insert_integrated_head(&store, action, &entry_hash, &entry, None).await;

        let result = store
            .as_read()
            .current_countersigning_session(&alice)
            .await
            .unwrap();
        assert!(
            result.is_none(),
            "a normal entry at the chain head is not a countersigning session"
        );
    }

    #[tokio::test]
    async fn current_countersigning_session_none_for_empty_chain() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let alice = AgentPubKey::from_raw_36(vec![6u8; 36]);

        let result = store
            .as_read()
            .current_countersigning_session(&alice)
            .await
            .unwrap();
        assert!(
            result.is_none(),
            "an empty chain has no countersigning session"
        );
    }

    #[tokio::test]
    async fn op_exists_returns_false_for_unknown_hash() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let unknown = DhtOpHash::from_raw_36(vec![99u8; 36]);
        let exists = store.as_read().op_exists(&unknown).await.unwrap();
        assert!(!exists);
    }

    #[tokio::test]
    async fn op_exists_returns_true_after_record_incoming_ops() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(1, false);
        let hash = op.0.as_hash().clone();

        store.record_incoming_ops(vec![op]).await.unwrap();

        let exists = store.as_read().op_exists(&hash).await.unwrap();
        assert!(exists, "op_exists should be true after record_incoming_ops");
    }

    #[tokio::test]
    async fn filter_existing_ops_removes_known_hashes() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let known = make_chain_op(2, false);
        let unknown = make_chain_op(3, false);
        let known_hash = known.0.as_hash().clone();
        let unknown_hash = unknown.0.as_hash().clone();

        store.record_incoming_ops(vec![known]).await.unwrap();

        let input = vec![
            make_chain_op_with_hash(20, known_hash.clone(), false),
            unknown,
        ];
        let filtered = store.as_read().filter_existing_ops(input).await.unwrap();
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0].0.as_hash(), &unknown_hash);
    }

    #[tokio::test]
    async fn ops_pending_sys_validation_returns_recorded_chain_op() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(10, false);
        let hash = op.0.as_hash().clone();

        store.record_incoming_ops(vec![op]).await.unwrap();

        let pending = store
            .as_read()
            .ops_pending_sys_validation(1_000)
            .await
            .unwrap();
        let hashes: Vec<_> = pending.iter().map(|o| o.as_hash().clone()).collect();
        assert!(hashes.contains(&hash));
    }

    #[tokio::test]
    async fn ops_pending_sys_validation_excludes_completed() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(11, false);
        let hash = op.0.as_hash().clone();

        store.record_incoming_ops(vec![op]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();

        let pending = store
            .as_read()
            .ops_pending_sys_validation(1_000)
            .await
            .unwrap();
        let hashes: Vec<_> = pending.iter().map(|o| o.as_hash().clone()).collect();
        assert!(!hashes.contains(&hash));
    }

    #[tokio::test]
    async fn ops_pending_sys_validation_respects_limit_across_union() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let ops: Vec<_> = (12..16).map(|seed| make_chain_op(seed, false)).collect();
        store.record_incoming_ops(ops).await.unwrap();

        let pending = store.as_read().ops_pending_sys_validation(2).await.unwrap();
        assert_eq!(pending.len(), 2);
    }

    #[tokio::test]
    async fn ops_pending_app_validation_returns_sys_validated_chain_op() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(50, false);
        let hash = op.0.as_hash().clone();

        store.record_incoming_ops(vec![op]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();

        let pending = store
            .as_read()
            .ops_pending_app_validation(1_000)
            .await
            .unwrap();
        let hashes: Vec<_> = pending.iter().map(|o| o.as_hash().clone()).collect();
        assert!(hashes.contains(&hash));
    }

    #[tokio::test]
    async fn ops_pending_app_validation_excludes_pending_sys() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(51, false);
        let hash = op.0.as_hash().clone();

        // Insert but don't record sys-validation outcome.
        store.record_incoming_ops(vec![op]).await.unwrap();

        let pending = store
            .as_read()
            .ops_pending_app_validation(1_000)
            .await
            .unwrap();
        let hashes: Vec<_> = pending.iter().map(|o| o.as_hash().clone()).collect();
        assert!(
            !hashes.contains(&hash),
            "op not yet sys-validated should not appear"
        );
    }

    #[tokio::test]
    async fn ops_pending_app_validation_excludes_app_validated() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(52, false);
        let hash = op.0.as_hash().clone();

        store.record_incoming_ops(vec![op]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(hash.clone(), AppOutcome::Accepted)])
            .await
            .unwrap();

        let pending = store
            .as_read()
            .ops_pending_app_validation(1_000)
            .await
            .unwrap();
        let hashes: Vec<_> = pending.iter().map(|o| o.as_hash().clone()).collect();
        assert!(
            !hashes.contains(&hash),
            "fully-validated op should not appear"
        );
    }

    #[tokio::test]
    async fn find_fork_for_action_returns_none_when_no_sibling() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(30, false);
        let action = match op.0.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().data().clone(),
            _ => unreachable!(),
        };

        let result = store.as_read().find_fork_for_action(&action).await.unwrap();
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn find_fork_for_action_returns_sibling() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let author = AgentPubKey::from_raw_36(vec![31u8; 36]);
        let prev_action_hash = ActionHash::from_raw_36(vec![231u8; 36]);

        // Two ops sharing the same prev_action and author but with different
        // timestamps/entry_hashes (seeds differ) — different hashes overall.
        let op_a = make_fork_op(&author, &prev_action_hash, 2, 32);
        let op_b = make_fork_op(&author, &prev_action_hash, 2, 33);

        // Capture op_a's action hash and signature before moving op_a into the store.
        let expected_hash = match op_a.as_content() {
            DhtOp::ChainOp(c) => ActionHash::with_data_sync(c.signed_action().data()),
            _ => unreachable!(),
        };
        let expected_sig = match op_a.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().signature().clone(),
            _ => unreachable!(),
        };

        let action_b = match op_b.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().data().clone(),
            _ => unreachable!(),
        };

        store
            .record_incoming_ops(vec![(op_a, false)])
            .await
            .unwrap();

        let result = store
            .as_read()
            .find_fork_for_action(&action_b)
            .await
            .unwrap();
        let (got_hash, got_sig) = result.expect("fork should be detected");
        assert_eq!(got_hash, expected_hash, "sibling hash should match op_a");
        assert_eq!(got_sig, expected_sig, "sibling signature should match op_a");
    }

    #[tokio::test]
    async fn pending_validation_receipts_returns_integrated_require_receipt_ops() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(60, true);
        let hash = op.0.as_hash().clone();
        let author = match op.0.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().data().author().clone(),
            _ => unreachable!(),
        };

        store.record_incoming_ops(vec![op]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(hash.clone(), AppOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(holochain_types::prelude::Timestamp::now())
            .await
            .unwrap();

        let validators = vec![AgentPubKey::from_raw_36(vec![0xFF; 36])];
        let receipts = store
            .as_read()
            .pending_validation_receipts(validators.clone())
            .await
            .unwrap();

        assert_eq!(receipts.len(), 1);
        assert_eq!(receipts[0].0.dht_op_hash, hash);
        assert_eq!(receipts[0].1, author);
        assert_eq!(receipts[0].0.validators, validators);
    }

    async fn integrate_activity(
        store: &crate::dht_store::DhtStore<DbWrite<Dht>>,
        op: DhtOpHashed,
        app: AppOutcome,
        when: i64,
    ) {
        let hash = op.as_hash().clone();
        store.record_incoming_ops(vec![(op, false)]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(hash, app)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(Timestamp::from_micros(when))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn get_agent_activity_valid_chain_hashes() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![42u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;
        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 1, 2),
            AppOutcome::Accepted,
            11,
        )
        .await;

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: true,
            include_warrants: false,
            include_full_records: false,
        };
        let resp: AgentActivityResponse = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();

        assert_eq!(resp.agent, author);
        match resp.valid_activity {
            ChainItems::Hashes(h) => {
                assert_eq!(h.len(), 2);
                assert_eq!(h[0].0, 0);
                assert_eq!(h[1].0, 1);
            }
            other => panic!("expected Hashes, got {other:?}"),
        }
        assert!(matches!(resp.rejected_activity, ChainItems::Hashes(ref h) if h.is_empty()));
        assert!(matches!(resp.status, ChainStatus::Valid(ref head) if head.action_seq == 1));
        let ho = resp.highest_observed.expect("highest observed");
        assert_eq!(ho.action_seq, 1);
    }

    #[tokio::test]
    async fn get_agent_activity_rejected_marks_invalid() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![7u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;
        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 1, 2),
            AppOutcome::Rejected,
            11,
        )
        .await;

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: true,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();

        assert!(matches!(resp.valid_activity, ChainItems::Hashes(ref h) if h.len() == 1));
        match resp.rejected_activity {
            ChainItems::Hashes(h) => {
                assert_eq!(h.len(), 1);
                assert_eq!(h[0].0, 1);
            }
            other => panic!("expected Hashes, got {other:?}"),
        }
        assert!(matches!(resp.status, ChainStatus::Invalid(ref head) if head.action_seq == 1));
    }

    #[tokio::test]
    async fn get_agent_activity_excludes_withheld_publish_ops() {
        // A self-authored op whose `ChainOpPublish.withhold_publish` is set
        // (an in-flight countersigning session) must NOT surface as live agent
        // activity until the session completes and the withhold flag is cleared.
        // Ordinary self-authored ops (withhold NULL) are always visible.
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![23u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        // seqs 0 and 1: ordinary self-authored ops (withhold NULL -> visible).
        store
            .test_insert_authored_chain_op(
                make_fork_op(&author, &prev, 0, 1),
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();
        store
            .test_insert_authored_chain_op(
                make_fork_op(&author, &prev, 1, 2),
                None,
                None,
                None,
                None,
            )
            .await
            .unwrap();
        // seq 2: a withheld countersigning op (withhold_publish = true -> hidden).
        let withheld = make_fork_op(&author, &prev, 2, 3);
        let withheld_hash = withheld.as_hash().clone();
        store
            .test_insert_authored_chain_op(withheld, None, None, None, Some(true))
            .await
            .unwrap();

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };

        // While withheld, only seqs 0 and 1 are reported.
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();
        match resp.valid_activity {
            ChainItems::Hashes(h) => {
                assert_eq!(h.len(), 2, "withheld op must be hidden");
                assert_eq!(h[0].0, 0);
                assert_eq!(h[1].0, 1);
            }
            other => panic!("expected Hashes, got {other:?}"),
        }

        // Clearing the withhold flag (as session completion does) reveals seq 2.
        store
            .test_set_chain_op_publish(&withheld_hash, None, None, None)
            .await
            .unwrap();
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();
        match resp.valid_activity {
            ChainItems::Hashes(h) => {
                assert_eq!(
                    h.len(),
                    3,
                    "revealed op must be visible after clearing withhold"
                );
                assert_eq!(h[2].0, 2);
            }
            other => panic!("expected Hashes, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn get_agent_activity_detects_fork() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![9u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;
        // Two valid actions both at seq 1 -> fork.
        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 1, 2),
            AppOutcome::Accepted,
            11,
        )
        .await;
        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 1, 3),
            AppOutcome::Accepted,
            12,
        )
        .await;

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();

        // The meaningful signal is fork detection: two valid actions at the
        // same seq yield Forked. (highest_observed reports the top sequence, not
        // necessarily every tip's hash.)
        match resp.status {
            ChainStatus::Forked(fork) => assert_eq!(fork.fork_seq, 1),
            other => panic!("expected Forked, got {other:?}"),
        }
        let ho = resp.highest_observed.expect("highest observed");
        assert_eq!(ho.action_seq, 1);
    }

    #[tokio::test]
    async fn get_agent_activity_full_returns_records() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![5u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: true,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();

        match resp.valid_activity {
            ChainItems::Full(records) => assert_eq!(records.len(), 1),
            other => panic!("expected Full, got {other:?}"),
        }
    }

    fn make_warrant_for(warrantee: &AgentPubKey, seed: u8) -> DhtOpHashed {
        use holochain_types::warrant::WarrantOp;
        use holochain_zome_types::prelude::{
            ChainIntegrityWarrant, ChainOpType, SignedWarrant, Warrant, WarrantProof,
        };

        let action_author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let action_hash = ActionHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let warrant = SignedWarrant::new(
            Warrant::new(
                WarrantProof::ChainIntegrity(ChainIntegrityWarrant::InvalidChainOp {
                    action_author,
                    action: (action_hash, Signature::from([seed; 64])),
                    chain_op_type: ChainOpType::CreateRecord,
                    reason: "test warrant".into(),
                }),
                AgentPubKey::from_raw_36(vec![seed.wrapping_add(10); 36]),
                Timestamp::from_micros(seed as i64 * 1000),
                warrantee.clone(),
            ),
            Signature::from([seed.wrapping_add(1); 64]),
        );
        DhtOpHashed::from_content_sync(DhtOp::WarrantOp(Box::new(WarrantOp::from(warrant))))
    }

    #[tokio::test]
    async fn get_agent_activity_attaches_warrants_when_requested() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![21u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;

        // Integrate a warrant whose warrantee is this author.
        let warrant = make_warrant_for(&author, 30);
        let wh = warrant.as_hash().clone();
        store
            .record_incoming_ops(vec![(warrant, false)])
            .await
            .unwrap();
        store
            .record_warrant_sys_validation_outcomes(vec![(wh, SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(Timestamp::from_micros(20))
            .await
            .unwrap();

        // include_warrants = true -> attached.
        let with = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: true,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &with)
            .await
            .unwrap();
        assert_eq!(resp.warrants.len(), 1);

        // include_warrants = false -> empty, and valid not requested.
        let without = GetAgentActivityOptions {
            include_valid_activity: false,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &without)
            .await
            .unwrap();
        assert!(resp.warrants.is_empty());
        assert!(matches!(resp.valid_activity, ChainItems::NotRequested));
    }

    // ---- get_agent_activity_with_scratch tests ----

    /// Build a scratch `SignedActionHashed` for a `Create` by `author`
    /// at the given `action_seq`, linked from `prev`.
    fn make_scratch_create(
        author: &AgentPubKey,
        seq: u32,
        prev: &ActionHash,
        seed: u8,
    ) -> SignedActionHashed {
        let action = build_action(
            ActionHeader {
                author: author.clone(),
                timestamp: Timestamp::from_micros(seed as i64 * 10_000),
                action_seq: seq,
                prev_action: Some(prev.clone()),
            },
            ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: EntryHash::from_raw_36(vec![seed.wrapping_add(200); 36]),
            }),
        );
        let action_hashed = holo_hash::HoloHashed::from_content_sync(action);
        SignedActionHashed::with_presigned(action_hashed, Signature::from([seed; 64]))
    }

    /// Build a scratch `SignedWarrant` whose `InvalidChainOp.action_author` == `warrantee`.
    fn make_scratch_warrant_for(warrantee: &AgentPubKey, seed: u8) -> SignedWarrant {
        use holochain_zome_types::prelude::{
            ChainIntegrityWarrant, ChainOpType, SignedWarrant, Warrant, WarrantProof,
        };

        let action_author = warrantee.clone();
        let action_hash = ActionHash::from_raw_36(vec![seed.wrapping_add(150); 36]);
        SignedWarrant::new(
            Warrant::new(
                WarrantProof::ChainIntegrity(ChainIntegrityWarrant::InvalidChainOp {
                    action_author,
                    action: (action_hash, Signature::from([seed; 64])),
                    chain_op_type: ChainOpType::CreateRecord,
                    reason: "scratch warrant".into(),
                }),
                AgentPubKey::from_raw_36(vec![seed.wrapping_add(50); 36]),
                Timestamp::from_micros(seed as i64 * 5_000),
                warrantee.clone(),
            ),
            Signature::from([seed.wrapping_add(1); 64]),
        )
    }

    /// A scratch action for `author` appears in the valid activity list of
    /// `get_agent_activity_with_scratch` and shifts `highest_observed`.
    #[tokio::test]
    async fn get_agent_activity_with_scratch_includes_scratch_action() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![110u8; 36]);
        let prev_hash = ActionHash::from_raw_36(vec![0u8; 36]);

        // One integrated action at seq 0.
        integrate_activity(
            &store,
            make_fork_op(&author, &prev_hash, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;

        // Scratch action at seq 1 (linked from some hash — only seq matters here).
        let scratch_prev = ActionHash::from_raw_36(vec![111u8; 36]);
        let scratch_sah = make_scratch_create(&author, 1, &scratch_prev, 111);
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        let sync_scratch = scratch.into_sync();

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity_with_scratch(
                &author,
                &ChainQueryFilter::new(),
                &opts,
                &sync_scratch,
            )
            .await
            .unwrap();

        // Both seq 0 (store) and seq 1 (scratch) should be valid.
        match &resp.valid_activity {
            ChainItems::Hashes(h) => {
                assert_eq!(h.len(), 2, "expected both store and scratch action");
                let seqs: Vec<u32> = h.iter().map(|(seq, _)| *seq).collect();
                assert!(seqs.contains(&0), "seq 0 (store) should be present");
                assert!(seqs.contains(&1), "seq 1 (scratch) should be present");
            }
            other => panic!("expected Hashes, got {other:?}"),
        }
        // highest_observed should reflect seq 1 from the scratch.
        let ho = resp
            .highest_observed
            .expect("highest_observed should be set");
        assert_eq!(
            ho.action_seq, 1,
            "highest_observed should include scratch action"
        );
    }

    /// A scratch action for `author` appears in the full-records list.
    #[tokio::test]
    async fn get_agent_activity_with_scratch_full_records_includes_scratch_action() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![112u8; 36]);
        let prev_hash = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev_hash, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;

        let scratch_prev = ActionHash::from_raw_36(vec![113u8; 36]);
        let scratch_sah = make_scratch_create(&author, 1, &scratch_prev, 113);
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        let sync_scratch = scratch.into_sync();

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: true,
        };
        let resp = store
            .as_read()
            .get_agent_activity_with_scratch(
                &author,
                &ChainQueryFilter::new(),
                &opts,
                &sync_scratch,
            )
            .await
            .unwrap();

        match &resp.valid_activity {
            ChainItems::Full(records) => {
                assert_eq!(records.len(), 2, "expected store record + scratch record");
            }
            other => panic!("expected Full, got {other:?}"),
        }
    }

    /// A scratch warrant targeting `author` appears in the response when
    /// `include_warrants = true`.
    #[tokio::test]
    async fn get_agent_activity_with_scratch_includes_scratch_warrant() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![114u8; 36]);
        let prev_hash = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev_hash, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;

        // Scratch contains one warrant for `author`.
        let signed_warrant = make_scratch_warrant_for(&author, 114);
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_warrant(signed_warrant);
        let sync_scratch = scratch.into_sync();

        let opts_with = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: true,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity_with_scratch(
                &author,
                &ChainQueryFilter::new(),
                &opts_with,
                &sync_scratch,
            )
            .await
            .unwrap();
        assert_eq!(resp.warrants.len(), 1, "scratch warrant should be present");

        // When include_warrants = false the scratch warrant should be absent.
        let opts_without = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity_with_scratch(
                &author,
                &ChainQueryFilter::new(),
                &opts_without,
                &sync_scratch,
            )
            .await
            .unwrap();
        assert!(
            resp.warrants.is_empty(),
            "warrants should be empty when not requested"
        );
    }

    /// Store-only `get_agent_activity` ignores a populated scratch.
    #[tokio::test]
    async fn get_agent_activity_store_only_ignores_scratch() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![115u8; 36]);
        let prev_hash = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev_hash, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;

        // Scratch action at seq 1 — must NOT appear in the store-only read.
        let scratch_prev = ActionHash::from_raw_36(vec![116u8; 36]);
        let scratch_sah = make_scratch_create(&author, 1, &scratch_prev, 116);
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        // (scratch is intentionally unused in the store-only call)
        let _ = scratch;

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();

        match &resp.valid_activity {
            ChainItems::Hashes(h) => {
                assert_eq!(
                    h.len(),
                    1,
                    "store-only read must not include scratch actions"
                );
                assert_eq!(h[0].0, 0, "only seq 0 from the store");
            }
            other => panic!("expected Hashes, got {other:?}"),
        }
    }

    /// Build a linked chain of `len` `AgentActivity` ops for `author`
    /// (seq 0..len, each `prev_action` = the previous action's hash). Returns the
    /// ops and the action hashes (index = seq).
    fn make_activity_chain(author: &AgentPubKey, len: u32) -> (Vec<DhtOpHashed>, Vec<ActionHash>) {
        let mut ops = Vec::new();
        let mut hashes = Vec::new();
        let mut prev = ActionHash::from_raw_36(vec![0u8; 36]);
        for seq in 0..len {
            let action = Action {
                header: ActionHeader {
                    author: author.clone(),
                    timestamp: Timestamp::from_micros((seq as i64 + 1) * 1000),
                    action_seq: seq,
                    prev_action: Some(prev.clone()),
                },
                data: ActionData::Create(CreateData {
                    entry_type: EntryType::App(AppEntryDef::new(
                        0.into(),
                        0.into(),
                        EntryVisibility::Public,
                    )),
                    entry_hash: EntryHash::from_raw_36(vec![(seq as u8).wrapping_add(100); 36]),
                }),
            };
            let action_hash = holo_hash::ActionHash::with_data_sync(&action);
            let signed_action = SignedAction::new(action, Signature::from([seq as u8; 64]));
            let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(
                ChainOp::AgentActivity(signed_action),
            )));
            prev = action_hash.clone();
            hashes.push(action_hash);
            ops.push(op);
        }
        (ops, hashes)
    }

    #[tokio::test]
    async fn must_get_agent_activity_to_genesis_complete() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![71u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 3);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }

        // A warrant for this author should be attached to a Complete response.
        let warrant = make_warrant_for(&author, 40);
        let wh = warrant.as_hash().clone();
        store
            .record_incoming_ops(vec![(warrant, false)])
            .await
            .unwrap();
        store
            .record_warrant_sys_validation_outcomes(vec![(wh, SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(Timestamp::from_micros(99))
            .await
            .unwrap();

        let filter = ChainFilter::new(hashes[2].clone());
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await
            .unwrap();

        match resp {
            MustGetAgentActivityResponse::Activity { activity, warrants } => {
                assert_eq!(activity.len(), 3);
                // Returned newest-first (seq DESC).
                assert_eq!(activity[0].action.action().action_seq(), 2);
                assert_eq!(activity[2].action.action().action_seq(), 0);
                assert_eq!(warrants.len(), 1);
            }
            other => panic!("expected Activity, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn must_get_agent_activity_chain_top_not_found() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![72u8; 36]);
        let (ops, _hashes) = make_activity_chain(&author, 2);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }
        let unknown = ActionHash::from_raw_36(vec![88u8; 36]);
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &ChainFilter::new(unknown.clone()))
            .await
            .unwrap();
        assert!(matches!(
            resp,
            MustGetAgentActivityResponse::ChainTopNotFound(h) if h == unknown
        ));
    }

    #[tokio::test]
    async fn must_get_agent_activity_until_hash_complete() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![73u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 3);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }
        // until = seq 1: expect seq 2 and 1, Complete.
        let filter = ChainFilter::until_hash(hashes[2].clone(), hashes[1].clone());
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await
            .unwrap();
        match resp {
            MustGetAgentActivityResponse::Activity { activity, .. } => {
                assert_eq!(activity.len(), 2);
                assert_eq!(activity[0].action.action().action_seq(), 2);
                assert_eq!(activity[1].action.action().action_seq(), 1);
            }
            other => panic!("expected Activity, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn must_get_agent_activity_until_hash_missing() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![74u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 2);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }
        // until_hash references an action not in this chain.
        let missing = ActionHash::from_raw_36(vec![77u8; 36]);
        let filter = ChainFilter::until_hash(hashes[1].clone(), missing.clone());
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await
            .unwrap();
        assert!(matches!(
            resp,
            MustGetAgentActivityResponse::UntilHashMissing(h) if h == missing
        ));
    }

    #[tokio::test]
    async fn must_get_agent_activity_take_complete() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![75u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 3);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }
        let filter = ChainFilter::take(hashes[2].clone(), 2);
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await
            .unwrap();
        match resp {
            MustGetAgentActivityResponse::Activity { activity, .. } => {
                assert_eq!(activity.len(), 2);
                assert_eq!(activity[0].action.action().action_seq(), 2);
                assert_eq!(activity[1].action.action().action_seq(), 1);
            }
            other => panic!("expected Activity, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn must_get_agent_activity_take_zero_errors() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![76u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 1);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }
        let filter = ChainFilter::take(hashes[0].clone(), 0);
        let result = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn must_get_agent_activity_gap_is_incomplete() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![78u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 3);
        // Integrate seq 0 and seq 2, skip seq 1 -> the chain top cannot reach genesis.
        integrate_activity(&store, ops[0].clone(), AppOutcome::Accepted, 10).await;
        integrate_activity(&store, ops[2].clone(), AppOutcome::Accepted, 12).await;
        let filter = ChainFilter::new(hashes[2].clone());
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await
            .unwrap();
        assert!(matches!(
            resp,
            MustGetAgentActivityResponse::IncompleteChain
        ));
    }

    // ---- must_get_agent_activity_with_scratch tests ----

    /// Scratch-only chain top resolves correctly: the store has no such action
    /// but the scratch does, so the response is not `ChainTopNotFound`.
    #[tokio::test]
    async fn must_get_agent_activity_with_scratch_chain_top_from_scratch() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![150u8; 36]);

        // Build a 1-action chain and integrate seq 0 into the store.
        let (ops, hashes) = make_activity_chain(&author, 1);
        integrate_activity(&store, ops[0].clone(), AppOutcome::Accepted, 10).await;

        // The scratch action at seq 1 is linked to hashes[0] (the store action).
        // Its content-derived hash is what we use as chain_top.
        let scratch_sah = make_scratch_create(&author, 1, &hashes[0], 150);
        let scratch_chain_top = scratch_sah.as_hash().clone();
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        let sync_scratch = scratch.into_sync();

        // filter.chain_top == scratch_chain_top (scratch-only action hash).
        let filter = ChainFilter::new(scratch_chain_top.clone());
        let resp = store
            .as_read()
            .must_get_agent_activity_with_scratch(&author, &filter, &sync_scratch)
            .await
            .unwrap();

        // The scratch resolved the chain top, so the result must not be
        // ChainTopNotFound.
        assert!(
            !matches!(resp, MustGetAgentActivityResponse::ChainTopNotFound(_)),
            "chain top should have been resolved from the scratch"
        );
    }

    /// A scratch-authored action within the bounded range appears in the
    /// merged activity of the response.
    #[tokio::test]
    async fn must_get_agent_activity_with_scratch_includes_scratch_activity() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![151u8; 36]);

        // Build a 3-action chain (seqs 0..=2) in the store.
        let (ops, hashes) = make_activity_chain(&author, 3);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }

        // Scratch action at seq 3, linked to the store action at seq 2 (hashes[2]).
        // The scratch action's content-derived hash becomes the chain_top for the filter.
        let scratch_sah = make_scratch_create(&author, 3, &hashes[2], 151);
        let scratch_hash = scratch_sah.as_hash().clone();
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        let sync_scratch = scratch.into_sync();

        // Use the scratch action's hash as chain_top so chain_top_seq == 3.
        let filter = ChainFilter::new(scratch_hash.clone());
        let resp = store
            .as_read()
            .must_get_agent_activity_with_scratch(&author, &filter, &sync_scratch)
            .await
            .unwrap();

        match resp {
            MustGetAgentActivityResponse::Activity { activity, .. } => {
                let seqs: Vec<u32> = activity
                    .iter()
                    .map(|a| a.action.action().action_seq())
                    .collect();
                assert!(
                    seqs.contains(&3),
                    "scratch action at seq 3 should be in merged activity; got {seqs:?}"
                );
                assert!(
                    seqs.contains(&2),
                    "store action at seq 2 should be in merged activity; got {seqs:?}"
                );
            }
            other => panic!("expected Activity, got {other:?}"),
        }
    }

    /// Store-only `must_get_agent_activity` ignores a populated scratch.
    #[tokio::test]
    async fn must_get_agent_activity_store_only_ignores_scratch() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![152u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 3);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }

        // Scratch action at seq 3 — must NOT appear in the store-only read.
        let scratch_sah = make_scratch_create(&author, 3, &hashes[2], 152);
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        // scratch intentionally not passed to the store-only call
        let _ = scratch;

        let filter = ChainFilter::new(hashes[2].clone());
        let resp = store
            .as_read()
            .must_get_agent_activity(&author, &filter)
            .await
            .unwrap();

        match resp {
            MustGetAgentActivityResponse::Activity { activity, .. } => {
                assert_eq!(
                    activity.len(),
                    3,
                    "store-only read must not include scratch actions"
                );
                let seqs: Vec<u32> = activity
                    .iter()
                    .map(|a| a.action.action().action_seq())
                    .collect();
                assert!(
                    !seqs.contains(&3),
                    "scratch action at seq 3 must not appear"
                );
            }
            other => panic!("expected Activity, got {other:?}"),
        }
    }

    /// take == 0 returns `InvalidInput` with a scratch present.
    #[tokio::test]
    async fn must_get_agent_activity_with_scratch_take_zero_errors() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![153u8; 36]);
        let (ops, hashes) = make_activity_chain(&author, 1);
        for (i, op) in ops.into_iter().enumerate() {
            integrate_activity(&store, op, AppOutcome::Accepted, 10 + i as i64).await;
        }
        let filter = ChainFilter::take(hashes[0].clone(), 0);
        let empty_scratch = crate::scratch::Scratch::new().into_sync();
        let result = store
            .as_read()
            .must_get_agent_activity_with_scratch(&author, &filter, &empty_scratch)
            .await;
        assert!(result.is_err(), "take == 0 must return an error");
    }

    /// chain_top absent from both store and scratch → `ChainTopNotFound`.
    #[tokio::test]
    async fn must_get_agent_activity_with_scratch_chain_top_absent_in_both() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![154u8; 36]);
        let unknown = ActionHash::from_raw_36(vec![99u8; 36]);
        let empty_scratch = crate::scratch::Scratch::new().into_sync();
        let resp = store
            .as_read()
            .must_get_agent_activity_with_scratch(
                &author,
                &ChainFilter::new(unknown.clone()),
                &empty_scratch,
            )
            .await
            .unwrap();
        assert!(
            matches!(resp, MustGetAgentActivityResponse::ChainTopNotFound(h) if h == unknown),
            "chain top absent from both sources must yield ChainTopNotFound"
        );
    }

    #[tokio::test]
    async fn pending_validation_receipts_excludes_ops_without_require_receipt() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(61, false);
        let hash = op.0.as_hash().clone();
        store.record_incoming_ops(vec![op]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(hash.clone(), AppOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(holochain_types::prelude::Timestamp::now())
            .await
            .unwrap();
        store
            .clear_require_receipts(vec![hash.clone()])
            .await
            .unwrap();

        let receipts = store
            .as_read()
            .pending_validation_receipts(vec![])
            .await
            .unwrap();
        assert!(
            receipts.iter().all(|(r, _)| r.dht_op_hash != hash),
            "op with cleared require_receipt should not appear"
        );
    }

    // ---- scratch overlay helpers ----

    /// Build a `SignedActionHashed` for a `Create` that references
    /// `entry_hash`. Useful for scratch overlay tests.
    fn make_signed_action_for_entry(seed: u8, entry_hash: EntryHash) -> SignedActionHashed {
        let action = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed; 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(200); 36])),
            },
            ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash,
            }),
        );
        let action_hashed = holo_hash::HoloHashed::from_content_sync(action);
        SignedActionHashed::with_presigned(action_hashed, Signature::from([seed; 64]))
    }

    /// Build a `SignedActionHashed` for a `Create` with no entry
    /// (simulates an action that is in the scratch but whose entry is missing).
    fn make_signed_action_no_entry(seed: u8) -> SignedActionHashed {
        // Use a Dna action — it carries no entry hash.
        let action = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed; 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 0,
                prev_action: None,
            },
            ActionData::Dna(DnaData {
                dna_hash: holo_hash::DnaHash::from_raw_36(vec![seed; 36]),
            }),
        );
        let action_hashed = holo_hash::HoloHashed::from_content_sync(action);
        SignedActionHashed::with_presigned(action_hashed, Signature::from([seed; 64]))
    }

    /// Build a scratch containing a single action.
    fn scratch_with_action(sah: SignedActionHashed) -> crate::scratch::SyncScratch {
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(sah, ChainTopOrdering::Relaxed);
        scratch.into_sync()
    }

    /// Build a scratch containing an action + its entry.
    fn scratch_with_action_and_entry(
        sah: SignedActionHashed,
        entry: holochain_types::prelude::Entry,
        entry_hash: EntryHash,
    ) -> crate::scratch::SyncScratch {
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(sah, ChainTopOrdering::Relaxed);
        let entry_hashed =
            holochain_types::prelude::EntryHashed::with_pre_hashed(entry, entry_hash);
        scratch.add_entry(entry_hashed, ChainTopOrdering::Relaxed);
        scratch.into_sync()
    }

    // ---- scratch overlay tests ----

    /// (a) action present only in store → `retrieve_action_with_scratch` returns it.
    #[tokio::test]
    async fn retrieve_action_with_scratch_finds_store_action() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let op = make_chain_op(90, false);
        let action = match op.0.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().data().clone(),
            _ => unreachable!(),
        };
        let action_hash = holo_hash::ActionHash::with_data_sync(&action);
        store.record_incoming_ops(vec![op]).await.unwrap();

        // Empty scratch — action lives only in the store.
        let empty_scratch = crate::scratch::Scratch::new().into_sync();
        let result = store
            .as_read()
            .retrieve_action_with_scratch(&action_hash, &empty_scratch)
            .await
            .unwrap();
        assert!(result.is_some(), "should find store-only action");
        assert_eq!(result.unwrap().as_hash(), &action_hash);
    }

    /// (b) action present only in scratch → `retrieve_action_with_scratch` returns it.
    #[tokio::test]
    async fn retrieve_action_with_scratch_finds_scratch_action() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let entry_hash = EntryHash::from_raw_36(vec![91u8.wrapping_add(100); 36]);
        let sah = make_signed_action_for_entry(91, entry_hash);
        let action_hash = sah.as_hash().clone();
        let scratch = scratch_with_action(sah);

        // The action was never written to the store.
        let result = store
            .as_read()
            .retrieve_action_with_scratch(&action_hash, &scratch)
            .await
            .unwrap();
        assert!(result.is_some(), "should find scratch-only action");
        assert_eq!(result.unwrap().as_hash(), &action_hash);
    }

    /// (d) store-only `retrieve_action` ignores a scratch-only action.
    #[tokio::test]
    async fn retrieve_action_ignores_scratch() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let entry_hash = EntryHash::from_raw_36(vec![92u8.wrapping_add(100); 36]);
        let sah = make_signed_action_for_entry(92, entry_hash);
        let action_hash = sah.as_hash().clone();

        // The action exists only in the scratch — not in the store.
        let result = store.as_read().retrieve_action(&action_hash).await.unwrap();
        assert!(
            result.is_none(),
            "store-only retrieve_action must not see scratch data"
        );
    }

    /// entry present only in store → `retrieve_entry_with_scratch` returns it.
    #[tokio::test]
    async fn retrieve_entry_with_scratch_finds_store_entry() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        // Build a CreateEntry op that carries a public entry.
        let seed = 93u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);

        let action = Action {
            header: ActionHeader {
                author,
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(200); 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: entry_hash.clone(),
            }),
        };
        let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
        let chain_op = ChainOp::CreateEntry(signed_action, OpEntry::Present(entry.clone()));
        let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)));
        store.record_incoming_ops(vec![(op, false)]).await.unwrap();

        let empty_scratch = crate::scratch::Scratch::new().into_sync();
        let result = store
            .as_read()
            .retrieve_entry_with_scratch(&entry_hash, None, &empty_scratch)
            .await
            .unwrap();
        assert!(result.is_some(), "should find store-only entry");
    }

    /// entry present only in scratch → `retrieve_entry_with_scratch` returns it.
    #[tokio::test]
    async fn retrieve_entry_with_scratch_finds_scratch_entry() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 94u8;
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let sah = make_signed_action_for_entry(seed, entry_hash.clone());
        let scratch = scratch_with_action_and_entry(sah, entry.clone(), entry_hash.clone());

        let result = store
            .as_read()
            .retrieve_entry_with_scratch(&entry_hash, None, &scratch)
            .await
            .unwrap();
        assert!(result.is_some(), "should find scratch-only entry");
    }

    /// `retrieve_record_with_scratch` with action in scratch + entry in scratch.
    #[tokio::test]
    async fn retrieve_record_with_scratch_action_and_entry_in_scratch() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 95u8;
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let sah = make_signed_action_for_entry(seed, entry_hash.clone());
        let action_hash = sah.as_hash().clone();
        let scratch = scratch_with_action_and_entry(sah, entry, entry_hash);

        let result = store
            .as_read()
            .retrieve_record_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        assert!(result.is_some(), "record with action+entry both in scratch");
    }

    /// (c) `retrieve_record_with_scratch` with action in scratch but its
    /// referenced entry nowhere → `None`.
    #[tokio::test]
    async fn retrieve_record_with_scratch_missing_entry_returns_none() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 96u8;
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        // Action references an entry, but the entry is not in scratch or store.
        let sah = make_signed_action_for_entry(seed, entry_hash);
        let action_hash = sah.as_hash().clone();
        let scratch = scratch_with_action(sah);

        let result = store
            .as_read()
            .retrieve_record_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        assert!(result.is_none(), "missing entry must yield None");
    }

    /// `retrieve_record_with_scratch` with an action that carries no entry
    /// (e.g. `Dna`) → returns the record.
    #[tokio::test]
    async fn retrieve_record_with_scratch_no_entry_action() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let sah = make_signed_action_no_entry(97);
        let action_hash = sah.as_hash().clone();
        let scratch = scratch_with_action(sah);

        let result = store
            .as_read()
            .retrieve_record_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        assert!(
            result.is_some(),
            "action with no entry should still produce a record"
        );
    }

    /// Action resolved from the store, but its referenced entry only in the
    /// scratch → record assembled from both sources. Exercises the cross-path
    /// where `retrieve_record` alone returns `None`.
    #[tokio::test]
    async fn retrieve_record_with_scratch_store_action_scratch_entry() {
        use holochain_types::prelude::{AppEntryBytes, Entry, EntryHashed};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        // `make_chain_op` records a AgentActivity op: the Create action
        // (referencing entry `seed+100`) lands in the store, but the entry does not.
        let seed = 94u8;
        let op = make_chain_op(seed, false);
        let action = match op.0.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().data().clone(),
            _ => unreachable!(),
        };
        let action_hash = holo_hash::ActionHash::with_data_sync(&action);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        store.record_incoming_ops(vec![op]).await.unwrap();

        // Store-only read returns None: the action references an entry the store
        // does not hold.
        let store_only = store
            .as_read()
            .retrieve_record(&action_hash, None)
            .await
            .unwrap();
        assert!(store_only.is_none(), "entry is absent from the store");

        // The entry lives only in the scratch.
        let entry = Entry::App(AppEntryBytes(
            holochain_serialized_bytes::SerializedBytes::from(
                holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
            ),
        ));
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_entry(
            EntryHashed::with_pre_hashed(entry, entry_hash),
            ChainTopOrdering::Relaxed,
        );
        let scratch = scratch.into_sync();

        // The overlay resolves the action from the store and the entry from the scratch.
        let result = store
            .as_read()
            .retrieve_record_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        let record = result.expect("record assembled from store action + scratch entry");
        assert_eq!(record.action_address(), &action_hash);
        assert!(
            record.entry().as_option().is_some(),
            "entry should be supplied by the scratch"
        );
    }

    // ---- get_live_{record,entry}_with_scratch helpers ----

    /// Build a public [`Entry::App`] with payload `vec![seed; 8]`.
    fn make_entry(seed: u8) -> holochain_types::prelude::Entry {
        use holochain_types::prelude::{AppEntryBytes, Entry};
        let bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        Entry::App(AppEntryBytes(bytes))
    }

    /// Build and integrate a `CreateEntry` op so that `entry_hash` has a live
    /// store create. Returns the action hash of the `Create` action.
    async fn integrate_store_entry_op(
        store: &crate::dht_store::DhtStore<DbWrite<Dht>>,
        seed: u8,
        author: &AgentPubKey,
        entry_hash: EntryHash,
        entry: holochain_types::prelude::Entry,
    ) -> ActionHash {
        use crate::dht_store::{AppOutcome, SysOutcome};

        let action = Action {
            header: ActionHeader {
                author: author.clone(),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(200); 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: entry_hash.clone(),
            }),
        };
        let action_hash = holo_hash::ActionHash::with_data_sync(&action);
        let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
        let chain_op = ChainOp::CreateEntry(signed_action, OpEntry::Present(entry.clone()));
        let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)));
        let op_hash = op.as_hash().clone();
        store.record_incoming_ops(vec![(op, false)]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(op_hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(op_hash.clone(), AppOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(Timestamp::from_micros(1000))
            .await
            .unwrap();
        action_hash
    }

    /// Build a scratch `Delete` action targeting `deletes_address` /
    /// `deletes_entry_address`, wrapped in a `SyncScratch`.
    fn scratch_with_delete(
        seed: u8,
        deletes_address: ActionHash,
        deletes_entry_address: EntryHash,
    ) -> crate::scratch::SyncScratch {
        let delete = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed; 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 2000),
                action_seq: 2,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(150); 36])),
            },
            ActionData::Delete(DeleteData {
                deletes_address,
                deletes_entry_address,
            }),
        );
        let action_hashed = holo_hash::HoloHashed::from_content_sync(delete);
        let sah = SignedActionHashed::with_presigned(action_hashed, Signature::from([seed; 64]));
        let mut scratch = crate::scratch::Scratch::new();
        scratch.add_action(sah, ChainTopOrdering::Relaxed);
        scratch.into_sync()
    }

    // ---- get_live_record_with_scratch tests ----

    /// (a) Store create + a scratch `Delete` targeting it →
    /// `get_live_record_with_scratch` returns `None`, but
    /// `get_live_record` still returns the record (it ignores the scratch).
    #[tokio::test]
    async fn get_live_record_with_scratch_scratch_delete_tombstones_store_record() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 110u8;
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry = make_entry(seed);

        // First insert a AgentActivity op so the action is in the store.
        let act_op = make_chain_op(seed, false);
        let action = match act_op.0.as_content() {
            DhtOp::ChainOp(c) => c.signed_action().data().clone(),
            _ => unreachable!(),
        };
        let action_hash = holo_hash::ActionHash::with_data_sync(&action);
        store.record_incoming_ops(vec![act_op]).await.unwrap();

        // Integrate a CreateEntry op so the record is fully live in the store.
        integrate_store_entry_op(&store, seed, &author, entry_hash.clone(), entry).await;

        // Store-only live-record sees it.
        let store_result = store
            .as_read()
            .get_live_record(&action_hash, None)
            .await
            .unwrap();
        assert!(
            store_result.is_some(),
            "store-only get_live_record should return the record"
        );

        // A scratch Delete targeting the action hash.
        let scratch = scratch_with_delete(seed, action_hash.clone(), entry_hash);

        // With-scratch path returns None due to the scratch tombstone.
        let result = store
            .as_read()
            .get_live_record_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        assert!(
            result.is_none(),
            "scratch delete should tombstone the record for get_live_record_with_scratch"
        );
    }

    /// (b) Scratch-only create → `get_live_record_with_scratch` and
    /// `get_live_entry_with_scratch` both return the live record.
    #[tokio::test]
    async fn get_live_with_scratch_scratch_only_create_is_live() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 111u8;
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let sah = make_signed_action_for_entry(seed, entry_hash.clone());
        let action_hash = sah.as_hash().clone();
        let scratch = scratch_with_action_and_entry(sah, entry, entry_hash.clone());

        // get_live_record_with_scratch should find the scratch-only create.
        let record_result = store
            .as_read()
            .get_live_record_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        assert!(
            record_result.is_some(),
            "scratch-only create should be live for get_live_record_with_scratch"
        );

        // get_live_entry_with_scratch should also find it.
        let entry_result = store
            .as_read()
            .get_live_entry_with_scratch(&entry_hash, None, &scratch)
            .await
            .unwrap();
        assert!(
            entry_result.is_some(),
            "scratch-only create should be live for get_live_entry_with_scratch"
        );
    }

    /// (c) Author-preference: both a store create and a scratch create exist for
    /// the same entry. Querying with the scratch-create author picks the scratch
    /// action; querying with a third author falls back to the store create first.
    #[tokio::test]
    async fn get_live_entry_with_scratch_author_preference() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        // Shared entry hash for both creates.
        let entry_hash = EntryHash::from_raw_36(vec![150u8; 36]);
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![1u8; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));

        // Store create by store_author (seed=112).
        let store_author = AgentPubKey::from_raw_36(vec![112u8; 36]);
        let store_action_hash = integrate_store_entry_op(
            &store,
            112u8,
            &store_author,
            entry_hash.clone(),
            entry.clone(),
        )
        .await;

        // Scratch create by scratch_author (seed=113).
        let seed = 113u8;
        let scratch_sah = make_signed_action_for_entry(seed, entry_hash.clone());
        let scratch_action_hash = scratch_sah.as_hash().clone();
        let scratch = scratch_with_action_and_entry(scratch_sah, entry.clone(), entry_hash.clone());

        // Querying with scratch_author → authored pick returns the scratch action.
        let scratch_author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let result_scratch_author = store
            .as_read()
            .get_live_entry_with_scratch(&entry_hash, Some(&scratch_author), &scratch)
            .await
            .unwrap()
            .expect("should find a live record for scratch_author");
        assert_eq!(
            result_scratch_author.action_address(),
            &scratch_action_hash,
            "scratch_author should pick the scratch create"
        );

        // Querying with store_author → authored pick returns the store action.
        let result_store_author = store
            .as_read()
            .get_live_entry_with_scratch(&entry_hash, Some(&store_author), &scratch)
            .await
            .unwrap()
            .expect("should find a live record for store_author");
        assert_eq!(
            result_store_author.action_address(),
            &store_action_hash,
            "store_author should pick the store create"
        );

        // Querying with an unmatched author → falls back to the first store
        // create (store before scratch).
        let other_author = AgentPubKey::from_raw_36(vec![200u8; 36]);
        let result_other = store
            .as_read()
            .get_live_entry_with_scratch(&entry_hash, Some(&other_author), &scratch)
            .await
            .unwrap()
            .expect("should find a live record for other_author");
        assert_eq!(
            result_other.action_address(),
            &store_action_hash,
            "unmatched author should fall back to first store create"
        );
    }

    /// A scratch `Delete` targeting a *store* entry create tombstones it, so
    /// `get_live_entry_with_scratch` returns `None` even though the store-only
    /// `get_live_entry` still returns the record.
    #[tokio::test]
    async fn get_live_entry_with_scratch_scratch_delete_tombstones_store_create() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 118u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let entry = make_entry(seed);

        // Integrate a CreateEntry op so the create is live in the store.
        let store_action_hash =
            integrate_store_entry_op(&store, seed, &author, entry_hash.clone(), entry).await;

        // Store-only read sees the live entry.
        let store_only = store
            .as_read()
            .get_live_entry(&entry_hash, None)
            .await
            .unwrap();
        assert!(
            store_only.is_some(),
            "store-only get_live_entry should return the create"
        );

        // A scratch Delete targeting the store create's action hash.
        let scratch = scratch_with_delete(seed, store_action_hash, entry_hash.clone());

        // The overlay must exclude the tombstoned store create → no live creates.
        let result = store
            .as_read()
            .get_live_entry_with_scratch(&entry_hash, None, &scratch)
            .await
            .unwrap();
        assert!(
            result.is_none(),
            "scratch delete of the store create should tombstone get_live_entry_with_scratch"
        );
    }

    /// (d) Store-only `get_live_record` and `get_live_entry` ignore a populated
    /// scratch (the requester-only invariant).
    #[tokio::test]
    async fn get_live_store_only_ignores_scratch() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 114u8;
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let sah = make_signed_action_for_entry(seed, entry_hash.clone());
        let action_hash = sah.as_hash().clone();

        // The action + entry live only in the scratch.
        // Store-only reads should return None.
        let result_record = store
            .as_read()
            .get_live_record(&action_hash, None)
            .await
            .unwrap();
        assert!(
            result_record.is_none(),
            "get_live_record must not see scratch-only data"
        );

        let result_entry = store
            .as_read()
            .get_live_entry(&entry_hash, None)
            .await
            .unwrap();
        assert!(
            result_entry.is_none(),
            "get_live_entry must not see scratch-only data"
        );

        // Confirm the scratch data is there (sanity check) but is irrelevant
        // to store-only queries.
        let scratch = scratch_with_action_and_entry(sah, entry, entry_hash);
        let _ = scratch; // scratch is intentionally not passed to the store-only methods
    }

    // ---- get_record_details_with_scratch / get_entry_details_with_scratch tests ----

    /// Build and integrate a `CreateRecord` op so that `action_hash` has a live
    /// store record. Returns the action hash. The action has no entry (uses `Dna`).
    async fn integrate_store_record_op(
        store: &crate::dht_store::DhtStore<DbWrite<Dht>>,
        seed: u8,
        author: &AgentPubKey,
        entry_hash: EntryHash,
        entry: holochain_types::prelude::Entry,
    ) -> ActionHash {
        use crate::dht_store::{AppOutcome, SysOutcome};

        let action = Action {
            header: ActionHeader {
                author: author.clone(),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 1,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(200); 36])),
            },
            data: ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: entry_hash.clone(),
            }),
        };
        let action_hash = holo_hash::ActionHash::with_data_sync(&action);
        let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
        let chain_op = ChainOp::CreateRecord(signed_action, OpEntry::Present(entry));
        let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)));
        let op_hash = op.as_hash().clone();
        store.record_incoming_ops(vec![(op, false)]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(op_hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(op_hash.clone(), AppOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(Timestamp::from_micros(1000))
            .await
            .unwrap();
        action_hash
    }

    /// Scratch `Update` and scratch `Delete` targeting an integrated store record
    /// appear in the returned `updates`/`deletes` lists.
    #[tokio::test]
    async fn get_record_details_with_scratch_shows_scratch_updates_and_deletes() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 120u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let entry = make_entry(seed);

        // Integrate a CreateRecord so the record exists in the store.
        let action_hash =
            integrate_store_record_op(&store, seed, &author, entry_hash.clone(), entry).await;

        // Also integrate a CreateEntry op so the entry is live (for the record retrieval).
        let entry2 = make_entry(seed);
        integrate_store_entry_op(
            &store,
            seed.wrapping_add(1),
            &author,
            entry_hash.clone(),
            entry2,
        )
        .await;

        // Build a scratch with one Delete and one Update targeting the store record.
        let new_entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(50); 36]);
        let mut scratch = crate::scratch::Scratch::new();

        // Delete targeting the store record's action hash.
        let delete = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed.wrapping_add(10); 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 3000),
                action_seq: 3,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(160); 36])),
            },
            ActionData::Delete(DeleteData {
                deletes_address: action_hash.clone(),
                deletes_entry_address: entry_hash.clone(),
            }),
        );
        let delete_hashed = holo_hash::HoloHashed::from_content_sync(delete);
        let delete_sah = SignedActionHashed::with_presigned(
            delete_hashed,
            Signature::from([seed.wrapping_add(10); 64]),
        );
        scratch.add_action(delete_sah, ChainTopOrdering::Relaxed);

        // Update targeting the store record's action hash.
        let update = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed.wrapping_add(20); 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 4000),
                action_seq: 4,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(170); 36])),
            },
            ActionData::Update(UpdateData {
                original_action_address: action_hash.clone(),
                original_entry_address: entry_hash.clone(),
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: new_entry_hash.clone(),
            }),
        );
        let update_hashed = holo_hash::HoloHashed::from_content_sync(update);
        let update_sah = SignedActionHashed::with_presigned(
            update_hashed,
            Signature::from([seed.wrapping_add(20); 64]),
        );
        scratch.add_action(update_sah, ChainTopOrdering::Relaxed);

        let scratch = scratch.into_sync();

        let details = store
            .as_read()
            .get_record_details_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap()
            .expect("should find details for integrated record");

        assert_eq!(
            details.deletes.len(),
            1,
            "scratch Delete should appear in deletes"
        );
        assert_eq!(
            details.updates.len(),
            1,
            "scratch Update should appear in updates"
        );
        assert_eq!(
            details.validation_status,
            holochain_zome_types::validate::ValidationStatus::Valid
        );
    }

    /// A scratch-only action (no integrated `CreateRecord` op) returns `None`.
    /// A scratch-only record (the author's own, not yet committed) has no
    /// integrated CreateRecord op, but is still returned with `Valid` status so
    /// host functions can resolve a just-authored record mid-call.
    #[tokio::test]
    async fn get_record_details_with_scratch_returns_valid_for_scratch_only_record() {
        use holochain_types::prelude::{AppEntryBytes, Entry};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 121u8;
        let entry_bytes = holochain_serialized_bytes::SerializedBytes::from(
            holochain_serialized_bytes::UnsafeBytes::from(vec![seed; 8]),
        );
        let entry = Entry::App(AppEntryBytes(entry_bytes));
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        // Action and its entry are only in the scratch — never written to the store.
        let sah = make_signed_action_for_entry(seed, entry_hash.clone());
        let action_hash = sah.as_hash().clone();
        let scratch = scratch_with_action_and_entry(sah, entry, entry_hash);

        let result = store
            .as_read()
            .get_record_details_with_scratch(&action_hash, None, &scratch)
            .await
            .unwrap();
        let details = result.expect("scratch-only record must return details");
        assert_eq!(
            details.validation_status,
            ValidationStatus::Valid,
            "the author's own uncommitted record is Valid"
        );
    }

    /// Store-only `get_record_details` ignores a populated scratch.
    #[tokio::test]
    async fn get_record_details_ignores_scratch() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 122u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let entry = make_entry(seed);

        // Integrate a CreateRecord and a CreateEntry.
        let action_hash =
            integrate_store_record_op(&store, seed, &author, entry_hash.clone(), entry.clone())
                .await;
        integrate_store_entry_op(
            &store,
            seed.wrapping_add(1),
            &author,
            entry_hash.clone(),
            entry,
        )
        .await;

        // Build a scratch with a Delete targeting the record (intentionally not
        // passed to the store-only read — it must be invisible to it).
        let _scratch = scratch_with_delete(seed, action_hash.clone(), entry_hash.clone());

        // Store-only path ignores the scratch delete.
        let details = store
            .as_read()
            .get_record_details(&action_hash, None)
            .await
            .unwrap()
            .expect("store-only get_record_details should find the record");
        assert!(
            details.deletes.is_empty(),
            "store-only get_record_details must not see scratch deletes"
        );
    }

    /// A scratch `Create` for an otherwise-Dead entry flips `entry_dht_status`
    /// to `Live`; scratch deletes and updates appear in the respective lists.
    #[tokio::test]
    async fn get_entry_details_with_scratch_scratch_create_flips_dead_to_live() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 125u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let entry = make_entry(seed);

        // Integrate a CreateEntry op that is then deleted (by a store delete) so
        // the entry is Dead in the store.
        let store_action_hash =
            integrate_store_entry_op(&store, seed, &author, entry_hash.clone(), entry.clone())
                .await;

        // Integrate a DeleteEntry so the store-create is deleted.
        {
            use crate::dht_store::{AppOutcome, SysOutcome};
            let delete_action = Action {
                header: ActionHeader {
                    author: author.clone(),
                    timestamp: Timestamp::from_micros(seed as i64 * 5000),
                    action_seq: 3,
                    prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(180); 36])),
                },
                data: ActionData::Delete(DeleteData {
                    deletes_address: store_action_hash.clone(),
                    deletes_entry_address: entry_hash.clone(),
                }),
            };
            let signed_action =
                SignedAction::new(delete_action, Signature::from([seed.wrapping_add(5); 64]));
            let chain_op = ChainOp::DeleteEntry(signed_action);
            let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)));
            let op_hash = op.as_hash().clone();
            store.record_incoming_ops(vec![(op, false)]).await.unwrap();
            store
                .record_chain_op_sys_validation_outcomes(vec![(
                    op_hash.clone(),
                    SysOutcome::Accepted,
                )])
                .await
                .unwrap();
            store
                .record_app_validation_outcomes(vec![(op_hash.clone(), AppOutcome::Accepted)])
                .await
                .unwrap();
            store
                .integrate_ready_ops(Timestamp::from_micros(2000))
                .await
                .unwrap();
        }

        // Confirm the entry is Dead in the store.
        let store_details = store
            .as_read()
            .get_entry_details(&entry_hash, None)
            .await
            .unwrap()
            .expect("entry exists in store");
        assert_eq!(
            store_details.entry_dht_status,
            holochain_zome_types::metadata::EntryDhtStatus::Dead,
            "entry should be Dead in store after store-delete"
        );

        // A scratch Create for the same entry_hash flips the status to Live.
        let scratch_sah = make_signed_action_for_entry(seed.wrapping_add(30), entry_hash.clone());
        let scratch = scratch_with_action_and_entry(scratch_sah, entry.clone(), entry_hash.clone());

        let details = store
            .as_read()
            .get_entry_details_with_scratch(&entry_hash, None, &scratch)
            .await
            .unwrap()
            .expect("entry should be found via scratch");

        assert_eq!(
            details.entry_dht_status,
            holochain_zome_types::metadata::EntryDhtStatus::Live,
            "scratch Create should flip Dead→Live"
        );
        // The scratch create appears in actions.
        assert!(
            !details.actions.is_empty(),
            "scratch Create should appear in actions"
        );
    }

    /// Scratch deletes and updates appear in `get_entry_details_with_scratch`
    /// alongside the store-only lists.
    #[tokio::test]
    async fn get_entry_details_with_scratch_shows_scratch_deletes_and_updates() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 126u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let entry = make_entry(seed);

        // Integrate a live CreateEntry op.
        let store_action_hash =
            integrate_store_entry_op(&store, seed, &author, entry_hash.clone(), entry.clone())
                .await;

        let new_entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(60); 36]);

        // Scratch with a Delete (deletes_entry_address == entry_hash) and an
        // Update (original_entry_address == entry_hash).
        let mut scratch = crate::scratch::Scratch::new();

        let delete = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed.wrapping_add(10); 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 3000),
                action_seq: 3,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(160); 36])),
            },
            ActionData::Delete(DeleteData {
                deletes_address: store_action_hash.clone(),
                deletes_entry_address: entry_hash.clone(),
            }),
        );
        scratch.add_action(
            SignedActionHashed::with_presigned(
                holo_hash::HoloHashed::from_content_sync(delete),
                Signature::from([seed.wrapping_add(10); 64]),
            ),
            ChainTopOrdering::Relaxed,
        );

        let update = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed.wrapping_add(20); 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 4000),
                action_seq: 4,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(170); 36])),
            },
            ActionData::Update(UpdateData {
                original_action_address: store_action_hash.clone(),
                original_entry_address: entry_hash.clone(),
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: new_entry_hash.clone(),
            }),
        );
        scratch.add_action(
            SignedActionHashed::with_presigned(
                holo_hash::HoloHashed::from_content_sync(update),
                Signature::from([seed.wrapping_add(20); 64]),
            ),
            ChainTopOrdering::Relaxed,
        );

        let scratch = scratch.into_sync();

        let details = store
            .as_read()
            .get_entry_details_with_scratch(&entry_hash, None, &scratch)
            .await
            .unwrap()
            .expect("entry should be found");

        assert_eq!(
            details.deletes.len(),
            1,
            "scratch Delete should appear in deletes"
        );
        assert_eq!(
            details.updates.len(),
            1,
            "scratch Update should appear in updates"
        );
        // The scratch Delete tombstones the only store create for this entry
        // (the scratch Update creates a *different* entry), so the entry flips
        // Live → Dead under the overlay.
        assert_eq!(
            details.entry_dht_status,
            holochain_zome_types::metadata::EntryDhtStatus::Dead,
            "scratch delete of the only create should make the entry Dead"
        );
    }

    /// Store-only `get_entry_details` ignores a populated scratch.
    #[tokio::test]
    async fn get_entry_details_ignores_scratch() {
        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let seed = 127u8;
        let author = AgentPubKey::from_raw_36(vec![seed; 36]);
        let entry_hash = EntryHash::from_raw_36(vec![seed.wrapping_add(100); 36]);
        let entry = make_entry(seed);

        // Integrate a live CreateEntry op.
        let store_action_hash =
            integrate_store_entry_op(&store, seed, &author, entry_hash.clone(), entry.clone())
                .await;

        // Build a scratch with a Delete targeting the store create (intentionally
        // not passed to the store-only read — it must be invisible to it).
        let _scratch = scratch_with_delete(seed, store_action_hash.clone(), entry_hash.clone());

        // Store-only path ignores the scratch delete.
        let details = store
            .as_read()
            .get_entry_details(&entry_hash, None)
            .await
            .unwrap()
            .expect("store-only get_entry_details should find the entry");
        assert!(
            details.deletes.is_empty(),
            "store-only get_entry_details must not see scratch deletes"
        );
        assert_eq!(
            details.entry_dht_status,
            holochain_zome_types::metadata::EntryDhtStatus::Live,
            "store-only get_entry_details must not see scratch deletes for status"
        );
    }

    // ---- get_links_with_scratch / get_link_details_with_scratch helpers ----

    /// Build and integrate a `CreateLink` op into the store.
    /// Returns `(action_hash, base_address)` of the created link.
    async fn integrate_link_op_for_base(
        store: &crate::dht_store::DhtStore<DbWrite<Dht>>,
        base: &holo_hash::AnyLinkableHash,
        zome_index: u8,
        link_type: u8,
        tag_bytes: Vec<u8>,
        seed: u8,
        when: i64,
    ) -> holo_hash::ActionHash {
        use crate::dht_store::{AppOutcome, SysOutcome};

        let action = Action {
            header: ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed; 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 2,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(60); 36])),
            },
            data: ActionData::CreateLink(CreateLinkData {
                base_address: base.clone(),
                target_address: holo_hash::AnyLinkableHash::from_raw_36_and_type(
                    vec![seed.wrapping_add(20); 36],
                    holo_hash::hash_type::AnyLinkable::Entry,
                ),
                zome_index: zome_index.into(),
                link_type: link_type.into(),
                tag: LinkTag(tag_bytes),
            }),
        };
        let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
        let op = DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(ChainOp::CreateLink(
            signed_action,
        ))));
        let op_hash = op.as_hash().clone();
        let action_hash = match op.as_content() {
            DhtOp::ChainOp(c) => holo_hash::ActionHash::with_data_sync(c.signed_action().data()),
            _ => unreachable!(),
        };
        store.record_incoming_ops(vec![(op, false)]).await.unwrap();
        store
            .record_chain_op_sys_validation_outcomes(vec![(op_hash.clone(), SysOutcome::Accepted)])
            .await
            .unwrap();
        store
            .record_app_validation_outcomes(vec![(op_hash, AppOutcome::Accepted)])
            .await
            .unwrap();
        store
            .integrate_ready_ops(Timestamp::from_micros(when))
            .await
            .unwrap();
        action_hash
    }

    /// Build a scratch `CreateLink` action for `base`.
    fn make_scratch_create_link(
        base: &holo_hash::AnyLinkableHash,
        zome_index: u8,
        link_type: u8,
        tag_bytes: Vec<u8>,
        seed: u8,
    ) -> SignedActionHashed {
        let action = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed; 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 1000),
                action_seq: 2,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(60); 36])),
            },
            ActionData::CreateLink(CreateLinkData {
                base_address: base.clone(),
                target_address: holo_hash::AnyLinkableHash::from_raw_36_and_type(
                    vec![seed.wrapping_add(20); 36],
                    holo_hash::hash_type::AnyLinkable::Entry,
                ),
                zome_index: zome_index.into(),
                link_type: link_type.into(),
                tag: LinkTag(tag_bytes),
            }),
        );
        let action_hashed = holo_hash::HoloHashed::from_content_sync(action);
        SignedActionHashed::with_presigned(action_hashed, Signature::from([seed; 64]))
    }

    /// Build a scratch `DeleteLink` action tombstoning `create_link_hash`.
    fn make_scratch_delete_link(
        base: &holo_hash::AnyLinkableHash,
        create_link_hash: holo_hash::ActionHash,
        seed: u8,
    ) -> SignedActionHashed {
        let action = build_action(
            ActionHeader {
                author: AgentPubKey::from_raw_36(vec![seed; 36]),
                timestamp: Timestamp::from_micros(seed as i64 * 1000 + 500),
                action_seq: 3,
                prev_action: Some(ActionHash::from_raw_36(vec![seed.wrapping_add(90); 36])),
            },
            ActionData::DeleteLink(DeleteLinkData {
                base_address: base.clone(),
                link_add_address: create_link_hash,
            }),
        );
        let action_hashed = holo_hash::HoloHashed::from_content_sync(action);
        SignedActionHashed::with_presigned(action_hashed, Signature::from([seed; 64]))
    }

    // ---- get_links_with_scratch tests ----

    /// A scratch `CreateLink` for `base` appears in `get_links_with_scratch`
    /// and is correctly filtered by type, tag, author, and time.
    #[tokio::test]
    async fn get_links_with_scratch_scratch_create_link_appears_and_is_filtered() {
        use crate::query::link::GetLinksFilter;
        use LinkTypeFilter;

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let base = holo_hash::AnyLinkableHash::from_raw_36_and_type(
            vec![130u8; 36],
            holo_hash::hash_type::AnyLinkable::Entry,
        );

        // Build a scratch CreateLink: zome 0, type 0, tag [1,2,3], seed 131.
        let scratch_sah = make_scratch_create_link(&base, 0, 0, vec![1, 2, 3], 131);
        let scratch_create_hash = scratch_sah.as_hash().clone();
        let mut scratch_inner = crate::scratch::Scratch::new();
        scratch_inner.add_action(scratch_sah, ChainTopOrdering::Relaxed);
        let scratch = scratch_inner.into_sync();

        let filter = GetLinksFilter {
            after: None,
            before: None,
            author: None,
        };

        // The link appears with a matching type+tag query.
        let links = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter,
                &scratch,
            )
            .await
            .unwrap();
        assert_eq!(links.len(), 1, "scratch CreateLink should appear");
        assert_eq!(links[0].create_link_hash, scratch_create_hash);

        // Filtered out by a different zome index.
        let links_no_type = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![1.into()]),
                None,
                &filter,
                &scratch,
            )
            .await
            .unwrap();
        assert!(
            links_no_type.is_empty(),
            "scratch CreateLink must be excluded by type filter"
        );

        // Filtered out by a non-matching tag prefix.
        let bad_tag = LinkTag(vec![9, 9, 9]);
        let links_no_tag = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                Some(&bad_tag),
                &filter,
                &scratch,
            )
            .await
            .unwrap();
        assert!(
            links_no_tag.is_empty(),
            "scratch CreateLink must be excluded by tag filter"
        );

        // Filtered out by author.
        let other_author = AgentPubKey::from_raw_36(vec![200u8; 36]);
        let filter_author = GetLinksFilter {
            after: None,
            before: None,
            author: Some(other_author),
        };
        let links_no_author = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter_author,
                &scratch,
            )
            .await
            .unwrap();
        assert!(
            links_no_author.is_empty(),
            "scratch CreateLink must be excluded by author filter"
        );

        // Filtered out by `before` (link timestamp is 131*1000; before=130*1000 excludes it).
        let filter_before = GetLinksFilter {
            after: None,
            before: Some(Timestamp::from_micros(130 * 1000)),
            author: None,
        };
        let links_before = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter_before,
                &scratch,
            )
            .await
            .unwrap();
        assert!(
            links_before.is_empty(),
            "scratch CreateLink must be excluded by before filter"
        );

        // Filtered out by `after` (link timestamp is 131*1000; after=132*1000 excludes it).
        let filter_after = GetLinksFilter {
            after: Some(Timestamp::from_micros(132 * 1000)),
            before: None,
            author: None,
        };
        let links_after = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter_after,
                &scratch,
            )
            .await
            .unwrap();
        assert!(
            links_after.is_empty(),
            "scratch CreateLink must be excluded by after filter"
        );
    }

    /// A scratch `DeleteLink` targeting a *store* `CreateLink` removes that
    /// link from `get_links_with_scratch`, but both the create and the delete
    /// appear in `get_link_details_with_scratch`.
    #[tokio::test]
    async fn scratch_delete_link_tombstones_store_create_in_get_links_but_not_details() {
        use crate::query::link::GetLinksFilter;
        use LinkTypeFilter;

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let base = holo_hash::AnyLinkableHash::from_raw_36_and_type(
            vec![140u8; 36],
            holo_hash::hash_type::AnyLinkable::Entry,
        );

        // Integrate a store CreateLink.
        let store_create_hash =
            integrate_link_op_for_base(&store, &base, 0, 0, vec![1, 2, 3], 141, 100).await;

        // Confirm it appears in store-only get_links.
        let filter = GetLinksFilter {
            after: None,
            before: None,
            author: None,
        };
        let store_links = store
            .as_read()
            .get_links(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter,
            )
            .await
            .unwrap();
        assert_eq!(store_links.len(), 1, "store CreateLink should be live");

        // Add a scratch DeleteLink targeting the store create.
        let dl_sah = make_scratch_delete_link(&base, store_create_hash.clone(), 142);
        let mut scratch_inner = crate::scratch::Scratch::new();
        scratch_inner.add_action(dl_sah, ChainTopOrdering::Relaxed);
        let scratch = scratch_inner.into_sync();

        // get_links_with_scratch must exclude the tombstoned store link.
        let links = store
            .as_read()
            .get_links_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter,
                &scratch,
            )
            .await
            .unwrap();
        assert!(
            links.is_empty(),
            "scratch DeleteLink must tombstone the store CreateLink in get_links_with_scratch"
        );

        // get_link_details_with_scratch must show the create AND the scratch delete.
        let details = store
            .as_read()
            .get_link_details_with_scratch(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &scratch,
            )
            .await
            .unwrap();
        assert_eq!(
            details.len(),
            1,
            "store CreateLink should appear in details"
        );
        let (create_sah, deletes) = &details[0];
        assert_eq!(
            create_sah.as_hash(),
            &store_create_hash,
            "details create should be the store CreateLink"
        );
        assert_eq!(
            deletes.len(),
            1,
            "scratch DeleteLink must appear in the details delete list"
        );
    }

    /// Store-only `get_links` and `get_link_details` ignore a populated scratch
    /// (requester-only invariant).
    #[tokio::test]
    async fn get_links_store_only_ignores_scratch() {
        use crate::query::link::GetLinksFilter;
        use LinkTypeFilter;

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let base = holo_hash::AnyLinkableHash::from_raw_36_and_type(
            vec![150u8; 36],
            holo_hash::hash_type::AnyLinkable::Entry,
        );

        // Integrate a store CreateLink.
        let store_create_hash =
            integrate_link_op_for_base(&store, &base, 0, 0, vec![1, 2, 3], 151, 100).await;

        // Build a scratch with: one new CreateLink + one DeleteLink tombstoning the store create.
        let scratch_cl_sah = make_scratch_create_link(&base, 0, 0, vec![4, 5, 6], 152);
        let scratch_dl_sah = make_scratch_delete_link(&base, store_create_hash.clone(), 153);
        let mut scratch_inner = crate::scratch::Scratch::new();
        scratch_inner.add_action(scratch_cl_sah, ChainTopOrdering::Relaxed);
        scratch_inner.add_action(scratch_dl_sah, ChainTopOrdering::Relaxed);
        // Scratch is built but NOT passed to the store-only methods.
        let _scratch = scratch_inner.into_sync();

        // Store-only get_links: should see only the store link (no scratch create, no exclusion).
        let filter = GetLinksFilter {
            after: None,
            before: None,
            author: None,
        };
        let links = store
            .as_read()
            .get_links(
                &base,
                &LinkTypeFilter::Dependencies(vec![0.into()]),
                None,
                &filter,
            )
            .await
            .unwrap();
        assert_eq!(
            links.len(),
            1,
            "store-only get_links must not see scratch CreateLink"
        );
        assert_eq!(
            links[0].create_link_hash, store_create_hash,
            "store-only get_links must return the store link unaffected by scratch delete"
        );

        // Store-only get_link_details: should see only the store create, with no deletes.
        let details = store
            .as_read()
            .get_link_details(&base, &LinkTypeFilter::Dependencies(vec![0.into()]), None)
            .await
            .unwrap();
        assert_eq!(
            details.len(),
            1,
            "store-only get_link_details must not see scratch CreateLink"
        );
        assert_eq!(
            details[0].1.len(),
            0,
            "store-only get_link_details must not see scratch DeleteLink"
        );
    }

    fn mk_action(
        author: &AgentPubKey,
        seq: u32,
        prev: Option<ActionHash>,
        data: ActionData,
    ) -> HoloHashed<Action> {
        HoloHashed::from_content_sync(Action {
            header: ActionHeader {
                author: author.clone(),
                timestamp: Timestamp::from_micros(seq as i64 * 1000),
                action_seq: seq,
                prev_action: prev,
            },
            data,
        })
    }

    fn mk_dna(author: &AgentPubKey) -> HoloHashed<Action> {
        mk_action(
            author,
            0,
            None,
            ActionData::Dna(DnaData {
                dna_hash: holo_hash::DnaHash::from_raw_36(vec![0u8; 36]),
            }),
        )
    }

    fn mk_create(author: &AgentPubKey, seq: u32) -> HoloHashed<Action> {
        mk_action(
            author,
            seq,
            Some(ActionHash::from_raw_36(vec![seq as u8; 36])),
            ActionData::Create(CreateData {
                entry_type: EntryType::App(AppEntryDef::new(
                    0.into(),
                    0.into(),
                    EntryVisibility::Public,
                )),
                entry_hash: EntryHash::from_raw_36(vec![seq as u8; 36]),
            }),
        )
    }

    fn mk_close(author: &AgentPubKey, seq: u32) -> HoloHashed<Action> {
        mk_action(
            author,
            seq,
            Some(ActionHash::from_raw_36(vec![seq as u8; 36])),
            ActionData::CloseChain(CloseChainData { new_target: None }),
        )
    }

    fn status_of(valid: Vec<HoloHashed<Action>>, rejected: Vec<HoloHashed<Action>>) -> ChainStatus {
        let agent = AgentPubKey::from_raw_36(vec![0u8; 36]);
        let options = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: true,
            include_warrants: false,
            include_full_records: true,
        };
        build_agent_activity_response(
            agent,
            valid,
            rejected,
            vec![],
            &ChainQueryFilter::new(),
            &options,
        )
        .status
    }

    #[test]
    fn closed_chain_reports_closed() {
        let agent = AgentPubKey::from_raw_36(vec![1u8; 36]);
        let close = mk_close(&agent, 2);
        let close_head = ChainHead {
            action_seq: 2,
            hash: close.as_hash().clone(),
        };
        let status = status_of(vec![mk_dna(&agent), mk_create(&agent, 1), close], vec![]);
        assert_eq!(status, ChainStatus::Closed(close_head));
    }

    #[test]
    fn forked_and_closed_reports_forked() {
        let agent = AgentPubKey::from_raw_36(vec![1u8; 36]);
        let status = status_of(
            vec![
                mk_dna(&agent),
                mk_create(&agent, 1),
                mk_create(&agent, 1), // fork at seq 1
                mk_close(&agent, 2),
            ],
            vec![],
        );
        assert!(matches!(status, ChainStatus::Forked(_)));
    }

    #[test]
    fn forked_on_close_reports_forked() {
        let agent = AgentPubKey::from_raw_36(vec![1u8; 36]);
        let status = status_of(
            vec![
                mk_dna(&agent),
                mk_create(&agent, 1),
                mk_create(&agent, 2),
                mk_close(&agent, 2), // fork at seq 2, one branch closes the chain
            ],
            vec![],
        );
        assert!(matches!(status, ChainStatus::Forked(_)));
    }

    #[test]
    fn invalid_and_closed_reports_invalid() {
        let agent = AgentPubKey::from_raw_36(vec![1u8; 36]);
        let status = status_of(
            vec![mk_dna(&agent), mk_create(&agent, 1), mk_close(&agent, 2)],
            vec![mk_create(&agent, 1)], // a rejected action
        );
        assert!(matches!(status, ChainStatus::Invalid(_)));
    }

    /// End-to-end: a `CloseChain` head integrated into the real DHT store is
    /// returned as `ChainStatus::Closed` by `get_agent_activity`.
    #[tokio::test]
    async fn get_agent_activity_closed_chain_reports_closed() {
        fn make_close_op(
            author: &AgentPubKey,
            prev: &ActionHash,
            seq: u32,
            seed: u8,
        ) -> DhtOpHashed {
            let action = Action {
                header: ActionHeader {
                    author: author.clone(),
                    timestamp: Timestamp::from_micros(seed as i64 * 1000),
                    action_seq: seq,
                    prev_action: Some(prev.clone()),
                },
                data: ActionData::CloseChain(CloseChainData { new_target: None }),
            };
            let signed_action = SignedAction::new(action, Signature::from([seed; 64]));
            let chain_op = ChainOp::AgentActivity(signed_action);
            DhtOpHashed::from_content_sync(DhtOp::ChainOp(Box::new(chain_op)))
        }

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();
        let author = AgentPubKey::from_raw_36(vec![7u8; 36]);
        let prev = ActionHash::from_raw_36(vec![0u8; 36]);

        integrate_activity(
            &store,
            make_fork_op(&author, &prev, 0, 1),
            AppOutcome::Accepted,
            10,
        )
        .await;
        integrate_activity(
            &store,
            make_close_op(&author, &prev, 1, 2),
            AppOutcome::Accepted,
            11,
        )
        .await;

        let opts = GetAgentActivityOptions {
            include_valid_activity: true,
            include_rejected_activity: false,
            include_warrants: false,
            include_full_records: false,
        };
        let resp = store
            .as_read()
            .get_agent_activity(&author, &ChainQueryFilter::new(), &opts)
            .await
            .unwrap();

        assert!(matches!(resp.status, ChainStatus::Closed(ref head) if head.action_seq == 1));
    }

    /// `record_incoming_ops` must respect the request validation receipt flag.
    /// A published op (flag = true) should request a validation receipt once integrated,
    /// while an op received through gossip (flag = false) should not.
    #[tokio::test]
    async fn record_incoming_ops_respects_require_receipt_flag() {
        use crate::dht_store::{AppOutcome, SysOutcome};

        let store = crate::dht_store::DhtStore::new_test(dht_id())
            .await
            .unwrap();

        let published = make_chain_op(70, true);
        let gossiped = make_chain_op(71, false);
        let published_hash = published.0.as_hash().clone();
        let gossiped_hash = gossiped.0.as_hash().clone();

        // Published op requests a validation receipt; gossiped op does not.
        store
            .record_incoming_ops(vec![published, gossiped])
            .await
            .unwrap();

        // Integrate both so they are eligible to have receipts sent.
        for hash in [&published_hash, &gossiped_hash] {
            store
                .record_chain_op_sys_validation_outcomes(vec![(hash.clone(), SysOutcome::Accepted)])
                .await
                .unwrap();
            store
                .record_app_validation_outcomes(vec![(hash.clone(), AppOutcome::Accepted)])
                .await
                .unwrap();
        }
        store
            .integrate_ready_ops(holochain_types::prelude::Timestamp::now())
            .await
            .unwrap();

        let receipts = store
            .as_read()
            .pending_validation_receipts(vec![])
            .await
            .unwrap();

        assert!(
            receipts
                .iter()
                .any(|(r, _)| r.dht_op_hash == published_hash),
            "published op (require_receipt) should be pending a validation receipt"
        );
        assert!(
            receipts.iter().all(|(r, _)| r.dht_op_hash != gossiped_hash),
            "gossiped op (no require_receipt) should not be pending a validation receipt"
        );
    }
}