khive_runtime/

atomic_plan.rs

//! ADR-099 (cross-op atomicity for bulk apply) — prepared write-plan types.
//!
//! Async prepare materializes a synchronous write plan outside any
//! transaction; commit later applies DML or a typed GTD no-op read assertion under a per-op
//! SAVEPOINT. This module defines the plan *shapes* only, one family per
//! admissible verb group (`update`, `delete`, `link`, `merge`,
//! `gtd.transition`, `gtd.complete`, the governance verbs) — not yet wired
//! into a live handler or the dispatch path. Every plan is deliberately
//! inert (plain data, no async, no embedding reference).
//!
//! Two validation-staleness invariants every plan must satisfy:
//! 1. **Predicate-based plans** carry an "all rows matching a condition"
//!    effect as a statement evaluated inside the transaction
//!    (`PlanPredicate`), never as a prepare-time-enumerated row list.
//! 2. **Affected-row guards** (`PlanStatement::guard`) are attached to the
//!    exact statement they validate, checked in-transaction; a mismatch
//!    fails the op and rolls back the whole unit.
//!
//! See `docs/atomic-plan.md` for why guards are per-statement rather than
//! per-plan.

use uuid::Uuid;

use khive_storage::SqlStatement;

/// One statement in a plan, paired with the guard (if any) that validates
/// it. **Runner contract:** a present `guard` is checked against the
/// affected-row count of applying `statement` alone (`SqlWriter::execute`'s
/// return value), or the result-row count for a GTD no-op read assertion,
/// not a batch total and not another statement's count.
/// `guard: None` means prepare made no row-existence
/// assumption about this particular statement (e.g. a cascade delete that
/// may legitimately touch zero rows).
#[derive(Debug, Clone)]
pub struct PlanStatement {
    /// The DML, or GTD no-op SELECT assertion, to apply inside the atomic unit.
    pub statement: SqlStatement,
    /// The expected-effect guard for `statement`, if prepare's validation
    /// assumed a target row exists for it.
    pub guard: Option<AffectedRowGuard>,
}

/// The predicate a prepare pass validated a plan's target against, replayed
/// as a statement evaluated **inside** the transaction (ADR-099 D1, rule 1:
/// "predicate-based plans wherever a write's scope depends on current
/// state"). Carrying the predicate rather than a prepare-time-enumerated row
/// list is what lets a later op in the same file (e.g. an intervening
/// `link`) be visible to this plan's apply.
#[derive(Debug, Clone)]
pub struct PlanPredicate {
    /// Human-readable description of the condition, for diagnostics (e.g.
    /// `"source_id = :from"`).
    pub description: String,
    /// The in-transaction statement whose scope is evaluated against
    /// current (committed-so-far) state, not prepare-time state.
    pub statement: SqlStatement,
}

/// An affected-row guard (ADR-099 D1, rule 2): the row-count prepare assumed
/// its target write would affect, re-verified in-transaction. A prepare-time
/// validation is a plan *hypothesis*, never a commitment — if the guard does
/// not hold at apply time, the op fails inside the atomic unit and the whole
/// unit rolls back (ADR-099 acceptance criteria: "zero-row apply fails the
/// unit").
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AffectedRowGuard {
    /// Minimum affected-row count for the guard to hold (inclusive).
    pub expected_min: u64,
    /// Maximum affected-row count for the guard to hold (inclusive), or
    /// `None` for "no upper bound" (e.g. a predicate-based rewire that may
    /// touch any number of rows).
    pub expected_max: Option<u64>,
}

impl AffectedRowGuard {
    /// A guard requiring exactly `n` affected rows (the common case for a
    /// single-target `update`/`delete`/`link` statement).
    pub fn exactly(n: u64) -> Self {
        Self {
            expected_min: n,
            expected_max: Some(n),
        }
    }

    /// A guard requiring at least one affected row and no upper bound (the
    /// shape for a predicate-based rewire that may touch any number of rows,
    /// e.g. `merge`'s edge rewire).
    pub fn at_least_one() -> Self {
        Self {
            expected_min: 1,
            expected_max: None,
        }
    }

