kiromi-ai-memory 0.2.2

// SPDX-License-Identifier: Apache-2.0 OR MIT
//! Public engine handle.

use std::sync::Arc;

use crate::embedder::Embedder;
use crate::error::Error;
use crate::metadata::MetadataStore;
use crate::partition::PartitionScheme;
use crate::regen::RegenSubjectOpts;
use crate::storage::Storage;
use crate::tenancy::TenantLocks;
use crate::tenant::TenantId;
use crate::util::Clock;

/// Engine handle. Cheaply cloneable.
#[derive(Clone)]
pub struct Memory {
    pub(crate) inner: Arc<MemoryInner>,
}

pub(crate) struct MemoryInner {
    pub(crate) storage: Box<dyn Storage>,
    pub(crate) metadata: Box<dyn MetadataStore>,
    pub(crate) embedder: Option<Box<dyn Embedder>>,
    pub(crate) tenant: TenantId,
    pub(crate) scheme: PartitionScheme,
    pub(crate) actor: Option<String>,
    pub(crate) clock: Box<dyn Clock>,
    pub(crate) locks: TenantLocks,
    /// Plan 18 phase D: pluggable vector-index surface. Defaults to
    /// [`crate::index::SqliteVecIndex`] when the metadata store exposes
    /// a SQLite pool.
    pub(crate) vector_index: Box<dyn crate::index::VectorIndex>,
    /// Plan 18 phase D: pluggable lexical-index surface. Defaults to
    /// [`crate::index::Fts5Index`].
    pub(crate) lexical_index: Box<dyn crate::index::LexicalIndex>,
    /// `true` once `schema_meta.embedder_id` has been bound (either from the
    /// configured Embedder at open() or from the first caller-provided vector).
    pub(crate) schema_meta_locked: parking_lot::Mutex<bool>,
    /// Push-side event stream. Subscribers receive after-commit events.
    pub(crate) event_tx: tokio::sync::broadcast::Sender<crate::event::MemoryEvent>,
}

impl std::fmt::Debug for MemoryInner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MemoryInner")
            .field("storage", &self.storage.id())
            .field("metadata", &self.metadata.id())
            .field(
                "embedder",
                &self.embedder.as_ref().map(|e| e.id().to_string()),
            )
            .field("tenant", &self.tenant.as_str())
            .field("scheme", &self.scheme.to_string())
            .field("actor", &self.actor)
            .field("vector_index", &self.vector_index.id())
            .field("lexical_index", &self.lexical_index.id())
            .field("event_tx_subscribers", &self.event_tx.receiver_count())
            .finish()
    }
}

impl std::fmt::Debug for Memory {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Debug::fmt(&self.inner, f)
    }
}

impl Memory {
    /// Start a new builder. Configure storage / metadata / embedder /
    /// tenant / partition scheme, then call [`crate::Builder::open`].
    ///
    /// ```no_run
    /// # async fn _ex() -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{InMemoryBackend, Memory, SqliteMetadata, TenantId};
    ///
    /// let mem = Memory::builder()
    ///     .storage(InMemoryBackend::new())
    ///     .metadata(SqliteMetadata::connect_memory().await?)
    ///     .tenant(TenantId::new("local").unwrap())
    ///     .partition_scheme("user={user}").unwrap()
    ///     .open()
    ///     .await?;
    /// # let _ = mem; Ok(()) }
    /// ```
    #[must_use]
    pub fn builder() -> crate::builder::Builder {
        crate::builder::Builder::default()
    }

    /// Test-only accessor: borrow the `MetadataStore` for direct SQL probing.
    /// Hidden from rustdoc.
    #[doc(hidden)]
    #[must_use]
    pub fn metadata_for_test(&self) -> &dyn crate::metadata::MetadataStore {
        self.inner.metadata.as_ref()
    }

    /// Tenant the handle is bound to.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) {
    /// assert_eq!(mem.tenant().as_str(), "local");
    /// # }
    /// ```
    #[must_use]
    pub fn tenant(&self) -> &TenantId {
        &self.inner.tenant
    }

    /// Partition scheme captured at open.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) {
    /// let _scheme = mem.scheme();
    /// # }
    /// ```
    #[must_use]
    pub fn scheme(&self) -> &PartitionScheme {
        &self.inner.scheme
    }

    /// Subscribe to the live event stream. The returned receiver gets every
    /// `MemoryEvent` published *after* this call. Slow subscribers receive
    /// `RecvError::Lagged(n)` and continue. Default channel capacity is
    /// `DEFAULT_EVENT_CAPACITY` (1024); override with `Builder::event_capacity`.
    ///
    /// See spec § 18.5.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) {
    /// let mut rx = mem.subscribe();
    /// // Drive any write op, then poll: rx.recv().await
    /// let _ = rx; // silence unused
    /// # }
    /// ```
    #[must_use]
    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<crate::event::MemoryEvent> {
        self.inner.event_tx.subscribe()
    }

    /// Graceful shutdown.
    ///
    /// Plan 18 dispatch 4 collapsed close to a no-op: every write commits
    /// to vec0 + FTS5 inside the same SQL transaction as the catalog row,
    /// so the only outstanding resource at drop time is the SQLite pool —
    /// and that closes when the last `Memory` clone goes away. We retain
    /// `close()` for source compatibility and to leave a hook if a future
    /// dispatch needs to drain async work before the pool is taken away.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// mem.close().await?;
    /// # Ok(()) }
    /// ```
    pub async fn close(self) -> crate::error::Result<()> {
        Ok(())
    }

    /// Append a memory to a partition.
    ///
    /// Embeds the body synchronously (or accepts a pre-computed vector via
    /// `AppendOpts::with_embedding`), writes the data file, inserts metadata,
    /// inserts explicit links, and writes an audit-log entry — all in one SQL
    /// transaction. After the transaction commits, the new vector is pushed
    /// into the leaf's in-memory index delta and an early flush may be
    /// triggered.
    ///
    /// **Cost:** one storage `put` (body), one SQL transaction (memory plus
    /// links plus audit), and the embed call itself (skipped when
    /// `AppendOpts::with_embedding` is set). Per-attribute writes each cost an
    /// extra `set_attribute` SQL transaction.
    ///
    /// **Errors:** [`Error::Storage`] if the body write fails;
    /// [`Error::Metadata`] for SQL failures; [`Error::Embedder`] if a
    /// configured embedder returns an empty/wrong-dim vector;
    /// [`Error::EmbedderMismatch`] / [`Error::EmbedderDimMismatch`] if the
    /// caller-supplied embedding's id or length contradicts `schema_meta`;
    /// [`Error::PartitionInvalid`] / [`Error::PartitionSchemeInvalid`] for
    /// partition resolution failures; [`Error::Tombstoned`] when a link
    /// target is soft-deleted; [`Error::Config`] when neither an embedder
    /// nor `with_embedding` is supplied.
    ///
    /// [`Error::Storage`]: crate::error::Error::Storage
    /// [`Error::Metadata`]: crate::error::Error::Metadata
    /// [`Error::Embedder`]: crate::error::Error::Embedder
    /// [`Error::EmbedderMismatch`]: crate::error::Error::EmbedderMismatch
    /// [`Error::EmbedderDimMismatch`]: crate::error::Error::EmbedderDimMismatch
    /// [`Error::PartitionInvalid`]: crate::error::Error::PartitionInvalid
    /// [`Error::PartitionSchemeInvalid`]: crate::error::Error::PartitionSchemeInvalid
    /// [`Error::Tombstoned`]: crate::error::Error::Tombstoned
    /// [`Error::Config`]: crate::error::Error::Config
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{AppendOpts, Content, Partitions};
    /// let parts = Partitions::new().with("user", "alex");
    /// let r = mem.append(parts, Content::markdown("hello"), AppendOpts::default()).await?;
    /// # let _ = r; Ok(()) }
    /// ```
    pub async fn append(
        &self,
        partitions: crate::partition::Partitions,
        content: crate::content::Content,
        mut opts: crate::opts::AppendOpts,
    ) -> crate::error::Result<crate::memory::MemoryRef> {
        use crate::audit::AuditOp;
        use crate::content::ContentHash;
        use crate::memory::{MemoryId, MemoryRef};
        use crate::metadata::{AppendMemoryRequest, AuditEntry, MemoryRow};
        use crate::storage::StorageKey;

        let path = partitions.resolve(&self.inner.scheme)?;
        let body = content.as_str();

        // ----- Resolve the embedding via either pathway. -----
        let (embedding, embedder_id_for_row): (Vec<f32>, String) = if let Some(precomputed) =
            opts.embedding.clone()
        {
            // Caller-provided vector path.
            let id_str = match self.inner.embedder.as_ref() {
                Some(_e) => {
                    return Err(Error::Config(
                        "AppendOpts::with_embedding cannot be used together with a configured Embedder"
                            .into(),
                    ));
                }
                None => crate::embedder::CALLER_PROVIDED_EMBEDDER_ID.to_string(),
            };
            (precomputed, id_str)
        } else if let Some(embedder) = self.inner.embedder.as_ref() {
            let mut got = embedder
                .embed(crate::embedder::EmbedRole::Document, &[body])
                .await?;
            if got.len() != 1 {
                return Err(Error::embedder(
                    "embedder returned wrong shape",
                    std::io::Error::new(std::io::ErrorKind::InvalidData, "shape"),
                ));
            }
            let v = got.pop().ok_or_else(|| {
                Error::embedder(
                    "embedder returned 0 vectors",
                    std::io::Error::new(std::io::ErrorKind::InvalidData, "empty"),
                )
            })?;
            if v.len() != embedder.dimensions() {
                return Err(Error::embedder(
                    "embedder returned wrong dim",
                    std::io::Error::new(std::io::ErrorKind::InvalidData, "dim"),
                ));
            }
            (v, embedder.id().to_string())
        } else {
            return Err(Error::Config(
                "no embedder configured and AppendOpts::with_embedding was not set".into(),
            ));
        };

        // ----- Lock dim on first append (caller-provided mode). -----
        // If schema_meta has no embedder_id yet (first append after open with
        // no embedder), bind it now. If it already has one, validate.
        {
            let dims = embedding.len();
            let meta = self.inner.metadata.read_schema_meta().await?;
            if let Some(meta) = meta {
                match &meta.embedder_id {
                    None => {
                        // Bind on first append.
                        let new_meta = crate::metadata::SchemaMeta {
                            partition_scheme: meta.partition_scheme,
                            scheme_version: meta.scheme_version,
                            embedder_id: Some(embedder_id_for_row.clone()),
                            embedder_dims: Some(i64::try_from(dims).unwrap_or(0)),
                            created_at_ms: meta.created_at_ms,
                        };
                        self.inner.metadata.write_schema_meta(&new_meta).await?;
                        *self.inner.schema_meta_locked.lock() = true;
                        // Plan 18: in caller-provided mode the open() call
                        // didn't know the dim, so it deferred creating the
                        // vector / FTS virtual tables. First append knows dim
                        // — bootstrap them now so the same-transaction
                        // memory_vec / memory_fts inserts have a target.
                        self.inner.metadata.create_indices_if_missing(dims).await?;
                    }
                    Some(stored) => {
                        let stored_dims = meta.embedder_dims.unwrap_or(0);
                        if stored != &embedder_id_for_row {
                            return Err(Error::EmbedderMismatch {
                                expected: stored.clone(),
                                expected_dims: usize::try_from(stored_dims).unwrap_or(0),
                                got: embedder_id_for_row.clone(),
                                got_dims: dims,
                            });
                        }
                        if stored_dims != i64::try_from(dims).unwrap_or(0) {
                            return Err(Error::EmbedderMismatch {
                                expected: stored.clone(),
                                expected_dims: usize::try_from(stored_dims).unwrap_or(0),
                                got: embedder_id_for_row.clone(),
                                got_dims: dims,
                            });
                        }
                    }
                }
            }
        }

        // NOTE: spec § 12.13 reserves a per-row `memory.embedder_id` consistency
        // check between schema_meta.embedder_id and the embedder_id_for_row above.
        // Slice 1 enforces single-embedder semantics at the schema_meta level
        // (the lock-and-check above); a future migration plan will introduce
        // both the per-row check and the opt-in toggle that lets it relax.
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        self.inner
            .metadata
            .ensure_partition_chain(&path, true, now)
            .await?;

        let id = MemoryId::generate();
        let hash = ContentHash::of_content(&content);
        let key = StorageKey::new(format!(
            "{}/data/{}/{}.{}",
            self.inner.tenant.as_str(),
            path.as_str(),
            id,
            content.kind().extension(),
        ));
        let data_path = key.as_str().to_string();
        self.inner
            .storage
            .put(&key, bytes::Bytes::copy_from_slice(body.as_bytes()))
            .await?;

        // Plan 11: attribute writes happen post-commit through the trait's
        // dedicated `set_attribute` (one row + one audit per attribute) so we
        // hold a snapshot of the map before the request consumes `opts`.
        let pending_attributes = std::mem::take(&mut opts.attributes);
        // Plan 15: default new appends to MemoryKind::Episodic when the
        // caller didn't specify; keeps legacy on-disk rows readable as
        // None while new rows always carry an explicit kind.
        let memory_kind = Some(opts.kind.unwrap_or_default());
        let row = MemoryRow {
            id,
            partition_path: path.clone(),
            data_path,
            content_kind: content.kind().extension().to_string(),
            content_hash: hash,
            bytes: i64::try_from(content.byte_len()).unwrap_or(0),
            embedder_id: embedder_id_for_row.clone(),
            tombstoned: false,
            created_at_ms: now,
            updated_at_ms: now,
            valid_from_ms: opts.valid_from_ms,
            valid_until_ms: opts.valid_until_ms,
            kind: memory_kind,
        };
        let audit = AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: AuditOp::Append,
            partition_path: Some(path.clone()),
            memory_id: Some(id),
            detail: serde_json::json!({
                "kind": content.kind().extension(),
                "bytes": row.bytes,
                "embedding_dim": embedding.len(),
                "links": opts.links.iter().map(|m| m.id.to_string()).collect::<Vec<_>>(),
            }),
        };
        // Plan 17: persist the embedding bytes inside the same SQL
        // transaction as the memory row + audit. Little-endian f32 packed,
        // length = dims * 4. `bytemuck::cast_slice` is the zero-copy cast;
        // we materialise into a `Vec<u8>` because the request crosses an
        // await boundary.
        let embedding_blob: Vec<u8> = bytemuck::cast_slice(&embedding).to_vec();
        let _outcome = self
            .inner
            .metadata
            .append_memory(AppendMemoryRequest {
                row,
                explicit_links: opts.links,
                audit,
                embedding_blob,
                content_for_index: body.to_string(),
            })
            .await?;

