agentstatedeveloper_core/

ledger.rs

use std::collections::HashSet;

use agentstategraph::{CommitOptions, Repository};
use agentstategraph_core::IntentCategory;

use crate::engine::Engine;
use crate::error::{AsdError, Result};
use crate::paths;
use crate::schema::LedgerEntry;
use crate::search_fts::SearchFtsDb;

// ---------------------------------------------------------------------------
// RatifyOps — narrower trait for the approve/reject/withdraw operations.
// Lives in core so Engine can hold an Arc<dyn RatifyOps> without depending
// on the commercial ratify crate. agentstatedeveloper-ratify implements this.
// ---------------------------------------------------------------------------

/// The three ledger review operations. Implemented by `RatifyOpsImpl`
/// in `agentstatedeveloper-ratify` (Team tier). Methods take an explicit
/// `repo: &Repository` so the impl can be a zero-sized struct — no lifetime
/// or clone needed when stored as `Arc<dyn RatifyOps>` in `Engine`.
pub trait RatifyOps: Send + Sync {
    fn approve_entry(
        &self,
        repo: &Repository,
        ref_name: &str,
        entry_id: &str,
        approver_id: &str,
        approver_kind: &str,
        message: Option<&str>,
        agent_id: &str,
    ) -> Result<ApprovalOutcome>;

    fn reject_entry(
        &self,
        repo: &Repository,
        ref_name: &str,
        entry_id: &str,
        reviewer_id: &str,
        reviewer_kind: &str,
        reason: &str,
        agent_id: &str,
    ) -> Result<ReviewOutcome>;

    fn withdraw_entry(
        &self,
        repo: &Repository,
        ref_name: &str,
        entry_id: &str,
        author_id: &str,
        agent_id: &str,
    ) -> Result<ReviewOutcome>;
}

/// Outcome of [`LedgerStore::approve_entry`]. Carries the updated entry
/// and whether it was already approved (caller idempotency hint).
#[derive(Debug, Clone)]
pub struct ApprovalOutcome {
    pub entry: LedgerEntry,
    pub already_approved: bool,
}

/// Outcome of [`LedgerStore::reject_entry`] / `withdraw_entry`. Same
/// idempotency shape as [`ApprovalOutcome`].
#[derive(Debug, Clone)]
pub struct ReviewOutcome {
    pub entry: LedgerEntry,
    pub already_resolved: bool,
}

pub trait LedgerStore {
    fn append_entry(&self, ref_name: &str, entry: &LedgerEntry, agent_id: &str) -> Result<()>;

    /// List entries for a symbol, newest first. By default, entries that
    /// have been superseded by a later entry are filtered out.
    fn list_entries(&self, ref_name: &str, symbol_id: &str) -> Result<Vec<LedgerEntry>> {
        let all = self.list_entries_with_superseded(ref_name, symbol_id)?;
        let superseded: HashSet<String> = all
            .iter()
            .flat_map(|e| e.supersedes.iter().cloned())
            .collect();
        Ok(all
            .into_iter()
            .filter(|e| !superseded.contains(&e.entry_id))
            .collect())
    }

    /// Full list including superseded entries.
    fn list_entries_with_superseded(
        &self,
        ref_name: &str,
        symbol_id: &str,
    ) -> Result<Vec<LedgerEntry>>;

    /// Approve an awaiting-approval entry. The OSS default returns a
    /// "commercial feature" error — the real implementation lives in
    /// the `agentstatedeveloper-ratify` crate (Team-tier).
    fn approve_entry(
        &self,
        _ref_name: &str,
        _entry_id: &str,
        _approver_id: &str,
        _approver_kind: &str,
        _message: Option<&str>,
        _agent_id: &str,
    ) -> Result<ApprovalOutcome> {
        Err(AsdError::Other(
            "ledger approve is a commercial feature (Team tier) — \
             install asd-pro to enable. See https://agentstatedeveloper.dev/pricing"
                .into(),
        ))
    }