    /// Whether an observed affected-row count satisfies this guard.
    pub fn holds_for(&self, affected: u64) -> bool {
        affected >= self.expected_min
            && self.expected_max.map(|max| affected <= max).unwrap_or(true)
    }
}

/// A deferred side effect recorded during prepare and run once, after the
/// atomic unit commits (ADR-099 D1, "post-commit pass"). v1's admissible set
/// computes no embeddings during prepare (D3's `update`/`merge` caveat), so
/// the only post-commit effects are reindex kicks computed from the
/// **committed** row content, plus the GAP-5 addition under B3: the
/// best-effort GTD lifecycle audit row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PostCommitEffect {
    /// No deferred side effect for this op.
    None,
    /// Re-embed and re-warm the given entity's vector row from its committed
    /// content (ADR-099 D3 `update` caveat: entity name/description change).
    ReindexEntity { entity_id: Uuid },
    /// Re-embed and re-warm the given note's vector row from its committed
    /// content (ADR-099 D3 `update` caveat: note name/content change).
    ReindexNote { note_id: Uuid, version: i64 },
    /// Invalidate consumers without recreating explicitly removed vectors.
    NoteChanged { note_id: Uuid, kind: String },
    /// Append one `gtd_lifecycle_audit` row for a committed `gtd.transition`
    /// or `gtd.complete` (ADR-099 B3, GAP-5): canonical `handle_transition`/
    /// `handle_complete` call `ensure_audit_schema` +
    /// `write_audit_record_with_status`
    /// (`khive-pack-gtd::handlers`) as a best-effort side write — a failed
    /// audit insert must never roll back an already-committed transition.
    /// Carries exactly the fields the lifecycle-audit helper needs. Applied
    /// outside `khive-runtime` (crate-direction: `khive-pack-gtd` depends on
    /// `khive-runtime`, not the other way around) — this crate's own
    /// `apply_post_commit_effects` treats this variant as a no-op; the
    /// `kkernel` caller that owns both crates applies it by calling the
    /// canonical `ensure_audit_schema`/`write_audit_record_with_status`
    /// functions
    /// directly.
    GtdAudit {
        task_id: Uuid,
        from_status: String,
        to_status: String,
        note: Option<String>,
        namespace: String,
    },
    /// A committed note delete (soft or hard) — fire the pack-installed
    /// note-mutation hook with the deleted note's kind (#750:
    /// `DeletePlan` previously carried no `post_commit`
    /// slot at all, so an atomic note delete never reached
    /// `KhiveRuntime::fire_note_mutation_hook`, unlike `operations.rs`'s
    /// `delete_note`, which fires it directly after a successful row
    /// delete). Entity deletes have no equivalent — the hook system is
    /// note-only (`khive-pack-memory`'s warm ANN cache is the only
    /// installed consumer today).
    NoteDeleted { note_id: Uuid, kind: String },
}

/// The natural key a committed symmetric edge update's surviving row must
/// be looked up by (ADR-099 B3, second
/// half). `khive-db`'s `edge_symmetric_absorb_or_update_inplace_statement`
/// pair never trusts a prepare-time-computed target id (see that builder's
/// doc comment); a caller rendering this op's result derives the actual
/// surviving id by querying `graph_edges`'s own
/// `UNIQUE(namespace, source_id, target_id, relation)` constraint via
/// `KhiveRuntime::get_edge_by_natural_key_including_deleted`, which — unlike
/// `list_edges` — includes soft-deleted rows: the ADR-039 DO NOTHING absorption
/// arm can commit leaving the surviving canonical row tombstoned
/// (khive#1213/#1214), and `list_edges` would report "not found" for exactly
/// that row. Looked up strictly after commit.
#[derive(Debug, Clone)]
pub struct EdgeNaturalKey {
    pub(crate) namespace: String,
    pub(crate) canon_source_id: Uuid,
    pub(crate) canon_target_id: Uuid,
    pub(crate) relation: khive_storage::EdgeRelation,
}

