khive_runtime/curation/

note_curation.rs

use super::{
    kind_owned_properties, merge_properties, note_update_values_equal,
    owner_established_property_named_in, reject_pack_managed_schedule_mutation,
    stale_note_snapshot_error, EntityDedupMergePolicy, KhiveRuntime, NamespaceToken, NotePatch,
    RuntimeError, RuntimeResult, Uuid, Value,
};

impl KhiveRuntime {
    /// Apply a note patch to exactly the supplied read snapshot without
    /// fetching the row again. The caller must persist it through
    /// [`Self::update_note_from_snapshot_with_embedding_report`] or a write
    /// plan guarded by the snapshot's `updated_at`/`deleted_at` values.
    pub(crate) async fn prepare_update_note_from_snapshot(
        &self,
        _token: &NamespaceToken,
        note: khive_storage::note::Note,
        patch: NotePatch,
    ) -> RuntimeResult<(khive_storage::note::Note, bool, bool)> {
        if patch.content.is_some() || patch.properties.is_some() {
            if let Some(error) = self.stream_member_error(&note).await? {
                return Err(error);
            }
        }
        let pack_owned = self.is_pack_owned_note_kind(&note.kind);
        Self::apply_note_update_fields(
            note,
            patch,
            pack_owned,
            false,
            false,
            chrono::Utc::now().timestamp_micros(),
        )
    }