    /// Reject an awaiting-approval entry. See `approve_entry` — OSS
    /// default errors; real impl in `agentstatedeveloper-ratify`.
    fn reject_entry(
        &self,
        _ref_name: &str,
        _entry_id: &str,
        _reviewer_id: &str,
        _reviewer_kind: &str,
        _reason: &str,
        _agent_id: &str,
    ) -> Result<ReviewOutcome> {
        Err(AsdError::Other(
            "ledger reject is a commercial feature (Team tier) — \
             install asd-pro to enable. See https://agentstatedeveloper.dev/pricing"
                .into(),
        ))
    }

    /// Withdraw an awaiting-approval entry. See `approve_entry` — OSS
    /// default errors; real impl in `agentstatedeveloper-ratify`.
    fn withdraw_entry(
        &self,
        _ref_name: &str,
        _entry_id: &str,
        _author_id: &str,
        _agent_id: &str,
    ) -> Result<ReviewOutcome> {
        Err(AsdError::Other(
            "ledger withdraw is a commercial feature (Team tier) — \
             install asd-pro to enable. See https://agentstatedeveloper.dev/pricing"
                .into(),
        ))
    }
}

pub struct AsgLedgerStore<'a> {
    pub repo: &'a Repository,
    /// Borrowed FTS connection from the owning `Engine`.  When `Some`,
    /// enables the SQLite write-through cache: `list_entries` reads from
    /// SQLite when populated, `append_entry` writes to SQLite after the git
    /// commit.  No `Connection::open` on every method call.
    pub fts: Option<&'a SearchFtsDb>,
}

impl<'a> AsgLedgerStore<'a> {
    /// Construct without SQLite caching (tests, internal engine calls).
    pub fn new(repo: &'a Repository) -> Self {
        Self { repo, fts: None }
    }
    /// Convenience: borrow the FTS connection already open in `engine`.
    pub fn from_engine(engine: &'a Engine) -> Self {
        Self {
            repo: &engine.repo,
            fts: engine.fts.as_ref(),
        }
    }
}

impl<'a> LedgerStore for AsgLedgerStore<'a> {
    fn append_entry(&self, ref_name: &str, entry: &LedgerEntry, agent_id: &str) -> Result<()> {
        let path = paths::ledger_entry_path(&entry.symbol_id, &entry.entry_id);
        let value = serde_json::to_value(entry)?;
        let mut opts = CommitOptions::new(
            agent_id,
            IntentCategory::Refine,
            format!("ledger {} for {}", entry.kind.as_str(), entry.symbol_id),
        );
        // Carry the decision's own rationale and confidence onto the commit as
        // provenance (Plan C t-001). Prefer the detailed body; fall back to the
        // summary so `reasoning` is never emptier than what the entry records.
        let reasoning = entry.body.clone().unwrap_or_else(|| entry.summary.clone());
        if !reasoning.is_empty() {
            opts = opts.with_reasoning(reasoning);
        }
        if let Some(confidence) = entry.confidence {
            opts = opts.with_confidence(confidence);
        }
        self.repo.set_json(ref_name, &path, &value, opts)?;

        // Write reverse index: entry_id → symbol_id for O(1) find_entry.
        let idx_path = paths::ledger_entry_index_path(&entry.entry_id);
        let idx_val = serde_json::Value::String(entry.symbol_id.clone());
        let idx_opts = CommitOptions::new(
            agent_id,
            IntentCategory::Refine,
            format!("ledger-idx {}", entry.entry_id),
        );
        self.repo
            .set_json(ref_name, &idx_path, &idx_val, idx_opts)?;

        // Best-effort SQLite write-through; failures are non-fatal.
        if let Some(fts) = self.fts {
            let _ = fts.upsert_ledger_entry(entry, ref_name);
        }
        Ok(())
    }