impl EdgeNaturalKey {
    /// The namespace containing the surviving edge.
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// The canonical source endpoint of the surviving edge.
    pub fn canon_source_id(&self) -> Uuid {
        self.canon_source_id
    }

    /// The canonical target endpoint of the surviving edge.
    pub fn canon_target_id(&self) -> Uuid {
        self.canon_target_id
    }

    /// The surviving edge's relation.
    pub fn relation(&self) -> khive_storage::EdgeRelation {
        self.relation
    }
}

/// Write plan for an `update` op (entity or note shape — ADR-099 D3's
/// `update` caveat covers both substrates the same way: row/FTS DML in the
/// plan, any reindex deferred to `post_commit`).
///
/// Deferred effects are assigned by this crate's prepare pass and cannot be
/// supplied by callers constructing a plan directly:
///
/// ```compile_fail
/// use khive_runtime::{PostCommitEffect, UpdatePlan};
/// use uuid::Uuid;
///
/// let id = Uuid::nil();
/// let plan = UpdatePlan {
///     target_id: id,
///     statements: Vec::new(),
///     post_commit: PostCommitEffect::ReindexEntity { entity_id: id },
///     edge_natural_key: None,
///     idempotent_noop: false,
/// };
/// ```
///
/// A plan returned by a prepare function cannot have its validated
/// statements cleared or replaced before it reaches the runner:
///
/// ```compile_fail
/// use khive_runtime::UpdatePlan;
///
/// fn clear_prepared_statements(mut prepared: UpdatePlan) {
///     prepared.statements.clear();
/// }
/// ```
#[derive(Debug, Clone)]
pub struct UpdatePlan {
    /// The id of the entity or note being updated. For a symmetric edge
    /// update this is the CALLER's requested id — advisory only, never the
    /// basis for post-commit result rendering (see [`EdgeNaturalKey`]).
    pub(crate) target_id: Uuid,
    /// Row + FTS DML statements to apply inside the atomic unit, in order.
    /// The row-update statement carries the existence guard; any FTS-mirror
    /// statement that follows it is unguarded (its target row's existence
    /// was already asserted by the row-update statement's own guard).
    pub(crate) statements: Vec<PlanStatement>,
    /// Deferred reindex assigned by the prepare pass when the update changed
    /// name, description, or content.
    pub(crate) post_commit: PostCommitEffect,
    /// `Some` only for a symmetric edge update — the natural key a caller
    /// must use to derive the committed surviving row post-commit, rather
    /// than trusting `target_id`. `None` for every other update shape
    /// (entity, note, non-symmetric edge), where `target_id` alone is
    /// already an exact, non-advisory identifier.
    pub(crate) edge_natural_key: Option<EdgeNaturalKey>,
    /// True for a note patch whose normalized values already equal the
    /// snapshot. Such plans carry a guarded SELECT assertion and must not
    /// execute DML that would advance the note revision.
    pub(crate) idempotent_noop: bool,
    pub(crate) entity_guard: Option<crate::entity_write::EntityWriteGuard>,
    pub(crate) note_guard: Option<crate::note_write::NoteWriteGuard>,
    pub(crate) note_vector_purge: Option<crate::note_write::NoteVectors>,
    pub(crate) note_embedding_inheritance: Option<crate::note_write::NoteEmbeddingInheritance>,
    /// Typed kind-owned graph operations, applied after the note statements in
    /// the same savepoint. Assertions never execute as writes.
    pub(crate) graph_effects: Vec<NoteUpdateStatement>,
}

/// Only runtime preparation can append these to an update. Packs return typed
/// graph requests, not SQL or execution flags.
#[derive(Debug, Clone)]
pub(crate) enum NoteUpdateStatement {
    Write(PlanStatement),
    Assert(PlanStatement),
}

impl UpdatePlan {
    /// The record id supplied to the prepare pass.
    pub fn target_id(&self) -> Uuid {
        self.target_id
    }

    /// The committed lookup key required for a symmetric edge update.
    pub fn edge_natural_key(&self) -> Option<&EdgeNaturalKey> {
        self.edge_natural_key.as_ref()
    }