        // Plan 18 dispatch 4: vec0 + FTS5 are written inside the same SQL
        // transaction as the memory row + audit (proved by
        // `tests/atomic_append.rs`). No further index bookkeeping is needed
        // here — the legacy `IndexHandle` delta + flusher are gone.

        // ----- Plan 11: persist typed attributes (one row + one audit each). -----
        for (key, value) in &pending_attributes {
            let attr_audit = AuditEntry {
                ts_ms: now,
                actor: self.inner.actor.clone(),
                op: AuditOp::AttributeSet,
                partition_path: Some(path.clone()),
                memory_id: Some(id),
                detail: serde_json::json!({
                    "key": key,
                    "kind": value.kind_str(),
                    "value": value,
                }),
            };
            self.inner
                .metadata
                .set_attribute(&id, key, value, attr_audit)
                .await?;
        }

        // ----- Plan 9: cascade summary_stale up the partition chain. -----
        // The leaf and every ancestor become stale; downstream summarizer
        // workers consume `subjects_needing_summary` to drive regeneration.
        let mut stale_chain: Vec<crate::partition::PartitionPath> = path.ancestors().collect();
        stale_chain.insert(0, path.clone());
        // Best-effort: never block append on stale-flag bookkeeping.
        if let Err(e) = self.inner.metadata.mark_summary_stale(&stale_chain).await {
            tracing::warn!(target: "kiromi-ai-memory.summary_stale", error=%e,
                "mark_summary_stale failed (non-fatal)");
        }