    fn list_entries_with_superseded(
        &self,
        ref_name: &str,
        symbol_id: &str,
    ) -> Result<Vec<LedgerEntry>> {
        // Fast path: SQLite cache — zero git tree walks when populated.
        if let Some(fts) = self.fts {
            if fts.ledger_entry_count_for(symbol_id, ref_name) > 0 {
                if let Ok(entries) = fts.list_ledger_entries_for(symbol_id, ref_name) {
                    return Ok(entries);
                }
            }
        }

        // Authoritative git path — also runs when SQLite is empty (first run
        // after `git pull`).  Populates the cache as a side effect.
        let parent = paths::ledger_symbol_path(symbol_id);
        let json = match self.repo.get_json(ref_name, &parent) {
            Ok(v) => v,
            Err(_) => return Ok(Vec::new()),
        };
        let mut entries = Vec::new();
        if let serde_json::Value::Object(map) = json {
            for (k, v) in map {
                match serde_json::from_value::<LedgerEntry>(v) {
                    Ok(e) => entries.push(e),
                    Err(err) => eprintln!(
                        "warning: skipping malformed ledger entry {}/{}: {}",
                        symbol_id, k, err
                    ),
                }
            }
        }
        entries.sort_by(|a, b| b.created_at.cmp(&a.created_at));

        // Populate cache for next call — best effort.
        if !entries.is_empty() {
            if let Some(fts) = self.fts {
                for entry in &entries {
                    let _ = fts.upsert_ledger_entry(entry, ref_name);
                }
            }
        }

        Ok(entries)
    }
}

// ---------------------------------------------------------------------------
// Orphan detection
// ---------------------------------------------------------------------------

/// Walk every ledger entry and tag those whose `symbol_id` is no longer
/// present in the qname index as `"orphaned"` and `"orphaned-at:<timestamp>"`;
/// clear those tags from entries whose symbol is back.
///
/// Returns the number of entries newly tagged. Already-orphaned entries
/// (already carrying the `"orphaned"` tag) are skipped. Runs in O(symbols +
/// ledger_entries) — safe to call at the end of `asd index` or from `health`.
pub fn detect_orphaned_entries(repo: &Repository, ref_name: &str, agent_id: &str) -> Result<usize> {
    tag_orphaned_entries(repo, ref_name, agent_id).map(|changed| changed.tagged.len())
}

/// The entries one orphan-tagging pass changed.
#[derive(Default)]
pub(crate) struct OrphanTags {
    /// Newly tagged: their symbol left the index.
    pub tagged: Vec<LedgerEntry>,
    /// Tags cleared: their symbol is back — a branch switched back to, a
    /// file restored.
    pub untagged: Vec<LedgerEntry>,
}

fn is_orphan_tag(tag: &str) -> bool {
    tag == "orphaned" || tag.starts_with("orphaned-at:")
}