    /// Shared field validation and mutation. Canonical callers replace the
    /// complete property map; ordinary updates retain their merge semantics.
    pub(crate) fn apply_note_update_fields(
        mut note: khive_storage::note::Note,
        patch: NotePatch,
        pack_owned: bool,
        replace_properties: bool,
        canonical: bool,
        now: i64,
    ) -> RuntimeResult<(khive_storage::note::Note, bool, bool)> {
        if note.properties.as_ref().is_some_and(|properties| {
            properties
                .as_object()
                .is_some_and(|map| map.contains_key(crate::secret_gate::RESERVED_WEB_RECEIPT_KEY))
        }) {
            return Err(RuntimeError::InvalidInput(
                "web receipt notes are immutable through generic update".into(),
            ));
        }
        // The stored row as read. A no-op answers with this, not with the
        // patched snapshot: the patch may differ from the row in ways the
        // no-op decision ignores (tag order), and nothing was written.
        let stored = note.clone();
        let original_name = note.name.clone();
        let original_content = note.content.clone();
        let original_salience = note.salience;
        let original_decay_factor = note.decay_factor;
        let original_properties = note.properties.clone();
        let original_status = note.status.clone();
        if patch
            .update_policy
            .kind
            .as_deref()
            .is_some_and(|kind| kind != note.kind)
        {
            return Err(RuntimeError::InvalidInput(
                "note update policy does not match the stored note kind".into(),
            ));
        }
        // Canonical writes follow ADR-115 §4: an incoming reserved value that echoes the
        // stored one is dropped, any other incoming reserved value is refused, and the stored
        // value is carried below only while the bytes it vouches for are unchanged.
        let mut patch = patch;
        let stored_reserved = if canonical {
            if let Some(properties) = patch.properties.as_mut() {
                crate::secret_gate::strip_reserved_echo(properties, note.properties.as_ref())?;
            }
            crate::secret_gate::reserved_properties(note.properties.as_ref())
        } else {
            Vec::new()
        };
        crate::secret_gate::reject_reserved_secret_gate_property(patch.properties.as_ref())?;
        if let Some(ref content) = patch.content {
            crate::secret_gate::check_at(content, "note", "content")?;
        }
        if let Some(Some(ref name)) = patch.name {
            crate::secret_gate::check_at(name, "note", "name")?;
        }
        if let Some(ref props) = patch.properties {
            crate::secret_gate::check_json_at(props, "note", "properties")?;
        }

        reject_pack_managed_schedule_mutation(&note, "update")?;

        let mut text_changed = false;

        if let Some(name_patch) = patch.name {
            text_changed |= note.name != name_patch;
            note.name = name_patch;
        }
        if let Some(content) = patch.content {
            text_changed |= note.content != content;
            note.content = content;
        }
        if let Some(salience_patch) = patch.salience {
            // Reject invalid salience rather than silently clamping caller input.
            if let Some(s) = salience_patch {
                if !s.is_finite() || !(0.0..=1.0).contains(&s) {
                    return Err(crate::RuntimeError::InvalidInput(format!(
                        "salience must be a finite value in [0.0, 1.0]; got {s}"
                    )));
                }
            }
            note.salience = salience_patch;
        }
        if let Some(decay_patch) = patch.decay_factor {
            // Reject invalid decay_factor rather than silently clamping caller input.
            if let Some(d) = decay_patch {
                if !d.is_finite() || d < 0.0 {
                    return Err(crate::RuntimeError::InvalidInput(format!(
                        "decay_factor must be a finite value >= 0.0; got {d}"
                    )));
                }
            }
            note.decay_factor = decay_patch;
        }
        if let Some(props) = patch.properties {
            if replace_properties {
                let replacement = props.as_object().ok_or_else(|| {
                    RuntimeError::InvalidInput("canonical note properties must be an object".into())
                })?;
                let original = note.properties.as_ref().and_then(Value::as_object);
                for key in kind_owned_properties(&note.kind).iter().copied().chain(
                    super::OWNER_ESTABLISHED_PROPERTIES
                        .iter()
                        .copied()
                        .filter(|_| pack_owned),
                ) {
                    if original.and_then(|map| map.get(key)) != replacement.get(key) {
                        return Err(RuntimeError::InvalidInput(format!(
                            "canonical replacement cannot change owner-established property `{key}` on a `{}` note",
                            note.kind
                        )));
                    }
                }
                note.properties = Some(props);
            } else {
                // Kind-owned identity is protected below pack hooks, including
                // direct runtime and atomic/proposal update preparation. The merge
                // path restores these same keys on its surviving row.
                let owned_keys = kind_owned_properties(&note.kind);
                if !owned_keys.is_empty() {
                    let object = props.as_object().ok_or_else(|| {
                        if note.kind == "message" {
                            RuntimeError::InvalidInput(
                                "properties on a `message` note must be patched with an object: a \
                             non-object patch would replace the transport-owned quarantine and \
                             channel provenance established by `comm.ingest`"
                                    .into(),
                            )
                        } else {
                            RuntimeError::InvalidInput(format!(
                                "properties on a `{}` note must be patched with an object; \
                             a non-object patch would erase its owner-established identity",
                                note.kind
                            ))
                        }
                    })?;
                    if let Some(named) = owned_keys.iter().find(|key| object.contains_key(**key)) {
                        if note.kind == "message" {
                            return Err(RuntimeError::InvalidInput(format!(
                            "`{named}` is transport-owned on a `message` note and cannot be patched; \
                             only `comm.ingest` may establish quarantine disposition and channel \
                             provenance"
                        )));
                        }
                        return Err(RuntimeError::InvalidInput(format!(
                            "`{named}` is not patchable on a `{}` note; \
                         use `comm.heartbeat` to report health without changing the row's identity",
                            note.kind
                        )));
                    }
                }
                // On a pack-owned note kind, the properties in
                // `OWNER_ESTABLISHED_PROPERTIES` are established by the owning pack
                // and read back by it to decide something structural — who wrote
                // the record and when, which author-side record it copies, which
                // conversation it belongs to. A caller cannot patch them here.
                // Only a patch that *names* one of them is refused, and naming is
                // the exact test: the merge below is `PreferFrom`, so a patch that
                // names an owned key would overwrite it while a patch that does
                // not name it leaves it intact. Every other key still merges
                // normally — arbitrary metadata on a pack-owned record (a
                // `blocked_on` note on a `task`) has no other write path and must
                // keep working.
                if pack_owned {
                    // A non-object patch names nothing, so it slips past the
                    // named-key check below and then takes `merge_json`'s
                    // non-object `PreferFrom` arm, which replaces the whole
                    // property object rather than merging into it — erasing
                    // every owned key. Refused on every pack-owned kind, not only
                    // rows that currently carry an owned key, so an identical
                    // call cannot succeed or fail on state the caller cannot see.
                    if !props.is_object() {
                        return Err(RuntimeError::InvalidInput(format!(
                        "properties on a `{}` note must be patched with an object: a non-object \
                         patch names no key, so it would replace the whole property object rather \
                         than merging into it. Pass an object containing the keys you intend to \
                         set.",
                        note.kind
                    )));
                    }
                    if let Some(named) = owner_established_property_named_in(&props) {
                        return Err(RuntimeError::InvalidInput(format!(
                            "`{named}` is not patchable on a `{}` note: the pack that owns this \
                         kind establishes it and reads it back — to decide how the record is \
                         attributed and grouped, or to reproduce it verbatim when the record \
                         is re-emitted — so it is written by the owner and immutable to a \
                         caller patch. Patch any other property key here, or omit \
                         `{named}` from this patch.",
                            note.kind
                        )));
                    }
                }
                let incoming_properties = Some(props);
                let (mut merged, _) = merge_properties(
                    &note.properties,
                    &incoming_properties,
                    EntityDedupMergePolicy::PreferFrom,
                );
                if let Some(properties) = merged.as_mut().and_then(Value::as_object_mut) {
                    for key in patch.update_policy.null_clearing_properties {
                        if incoming_properties
                            .as_ref()
                            .and_then(|incoming| incoming.get(*key))
                            .is_some_and(Value::is_null)
                        {
                            properties.remove(*key);
                        }
                    }
                }
                note.properties = merged;
            }
        }
        if let Some(status) = patch.kind_status {
            note.status = status;
        }

        if !stored_reserved.is_empty() {
            let unchanged = original_name == note.name
                && original_content == note.content
                && crate::secret_gate::without_reserved_properties(original_properties.as_ref())
                    == crate::secret_gate::without_reserved_properties(note.properties.as_ref());
            let properties = note
                .properties
                .get_or_insert_with(|| Value::Object(Default::default()))
                .as_object_mut()
                .ok_or_else(|| {
                    RuntimeError::InvalidInput("canonical note properties must be an object".into())
                })?;
            if unchanged {
                properties.extend(
                    stored_reserved
                        .into_iter()
                        .map(|(key, value)| (key.to_string(), value)),
                );
            } else {
                // Canonical writes take no manifest exemption, so a changed row loses the
                // stamp and must pass the strict scans as a whole.
                for key in crate::secret_gate::RESERVED_PROPERTY_KEYS {
                    properties.remove(key);
                }
                crate::secret_gate::check_at(&note.content, "note", "content")?;
                if let Some(name) = &note.name {
                    crate::secret_gate::check_at(name, "note", "name")?;
                }
                if let Some(properties) = &note.properties {
                    crate::secret_gate::check_json_at(properties, "note", "properties")?;
                }
            }
        } else {
            // The whole-note CAS persists the merged properties, including keys
            // carried from the snapshot when the patch changes another field.
            crate::secret_gate::reject_reserved_secret_gate_property(note.properties.as_ref())?;
        }

        // JSON object key order is not meaningful to callers. Tags are also
        // set-like in every existing note reader, so their order is ignored
        // for the no-op decision while duplicate entries remain meaningful.
        // All other arrays retain ordinary JSON ordering semantics.
        let changed = original_name != note.name
            || original_content != note.content
            || original_salience != note.salience
            || original_decay_factor != note.decay_factor
            || if canonical {
                original_properties != note.properties
            } else {
                !note_update_values_equal(&original_properties, &note.properties)
            }
            || original_status != note.status;
        if !changed {
            return Ok((stored, text_changed, false));
        }

        // `updated_at` is also the optimistic-concurrency revision for
        // full-note replacement. Make it strictly advance even when two
        // operations land inside one clock microsecond. Saturation is not a
        // valid fallback: reusing i64::MAX would make the CAS accept a write
        // without advancing its revision.
        let minimum_updated_at = note.updated_at.checked_add(1).ok_or_else(|| {
            RuntimeError::Internal(format!(
                "note {} updated_at is already at i64::MAX and cannot advance",
                note.id
            ))
        })?;
        note.updated_at = now.max(minimum_updated_at);
        Ok((note, text_changed, true))
    }