        // ----- Emit the live event AFTER the SQL commit (spec § 18.5). -----
        // `let _` because the broadcast channel is lossy: a slow subscriber yields
        // `RecvError::Lagged` on the receiver side, never an error here. Engine
        // must NEVER fail an append because nobody is listening.
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::Appended {
                id,
                partition: path.clone(),
                ts_ms: now,
            });
        // Plan 9: SummaryNeeded fires for the leaf so worker loops wake up.
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::SummaryNeeded {
                subject: crate::summary::SummarySubject::Partition(path.clone()),
                ts_ms: now,
            });

        Ok(MemoryRef {
            id,
            partition: path,
        })
    }

    /// Fetch a single memory by ref. Returns [`Error::MemoryNotFound`] when
    /// no row exists, [`Error::Tombstoned`] when the row is soft-deleted,
    /// [`Error::IndexCorrupt`] when the body's content-hash diverges from
    /// the row's `content_hash`.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let rec = mem.get(&r).await?;
    /// assert_eq!(rec.r#ref.id, r.id);
    /// # Ok(()) }
    /// ```
    pub async fn get(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<crate::memory::MemoryRecord> {
        let row = self
            .inner
            .metadata
            .get_memory(&r.id)
            .await?
            .ok_or_else(|| Error::MemoryNotFound(r.id.to_string()))?;
        if row.tombstoned {
            return Err(Error::Tombstoned(r.id.to_string()));
        }
        let body = self
            .inner
            .storage
            .get(&crate::storage::StorageKey::new(row.data_path.clone()))
            .await?;
        let body_str =
            String::from_utf8(body.to_vec()).map_err(|e| Error::storage("non-utf8 body", e))?;
        let content = match row.content_kind.as_str() {
            "md" => crate::content::Content::markdown(body_str),
            _ => crate::content::Content::text(body_str),
        };
        let actual = crate::content::ContentHash::of_content(&content);
        if actual != row.content_hash {
            return Err(Error::IndexCorrupt(format!(
                "content hash mismatch for {}",
                r.id
            )));
        }
        Ok(crate::memory::MemoryRecord {
            r#ref: crate::memory::MemoryRef {
                id: row.id,
                partition: row.partition_path,
            },
            content,
            hash: row.content_hash,
            created_at_ms: row.created_at_ms,
            updated_at_ms: row.updated_at_ms,
            tombstoned: row.tombstoned,
            valid_from_ms: row.valid_from_ms,
            valid_until_ms: row.valid_until_ms,
            kind: row.kind,
        })
    }

    /// List live memories under a partition. Use [`ListOpts`](crate::ListOpts)
    /// to opt into tombstones, set a page limit, or pass a cursor from a
    /// previous [`Page`](crate::Page).
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{ListOpts, Partitions};
    /// let parts = Partitions::new().with("user", "alex");
    /// let page = mem.list(parts, ListOpts::default()).await?;
    /// for r in &page.items { let _ = r; }
    /// # Ok(()) }
    /// ```
    pub async fn list(
        &self,
        partitions: crate::partition::Partitions,
        opts: crate::opts::ListOpts,
    ) -> crate::error::Result<crate::opts::Page<crate::memory::MemoryRef>> {
        let path = partitions.resolve(&self.inner.scheme)?;
        let (rows, mut next_cursor) = self
            .inner
            .metadata
            .list_memories(
                &path,
                opts.limit,
                opts.cursor.as_deref(),
                opts.include_tombstoned,
            )
            .await?;
        // Plan 15: bi-temporal validity post-filter. A row is "valid at
        // t" when (valid_from <= t || NULL) AND (valid_until > t || NULL).
        let items: Vec<crate::memory::MemoryRef> = if let Some(t) = opts.valid_at_ms {
            // Filtering shrinks the page so the cursor inherited from
            // the unfiltered fetch may overshoot; drop it when the
            // filter trims anything.
            let pre_len = rows.len();
            let kept: Vec<_> = rows
                .into_iter()
                .filter(|r| {
                    let from_ok = r.valid_from_ms.is_none_or(|f| f <= t);
                    let until_ok = r.valid_until_ms.is_none_or(|u| u > t);
                    from_ok && until_ok
                })
                .map(|r| crate::memory::MemoryRef {
                    id: r.id,
                    partition: r.partition_path,
                })
                .collect();
            if kept.len() != pre_len {
                next_cursor = None;
            }
            kept
        } else {
            rows.into_iter()
                .map(|r| crate::memory::MemoryRef {
                    id: r.id,
                    partition: r.partition_path,
                })
                .collect()
        };
        Ok(crate::opts::Page { items, next_cursor })
    }

    /// Soft-delete a single memory. Idempotent — repeated calls succeed.
    /// Emits [`MemoryEvent::Deleted`](crate::MemoryEvent::Deleted) +
    /// [`MemoryEvent::SummaryNeeded`](crate::MemoryEvent::SummaryNeeded)
    /// after the SQL transaction commits.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.delete(&r).await?;
    /// # Ok(()) }
    /// ```
    pub async fn delete(&self, r: &crate::memory::MemoryRef) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::Delete,
            partition_path: Some(r.partition.clone()),
            memory_id: Some(r.id),
            detail: serde_json::Value::Null,
        };
        self.inner.metadata.tombstone_memory(&r.id, audit).await?;
        // Plan 9: cascade summary_stale on the deleted memory's chain.
        let mut chain: Vec<crate::partition::PartitionPath> = r.partition.ancestors().collect();
        chain.insert(0, r.partition.clone());
        if let Err(e) = self.inner.metadata.mark_summary_stale(&chain).await {
            tracing::warn!(target: "kiromi-ai-memory.summary_stale", error=%e,
                "mark_summary_stale failed on delete (non-fatal)");
        }
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::Deleted {
                id: r.id,
                ts_ms: now,
            });
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::SummaryNeeded {
                subject: crate::summary::SummarySubject::Partition(r.partition.clone()),
                ts_ms: now,
            });
        Ok(())
    }

    /// Soft-delete every live memory under a partition path. Returns the
    /// number of rows tombstoned. Emits
    /// [`MemoryEvent::PartitionDeleted`](crate::MemoryEvent::PartitionDeleted)
    /// after the SQL commit.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{DeleteOpts, Partitions};
    /// let parts = Partitions::new().with("user", "alex");
    /// let n = mem.delete_partition(parts, DeleteOpts::soft()).await?;
    /// # let _ = n; Ok(()) }
    /// ```
    pub async fn delete_partition(
        &self,
        partitions: crate::partition::Partitions,
        _opts: crate::opts::DeleteOpts,
    ) -> crate::error::Result<u64> {
        let path = partitions.resolve(&self.inner.scheme)?;
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::DeletePartition,
            partition_path: Some(path.clone()),
            memory_id: None,
            detail: serde_json::Value::Null,
        };
        let count = self
            .inner
            .metadata
            .tombstone_partition(&path, audit)
            .await?;
        // Plan 9: cascade stale up.
        let mut chain: Vec<crate::partition::PartitionPath> = path.ancestors().collect();
        chain.insert(0, path.clone());
        if let Err(e) = self.inner.metadata.mark_summary_stale(&chain).await {
            tracing::warn!(target: "kiromi-ai-memory.summary_stale", error=%e,
                "mark_summary_stale failed on delete_partition (non-fatal)");
        }
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::PartitionDeleted {
                partition: path.clone(),
                count,
                ts_ms: now,
            });
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::SummaryNeeded {
                subject: crate::summary::SummarySubject::Partition(path),
                ts_ms: now,
            });
        Ok(count)
    }

    /// Record an explicit link `src → dst` (and its reverse).
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, a: kiromi_ai_memory::MemoryRef, b: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.add_link(&a, &b).await?;
    /// # Ok(()) }
    /// ```
    pub async fn add_link(
        &self,
        src: &crate::memory::MemoryRef,
        dst: &crate::memory::MemoryRef,
    ) -> crate::error::Result<()> {
        self.add_link_typed(src, dst, crate::link::LinkKind::Explicit)
            .await
    }

    /// Plan 15: record a typed link `src → dst` (and its reverse).
    ///
    /// Behaves like [`Memory::add_link`] except the persisted edge
    /// carries the supplied [`crate::link::LinkKind`]. See the
    /// [`crate::link::LinkKind`] doc for the meaning of each kind.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, a: kiromi_ai_memory::MemoryRef, b: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::LinkKind;
    /// mem.add_link_typed(&a, &b, LinkKind::Supersedes).await?;
    /// # Ok(()) }
    /// ```
    pub async fn add_link_typed(
        &self,
        src: &crate::memory::MemoryRef,
        dst: &crate::memory::MemoryRef,
        kind: crate::link::LinkKind,
    ) -> crate::error::Result<()> {
        if src.id == dst.id {
            return Err(Error::LinkInvalid("self-link".into()));
        }
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let link = crate::link::Link {
            src: src.id,
            dst: dst.id,
            kind,
            created_at_ms: now,
        };
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::LinkAdd,
            partition_path: None,
            memory_id: Some(src.id),
            detail: serde_json::json!({
                "dst": dst.id.to_string(),
                "kind": kind.as_persisted_str(),
            }),
        };
        self.inner.metadata.add_link(&link, audit).await?;
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::LinkAdded {
                src: src.id,
                dst: dst.id,
                ts_ms: now,
            });
        Ok(())
    }

    /// Remove an explicit link.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, a: kiromi_ai_memory::MemoryRef, b: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.remove_link(&a, &b).await?;
    /// # Ok(()) }
    /// ```
    pub async fn remove_link(
        &self,
        src: &crate::memory::MemoryRef,
        dst: &crate::memory::MemoryRef,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::LinkRemove,
            partition_path: None,
            memory_id: Some(src.id),
            detail: serde_json::json!({ "dst": dst.id.to_string() }),
        };
        self.inner
            .metadata
            .remove_link(&src.id, &dst.id, audit)
            .await?;
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::LinkRemoved {
                src: src.id,
                dst: dst.id,
                ts_ms: now,
            });
        Ok(())
    }

    /// All links sourced at the given memory.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let links = mem.links_of(&r).await?;
    /// # let _ = links; Ok(()) }
    /// ```
    pub async fn links_of(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<Vec<crate::link::Link>> {
        self.inner.metadata.links_of(&r.id).await
    }

    /// Plan 15: list live memory refs whose [`crate::MemoryKind`] tag
    /// matches, scoped by [`crate::Scope`].
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{MemoryKind, Scope};
    /// let runbooks = mem.find_by_kind(MemoryKind::Procedural, Scope::All).await?;
    /// # let _ = runbooks; Ok(()) }
    /// ```
    pub async fn find_by_kind(
        &self,
        kind: crate::memory::MemoryKind,
        scope: crate::summary::Scope,
    ) -> crate::error::Result<Vec<crate::memory::MemoryRef>> {
        let rows = self
            .inner
            .metadata
            .find_memories_by_kind(kind, &scope)
            .await?;
        Ok(rows
            .into_iter()
            .map(|r| crate::memory::MemoryRef {
                id: r.id,
                partition: r.partition_path,
            })
            .collect())
    }

    /// Plan 15: update the bi-temporal validity range on a memory.
    ///
    /// Both bounds are optional. `None` means "no bound on that side".
    /// A `validity_set` audit row is written and a
    /// [`MemoryEvent::ValidityUpdated`](crate::MemoryEvent::ValidityUpdated)
    /// event fires after the SQL commit.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.set_validity(&r, Some(1_000), Some(2_000)).await?;
    /// # Ok(()) }
    /// ```
    pub async fn set_validity(
        &self,
        r: &crate::memory::MemoryRef,
        valid_from_ms: Option<i64>,
        valid_until_ms: Option<i64>,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::ValiditySet,
            partition_path: Some(r.partition.clone()),
            memory_id: Some(r.id),
            detail: serde_json::json!({
                "valid_from_ms": valid_from_ms,
                "valid_until_ms": valid_until_ms,
            }),
        };
        self.inner
            .metadata
            .set_memory_validity(&r.id, valid_from_ms, valid_until_ms, audit)
            .await?;
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::ValidityUpdated {
                id: r.id,
                ts_ms: now,
            });
        Ok(())
    }

    /// Plan 16: record a typed link between any two nodes (memory,
    /// summary, partition).
    ///
    /// The edge is recorded once (NOT mirrored — pass both `(a, b)`
    /// and `(b, a)` if you want symmetric semantics; the legacy
    /// memory↔memory [`Memory::add_link`] keeps mirroring for
    /// backwards compatibility). Idempotent on the composite key
    /// `(src_kind, src_id, dst_kind, dst_id, kind)`.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{LinkKind, NodeRef};
    /// # let m: kiromi_ai_memory::MemoryRef = unimplemented!();
    /// # let s: kiromi_ai_memory::summary::SummaryRef = unimplemented!();
    /// mem.link(NodeRef::Memory(m), NodeRef::Summary(s), LinkKind::Derived).await?;
    /// # Ok(()) }
    /// ```
    pub async fn link(
        &self,
        src: crate::graph::NodeRef,
        dst: crate::graph::NodeRef,
        kind: crate::link::LinkKind,
    ) -> crate::error::Result<()> {
        if src == dst {
            return Err(Error::LinkInvalid("self-link".into()));
        }
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::NodeLinkAdd,
            partition_path: None,
            memory_id: match &src {
                crate::graph::NodeRef::Memory(m) => Some(m.id),
                _ => None,
            },
            detail: serde_json::json!({
                "src": src,
                "dst": dst,
                "kind": kind.as_persisted_str(),
            }),
        };
        self.inner
            .metadata
            .add_node_link(&src, &dst, kind, now, None, audit)
            .await?;
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::LinkAdded {
                src: match &src {
                    crate::graph::NodeRef::Memory(m) => m.id,
                    _ => crate::memory::MemoryId::generate(),
                },
                dst: match &dst {
                    crate::graph::NodeRef::Memory(m) => m.id,
                    _ => crate::memory::MemoryId::generate(),
                },
                ts_ms: now,
            });
        Ok(())
    }

    /// Plan 16: remove a typed node↔node link. Idempotent.
    pub async fn unlink(
        &self,
        src: crate::graph::NodeRef,
        dst: crate::graph::NodeRef,
        kind: crate::link::LinkKind,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::NodeLinkRemove,
            partition_path: None,
            memory_id: match &src {
                crate::graph::NodeRef::Memory(m) => Some(m.id),
                _ => None,
            },
            detail: serde_json::json!({
                "src": src,
                "dst": dst,
                "kind": kind.as_persisted_str(),
            }),
        };
        self.inner
            .metadata
            .remove_node_link(&src, &dst, kind, audit)
            .await?;
        Ok(())
    }

    /// Plan 16: apply a coordinated batch of catalog updates triggered
    /// by the arrival of `ops.trigger`.
    ///
    /// Single SQL transaction — every op succeeds or every op rolls
    /// back. Raw memory and summary blobs are never touched. Emits a
    /// single [`MemoryEvent::Evolved`](crate::MemoryEvent::Evolved)
    /// post-commit.
    ///
    /// Returns [`EvolutionReport`](crate::EvolutionReport) with the
    /// applied count, the new audit-log seq, and the count of
    /// post-commit events broadcast.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, m_old: kiromi_ai_memory::MemoryRef, m_new: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{
    ///     canonical_keys, AttributeValue, EvolutionOps, LinkKind, MemoryKind, NodeRef,
    /// };
    /// let report = mem
    ///     .evolve(
    ///         EvolutionOps::triggered_by(m_new.clone())
    ///             .with_note("agent: m_new supersedes m_old")
    ///             .add_link(
    ///                 NodeRef::Memory(m_new.clone()),
    ///                 NodeRef::Memory(m_old.clone()),
    ///                 LinkKind::Supersedes,
    ///             )
    ///             .close_validity(m_old.clone(), None, Some(1_700_000_000_000))
    ///             .change_kind(m_old.clone(), MemoryKind::Archival)
    ///             .set_memory_attribute(
    ///                 m_new,
    ///                 canonical_keys::HEADLINE,
    ///                 AttributeValue::String("supersedes m_old".into()),
    ///             ),
    ///     )
    ///     .await?;
    /// # let _ = report; Ok(()) }
    /// ```
    pub async fn evolve(
        &self,
        ops: crate::evolve::EvolutionOps,
    ) -> crate::error::Result<crate::evolve::EvolutionReport> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let trigger_id = ops.trigger.as_ref().map(|t| t.id);
        // Audit detail: stash the JSON form of the ops for replay /
        // inspection. We deliberately serialize the whole struct so
        // the audit row is self-contained.
        let detail = serde_json::json!({
            "trigger": trigger_id.map(|id| id.to_string()),
            "note": ops.note,
            "memory_attributes": ops.memory_attributes.len(),
            "summary_attributes": ops.summary_attributes.len(),
            "links_added": ops.links_added.len(),
            "links_removed": ops.links_removed.len(),
            "validity_updates": ops.validity_updates.len(),
            "kind_updates": ops.kind_updates.len(),
        });
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::Evolve,
            partition_path: ops.trigger.as_ref().map(|t| t.partition.clone()),
            memory_id: trigger_id,
            detail,
        };
        let outcome = self.inner.metadata.apply_evolution(&ops, audit).await?;
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::Evolved {
                trigger: trigger_id,
                applied: outcome.applied,
                audit_seq: outcome.audit_seq,
                ts_ms: now,
            });
        Ok(crate::evolve::EvolutionReport {
            applied: outcome.applied,
            audit_seq: outcome.audit_seq,
            events_emitted: 1,
        })
    }

    /// Plan 16: every edge sourced at `src`, regardless of destination
    /// kind.
    pub async fn edges_from(
        &self,
        src: crate::graph::NodeRef,
    ) -> crate::error::Result<Vec<crate::link::Edge>> {
        self.inner.metadata.node_links_from(&src).await
    }

    /// Plan 16: every edge targeting `dst`, regardless of source
    /// kind.
    pub async fn edges_to(
        &self,
        dst: crate::graph::NodeRef,
    ) -> crate::error::Result<Vec<crate::link::Edge>> {
        self.inner.metadata.node_links_to(&dst).await
    }

    /// Plan 15: links sourced at `r`, optionally filtered by kind.
    ///
    /// Equivalent to [`Memory::links_of`] when `opts.kind_filter` is
    /// `None`; otherwise drops every link whose `kind` does not match.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{LinkKind, LinksOpts};
    /// let supers = mem
    ///     .links_of_with_opts(&r, LinksOpts::default().with_kind(LinkKind::Supersedes))
    ///     .await?;
    /// # let _ = supers; Ok(()) }
    /// ```
    pub async fn links_of_with_opts(
        &self,
        r: &crate::memory::MemoryRef,
        opts: crate::opts::LinksOpts,
    ) -> crate::error::Result<Vec<crate::link::Link>> {
        let all = self.inner.metadata.links_of(&r.id).await?;
        Ok(match opts.kind_filter {
            None => all,
            Some(k) => all.into_iter().filter(|l| l.kind == k).collect(),
        })
    }

    /// Inspector helper: SQL row for one memory (any tombstone state).
    ///
    /// Returns `None` when no row exists. Used by the CLI's `inspect`
    /// subcommand; not part of the regular read path.
    pub async fn metadata_inspect_memory(
        &self,
        id: &crate::memory::MemoryId,
    ) -> crate::error::Result<Option<crate::metadata::MemoryRow>> {
        self.inner.metadata.get_memory(id).await
    }

    /// Inspector helper: every row under a partition (any tombstone state),
    /// ordered as the underlying metadata store returns them. Used by the
    /// CLI's `inspect --partition` subcommand. Page size capped at 1000.
    pub async fn metadata_inspect_partition(
        &self,
        path: &crate::partition::PartitionPath,
    ) -> crate::error::Result<Vec<crate::metadata::MemoryRow>> {
        let (rows, _cursor) = self
            .inner
            .metadata
            .list_memories(path, 1000, None, true)
            .await?;
        Ok(rows)
    }

    /// List partitions under an optional prefix. Pass `None` to walk the
    /// full tenant tree; pass `Some(path)` to walk that subtree (the
    /// supplied path is included in the result).
    ///
    /// Each [`crate::metadata::PartitionInfo`] carries the partition's
    /// level, leaf flag, creation timestamp, and the count of *live*
    /// memories whose `partition_path` equals that exact path (not
    /// recursive). Mind-map / tree-view callers compose this with
    /// [`Memory::list`] to enumerate the tenant's content tree without
    /// having to scrape the SQL store directly.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// let infos = mem.list_partitions(None).await?;
    /// for info in &infos { let _ = info.path.as_str(); }
    /// # Ok(()) }
    /// ```
    pub async fn list_partitions(
        &self,
        prefix: Option<&crate::partition::PartitionPath>,
    ) -> crate::error::Result<Vec<crate::metadata::PartitionInfo>> {
        self.inner.metadata.list_partitions(prefix).await
    }

    // ===== Plan 9: summaries first-class =====

    /// Attach a caller-computed summary to the supplied subject.
    ///
    /// Writes the markdown body to storage under `metadata/<path>/summaries/...`,
    /// inserts a `summary` row, and — if a previous summary exists for the same
    /// `(subject, style)` — sets that row's `superseded_by` pointer. Cascades
    /// `partition.summary_stale` flags upward (parent partitions become stale)
    /// and clears it on the subject's own partition (when applicable). Emits
    /// [`crate::event::MemoryEvent::SummaryAttached`] after the SQL transaction
    /// commits.
    ///
    /// `inputs` records what the summarizer consumed. The engine never inspects
    /// it; it round-trips through JSON for downstream tooling.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::SummarySubject;
    /// use kiromi_ai_memory::summarizer::SummaryStyle;
    /// let sref = mem.attach_summary(
    ///     SummarySubject::Tenant, SummaryStyle::Compact,
    ///     "openai:gpt-4.1:v1", "rolled-up tenant memo".to_string(), vec![],
    /// ).await?;
    /// # let _ = sref; Ok(()) }
    /// ```
    pub async fn attach_summary(
        &self,
        subject: crate::summary::SummarySubject,
        style: crate::summarizer::SummaryStyle,
        summarizer_id: impl Into<String>,
        content: impl Into<crate::summary::content::SummaryContent>,
        inputs: Vec<crate::summary::SummarySubject>,
    ) -> crate::error::Result<crate::summary::SummaryRef> {
        use crate::audit::AuditOp;
        use crate::content::ContentHash;
        use crate::metadata::{AuditEntry, SummaryRow};
        use crate::storage::StorageKey;
        use crate::summary::SummaryId;

        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let summarizer_id = summarizer_id.into();
        let content: crate::summary::content::SummaryContent = content.into();
        let id = SummaryId::generate();

        // Latest version → next version.
        let prior = self.inner.metadata.latest_summary(&subject, &style).await?;
        let next_version = prior.as_ref().map(|p| p.version + 1).unwrap_or(1);

        let data_path = match &subject {
            crate::summary::SummarySubject::Memory(r) => format!(
                "{}/metadata/{}/summaries/memory/{}/{}.v{}.md",
                self.inner.tenant.as_str(),
                r.partition.as_str(),
                r.id,
                id,
                next_version
            ),
            crate::summary::SummarySubject::Partition(p) => format!(
                "{}/metadata/{}/summaries/partition/{}.v{}.md",
                self.inner.tenant.as_str(),
                p.as_str(),
                id,
                next_version
            ),
            crate::summary::SummarySubject::Tenant => format!(
                "{}/metadata/tenant/summaries/{}.v{}.md",
                self.inner.tenant.as_str(),
                id,
                next_version
            ),
        };

        // Plan 11: the Markdown blob carries `prose` verbatim; the JSON
        // sidecar carries the full structured content.
        let key = StorageKey::new(&data_path);
        let body_bytes = content.prose.as_bytes();
        self.inner
            .storage
            .put(&key, bytes::Bytes::copy_from_slice(body_bytes))
            .await?;
        if !content.blocks.is_empty()
            && let Some(sidecar_path) = Self::sidecar_json_key(&data_path)
        {
            let json = serde_json::to_vec(&content)
                .map_err(|e| Error::storage("encode summary sidecar", e))?;
            self.inner
                .storage
                .put(&StorageKey::new(&sidecar_path), bytes::Bytes::from(json))
                .await?;
        }

        let hash = ContentHash::of_bytes(body_bytes);
        let row = SummaryRow {
            id,
            subject_kind: subject.kind_str().to_string(),
            subject_path: subject.partition_path().cloned(),
            subject_memory: subject.memory_id(),
            style: style.as_str().into_owned(),
            version: next_version,
            data_path,
            content_hash: hash,
            bytes: i64::try_from(body_bytes.len()).unwrap_or(0),
            summarizer_id: summarizer_id.clone(),
            inputs: inputs.clone(),
            superseded_by: None,
            tombstoned: false,
            created_at_ms: now,
        };

        let audit_partition = subject.partition_path().cloned();
        let audit_memory = subject.memory_id();
        let audit = AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: AuditOp::SummaryAttach,
            partition_path: audit_partition,
            memory_id: audit_memory,
            detail: serde_json::json!({
                "summary_id": id.to_string(),
                "subject_kind": subject.kind_str(),
                "style": style.as_str(),
                "version": next_version,
                "summarizer_id": summarizer_id,
                "bytes": body_bytes.len(),
                "blocks": content.blocks.len(),
            }),
        };

        // Plan 18: embed the summary body once, up front, so the index inserts
        // can run inside the same SQL transaction as the catalog row. The
        // returned `Option` is `None` only in caller-provided mode without an
        // embedder; in that case the FTS5 row still goes in but `summary_vec`
        // is left empty — `regenerate_embeddings` backfills later.
        let summary_embedding = self.embed_summary_body(&content.prose).await?;
        let summary_blob: Option<Vec<u8>> = summary_embedding
            .as_ref()
            .map(|v| bytemuck::cast_slice(v.as_slice()).to_vec());

        // Plan 18: parent_path convention used by `summary_vec` / `summary_fts`:
        // tenant subjects map to the literal `<root>` sentinel, partition
        // subjects use their own path, memory subjects use the memory's
        // owning partition path.
        let parent_path_for_index: String = match &subject {
            crate::summary::SummarySubject::Memory(r) => r.partition.as_str().to_string(),
            crate::summary::SummarySubject::Partition(p) => p.as_str().to_string(),
            crate::summary::SummarySubject::Tenant => "<root>".to_string(),
        };

        self.inner
            .metadata
            .insert_summary(crate::metadata::InsertSummaryRequest {
                row,
                audit,
                embedding_blob: summary_blob,
                content_for_index: content.prose.clone(),
                parent_path: parent_path_for_index,
            })
            .await?;
        if let Some(p) = prior.as_ref() {
            self.inner.metadata.supersede_summary(&p.id, &id).await?;
        }

        // Plan 10: denormalise inputs into summary_input for traverse +
        // reverse-edge lookups. Idempotent at the SQL level.
        for input in &inputs {
            let (kind, idstr) = match input {
                crate::summary::SummarySubject::Memory(r) => ("memory", r.id.to_string()),
                crate::summary::SummarySubject::Partition(p) => {
                    ("partition", p.as_str().to_string())
                }
                crate::summary::SummarySubject::Tenant => ("tenant", "<root>".to_string()),
            };
            self.inner
                .metadata
                .insert_summary_input(&id, kind, &idstr)
                .await?;
        }

        // Plan 11: walk the structured blocks for inline citations and write
        // one `summary_input` row per `DataPointRef` (with sub-position info)
        // and per `PartitionRef`. These rows feed `references_to` lookups.
        for dp in content.data_point_refs() {
            self.inner
                .metadata
                .insert_summary_input_with_range(
                    &id,
                    "memory",
                    &dp.memory_id.to_string(),
                    &dp.as_input_range(),
                )
                .await?;
        }
        for pr in content.partition_refs() {
            let range = crate::summary::content::SummaryInputRange {
                note: pr.note.clone(),
                ..Default::default()
            };
            self.inner
                .metadata
                .insert_summary_input_with_range(&id, "partition", pr.path.as_str(), &range)
                .await?;
        }

        // Plan 18 dispatch 2 wired summary_vec + summary_fts inserts into the
        // `insert_summary` SQL transaction; dispatch 4 deleted the legacy
        // `ChildSummaries` IndexHandle propagation that previously rode
        // alongside it. Hierarchical descent reads straight from the SQL
        // virtual tables.

        // Stale cascade.
        self.cascade_summary_stale_for_subject(&subject, true)
            .await?;

        let version_u32 = u32::try_from(next_version).unwrap_or(0);
        let r#ref = crate::summary::SummaryRef {
            id,
            subject: subject.clone(),
            style: style.clone(),
            version: version_u32,
        };
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::SummaryAttached {
                id,
                subject,
                style,
                version: version_u32,
                ts_ms: now,
            });
        Ok(r#ref)
    }

    /// Plan 16: thin wrapper around [`Memory::attach_summary`] that also
    /// applies a batch of typed attributes from [`crate::AttachSummaryOpts`]
    /// after the summary row commits.
    ///
    /// Each attribute is written in its own transaction (one
    /// `summary_attribute_set` audit entry per call). For atomic
    /// multi-attribute writes, attach first then call
    /// [`crate::Memory::evolve`] with the desired set.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::SummarySubject;
    /// use kiromi_ai_memory::summarizer::SummaryStyle;
    /// use kiromi_ai_memory::AttachSummaryOpts;
    /// let sref = mem
    ///     .attach_summary_with_opts(
    ///         SummarySubject::Tenant,
    ///         SummaryStyle::Compact,
    ///         "openai:gpt-4.1",
    ///         "weekly".to_string(),
    ///         vec![],
    ///         AttachSummaryOpts::default()
    ///             .with_keywords(["weekly", "rollup"])
    ///             .with_headline("Weekly rollup"),
    ///     )
    ///     .await?;
    /// # let _ = sref; Ok(()) }
    /// ```
    pub async fn attach_summary_with_opts(
        &self,
        subject: crate::summary::SummarySubject,
        style: crate::summarizer::SummaryStyle,
        summarizer_id: impl Into<String>,
        content: impl Into<crate::summary::content::SummaryContent>,
        inputs: Vec<crate::summary::SummarySubject>,
        opts: crate::opts::AttachSummaryOpts,
    ) -> crate::error::Result<crate::summary::SummaryRef> {
        let sref = self
            .attach_summary(subject, style, summarizer_id, content, inputs)
            .await?;
        for (k, v) in opts.attributes {
            self.summary_set_attribute(&sref, &k, v).await?;
        }
        Ok(sref)
    }

    /// Look up a summary by [`crate::summary::SummaryRef`]. Reads the body
    /// blob from storage. Returns [`crate::error::Error::SummaryNotFound`]
    /// when the row is absent or tombstoned.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, sr: kiromi_ai_memory::summary::SummaryRef) -> kiromi_ai_memory::Result<()> {
    /// let rec = mem.get_summary(&sr).await?;
    /// # let _ = rec.content.prose; Ok(()) }
    /// ```
    pub async fn get_summary(
        &self,
        r: &crate::summary::SummaryRef,
    ) -> crate::error::Result<crate::summary::SummaryRecord> {
        let row = self
            .inner
            .metadata
            .get_summary(&r.id)
            .await?
            .ok_or_else(|| Error::SummaryNotFound(r.id.to_string()))?;
        if row.tombstoned {
            return Err(Error::SummaryNotFound(r.id.to_string()));
        }
        let body = self
            .inner
            .storage
            .get(&crate::storage::StorageKey::new(row.data_path.clone()))
            .await?;
        let sidecar = self.try_get_sidecar(&row.data_path).await?;
        Self::row_to_summary_record(row, body.as_ref(), sidecar.as_deref())
    }

    /// Latest live summary for `(subject, style)`, if any.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{summary::SummarySubject, summarizer::SummaryStyle};
    /// let s = mem.latest_summary(&SummarySubject::Tenant, &SummaryStyle::Compact).await?;
    /// # let _ = s; Ok(()) }
    /// ```
    pub async fn latest_summary(
        &self,
        subject: &crate::summary::SummarySubject,
        style: &crate::summarizer::SummaryStyle,
    ) -> crate::error::Result<Option<crate::summary::SummaryRecord>> {
        let row = self.inner.metadata.latest_summary(subject, style).await?;
        if let Some(row) = row {
            let body = self
                .inner
                .storage
                .get(&crate::storage::StorageKey::new(row.data_path.clone()))
                .await?;
            let sidecar = self.try_get_sidecar(&row.data_path).await?;
            Ok(Some(Self::row_to_summary_record(
                row,
                body.as_ref(),
                sidecar.as_deref(),
            )?))
        } else {
            Ok(None)
        }
    }

    /// All live summaries for a subject, ordered version DESC then created DESC.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::SummarySubject;
    /// let refs = mem.summaries_of(&SummarySubject::Tenant).await?;
    /// # let _ = refs; Ok(()) }
    /// ```
    pub async fn summaries_of(
        &self,
        subject: &crate::summary::SummarySubject,
    ) -> crate::error::Result<Vec<crate::summary::SummaryRef>> {
        let rows = self.inner.metadata.list_summaries_of(subject).await?;
        Ok(rows
            .into_iter()
            .map(|row| crate::summary::SummaryRef {
                id: row.id,
                subject: subject.clone(),
                style: crate::summarizer::SummaryStyle::from_persisted(&row.style),
                version: u32::try_from(row.version).unwrap_or(0),
            })
            .collect())
    }

    /// Soft-tombstone a summary. Best-effort blob deletion. Emits
    /// [`crate::event::MemoryEvent::SummaryDeleted`].
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, sr: kiromi_ai_memory::summary::SummaryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.delete_summary(&sr).await?;
    /// # Ok(()) }
    /// ```
    pub async fn delete_summary(&self, r: &crate::summary::SummaryRef) -> crate::error::Result<()> {
        use crate::audit::AuditOp;
        use crate::metadata::AuditEntry;

        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        // Look up before we tombstone so we know the storage key.
        let row = self.inner.metadata.get_summary(&r.id).await?;
        let audit = AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: AuditOp::SummaryDelete,
            partition_path: r.subject.partition_path().cloned(),
            memory_id: r.subject.memory_id(),
            detail: serde_json::json!({ "summary_id": r.id.to_string() }),
        };
        self.inner.metadata.delete_summary(&r.id, audit).await?;
        if let Some(row) = row {
            let _ = self
                .inner
                .storage
                .delete(&crate::storage::StorageKey::new(row.data_path))
                .await;
        }
        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::SummaryDeleted {
                id: r.id,
                ts_ms: now,
            });
        Ok(())
    }

    // ===== Plan 11: citation reverse edges =====

    /// Every summary that cites the supplied memory, with each citation's
    /// per-row sub-position info (when the citation carried one). Order
    /// is creation-time DESC.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let refs = mem.references_to(&r).await?;
    /// # let _ = refs; Ok(()) }
    /// ```
    pub async fn references_to(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<
        Vec<(
            crate::summary::SummaryRef,
            crate::summary::content::DataPointRef,
        )>,
    > {
        let rows = self
            .inner
            .metadata
            .summaries_citing_with_ranges("memory", &r.id.to_string())
            .await?;
        let mut out = Vec::with_capacity(rows.len());
        for (sid, range) in rows {
            let Some(row) = self.inner.metadata.get_summary(&sid).await? else {
                continue;
            };
            if row.tombstoned {
                continue;
            }
            let subject = subject_from_summary_row(&row)?;
            let s_ref = crate::summary::SummaryRef {
                id: row.id,
                subject,
                style: crate::summarizer::SummaryStyle::from_persisted(&row.style),
                version: u32::try_from(row.version).unwrap_or(0),
            };
            let dp = crate::summary::content::DataPointRef::from_input_range(r.id, &range);
            out.push((s_ref, dp));
        }
        Ok(out)
    }

    /// Every summary that cites the supplied partition (whole-partition
    /// citations only — partition citations don't carry sub-positions).
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, p: kiromi_ai_memory::PartitionPath) -> kiromi_ai_memory::Result<()> {
    /// let refs = mem.references_to_partition(&p).await?;
    /// # let _ = refs; Ok(()) }
    /// ```
    pub async fn references_to_partition(
        &self,
        path: &crate::partition::PartitionPath,
    ) -> crate::error::Result<Vec<crate::summary::SummaryRef>> {
        let rows = self
            .inner
            .metadata
            .summaries_citing_with_ranges("partition", path.as_str())
            .await?;
        let mut out = Vec::with_capacity(rows.len());
        for (sid, _) in rows {
            let Some(row) = self.inner.metadata.get_summary(&sid).await? else {
                continue;
            };
            if row.tombstoned {
                continue;
            }
            let subject = subject_from_summary_row(&row)?;
            out.push(crate::summary::SummaryRef {
                id: row.id,
                subject,
                style: crate::summarizer::SummaryStyle::from_persisted(&row.style),
                version: u32::try_from(row.version).unwrap_or(0),
            });
        }
        Ok(out)
    }

    // ===== Plan 11: typed memory attributes =====

    /// Attach (or overwrite) one typed attribute on an existing memory.
    /// Writes an `attribute_set` audit entry in the same SQL transaction.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.set_attribute(&r, "speaker", "alex".into()).await?;
    /// # Ok(()) }
    /// ```
    pub async fn set_attribute(
        &self,
        r: &crate::memory::MemoryRef,
        key: &str,
        value: crate::attribute::AttributeValue,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::AttributeSet,
            partition_path: Some(r.partition.clone()),
            memory_id: Some(r.id),
            detail: serde_json::json!({
                "key": key,
                "kind": value.kind_str(),
                "value": value,
            }),
        };
        self.inner
            .metadata
            .set_attribute(&r.id, key, &value, audit)
            .await
    }

    /// Clear one attribute. Idempotent — silently no-ops when the row was
    /// absent. Writes an `attribute_clear` audit entry either way.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.clear_attribute(&r, "speaker").await?;
    /// # Ok(()) }
    /// ```
    pub async fn clear_attribute(
        &self,
        r: &crate::memory::MemoryRef,
        key: &str,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::AttributeClear,
            partition_path: Some(r.partition.clone()),
            memory_id: Some(r.id),
            detail: serde_json::json!({ "key": key }),
        };
        self.inner.metadata.clear_attribute(&r.id, key, audit).await
    }

    /// Read one attribute.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let v = mem.get_attribute(&r, "speaker").await?;
    /// # let _ = v; Ok(()) }
    /// ```
    pub async fn get_attribute(
        &self,
        r: &crate::memory::MemoryRef,
        key: &str,
    ) -> crate::error::Result<Option<crate::attribute::AttributeValue>> {
        self.inner.metadata.get_attribute(&r.id, key).await
    }

    /// All attributes attached to a memory, sorted by key.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let attrs = mem.attributes_of(&r).await?;
    /// for (k, v) in &attrs { let _ = (k, v); }
    /// # Ok(()) }
    /// ```
    pub async fn attributes_of(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<std::collections::BTreeMap<String, crate::attribute::AttributeValue>>
    {
        self.inner.metadata.list_attributes(&r.id).await
    }

    /// Find every memory whose `(key, value)` matches exactly. Backed by
    /// the kind-specific partial index on `memory_attribute`.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// let rs = mem.find_by_attribute("speaker", &"alex".into()).await?;
    /// # let _ = rs; Ok(()) }
    /// ```
    pub async fn find_by_attribute(
        &self,
        key: &str,
        value: &crate::attribute::AttributeValue,
    ) -> crate::error::Result<Vec<crate::memory::MemoryRef>> {
        let ids = self.inner.metadata.find_by_attribute(key, value).await?;
        self.ids_to_memory_refs(ids).await
    }

    /// Find every memory whose attribute lies in `[min, max]`. Both ends
    /// must share the same orderable kind (`Int`, `Decimal`, `Timestamp`);
    /// otherwise [`crate::error::Error::InvalidAttribute`].
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::AttributeValue;
    /// let rs = mem.find_by_attribute_range("ts_ms",
    ///     &AttributeValue::Timestamp(0), &AttributeValue::Timestamp(i64::MAX)).await?;
    /// # let _ = rs; Ok(()) }
    /// ```
    pub async fn find_by_attribute_range(
        &self,
        key: &str,
        min: &crate::attribute::AttributeValue,
        max: &crate::attribute::AttributeValue,
    ) -> crate::error::Result<Vec<crate::memory::MemoryRef>> {
        let ids = self
            .inner
            .metadata
            .find_by_attribute_range(key, min, max)
            .await?;
        self.ids_to_memory_refs(ids).await
    }

    /// Internal: hydrate `MemoryId`s into `MemoryRef`s by joining against
    /// the `memory` table to recover each row's partition path.
    async fn ids_to_memory_refs(
        &self,
        ids: Vec<crate::memory::MemoryId>,
    ) -> crate::error::Result<Vec<crate::memory::MemoryRef>> {
        let mut out = Vec::with_capacity(ids.len());
        for id in ids {
            if let Some(row) = self.inner.metadata.get_memory(&id).await?
                && !row.tombstoned
            {
                out.push(crate::memory::MemoryRef {
                    id: row.id,
                    partition: row.partition_path,
                });
            }
        }
        Ok(out)
    }

    /// Plan 16: upsert one typed attribute on a summary row. Audit op
    /// is `summary_attribute_set`.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, sref: kiromi_ai_memory::SummaryRef) -> kiromi_ai_memory::Result<()> {
    /// mem.summary_set_attribute(&sref, "tags", vec!["weekly", "rollup"]).await?;
    /// # Ok(()) }
    /// ```
    pub async fn summary_set_attribute(
        &self,
        sref: &crate::summary::SummaryRef,
        key: &str,
        value: crate::attribute::AttributeValue,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::SummaryAttributeSet,
            partition_path: sref.subject.partition_path().cloned(),
            memory_id: sref.subject.memory_id(),
            detail: serde_json::json!({
                "summary_id": sref.id.to_string(),
                "key": key,
                "kind": value.kind_str(),
                "value": value,
            }),
        };
        self.inner
            .metadata
            .set_summary_attribute(&sref.id, key, &value, audit)
            .await
    }

    /// Plan 16: clear one typed attribute on a summary row. Idempotent.
    pub async fn summary_clear_attribute(
        &self,
        sref: &crate::summary::SummaryRef,
        key: &str,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::SummaryAttributeClear,
            partition_path: sref.subject.partition_path().cloned(),
            memory_id: sref.subject.memory_id(),
            detail: serde_json::json!({
                "summary_id": sref.id.to_string(),
                "key": key,
            }),
        };
        self.inner
            .metadata
            .clear_summary_attribute(&sref.id, key, audit)
            .await
    }

    /// Plan 16: read one typed attribute on a summary row.
    pub async fn summary_get_attribute(
        &self,
        sref: &crate::summary::SummaryRef,
        key: &str,
    ) -> crate::error::Result<Option<crate::attribute::AttributeValue>> {
        self.inner
            .metadata
            .get_summary_attribute(&sref.id, key)
            .await
    }

    /// Plan 16: every typed attribute on a summary, sorted by key.
    pub async fn summary_attributes_of(
        &self,
        sref: &crate::summary::SummaryRef,
    ) -> crate::error::Result<std::collections::BTreeMap<String, crate::attribute::AttributeValue>>
    {
        self.inner.metadata.list_summary_attributes(&sref.id).await
    }

    /// Plan 16: find every summary whose `(key, value)` matches exactly.
    pub async fn find_summaries_by_attribute(
        &self,
        key: &str,
        value: &crate::attribute::AttributeValue,
    ) -> crate::error::Result<Vec<crate::summary::SummaryRef>> {
        let ids = self
            .inner
            .metadata
            .find_summaries_by_attribute(key, value)
            .await?;
        let mut out = Vec::with_capacity(ids.len());
        for sid in ids {
            if let Some(row) = self.inner.metadata.get_summary(&sid).await? {
                if row.tombstoned {
                    continue;
                }
                let subject = subject_from_summary_row(&row)?;
                out.push(crate::summary::SummaryRef {
                    id: row.id,
                    subject,
                    style: crate::summarizer::SummaryStyle::from_persisted(&row.style),
                    version: u32::try_from(row.version).unwrap_or(0),
                });
            }
        }
        Ok(out)
    }

    /// Tenant-level memo convenience — equivalent to
    /// `latest_summary(Tenant, Detailed)`. Returns the rendered prose.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// let memo = mem.tenant_memo().await?;
    /// # let _ = memo; Ok(()) }
    /// ```
    pub async fn tenant_memo(&self) -> crate::error::Result<Option<String>> {
        let rec = self
            .latest_summary(
                &crate::summary::SummarySubject::Tenant,
                &crate::summarizer::SummaryStyle::Detailed,
            )
            .await?;
        Ok(rec.map(|r| r.content.prose))
    }

    /// Partition-level convenience — returns the latest live summary's prose.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, path: kiromi_ai_memory::PartitionPath) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summarizer::SummaryStyle;
    /// let p = mem.partition_summary(&path, &SummaryStyle::Compact).await?;
    /// # let _ = p; Ok(()) }
    /// ```
    pub async fn partition_summary(
        &self,
        path: &crate::partition::PartitionPath,
        style: &crate::summarizer::SummaryStyle,
    ) -> crate::error::Result<Option<String>> {
        let rec = self
            .latest_summary(
                &crate::summary::SummarySubject::Partition(path.clone()),
                style,
            )
            .await?;
        Ok(rec.map(|r| r.content.prose))
    }

    /// Explicit "I changed something out-of-band; please mark stale".
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, path: kiromi_ai_memory::PartitionPath) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::StaleKind;
    /// mem.mark_partition_stale(&path, StaleKind::Both).await?;
    /// # Ok(()) }
    /// ```
    pub async fn mark_partition_stale(
        &self,
        path: &crate::partition::PartitionPath,
        what: crate::summary::StaleKind,
    ) -> crate::error::Result<()> {
        let now = self.inner.clock.now_ms();
        let mut paths: Vec<crate::partition::PartitionPath> = path.ancestors().collect();
        paths.insert(0, path.clone());
        match what {
            crate::summary::StaleKind::Summary => {
                self.inner.metadata.mark_summary_stale(&paths).await?;
                let _ = self
                    .inner
                    .event_tx
                    .send(crate::event::MemoryEvent::SummaryNeeded {
                        subject: crate::summary::SummarySubject::Partition(path.clone()),
                        ts_ms: now,
                    });
            }
            crate::summary::StaleKind::ChildIndex => {
                self.inner.metadata.mark_child_index_stale(&paths).await?;
            }
            crate::summary::StaleKind::Both => {
                self.inner.metadata.mark_summary_stale(&paths).await?;
                self.inner.metadata.mark_child_index_stale(&paths).await?;
                let _ = self
                    .inner
                    .event_tx
                    .send(crate::event::MemoryEvent::SummaryNeeded {
                        subject: crate::summary::SummarySubject::Partition(path.clone()),
                        ts_ms: now,
                    });
            }
        }
        Ok(())
    }

    /// Internal: embed a summary body via the configured embedder under the
    /// `Document` role. Returns `None` when no embedder is configured.
    async fn embed_summary_body(&self, body: &str) -> crate::error::Result<Option<Vec<f32>>> {
        let Some(embedder) = self.inner.embedder.as_ref() else {
            return Ok(None);
        };
        let mut got = embedder
            .embed(crate::embedder::EmbedRole::Document, &[body])
            .await?;
        if got.len() != 1 {
            return Err(Error::embedder(
                "embedder returned wrong shape for summary",
                std::io::Error::new(std::io::ErrorKind::InvalidData, "shape"),
            ));
        }
        let v = got.pop().ok_or_else(|| {
            Error::embedder(
                "embedder returned 0 vectors for summary",
                std::io::Error::new(std::io::ErrorKind::InvalidData, "empty"),
            )
        })?;
        Ok(Some(v))
    }

    /// Internal: cascade `summary_stale` flags after `attach_summary`.
    /// Clears the subject's own partition (when applicable) and marks the
    /// parent chain stale.
    async fn cascade_summary_stale_for_subject(
        &self,
        subject: &crate::summary::SummarySubject,
        clear_self: bool,
    ) -> crate::error::Result<()> {
        match subject {
            crate::summary::SummarySubject::Memory(r) => {
                // Memory summary doesn't clear the partition — the partition
                // summary is independent. But the memory's partition + ancestors
                // become stale: leaf rollup needs to re-pick up this memory's
                // new compact body.
                let mut chain: Vec<crate::partition::PartitionPath> =
                    r.partition.ancestors().collect();
                chain.insert(0, r.partition.clone());
                self.inner.metadata.mark_summary_stale(&chain).await?;
            }
            crate::summary::SummarySubject::Partition(p) => {
                if clear_self {
                    self.inner.metadata.clear_summary_stale(p).await?;
                }
                let parents: Vec<crate::partition::PartitionPath> = p.ancestors().collect();
                if !parents.is_empty() {
                    self.inner.metadata.mark_summary_stale(&parents).await?;
                }
            }
            crate::summary::SummarySubject::Tenant => {
                // Tenant memo regenerated; nothing to mark stale further.
            }
        }
        Ok(())
    }

    /// Internal: row → public record.
    ///
    /// Plan 11: when a structured-content sidecar is provided
    /// (`{id}.v{N}.json`), it overrides the Markdown blob's role; otherwise
    /// the record is prose-only from the body.
    fn row_to_summary_record(
        row: crate::metadata::SummaryRow,
        body: &[u8],
        sidecar_json: Option<&[u8]>,
    ) -> crate::error::Result<crate::summary::SummaryRecord> {
        let body = std::str::from_utf8(body)
            .map_err(|e| Error::storage("non-utf8 summary body", e))?
            .to_string();
        let content: crate::summary::content::SummaryContent = match sidecar_json {
            Some(j) => serde_json::from_slice(j).unwrap_or_else(|_| body.clone().into()),
            None => body.clone().into(),
        };
        let style = crate::summarizer::SummaryStyle::from_persisted(&row.style);
        let subject = if row.subject_kind == "memory" {
            let mid = row
                .subject_memory
                .ok_or_else(|| Error::IndexCorrupt("memory subject missing memory id".into()))?;
            let path = row.subject_path.clone().ok_or_else(|| {
                Error::IndexCorrupt("memory subject missing partition path".into())
            })?;
            crate::summary::SummarySubject::Memory(crate::memory::MemoryRef {
                id: mid,
                partition: path,
            })
        } else if row.subject_kind == "partition" {
            let p = row
                .subject_path
                .clone()
                .ok_or_else(|| Error::IndexCorrupt("partition subject missing path".into()))?;
            crate::summary::SummarySubject::Partition(p)
        } else {
            crate::summary::SummarySubject::Tenant
        };
        Ok(crate::summary::SummaryRecord {
            r#ref: crate::summary::SummaryRef {
                id: row.id,
                subject,
                style,
                version: u32::try_from(row.version).unwrap_or(0),
            },
            content,
            summarizer_id: row.summarizer_id,
            inputs: row.inputs,
            superseded_by: row.superseded_by,
            created_at_ms: row.created_at_ms,
        })
    }

    /// Internal: derive the JSON sidecar storage key from a `.md` data path.
    /// Returns `None` if the path does not end in `.md` (e.g. legacy rows).
    pub(crate) fn sidecar_json_key(data_path: &str) -> Option<String> {
        data_path.strip_suffix(".md").map(|p| format!("{p}.json"))
    }

    /// Internal: best-effort fetch of the structured sidecar. Returns
    /// `Ok(None)` when the sidecar is missing — we treat that as the
    /// "prose-only" pathway. Other errors propagate.
    async fn try_get_sidecar(&self, data_path: &str) -> crate::error::Result<Option<Vec<u8>>> {
        let Some(key) = Self::sidecar_json_key(data_path) else {
            return Ok(None);
        };
        match self
            .inner
            .storage
            .get(&crate::storage::StorageKey::new(key))
            .await
        {
            Ok(b) => Ok(Some(b.to_vec())),
            // Treat any storage error as "no sidecar"; structured content is
            // optional. The Markdown blob is the source of truth for prose.
            Err(_) => Ok(None),
        }
    }

    /// Stream subjects whose `summary_stale` flag is on, deepest-first.
    /// Caller drives a worker loop that consumes this stream and calls
    /// `attach_summary` for each one.
    ///
    /// Returns a one-shot stream materialised from a single SQL snapshot.
    /// When `scope` is [`crate::summary::Scope::All`] or
    /// [`crate::summary::Scope::Tenant`], the tenant subject is appended at
    /// the end so child rollups regenerate before the tenant memo.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) {
    /// use kiromi_ai_memory::summary::Scope;
    /// use kiromi_ai_memory::summarizer::SummaryStyle;
    /// let _stream = mem.subjects_needing_summary(Scope::All, SummaryStyle::Compact);
    /// # }
    /// ```
    pub fn subjects_needing_summary(
        &self,
        scope: crate::summary::Scope,
        style: crate::summarizer::SummaryStyle,
    ) -> impl futures::Stream<Item = crate::error::Result<crate::summary::SummarySubject>> + 'static
    {
        use futures::stream::{self, StreamExt};
        let inner = self.inner.clone();
        stream::once(async move {
            inner
                .metadata
                .subjects_needing_summary(&scope, &style)
                .await
        })
        .flat_map(|res| match res {
            Ok(items) => stream::iter(items.into_iter().map(Ok)).left_stream(),
            Err(e) => stream::iter(std::iter::once(Err(e))).right_stream(),
        })
    }

    /// Plan 12: stream subjects under `scope` that the caller's
    /// summarizer should regenerate. When `opts.force == false` this
    /// is a thin wrapper over [`Self::subjects_needing_summary`]
    /// (only stale subjects). When `opts.force == true` every subject
    /// under scope (memories + partitions + tenant memo) is yielded
    /// regardless of stale flag — useful for mass re-summarisation
    /// after a model upgrade.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) {
    /// use kiromi_ai_memory::summary::Scope;
    /// use kiromi_ai_memory::regen::RegenSubjectOpts;
    /// let _stream = mem.subjects_to_regenerate(Scope::All, RegenSubjectOpts::default());
    /// # }
    /// ```
    pub fn subjects_to_regenerate(
        &self,
        scope: crate::summary::Scope,
        opts: RegenSubjectOpts,
    ) -> impl futures::Stream<Item = crate::error::Result<crate::summary::SummarySubject>> + 'static
    {
        use futures::stream::{self, StreamExt};
        let inner = self.inner.clone();
        let style = opts.style.clone();
        let force = opts.force;
        stream::once(async move {
            if force {
                subjects_in_scope(&inner, &scope).await
            } else {
                inner
                    .metadata
                    .subjects_needing_summary(&scope, &style)
                    .await
            }
        })
        .flat_map(|res| match res {
            Ok(items) => stream::iter(items.into_iter().map(Ok)).left_stream(),
            Err(e) => stream::iter(std::iter::once(Err(e))).right_stream(),
        })
    }

    /// Inputs the caller's summarizer should consume for `subject`.
    /// - Memory → the memory's own body.
    /// - Partition (leaf) → live memories' bodies, preferring each memory's
    ///   compact summary when one already exists.
    /// - Partition (internal) → children's compact summaries (deepest-first
    ///   regen ensures freshness).
    /// - Tenant → top-level partitions' compact summaries.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::SummarySubject;
    /// let inputs = mem.subject_inputs(&SummarySubject::Tenant).await?;
    /// # let _ = inputs; Ok(()) }
    /// ```
    pub async fn subject_inputs(
        &self,
        subject: &crate::summary::SummarySubject,
    ) -> crate::error::Result<Vec<String>> {
        match subject {
            crate::summary::SummarySubject::Memory(r) => {
                let rec = self.get(r).await?;
                Ok(vec![rec.content.as_str().to_string()])
            }
            crate::summary::SummarySubject::Partition(p) => {
                let is_leaf = self.inner.metadata.partition_is_leaf(p).await?;
                if is_leaf {
                    let (rows, _) = self.inner.metadata.list_memories(p, 0, None, false).await?;
                    // Plan-18 review I-7: coalesce the per-memory `latest_summary`
                    // lookups into a single SQL roundtrip. With dozens of memories
                    // per leaf this halves the wall-time on every summarisation
                    // tick driven by `subjects_needing_summary`.
                    let ids: Vec<_> = rows.iter().map(|r| r.id).collect();
                    let paths: Vec<_> = rows.iter().map(|r| r.partition_path.clone()).collect();
                    let summary_by_id = self
                        .inner
                        .metadata
                        .latest_memory_summaries_batch(
                            &ids,
                            &paths,
                            &crate::summarizer::SummaryStyle::Compact,
                        )
                        .await?;
                    let mut out = Vec::with_capacity(rows.len());
                    for row in rows {
                        let content = match summary_by_id.get(&row.id) {
                            Some(s) => {
                                self.inner
                                    .storage
                                    .get(&crate::storage::StorageKey::new(s.data_path.clone()))
                                    .await?
                            }
                            None => {
                                self.inner
                                    .storage
                                    .get(&crate::storage::StorageKey::new(row.data_path))
                                    .await?
                            }
                        };
                        out.push(String::from_utf8_lossy(&content).into_owned());
                    }
                    Ok(out)
                } else {
                    let children = self.inner.metadata.children_of(p).await?;
                    let mut out = Vec::with_capacity(children.len());
                    for c in children {
                        if let Some(latest) = self
                            .inner
                            .metadata
                            .latest_summary(
                                &crate::summary::SummarySubject::Partition(c),
                                &crate::summarizer::SummaryStyle::Compact,
                            )
                            .await?
                        {
                            let body = self
                                .inner
                                .storage
                                .get(&crate::storage::StorageKey::new(latest.data_path))
                                .await?;
                            out.push(String::from_utf8_lossy(&body).into_owned());
                        }
                    }
                    Ok(out)
                }
            }
            crate::summary::SummarySubject::Tenant => {
                let tops = self.inner.metadata.top_level_partitions().await?;
                let mut out = Vec::with_capacity(tops.len());
                for t in tops {
                    if let Some(latest) = self
                        .inner
                        .metadata
                        .latest_summary(
                            &crate::summary::SummarySubject::Partition(t),
                            &crate::summarizer::SummaryStyle::Compact,
                        )
                        .await?
                    {
                        let body = self
                            .inner
                            .storage
                            .get(&crate::storage::StorageKey::new(latest.data_path))
                            .await?;
                        out.push(String::from_utf8_lossy(&body).into_owned());
                    }
                }
                Ok(out)
            }
        }
    }

    /// Run a top-K vector search using a caller-supplied query vector.
    ///
    /// Typed alternative to `Query::semantic("").with_embedding(vec)`
    /// for code paths where there is no natural query text — e.g.
    /// "find memories similar to this thing" navigation in a mind-map
    /// UI, or LLM-backed agentic retrieval that is already working in
    /// vector space.
    ///
    /// `scope` restricts the search to a partition subtree (and its
    /// descendants); `None` searches every leaf in the tenant.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// let hits = mem.search_by_embedding(vec![0.0; 384], 10, None).await?;
    /// # let _ = hits; Ok(()) }
    /// ```
    pub async fn search_by_embedding(
        &self,
        vector: Vec<f32>,
        k: usize,
        scope: Option<crate::partition::PartitionPath>,
    ) -> crate::error::Result<Vec<crate::query::SearchHit>> {
        let mut q = crate::query::Query::semantic("").with_embedding(vector);
        if let Some(p) = scope {
            q = q.within(p);
        }
        self.search(q, k).await
    }

    // ===== Plan 10: bulk in-scope fetchers =====

    /// Every live link whose endpoints both fall under `scope`.
    ///
    /// SQL-only: no storage hit. Mind-map renderers compose this with
    /// [`Memory::summaries_in_scope`] and [`Memory::memories_in_scope`] to
    /// pull the full graph state in three round trips before assembling
    /// client-side.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::Scope;
    /// let links = mem.links_in_scope(Scope::All).await?;
    /// # let _ = links; Ok(()) }
    /// ```
    pub async fn links_in_scope(
        &self,
        scope: crate::summary::Scope,
    ) -> crate::error::Result<Vec<crate::link::Link>> {
        self.inner.metadata.links_in_subtree(&scope).await
    }

    /// Every live summary whose subject falls under `scope`.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::Scope;
    /// let refs = mem.summaries_in_scope(Scope::Tenant).await?;
    /// # let _ = refs; Ok(()) }
    /// ```
    pub async fn summaries_in_scope(
        &self,
        scope: crate::summary::Scope,
    ) -> crate::error::Result<Vec<crate::summary::SummaryRef>> {
        let rows = self.inner.metadata.summaries_in_subtree(&scope).await?;
        Ok(rows
            .into_iter()
            .map(|row| {
                let subject = subject_from_summary_row(&row)
                    .unwrap_or(crate::summary::SummarySubject::Tenant);
                crate::summary::SummaryRef {
                    id: row.id,
                    subject,
                    style: crate::summarizer::SummaryStyle::from_persisted(&row.style),
                    version: u32::try_from(row.version).unwrap_or(0),
                }
            })
            .collect())
    }

    /// Every live memory whose partition falls under `scope`.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::summary::Scope;
    /// let refs = mem.memories_in_scope(Scope::All).await?;
    /// # let _ = refs; Ok(()) }
    /// ```
    pub async fn memories_in_scope(
        &self,
        scope: crate::summary::Scope,
    ) -> crate::error::Result<Vec<crate::memory::MemoryRef>> {
        let rows = self.inner.metadata.memories_in_subtree(&scope).await?;
        Ok(rows
            .into_iter()
            .map(|r| crate::memory::MemoryRef {
                id: r.id,
                partition: r.partition_path,
            })
            .collect())
    }

    // ===== Plan 10: multi-hop traversal =====

    /// BFS-walk the memory ↔ summary ↔ partition graph from `start`, up to
    /// `max_hops` deep, capped by `opts.max_nodes`. Returns a [`crate::graph::Graph`] that
    /// callers (mind-map renderers, debug UIs, future MCP `read_graph` impl)
    /// can consume directly.
    ///
    /// Edges followed:
    /// - `Memory ↔ Memory` via the `link` table (both directions).
    /// - `Summary → Memory/Partition/Summary` via the denormalised
    ///   `summary_input` table (and the reverse via `summaries_citing`).
    /// - `Memory ↔ Partition` via the row's `partition_path`.
    /// - `Partition → child Partition / contained Memory` via the SQL
    ///   `partition` + `memory` tables.
    ///
    /// `opts.include_kinds` filters which edge kinds are followed; the
    /// default is all of them. `opts.include_partition_edges = false` is a
    /// convenience escape hatch for "memory + link only" queries.
    ///
    /// The result is deterministic: nodes are de-duplicated (in insertion
    /// order) and edges are appended in the order their source node was
    /// expanded.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::graph::{NodeRef, TraverseOpts};
    /// let g = mem.traverse(NodeRef::Memory(r), 3, TraverseOpts::default()).await?;
    /// # let _ = g.nodes.len(); Ok(()) }
    /// ```
    pub async fn traverse(
        &self,
        start: crate::graph::NodeRef,
        max_hops: u32,
        opts: crate::graph::TraverseOpts,
    ) -> crate::error::Result<crate::graph::Graph> {
        use crate::graph::{Graph, GraphNode, NodeRef};
        use std::collections::{HashSet, VecDeque};

        let mut graph = Graph::default();
        let mut seen: HashSet<NodeRef> = HashSet::new();
        let mut frontier: VecDeque<(NodeRef, u32)> = VecDeque::new();

        seen.insert(start.clone());
        graph.nodes.push(GraphNode {
            r#ref: start.clone(),
        });
        frontier.push_back((start, 0));

        let max_nodes = opts.max_nodes as usize;

        while let Some((node, hop)) = frontier.pop_front() {
            if u32::try_from(graph.nodes.len()).unwrap_or(u32::MAX)
                >= u32::try_from(max_nodes).unwrap_or(u32::MAX)
            {
                break;
            }
            let edges = self.expand_node(&node, &opts).await?;
            for edge in edges {
                if !opts.permits(edge.kind.tag()) {
                    continue;
                }
                let to = edge.to.clone();
                if seen.insert(to.clone()) {
                    graph.nodes.push(GraphNode { r#ref: to.clone() });
                    if u32::try_from(graph.nodes.len()).unwrap_or(u32::MAX)
                        >= u32::try_from(max_nodes).unwrap_or(u32::MAX)
                    {
                        graph.edges.push(edge);
                        break;
                    }
                    if hop + 1 < max_hops {
                        frontier.push_back((to, hop + 1));
                    }
                }
                graph.edges.push(edge);
            }
        }
        Ok(graph)
    }

    /// JSON-serialised [`Memory::traverse`]. Convenience for FFI consumers
    /// (Swift, future MCP server) that don't want to round-trip the full
    /// [`crate::graph::Graph`] type through their bridge.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::graph::{NodeRef, TraverseOpts};
    /// let json = mem.traverse_json(NodeRef::Memory(r), 3, TraverseOpts::default()).await?;
    /// # let _ = json; Ok(()) }
    /// ```
    pub async fn traverse_json(
        &self,
        start: crate::graph::NodeRef,
        max_hops: u32,
        opts: crate::graph::TraverseOpts,
    ) -> crate::error::Result<String> {
        let g = self.traverse(start, max_hops, opts).await?;
        serde_json::to_string(&g).map_err(|e| Error::storage("graph json", e))
    }

    /// Internal: expand one node into its outgoing edges.
    async fn expand_node(
        &self,
        node: &crate::graph::NodeRef,
        opts: &crate::graph::TraverseOpts,
    ) -> crate::error::Result<Vec<crate::graph::GraphEdge>> {
        use crate::graph::{EdgeKind, EdgeKindTag, GraphEdge, NodeRef};
        let mut out: Vec<GraphEdge> = Vec::new();

        match node {
            NodeRef::Memory(r) => {
                if opts.permits(EdgeKindTag::Link) {
                    // Plan 16: pull explicit edges from node_link so any
                    // memory→{memory,summary,partition} edge surfaces.
                    let edges = self
                        .inner
                        .metadata
                        .node_links_from(&NodeRef::Memory(r.clone()))
                        .await?;
                    for e in edges {
                        let to = e.dst;
                        // For memory→memory edges, drop tombstoned destinations.
                        if let NodeRef::Memory(m) = &to {
                            let dst_row = self.inner.metadata.get_memory(&m.id).await?;
                            let Some(dst_row) = dst_row else { continue };
                            if dst_row.tombstoned {
                                continue;
                            }
                        }
                        out.push(GraphEdge {
                            from: node.clone(),
                            to,
                            kind: EdgeKind::Link,
                        });
                    }
                }
                if opts.permits(EdgeKindTag::SummaryInput) {
                    let citing = self
                        .inner
                        .metadata
                        .summaries_citing("memory", &r.id.to_string())
                        .await?;
                    for sid in citing {
                        let row = self.inner.metadata.get_summary(&sid).await?;
                        let Some(row) = row else { continue };
                        if row.tombstoned {
                            continue;
                        }
                        let subject = subject_from_summary_row(&row)?;
                        let s_ref = crate::summary::SummaryRef {
                            id: row.id,
                            subject: subject.clone(),
                            style: crate::summarizer::SummaryStyle::from_persisted(&row.style),
                            version: u32::try_from(row.version).unwrap_or(0),
                        };
                        // We're expanding the memory, so the edge points OUT
                        // of it to the citing summary (BFS only walks `to`).
                        // Renderers can still see the `from_subject` on the
                        // edge's kind to know what subject the summary covers.
                        out.push(GraphEdge {
                            from: node.clone(),
                            to: NodeRef::Summary(s_ref),
                            kind: EdgeKind::SummaryInput {
                                from_subject: subject,
                            },
                        });
                    }
                }
                if opts.permits(EdgeKindTag::ParentPartition) {
                    out.push(GraphEdge {
                        from: node.clone(),
                        to: NodeRef::Partition(r.partition.clone()),
                        kind: EdgeKind::ParentPartition,
                    });
                }
            }

            NodeRef::Summary(s) => {
                if opts.permits(EdgeKindTag::Link) {
                    // Plan 16: explicit edges sourced at this summary.
                    let edges = self
                        .inner
                        .metadata
                        .node_links_from(&NodeRef::Summary(s.clone()))
                        .await?;
                    for e in edges {
                        out.push(GraphEdge {
                            from: node.clone(),
                            to: e.dst,
                            kind: EdgeKind::Link,
                        });
                    }
                }
                if opts.permits(EdgeKindTag::SummaryInput) {
                    let row = self.inner.metadata.get_summary(&s.id).await?;
                    if let Some(row) = row {
                        for input in &row.inputs {
                            let to = match input {
                                crate::summary::SummarySubject::Memory(m) => {
                                    NodeRef::Memory(m.clone())
                                }
                                crate::summary::SummarySubject::Partition(p) => {
                                    NodeRef::Partition(p.clone())
                                }
                                crate::summary::SummarySubject::Tenant => {
                                    NodeRef::Partition(crate::partition::tenant_root_path())
                                }
                            };
                            out.push(GraphEdge {
                                from: node.clone(),
                                to,
                                kind: EdgeKind::SummaryInput {
                                    from_subject: s.subject.clone(),
                                },
                            });
                        }
                    }
                }
                if opts.permits(EdgeKindTag::ParentPartition) {
                    match &s.subject {
                        crate::summary::SummarySubject::Memory(m) => {
                            out.push(GraphEdge {
                                from: node.clone(),
                                to: NodeRef::Memory(m.clone()),
                                kind: EdgeKind::ParentPartition,
                            });
                        }
                        crate::summary::SummarySubject::Partition(p) => {
                            out.push(GraphEdge {
                                from: node.clone(),
                                to: NodeRef::Partition(p.clone()),
                                kind: EdgeKind::ParentPartition,
                            });
                        }
                        crate::summary::SummarySubject::Tenant => {
                            out.push(GraphEdge {
                                from: node.clone(),
                                to: NodeRef::Partition(crate::partition::tenant_root_path()),
                                kind: EdgeKind::ParentPartition,
                            });
                        }
                    }
                }
            }

            NodeRef::Partition(p) => {
                if opts.permits(EdgeKindTag::Link) {
                    // Plan 16: explicit edges sourced at this partition.
                    let edges = self
                        .inner
                        .metadata
                        .node_links_from(&NodeRef::Partition(p.clone()))
                        .await?;
                    for e in edges {
                        out.push(GraphEdge {
                            from: node.clone(),
                            to: e.dst,
                            kind: EdgeKind::Link,
                        });
                    }
                }
                if opts.permits(EdgeKindTag::PartitionContains) {
                    // Children of this partition.
                    let children = self.inner.metadata.children_of(p).await?;
                    for c in children {
                        out.push(GraphEdge {
                            from: node.clone(),
                            to: NodeRef::Partition(c),
                            kind: EdgeKind::PartitionContains,
                        });
                    }
                    // Memories in this partition.
                    let (memories, _) =
                        self.inner.metadata.list_memories(p, 0, None, false).await?;
                    for m in memories {
                        out.push(GraphEdge {
                            from: node.clone(),
                            to: NodeRef::Memory(crate::memory::MemoryRef {
                                id: m.id,
                                partition: m.partition_path,
                            }),
                            kind: EdgeKind::PartitionContains,
                        });
                    }
                }
                if opts.permits(EdgeKindTag::ParentPartition) {
                    let parents: Vec<crate::partition::PartitionPath> = p.ancestors().collect();
                    if let Some(parent) = parents.first().cloned() {
                        out.push(GraphEdge {
                            from: node.clone(),
                            to: NodeRef::Partition(parent),
                            kind: EdgeKind::ParentPartition,
                        });
                    }
                }
            }
        }

        Ok(out)
    }

    // ---- Plan 12: snapshots ----

    /// Take a point-in-time snapshot of the live tenant state.
    ///
    /// Records a [`crate::snapshot::SnapshotManifest`] under
    /// `metadata/snapshots/{id}.manifest.json` carrying the primary keys
    /// of every live memory and link at the snapshot's
    /// `audit_log.seq`, and inserts a [`crate::SnapshotRow`] + audit
    /// entry in one SQL transaction.
    ///
    /// Storage blobs are append-only; the snapshot never copies blob
    /// bytes — only enumerates the keys that were live. A future
    /// `restore` reconciles the live set against the manifest.
    ///
    /// **Cost:** O(live set) — one SELECT for memories, one for links, one
    /// storage `put` for the manifest blob, one SQL transaction for the row +
    /// audit entry. Crash-safety: the manifest blob is written *before* the
    /// SQL row, so a crash between the two leaves a recoverable orphan that
    /// `gc()` reaps. **Errors:** [`Error::Storage`]
    /// or [`Error::Metadata`] for backend
    /// failures.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::snapshot::SnapshotOpts;
    /// let s = mem.snapshot(SnapshotOpts::default().with_tag("nightly")).await?;
    /// # let _ = s; Ok(()) }
    /// ```
    pub async fn snapshot(
        &self,
        opts: crate::snapshot::SnapshotOpts,
    ) -> crate::error::Result<crate::snapshot::SnapshotRef> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let id = crate::snapshot::SnapshotId::generate();
        let seq = self.inner.metadata.current_audit_seq().await?;

        // Enumerate the live set under Scope::All.
        let mems = self
            .inner
            .metadata
            .memories_in_subtree(&crate::summary::Scope::All)
            .await?;
        let mut memory_ids: Vec<crate::memory::MemoryId> = mems.iter().map(|r| r.id).collect();
        memory_ids.sort_unstable();

        let summaries = self
            .inner
            .metadata
            .summaries_in_subtree(&crate::summary::Scope::All)
            .await?;
        let mut summary_ids: Vec<crate::summary::SummaryId> =
            summaries.iter().map(|r| r.id).collect();
        summary_ids.sort_unstable();

        let links = self
            .inner
            .metadata
            .links_in_subtree(&crate::summary::Scope::All)
            .await?;
        let mut link_pairs: Vec<(crate::memory::MemoryId, crate::memory::MemoryId)> =
            links.iter().map(|l| (l.src, l.dst)).collect();
        link_pairs.sort_unstable();
        link_pairs.dedup();

        let mut attrs_raw = self.inner.metadata.list_all_attributes().await?;
        attrs_raw.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1)));
        let attributes: Vec<crate::snapshot::SnapshotAttribute> = attrs_raw
            .into_iter()
            .map(
                |(memory_id, key, value)| crate::snapshot::SnapshotAttribute {
                    memory_id,
                    key,
                    value,
                },
            )
            .collect();

        let manifest = crate::snapshot::SnapshotManifest {
            snapshot_id: id,
            seq,
            created_at_ms: now,
            memory_ids,
            summary_ids,
            link_pairs,
            attributes,
        };
        let manifest_path = crate::snapshot::SnapshotManifest::manifest_path_for(id);
        let body = serde_json::to_vec(&manifest).map_err(|e| {
            crate::error::Error::storage(
                "snapshot manifest serialise",
                std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()),
            )
        })?;
        let key = crate::storage::StorageKey::new(format!(
            "{}/metadata/{}",
            self.inner.tenant.as_str(),
            manifest_path,
        ));
        self.inner
            .storage
            .put(&key, bytes::Bytes::from(body))
            .await?;

        let row = crate::metadata::SnapshotRow {
            id: id.to_string(),
            seq,
            tag: opts.tag.clone(),
            reason: opts.reason.clone(),
            manifest_path,
            created_at_ms: now,
        };
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::Snapshot,
            partition_path: None,
            memory_id: None,
            detail: serde_json::json!({
                "snapshot_id": id.to_string(),
                "seq": seq,
                "tag": opts.tag,
                "reason": opts.reason,
            }),
        };
        self.inner.metadata.insert_snapshot(row, audit).await?;

        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::SnapshotTaken {
                id,
                seq,
                ts_ms: now,
            });

        Ok(crate::snapshot::SnapshotRef {
            id,
            seq,
            created_at_ms: now,
            tag: opts.tag,
            reason: opts.reason,
        })
    }

    /// All known snapshots, newest first.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory) -> kiromi_ai_memory::Result<()> {
    /// let snaps = mem.list_snapshots().await?;
    /// # let _ = snaps; Ok(()) }
    /// ```
    pub async fn list_snapshots(&self) -> crate::error::Result<Vec<crate::snapshot::SnapshotRef>> {
        let rows = self.inner.metadata.list_snapshots().await?;
        let mut out = Vec::with_capacity(rows.len());
        for r in rows {
            let id: crate::snapshot::SnapshotId =
                r.id.parse().map_err(|e: ulid::DecodeError| {
                    crate::error::Error::IndexCorrupt(format!("snapshot.id: {e}"))
                })?;
            out.push(crate::snapshot::SnapshotRef {
                id,
                seq: r.seq,
                created_at_ms: r.created_at_ms,
                tag: r.tag,
                reason: r.reason,
            });
        }
        Ok(out)
    }

    /// Delete a snapshot row. The manifest blob is **not** removed
    /// synchronously — a future `Memory::gc(opts)` (Plan 12,
    /// deferred) reaps it after the retention window so an accidental
    /// delete remains recoverable until then.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, s: kiromi_ai_memory::snapshot::SnapshotRef) -> kiromi_ai_memory::Result<()> {
    /// mem.delete_snapshot(&s).await?;
    /// # Ok(()) }
    /// ```
    pub async fn delete_snapshot(
        &self,
        sref: &crate::snapshot::SnapshotRef,
    ) -> crate::error::Result<()> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::SnapshotDelete,
            partition_path: None,
            memory_id: None,
            detail: serde_json::json!({ "snapshot_id": sref.id.to_string() }),
        };
        self.inner
            .metadata
            .delete_snapshot(&sref.id.to_string(), audit)
            .await
    }

    /// Open a read-only [`crate::MemoryView`] anchored at `sref`. Reads
    /// through the view filter against the snapshot's manifest, so
    /// only rows that were live at snapshot time are visible.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, s: kiromi_ai_memory::snapshot::SnapshotRef) -> kiromi_ai_memory::Result<()> {
    /// let view = mem.at(&s).await?;
    /// # let _ = view.snapshot(); Ok(()) }
    /// ```
    pub async fn at(
        &self,
        sref: &crate::snapshot::SnapshotRef,
    ) -> crate::error::Result<crate::handle::MemoryView> {
        let manifest = self.read_snapshot_manifest(sref).await?;
        Ok(crate::handle::MemoryView {
            inner: std::sync::Arc::clone(&self.inner),
            snapshot: sref.clone(),
            manifest: std::sync::Arc::new(manifest),
        })
    }

    pub(crate) async fn read_snapshot_manifest(
        &self,
        sref: &crate::snapshot::SnapshotRef,
    ) -> crate::error::Result<crate::snapshot::SnapshotManifest> {
        let row = self
            .inner
            .metadata
            .get_snapshot(&sref.id.to_string())
            .await?
            .ok_or_else(|| Error::SnapshotNotFound {
                id: sref.id.to_string(),
            })?;
        let key = crate::storage::StorageKey::new(format!(
            "{}/metadata/{}",
            self.inner.tenant.as_str(),
            row.manifest_path,
        ));
        let body = self.inner.storage.get(&key).await?;
        let manifest: crate::snapshot::SnapshotManifest = serde_json::from_slice(&body)
            .map_err(|e| Error::IndexCorrupt(format!("snapshot manifest: {e}")))?;
        Ok(manifest)
    }

    /// Hard-rollback the live set to a snapshot. Storage blobs are
    /// untouched (append-only); SQL state is reconciled against the
    /// manifest in one batch:
    ///
    /// 1. Memories live now but absent from the snapshot are
    ///    re-tombstoned; memories tombstoned now but live in the
    ///    snapshot are un-tombstoned.
    /// 2. Same for summaries.
    /// 3. Links are reconciled against the snapshot's link-pair set.
    /// 4. Attributes (if `opts.also_restore_attributes`) are
    ///    re-applied verbatim from the manifest, and any keys that
    ///    are present now but not in the manifest are cleared.
    /// 5. Every touched partition is marked `summary_stale = 1`.
    /// 6. A `Restore` audit row is written.
    ///
    /// **Cost:** O(|now ∆ manifest|) row touches, one SQL transaction. The
    /// returned [`crate::snapshot::RestoreReport`] carries the per-bucket
    /// counts. **Errors:** [`Error::Storage`]
    /// or [`Error::Metadata`] on backend
    /// failure; [`Error::SnapshotNotFound`]
    /// if the manifest blob is missing.
    ///
    /// ```no_run
    /// # async fn _ex(mem: kiromi_ai_memory::Memory, s: kiromi_ai_memory::snapshot::SnapshotRef) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::snapshot::RestoreOpts;
    /// let report = mem.restore(&s, RestoreOpts::default()).await?;
    /// # let _ = report; Ok(()) }
    /// ```
    pub async fn restore(
        &self,
        sref: &crate::snapshot::SnapshotRef,
        opts: crate::snapshot::RestoreOpts,
    ) -> crate::error::Result<crate::snapshot::RestoreReport> {
        let _g = self.inner.locks.lock(&self.inner.tenant).await;
        let now = self.inner.clock.now_ms();
        let manifest = self.read_snapshot_manifest(sref).await?;
        let mut report = crate::snapshot::RestoreReport::default();
        let mut touched_partitions: std::collections::BTreeSet<crate::partition::PartitionPath> =
            std::collections::BTreeSet::new();

        // ---- Memories ----
        let live_now: std::collections::BTreeSet<crate::memory::MemoryId> = self
            .inner
            .metadata
            .memories_in_subtree(&crate::summary::Scope::All)
            .await?
            .into_iter()
            .map(|r| r.id)
            .collect();
        let manifest_mems: std::collections::BTreeSet<crate::memory::MemoryId> =
            manifest.memory_ids.iter().copied().collect();

        // Re-tombstone: live now but not in manifest.
        for id in live_now.difference(&manifest_mems) {
            if let Some(row) = self.inner.metadata.get_memory(id).await? {
                touched_partitions.insert(row.partition_path.clone());
            }
            self.inner
                .metadata
                .set_memory_tombstone(id, true, now)
                .await?;
            report.memories_re_tombstoned += 1;
        }
        // Un-tombstone: in manifest but tombstoned now.
        for id in manifest.memory_ids.iter() {
            if !live_now.contains(id) {
                // Could be tombstoned (revive) or absent. If absent, skip.
                if let Some(row) = self.inner.metadata.get_memory(id).await?
                    && row.tombstoned
                {
                    touched_partitions.insert(row.partition_path.clone());
                    self.inner
                        .metadata
                        .set_memory_tombstone(id, false, now)
                        .await?;
                    report.memories_un_tombstoned += 1;
                }
            }
        }

        // ---- Summaries ----
        let live_summaries_now: std::collections::BTreeSet<crate::summary::SummaryId> = self
            .inner
            .metadata
            .summaries_in_subtree(&crate::summary::Scope::All)
            .await?
            .into_iter()
            .map(|r| r.id)
            .collect();
        let manifest_summaries: std::collections::BTreeSet<crate::summary::SummaryId> =
            manifest.summary_ids.iter().copied().collect();
        for id in live_summaries_now.difference(&manifest_summaries) {
            self.inner.metadata.set_summary_tombstone(id, true).await?;
            report.summaries_re_tombstoned += 1;
        }
        for id in manifest.summary_ids.iter() {
            if !live_summaries_now.contains(id)
                && let Some(row) = self.inner.metadata.get_summary(id).await?
                && row.tombstoned
            {
                self.inner.metadata.set_summary_tombstone(id, false).await?;
                report.summaries_un_tombstoned += 1;
            }
        }

        // ---- Links ----
        let live_links_now: std::collections::BTreeSet<(
            crate::memory::MemoryId,
            crate::memory::MemoryId,
        )> = self
            .inner
            .metadata
            .links_in_subtree(&crate::summary::Scope::All)
            .await?
            .into_iter()
            .map(|l| (l.src, l.dst))
            .collect();
        let manifest_links: std::collections::BTreeSet<(
            crate::memory::MemoryId,
            crate::memory::MemoryId,
        )> = manifest.link_pairs.iter().copied().collect();
        // Remove links that exist now but not in manifest. Compare both
        // (a, b) and (b, a) treated as the same edge — manifest stores
        // both directions, so remove pairs where the canonical (min, max)
        // is missing from manifest_links.
        let mut canon_manifest: std::collections::BTreeSet<(
            crate::memory::MemoryId,
            crate::memory::MemoryId,
        )> = std::collections::BTreeSet::new();
        for (a, b) in &manifest_links {
            let (lo, hi) = if a < b { (*a, *b) } else { (*b, *a) };
            canon_manifest.insert((lo, hi));
        }
        let mut canon_now: std::collections::BTreeSet<(
            crate::memory::MemoryId,
            crate::memory::MemoryId,
        )> = std::collections::BTreeSet::new();
        for (a, b) in &live_links_now {
            let (lo, hi) = if a < b { (*a, *b) } else { (*b, *a) };
            canon_now.insert((lo, hi));
        }
        for (a, b) in canon_now.difference(&canon_manifest) {
            self.inner.metadata.remove_link_unaudited(a, b).await?;
            report.links_removed += 1;
        }
        for (a, b) in canon_manifest.difference(&canon_now) {
            self.inner.metadata.add_link_unaudited(a, b, now).await?;
            report.links_added += 1;
        }

        // ---- Attributes ----
        if opts.also_restore_attributes {
            let mut current_keys: std::collections::BTreeSet<(crate::memory::MemoryId, String)> =
                self.inner
                    .metadata
                    .list_all_attributes()
                    .await?
                    .into_iter()
                    .map(|(m, k, _)| (m, k))
                    .collect();
            for entry in &manifest.attributes {
                let manifest_audit = crate::metadata::AuditEntry {
                    ts_ms: now,
                    actor: self.inner.actor.clone(),
                    op: crate::audit::AuditOp::AttributeSet,
                    partition_path: None,
                    memory_id: Some(entry.memory_id),
                    detail: serde_json::json!({
                        "key": entry.key,
                        "kind": entry.value.kind_str(),
                        "via": "restore",
                    }),
                };
                self.inner
                    .metadata
                    .set_attribute(&entry.memory_id, &entry.key, &entry.value, manifest_audit)
                    .await?;
                report.attributes_set += 1;
                current_keys.remove(&(entry.memory_id, entry.key.clone()));
            }
            for (mid, key) in current_keys {
                let clear_audit = crate::metadata::AuditEntry {
                    ts_ms: now,
                    actor: self.inner.actor.clone(),
                    op: crate::audit::AuditOp::AttributeClear,
                    partition_path: None,
                    memory_id: Some(mid),
                    detail: serde_json::json!({ "key": key, "via": "restore" }),
                };
                self.inner
                    .metadata
                    .clear_attribute(&mid, &key, clear_audit)
                    .await?;
                report.attributes_cleared += 1;
            }
        }

        // ---- Mark partitions stale ----
        let mut all_paths: std::collections::BTreeSet<crate::partition::PartitionPath> =
            std::collections::BTreeSet::new();
        for p in touched_partitions {
            all_paths.insert(p.clone());
            for a in p.ancestors() {
                all_paths.insert(a);
            }
        }
        let stale_vec: Vec<_> = all_paths.iter().cloned().collect();
        if !stale_vec.is_empty() {
            self.inner.metadata.mark_summary_stale(&stale_vec).await?;
        }
        report.partitions_marked_stale = u64::try_from(stale_vec.len()).unwrap_or(0);

        // ---- Audit ----
        let restore_audit = crate::metadata::AuditEntry {
            ts_ms: now,
            actor: self.inner.actor.clone(),
            op: crate::audit::AuditOp::Restore,
            partition_path: None,
            memory_id: None,
            detail: serde_json::json!({
                "snapshot_id": sref.id.to_string(),
                "snapshot_seq": sref.seq,
                "memories_re_tombstoned": report.memories_re_tombstoned,
                "memories_un_tombstoned": report.memories_un_tombstoned,
                "summaries_re_tombstoned": report.summaries_re_tombstoned,
                "summaries_un_tombstoned": report.summaries_un_tombstoned,
                "links_added": report.links_added,
                "links_removed": report.links_removed,
                "attributes_set": report.attributes_set,
                "attributes_cleared": report.attributes_cleared,
            }),
        };
        let _ = self
            .inner
            .metadata
            .insert_restore_audit(restore_audit)
            .await?;

        let _ = self
            .inner
            .event_tx
            .send(crate::event::MemoryEvent::Restored {
                snapshot_id: sref.id,
                ts_ms: now,
            });

        Ok(report)
    }
}