/// [`detect_orphaned_entries`], returning the entries it changed so the
/// caller can bring the ledger cache up to date.
///
/// An entry whose symbol is back loses its orphan tags and gains
/// `reattached-at:<now>`. Conclusions import keeps whichever copy of an entry
/// was revised last, judged by its `*-at:` tags, so without a later tag a
/// copy exported while the entry was tagged would count as newer and tag it
/// again.
///
/// Every change lands in one commit. A commit per entry stored a fresh copy
/// of the ledger map each time.
pub(crate) fn tag_orphaned_entries(
    repo: &Repository,
    ref_name: &str,
    agent_id: &str,
) -> Result<OrphanTags> {
    use chrono::Utc;

    let (spec, fork) = crate::subtree::speculate_at_head(repo, ref_name, "asd-tag-orphans")?;
    let staged = (|| -> Result<OrphanTags> {
        // Every symbol_id currently in the index.
        let indexed: HashSet<String> =
            crate::subtree::read_seed(repo, &fork, &format!("{}/index/by-qname", paths::ASD_ROOT))?
                .values()
                .filter_map(|v| v.get("symbol_id")?.as_str().map(|s| s.to_string()))
                .collect();
        let mut ledger = crate::subtree::read_seed(repo, &fork, &paths::ledger_root())?;

        let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ");
        let orphaned_at = format!("orphaned-at:{now}");
        let reattached_at = format!("reattached-at:{now}");
        let mut changed = OrphanTags::default();
        for (sym_id, per_symbol) in ledger.iter_mut() {
            let live = indexed.contains(sym_id);
            let serde_json::Value::Object(entries_map) = per_symbol else {
                continue;
            };
            for value in entries_map.values_mut() {
                // Read the tags off the stored value: parsing every entry in
                // a 12,000-entry ledger on each index is wasted work.
                let tagged = value
                    .get("tags")
                    .and_then(serde_json::Value::as_array)
                    .is_some_and(|tags| {
                        tags.iter()
                            .filter_map(serde_json::Value::as_str)
                            .any(is_orphan_tag)
                    });
                if live != tagged {
                    continue;
                }
                let Ok(mut entry) = serde_json::from_value::<LedgerEntry>(value.clone()) else {
                    continue;
                };
                if live {
                    entry
                        .tags
                        .retain(|t| !is_orphan_tag(t) && !t.starts_with("reattached-at:"));
                    entry.tags.push(reattached_at.clone());
                } else {
                    entry.tags.push("orphaned".to_string());
                    entry.tags.push(orphaned_at.clone());
                }
                *value = serde_json::to_value(&entry)?;
                if live {
                    changed.untagged.push(entry);
                } else {
                    changed.tagged.push(entry);
                }
            }
        }
        if !changed.tagged.is_empty() || !changed.untagged.is_empty() {
            repo.spec_set_json(
                spec,
                &paths::ledger_root(),
                &serde_json::Value::Object(ledger),
            )
            .map_err(|e| AsdError::Other(e.to_string()))?;
        }
        Ok(changed)
    })();
    match staged {
        Ok(changed) if changed.tagged.is_empty() && changed.untagged.is_empty() => {
            let _ = repo.discard_speculation(spec);
            Ok(changed)
        }
        Ok(changed) => {
            let plural = |n: usize| if n == 1 { "y" } else { "ies" };
            let (t, u) = (changed.tagged.len(), changed.untagged.len());
            let opts = CommitOptions::new(
                agent_id,
                IntentCategory::Refine,
                match (t, u) {
                    (_, 0) => format!("tag {t} orphaned ledger entr{}", plural(t)),
                    (0, _) => format!("clear orphan tags from {u} ledger entr{}", plural(u)),
                    _ => format!(
                        "tag {t} orphaned ledger entr{}, clear orphan tags from {u}",
                        plural(t)
                    ),
                },
            )
            .with_reasoning(
                changed
                    .tagged
                    .iter()
                    .map(|e| format!("tagged {}", e.entry_id))
                    .chain(
                        changed
                            .untagged
                            .iter()
                            .map(|e| format!("cleared {}", e.entry_id)),
                    )
                    .collect::<Vec<_>>()
                    .join("\n"),
            );
            repo.commit_speculation(spec, opts)
                .map_err(|e| AsdError::Other(e.to_string()))?;
            Ok(changed)
        }
        Err(e) => {
            let _ = repo.discard_speculation(spec);
            Err(e)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    struct NopStore;

    impl LedgerStore for NopStore {
        fn append_entry(&self, _: &str, _: &LedgerEntry, _: &str) -> crate::error::Result<()> {
            Ok(())
        }
        fn list_entries_with_superseded(
            &self,
            _: &str,
            _: &str,
        ) -> crate::error::Result<Vec<LedgerEntry>> {
            Ok(vec![])
        }
    }

    #[test]
    fn oss_approve_returns_commercial_error_with_url() {
        let store = NopStore;
        let err = store
            .approve_entry("main", "e1", "alice", "human", None, "agent")
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("commercial feature"), "msg: {}", msg);
        assert!(
            msg.contains("agentstatedeveloper.dev/pricing"),
            "msg: {}",
            msg
        );
    }

    #[test]
    fn oss_reject_returns_commercial_error_with_url() {
        let store = NopStore;
        let err = store
            .reject_entry("main", "e1", "alice", "human", "reason", "agent")
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("commercial feature"), "msg: {}", msg);
        assert!(
            msg.contains("agentstatedeveloper.dev/pricing"),
            "msg: {}",
            msg
        );
    }

    #[test]
    fn oss_withdraw_returns_commercial_error_with_url() {
        let store = NopStore;
        let err = store
            .withdraw_entry("main", "e1", "alice", "agent")
            .unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("commercial feature"), "msg: {}", msg);
        assert!(
            msg.contains("agentstatedeveloper.dev/pricing"),
            "msg: {}",
            msg
        );
    }
}