    /// Patch-style note update.
    #[cfg(test)]
    pub(crate) async fn update_note(
        &self,
        token: &NamespaceToken,
        id: Uuid,
        patch: NotePatch,
    ) -> RuntimeResult<khive_storage::note::Note> {
        Ok(self
            .update_note_with_embedding_report(token, id, patch)
            .await?
            .0)
    }

    pub async fn update_note_with_embedding_report(
        &self,
        token: &NamespaceToken,
        id: Uuid,
        patch: NotePatch,
    ) -> RuntimeResult<(
        khive_storage::note::Note,
        crate::retrieval::EmbeddingTruncationReport,
    )> {
        let snapshot = self
            .notes(token)?
            .get_note(id)
            .await?
            .ok_or_else(|| RuntimeError::NotFound(format!("note {id}")))?;
        self.update_note_from_snapshot_with_embedding_report(token, snapshot, patch)
            .await
    }

    /// Patch and persist one note from a caller-owned read snapshot.
    ///
    /// This is the canonical seam for kind hooks that normalize coupled
    /// fields from the current note. The same snapshot feeds normalization,
    /// patch application, and the compare-and-swap write; a concurrent note
    /// change therefore refuses the write instead of persisting derivations
    /// computed from stale state.
    pub async fn update_note_from_snapshot_with_embedding_report(
        &self,
        token: &NamespaceToken,
        snapshot: khive_storage::note::Note,
        patch: NotePatch,
    ) -> RuntimeResult<(
        khive_storage::note::Note,
        crate::retrieval::EmbeddingTruncationReport,
    )> {
        let (note, plan) = self
            .prepare_versioned_note_update(token, snapshot, patch)
            .await?;
        self.commit_prepared_note_update(token, note, crate::AtomicOpPlan::Update(Box::new(plan)))
            .await
    }