    /// The potential deferred effect. The writer resolves inherited embedding
    /// membership before issuing the committed-effects token.
    pub fn post_commit(&self) -> &PostCommitEffect {
        &self.post_commit
    }

    /// Whether this plan is a mutation-free note update assertion.
    pub fn is_idempotent_noop(&self) -> bool {
        self.idempotent_noop
    }
}

/// Write plan for an `AddEntity` proposal change: a fresh entity row plus its
/// FTS document in the same atomic unit. Vector indexing remains a deferred
/// effect because embedding may suspend.
#[derive(Debug, Clone)]
pub struct AddEntityPlan {
    /// The freshly generated id of the entity being created.
    pub(crate) entity_id: Uuid,
    /// Row + FTS insert statements to apply inside the atomic unit, in
    /// order. The row-insert statement carries the existence guard; the
    /// FTS-insert statement that follows it is unguarded (an ordinary
    /// `INSERT` into a virtual table with no conflicting row).
    pub(crate) statements: Vec<PlanStatement>,
    /// Reindex the committed entity after the transaction closes, as
    /// assigned by the prepare pass.
    pub(crate) post_commit: PostCommitEffect,
}

impl AddEntityPlan {
    /// The id generated for the prepared entity.
    pub fn entity_id(&self) -> Uuid {
        self.entity_id
    }
}

/// Write plan for an `AddNote` proposal change: a fresh note row plus its FTS
/// document in the same atomic unit.
#[derive(Debug, Clone)]
pub struct AddNotePlan {
    /// The freshly generated id of the note being created.
    pub(crate) note_id: Uuid,
    pub(crate) note_guard: Option<crate::note_write::NoteWriteGuard>,
    /// Row + FTS insert statements to apply inside the atomic unit, in
    /// order, mirroring [`AddEntityPlan::statements`].
    pub(crate) statements: Vec<PlanStatement>,
    /// Reindex the committed note after the transaction closes, as assigned
    /// by the prepare pass.
    pub(crate) post_commit: PostCommitEffect,
}

impl AddNotePlan {
    /// The id generated for the prepared note.
    pub fn note_id(&self) -> Uuid {
        self.note_id
    }
}

/// Write plan for a `delete` op (soft or hard).
///
/// Deferred effects are assigned by this crate's prepare pass and cannot be
/// attached to a statement-free plan by external callers:
///
/// ```compile_fail
/// use khive_runtime::{DeletePlan, PostCommitEffect};
/// use uuid::Uuid;
///
/// let id = Uuid::nil();
/// let plan = DeletePlan {
///     target_id: id,
///     statements: Vec::new(),
///     post_commit: PostCommitEffect::NoteDeleted {
///         note_id: id,
///         kind: "observation".to_owned(),
///     },
/// };
/// ```
#[derive(Debug, Clone)]
pub struct DeletePlan {
    /// The id of the entity or note being deleted.
    pub(crate) target_id: Uuid,
    /// Row DML (and, for a hard delete, incident-edge cascade DML) to apply
    /// inside the atomic unit, in order. The target-row delete statement
    /// carries the existence guard; a cascade edge-delete statement (hard
    /// delete only) is unguarded — it may legitimately affect zero rows if
    /// the target had no incident edges.
    pub(crate) statements: Vec<PlanStatement>,
    /// Deferred note-mutation-hook fire assigned by the prepare pass for a
    /// note delete (#750 2). `PostCommitEffect::None` for entity and edge
    /// deletes — the hook system is note-only.
    pub(crate) post_commit: PostCommitEffect,
}

impl DeletePlan {
    /// The record id supplied to the prepare pass.
    pub fn target_id(&self) -> Uuid {
        self.target_id
    }
}