// ============================================================
// Plan 12 Phase C: MemoryView — read-only handle at a snapshot.
// ============================================================

/// Read-only view of a [`Memory`] anchored at a [`crate::SnapshotRef`].
///
/// Reads filter through the snapshot's manifest, so only rows that
/// were live at snapshot time are visible. There are no write methods —
/// the type system enforces immutability. Cheaply cloneable.
#[derive(Clone, Debug)]
pub struct MemoryView {
    pub(crate) inner: std::sync::Arc<MemoryInner>,
    pub(crate) snapshot: crate::snapshot::SnapshotRef,
    pub(crate) manifest: std::sync::Arc<crate::snapshot::SnapshotManifest>,
}

impl MemoryView {
    /// The snapshot this view is anchored at.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView) {
    /// let _sref = view.snapshot();
    /// # }
    /// ```
    #[must_use]
    pub fn snapshot(&self) -> &crate::snapshot::SnapshotRef {
        &self.snapshot
    }

    /// Borrow the underlying manifest (sorted vectors of live ids).
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView) {
    /// let m = view.manifest();
    /// let _ = m.memory_ids.len();
    /// # }
    /// ```
    #[must_use]
    pub fn manifest(&self) -> &crate::snapshot::SnapshotManifest {
        &self.manifest
    }

    fn memory_in_manifest(&self, id: &crate::memory::MemoryId) -> bool {
        self.manifest.memory_ids.binary_search(id).is_ok()
    }

    fn summary_in_manifest(&self, id: &crate::summary::SummaryId) -> bool {
        self.manifest.summary_ids.binary_search(id).is_ok()
    }

    /// Fetch a single memory at the snapshot. Returns
    /// [`crate::Error::MemoryNotFound`] if the memory was not live at
    /// snapshot time, regardless of its current tombstone state.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let rec = view.get(&r).await?;
    /// # let _ = rec.r#ref; Ok(()) }
    /// ```
    pub async fn get(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<crate::memory::MemoryRecord> {
        if !self.memory_in_manifest(&r.id) {
            return Err(Error::MemoryNotFound(r.id.to_string()));
        }
        let row = self
            .inner
            .metadata
            .get_memory(&r.id)
            .await?
            .ok_or_else(|| Error::MemoryNotFound(r.id.to_string()))?;
        let body = self
            .inner
            .storage
            .get(&crate::storage::StorageKey::new(row.data_path.clone()))
            .await?;
        let body_str =
            String::from_utf8(body.to_vec()).map_err(|e| Error::storage("non-utf8 body", e))?;
        let content = match row.content_kind.as_str() {
            "md" => crate::content::Content::markdown(body_str),
            _ => crate::content::Content::text(body_str),
        };
        let actual = crate::content::ContentHash::of_content(&content);
        if actual != row.content_hash {
            return Err(Error::IndexCorrupt(format!(
                "content hash mismatch for {}",
                r.id
            )));
        }
        Ok(crate::memory::MemoryRecord {
            r#ref: crate::memory::MemoryRef {
                id: row.id,
                partition: row.partition_path,
            },
            content,
            hash: row.content_hash,
            created_at_ms: row.created_at_ms,
            updated_at_ms: row.updated_at_ms,
            // From the snapshot's POV the row is "alive".
            tombstoned: false,
            valid_from_ms: row.valid_from_ms,
            valid_until_ms: row.valid_until_ms,
            kind: row.kind,
        })
    }

    /// List memories under a partition that were live at snapshot time.
    /// Pagination is in-memory-filtered on top of the engine's list.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{ListOpts, Partitions};
    /// let parts = Partitions::new().with("user", "alex");
    /// let page = view.list(parts, ListOpts::default()).await?;
    /// # let _ = page; Ok(()) }
    /// ```
    pub async fn list(
        &self,
        partitions: crate::partition::Partitions,
        opts: crate::opts::ListOpts,
    ) -> crate::error::Result<crate::opts::Page<crate::memory::MemoryRef>> {
        let path = partitions.resolve(&self.inner.scheme)?;
        // Always include tombstoned in the inner list so we can resurrect
        // rows that are tombstoned now but were live at snapshot time.
        let (rows, _) = self
            .inner
            .metadata
            .list_memories(&path, 0, None, true)
            .await?;
        let mut items: Vec<crate::memory::MemoryRef> = rows
            .into_iter()
            .filter(|r| self.memory_in_manifest(&r.id))
            .map(|r| crate::memory::MemoryRef {
                id: r.id,
                partition: r.partition_path,
            })
            .collect();
        // Apply caller's limit / cursor.
        if let Some(cursor) = opts.cursor.as_deref()
            && let Some(pos) = items
                .iter()
                .position(|m| m.id.to_string().as_str() == cursor)
        {
            items.drain(..=pos);
        }
        let next_cursor = if opts.limit > 0 && items.len() > opts.limit as usize {
            let cut = opts.limit as usize;
            let next = items.get(cut - 1).map(|m| m.id.to_string());
            items.truncate(cut);
            next
        } else {
            None
        };
        Ok(crate::opts::Page { items, next_cursor })
    }

    /// Run a search restricted to the snapshot's live set.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::Query;
    /// let hits = view.search(Query::text("alpha"), 10).await?;
    /// # let _ = hits; Ok(()) }
    /// ```
    pub async fn search(
        &self,
        query: crate::query::Query,
        k: usize,
    ) -> crate::error::Result<Vec<crate::query::SearchHit>> {
        // Reuse the engine's search and post-filter to the manifest.
        let mem = crate::handle::Memory {
            inner: std::sync::Arc::clone(&self.inner),
        };
        let hits = mem.search(query, k).await?;
        Ok(hits
            .into_iter()
            .filter(|h| self.memory_in_manifest(&h.r#ref.id))
            .collect())
    }

    /// Latest summary for a `(subject, style)` that was live at
    /// snapshot time, if any.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView) -> kiromi_ai_memory::Result<()> {
    /// use kiromi_ai_memory::{summary::SummarySubject, summarizer::SummaryStyle};
    /// let s = view.latest_summary(&SummarySubject::Tenant, &SummaryStyle::Compact).await?;
    /// # let _ = s; Ok(()) }
    /// ```
    pub async fn latest_summary(
        &self,
        subject: &crate::summary::SummarySubject,
        style: &crate::summarizer::SummaryStyle,
    ) -> crate::error::Result<Option<crate::summary::SummaryRecord>> {
        let mem = crate::handle::Memory {
            inner: std::sync::Arc::clone(&self.inner),
        };
        // Walk all summaries for the subject, filter through the manifest,
        // pick the highest version among live-at-snapshot ones.
        let rows = self.inner.metadata.list_summaries_of(subject).await?;
        let style_tag = style.as_str().into_owned();
        let mut candidates: Vec<_> = rows
            .into_iter()
            .filter(|r| r.style == style_tag)
            .filter(|r| self.summary_in_manifest(&r.id))
            .collect();
        candidates.sort_by_key(|c| std::cmp::Reverse(c.version));
        let Some(picked) = candidates.into_iter().next() else {
            return Ok(None);
        };
        // Hydrate via the engine's get_summary to share the body fetch
        // path.
        mem.get_summary(&crate::summary::SummaryRef {
            id: picked.id,
            subject: subject.clone(),
            style: style.clone(),
            version: u32::try_from(picked.version).unwrap_or(0),
        })
        .await
        .map(Some)
    }

    /// Read all attributes attached to `r` at snapshot time.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let attrs = view.attributes_of(&r).await?;
    /// # let _ = attrs; Ok(()) }
    /// ```
    pub async fn attributes_of(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<std::collections::BTreeMap<String, crate::attribute::AttributeValue>>
    {
        if !self.memory_in_manifest(&r.id) {
            return Err(Error::MemoryNotFound(r.id.to_string()));
        }
        let mut out = std::collections::BTreeMap::new();
        for entry in &self.manifest.attributes {
            if entry.memory_id == r.id {
                out.insert(entry.key.clone(), entry.value.clone());
            }
        }
        Ok(out)
    }

    /// Read one attribute at snapshot time.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let v = view.get_attribute(&r, "speaker").await?;
    /// # let _ = v; Ok(()) }
    /// ```
    pub async fn get_attribute(
        &self,
        r: &crate::memory::MemoryRef,
        key: &str,
    ) -> crate::error::Result<Option<crate::attribute::AttributeValue>> {
        if !self.memory_in_manifest(&r.id) {
            return Err(Error::MemoryNotFound(r.id.to_string()));
        }
        for entry in &self.manifest.attributes {
            if entry.memory_id == r.id && entry.key == key {
                return Ok(Some(entry.value.clone()));
            }
        }
        Ok(None)
    }

    /// Links sourced at `r` that were live at snapshot time.
    ///
    /// ```no_run
    /// # async fn _ex(view: kiromi_ai_memory::MemoryView, r: kiromi_ai_memory::MemoryRef) -> kiromi_ai_memory::Result<()> {
    /// let links = view.links_of(&r).await?;
    /// # let _ = links; Ok(()) }
    /// ```
    pub async fn links_of(
        &self,
        r: &crate::memory::MemoryRef,
    ) -> crate::error::Result<Vec<crate::link::Link>> {
        if !self.memory_in_manifest(&r.id) {
            return Err(Error::MemoryNotFound(r.id.to_string()));
        }
        // Collect manifest-resident edges adjacent to r.id.
        let mut out = Vec::new();
        for (a, b) in &self.manifest.link_pairs {
            if *a == r.id {
                out.push(crate::link::Link {
                    src: *a,
                    dst: *b,
                    kind: crate::link::LinkKind::Explicit,
                    created_at_ms: self.snapshot.created_at_ms,
                });
            }
        }
        Ok(out)
    }
}

/// Reconstruct a [`crate::summary::SummarySubject`] from a [`crate::metadata::SummaryRow`].
fn subject_from_summary_row(
    row: &crate::metadata::SummaryRow,
) -> crate::error::Result<crate::summary::SummarySubject> {
    use crate::error::Error;
    match row.subject_kind.as_str() {
        "memory" => {
            let mid = row
                .subject_memory
                .ok_or_else(|| Error::IndexCorrupt("memory subject missing id".into()))?;
            let path = row
                .subject_path
                .clone()
                .ok_or_else(|| Error::IndexCorrupt("memory subject missing path".into()))?;
            Ok(crate::summary::SummarySubject::Memory(
                crate::memory::MemoryRef {
                    id: mid,
                    partition: path,
                },
            ))
        }
        "partition" => Ok(crate::summary::SummarySubject::Partition(
            row.subject_path
                .clone()
                .ok_or_else(|| Error::IndexCorrupt("partition subject missing path".into()))?,
        )),
        _ => Ok(crate::summary::SummarySubject::Tenant),
    }
}

/// Plan 12 helper: enumerate every subject under `scope`, ignoring the
/// `summary_stale` flag. Yields memory-subjects + partition-subjects +
/// tenant subject (when scope is `All` or `Tenant`).
///
/// Used by [`Memory::subjects_to_regenerate`] under `force = true`.
async fn subjects_in_scope(
    inner: &std::sync::Arc<MemoryInner>,
    scope: &crate::summary::Scope,
) -> crate::error::Result<Vec<crate::summary::SummarySubject>> {
    use crate::summary::{Scope, SummarySubject};
    let mut out: Vec<SummarySubject> = Vec::new();

    // Memory subjects: every live memory under scope.
    let mems = inner.metadata.memories_in_subtree(scope).await?;
    for r in mems {
        out.push(SummarySubject::Memory(crate::memory::MemoryRef {
            id: r.id,
            partition: r.partition_path,
        }));
    }

    // Partition subjects: every partition under scope, deepest-first.
    match scope {
        Scope::All | Scope::Tenant => {
            let parts = inner.metadata.list_partitions(None).await?;
            // Deepest-first ordering.
            let mut parts: Vec<_> = parts;
            parts.sort_by(|a, b| b.level.cmp(&a.level).then_with(|| b.path.cmp(&a.path)));
            for p in parts {
                out.push(SummarySubject::Partition(p.path));
            }
            out.push(SummarySubject::Tenant);
        }
        Scope::Partition(p) => {
            let parts = inner.metadata.list_partitions(Some(p)).await?;
            let mut parts: Vec<_> = parts;
            parts.sort_by(|a, b| b.level.cmp(&a.level).then_with(|| b.path.cmp(&a.path)));
            for q in parts {
                out.push(SummarySubject::Partition(q.path));
            }
        }
        Scope::Memory(_) => {} // Already covered above.
    }

    Ok(out)
}