    /// Commit a normalized and validated kind-owned update, including its typed
    /// graph companions. Callers must first run `prepare_note_update_policy`
    /// against this exact snapshot and pass the policy it returned; the shared
    /// atomic prepare seam checks all patch fields before asking the kind hook
    /// to derive any graph effects.
    pub async fn update_note_from_snapshot_with_kind_effects(
        &self,
        token: &NamespaceToken,
        snapshot: khive_storage::Note,
        args: &Value,
        policy: crate::NoteUpdatePolicy,
        registry: &crate::VerbRegistry,
    ) -> RuntimeResult<(
        khive_storage::Note,
        crate::retrieval::EmbeddingTruncationReport,
    )> {
        let (note, plan) = crate::atomic_prepare::prepare_update_from_note_snapshot(
            self, token, args, None, snapshot, policy, registry,
        )
        .await?;
        self.commit_prepared_note_update(token, note, plan).await
    }

    async fn commit_prepared_note_update(
        &self,
        token: &NamespaceToken,
        note: khive_storage::Note,
        plan: crate::AtomicOpPlan,
    ) -> RuntimeResult<(
        khive_storage::Note,
        crate::retrieval::EmbeddingTruncationReport,
    )> {
        let id = note.id;
        use crate::atomic_runner::{run_atomic_unit, AtomicOpFailure, AtomicRunOutcome};
        match run_atomic_unit(self.sql().as_ref(), vec![plan]).await {
            Ok(AtomicRunOutcome::Committed { post_commit }) => {
                let outcomes = crate::atomic_prepare::apply_post_commit_effects_with_report(
                    self,
                    token,
                    post_commit,
                )
                .await?;
                let report = outcomes
                    .first()
                    .map(|outcome| outcome.truncation.clone())
                    .unwrap_or_default();
                let degradations = super::note_reindex::NoteReindexReport {
                    truncation: report.clone(),
                    failures: outcomes
                        .into_iter()
                        .flat_map(|outcome| outcome.failures)
                        .collect(),
                }
                .post_commit_degradations();
                crate::operations::legacy_post_commit_result(
                    "update_note",
                    id,
                    (note, report),
                    degradations,
                )
            }
            Ok(AtomicRunOutcome::RolledBack {
                failure: AtomicOpFailure::NoteConflict(conflict),
                ..
            }) => Err(conflict.into_error().into()),
            Ok(AtomicRunOutcome::RolledBack {
                failure: AtomicOpFailure::GuardFailed { .. },
                ..
            }) => Err(stale_note_snapshot_error(id)),
            Ok(AtomicRunOutcome::RolledBack { failure, .. }) => Err(RuntimeError::Internal(
                format!("note update rolled back: {failure:?}"),
            )),
            Err(error) => Err(RuntimeError::Storage(error.0)),
        }
    }
}