/// Write plan for a `link` op (create a typed directed edge). Endpoint
/// existence is checked **structurally**, not via an unanchored plan-level
/// guard: `statement` is a guarded `INSERT ... SELECT ... WHERE EXISTS`
/// shape whose `SELECT` re-probes both endpoints inside the transaction, so
/// the runner's affected-row check on this one statement *is* the
/// in-transaction existence probe (ADR-099 acceptance criteria's
/// dangling-edge case — `[delete(X, hard), link(A, X)]` — is closed by this
/// guard failing once X is gone, regardless of statement ordering
/// convention).
#[derive(Debug, Clone)]
pub struct LinkPlan {
    pub(crate) source_id: Uuid,
    pub(crate) target_id: Uuid,
    /// Guarded edge mutation followed by its event-plane append statements.
    /// The first statement's affected-row count is also the endpoint and
    /// compare-and-swap probe; event statements are reached only after it
    /// succeeds.
    pub(crate) statements: Vec<PlanStatement>,
    /// Prepare-time disposition, protected by the first statement's guard.
    pub(crate) disposition: khive_storage::EdgeUpsertDisposition,
}

impl LinkPlan {
    /// The canonical source endpoint used by the prepared statement.
    pub fn source_id(&self) -> Uuid {
        self.source_id
    }

    /// The canonical target endpoint used by the prepared statement.
    pub fn target_id(&self) -> Uuid {
        self.target_id
    }

    /// Whether this atomic link created, replaced, or explicitly resurrected
    /// its natural-key row.
    pub fn disposition(&self) -> khive_storage::EdgeUpsertDisposition {
        self.disposition
    }
}

/// Write plan for a `merge` op (deduplicate two entities). Rewires and
/// lifecycle writes are split into separate fields precisely so a guard is
/// never ambiguous between them: the edge rewire is **predicate-based**
/// (ADR-099 D1 rule 1) and may touch zero or many rows depending on earlier
/// in-file writes, so it is never guarded; the `from`/`into` entity
/// lifecycle write assumes both rows exist, so it always is.
#[derive(Debug, Clone)]
pub struct MergePlan {
    pub(crate) into_id: Uuid,
    pub(crate) from_id: Uuid,
    /// Predicate-based edge-rewire statement(s)
    /// (`UPDATE graph_edges SET source_id = :into WHERE source_id = :from`-
    /// shaped), evaluated inside the transaction so they structurally see
    /// any earlier op's edge writes in the same file (ADR-099 acceptance
    /// criteria: "merge rewires see earlier in-file writes"). Never
    /// guarded — a rewire touching zero rows is a legitimate outcome.
    pub(crate) rewires: Vec<PlanPredicate>,
    /// The `from` entity's soft-delete/tombstone DML (and any other
    /// lifecycle write prepare assumed a target row exists for). Always
    /// guarded — prepare validated `into`/`from` both exist.
    pub(crate) lifecycle: Vec<PlanStatement>,
}

impl MergePlan {
    /// The entity retained by the prepared merge.
    pub fn into_id(&self) -> Uuid {
        self.into_id
    }

    /// The entity retired by the prepared merge.
    pub fn from_id(&self) -> Uuid {
        self.from_id
    }
}

/// Write plan for a `gtd.transition` op (explicit task lifecycle change).
#[derive(Debug, Clone)]
pub struct GtdTransitionPlan {
    pub(crate) task_id: Uuid,
    /// Task-property DML to apply inside the atomic unit. Property-only status
    /// mutation triggers no reindex (ADR-099 D3). The transition
    /// statement carries the guard over the exact decision snapshot's note
    /// revision, deletion marker, and semantic status (prepare validated the
    /// current status and requested transition were legal). For an idempotent
    /// no-op (`current == target` after `normalize_status`) this contains one
    /// guarded SELECT assertion that revalidates the prepare snapshot
    /// under the commit transaction. Atomic v1 and canonical dispatch both
    /// persist no caller note.
    pub(crate) statements: Vec<PlanStatement>,
    /// Explicit result-shape and execution discriminator: true executes the
    /// snapshot assertion through the writer's read API and guards its result
    /// count; false executes DML and guards affected rows.
    pub(crate) idempotent_noop: bool,
    /// Deferred lifecycle audit row assigned by the prepare pass (GAP-5):
    /// `PostCommitEffect::None` for the idempotent no-op case. This matches
    /// canonical dispatch, which emits no audit row for a same-status request.
    pub(crate) post_commit: PostCommitEffect,
}

impl GtdTransitionPlan {
    /// Build a plan from its already-decided statements, no-op classification,
    /// and audit effect.
    /// Callers outside this crate (e.g. the atomic-apply layer wiring
    /// `gtd.transition` into a multi-op unit) cannot construct the struct
    /// literal directly since its fields are crate-private.
    pub fn new(
        task_id: Uuid,
        statements: Vec<PlanStatement>,
        idempotent_noop: bool,
        post_commit: PostCommitEffect,
    ) -> Self {
        Self {
            task_id,
            statements,
            idempotent_noop,
            post_commit,
        }
    }

    /// The task targeted by the prepared transition.
    pub fn task_id(&self) -> Uuid {
        self.task_id
    }

    /// Guarded task-property DML for this transition, or the guarded
    /// mutation-free snapshot assertion for an idempotent no-op.
    pub fn statements(&self) -> &[PlanStatement] {
        &self.statements
    }

    /// Whether prepare classified this transition as already at its target.
    pub fn is_idempotent_noop(&self) -> bool {
        self.idempotent_noop
    }

    /// The deferred lifecycle audit effect assigned by the prepare pass.
    pub fn post_commit(&self) -> &PostCommitEffect {
        &self.post_commit
    }
}

/// Write plan for a `gtd.complete` op (task lifecycle terminal transition).
#[derive(Debug, Clone)]
pub struct GtdCompletePlan {
    pub(crate) task_id: Uuid,
    /// Status + `completed_at` property DML to apply inside the atomic unit.
    /// The statement guards the exact decision snapshot's note revision,
    /// deletion marker, and semantic status after prepare validated that the
    /// task was in a completable state.
    pub(crate) statements: Vec<PlanStatement>,
    /// Deferred lifecycle audit row assigned by the prepare pass (GAP-5):
    /// mirrors `handle_complete`'s best-effort
    /// `write_audit_record_with_status` call.
    pub(crate) post_commit: PostCommitEffect,
}

impl GtdCompletePlan {
    /// Build a plan from its already-decided statements and audit effect.
    /// Callers outside this crate (e.g. the atomic-apply layer wiring
    /// `gtd.complete` into a multi-op unit) cannot construct the struct
    /// literal directly since its fields are crate-private.
    pub fn new(
        task_id: Uuid,
        statements: Vec<PlanStatement>,
        post_commit: PostCommitEffect,
    ) -> Self {
        Self {
            task_id,
            statements,
            post_commit,
        }
    }

    /// The task targeted by the prepared completion.
    pub fn task_id(&self) -> Uuid {
        self.task_id
    }

    /// Guarded status + `completed_at` property DML for this completion.
    pub fn statements(&self) -> &[PlanStatement] {
        &self.statements
    }

    /// The deferred lifecycle audit effect assigned by the prepare pass.
    pub fn post_commit(&self) -> &PostCommitEffect {
        &self.post_commit
    }
}

/// Which governance verb (`propose` / `review` / `withdraw`) a
/// [`GovernancePlan`] applies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GovernanceOp {
    Propose,
    Review,
    Withdraw,
}

/// Write plan for a governance op (`propose`, `review`, or `withdraw` — the
/// event-sourced change-proposal lifecycle, ADR-046).
#[derive(Debug, Clone)]
pub struct GovernancePlan {
    pub(crate) op: GovernanceOp,
    pub(crate) proposal_id: Uuid,
    /// Event-log + status DML to apply inside the atomic unit. The
    /// lifecycle-state-check statement carries the guard (prepare validated
    /// the proposal was in a state admitting this transition).
    pub(crate) statements: Vec<PlanStatement>,
}

impl GovernancePlan {
    /// The governance operation represented by this plan.
    pub fn op(&self) -> GovernanceOp {
        self.op
    }

    /// The proposal targeted by this plan.
    pub fn proposal_id(&self) -> Uuid {
        self.proposal_id
    }
}

#[cfg(test)]
#[path = "atomic_plan_tests.rs"]
mod tests;