diff --git a/AGENTS.md b/AGENTS.md index 28e236f4..38e8a560 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,10 +15,9 @@ khive gives your agent: 9. **Brain** — Bayesian profile tuning from feedback signals 10. **Session** — persist and resume agent-session records -All 9 packs load by default. **76 public verbs** across the packs — the `git` pack -(commit/issue/pull_request provenance notes, populated by a batch ingester, not agent-facing -verbs) contributes no new verb (regenerate via `request(ops="verbs()")` before editing this -line). +All 9 packs load by default. **77 public verbs** across the packs — the `git` pack +contributes the `git.digest` verb plus the commit/issue/pull_request provenance note kinds +and a batch ingester (regenerate via `request(ops="verbs()")` before editing this line). If you're working on khive itself (writing code in this repo), see `CLAUDE.md` instead. diff --git a/CLAUDE.md b/CLAUDE.md index d3e3fd3c..c8ee2360 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -168,8 +168,8 @@ request(ops="[{\"tool\":\"v1\",\"args\":{...}}, ...]") Verbs come from whichever packs are loaded via `KHIVE_PACKS` (env) or `--pack` (CLI). Default loads all 9 production packs: kg, gtd, memory, brain, comm, schedule, knowledge, session, git -(76 verbs total — git contributes commit/issue/pull_request note kinds and a batch ingester, -no new verbs; comm.probe (#644) added 2026-07-07; brain.event_counts (ADR-103 Stage 1, #724 +(77 verbs total — git contributes commit/issue/pull_request note kinds, a batch ingester, +and the git.digest verb (ADR-088 Amendment 1); comm.probe (#644) added 2026-07-07; brain.event_counts (ADR-103 Stage 1, #724 Ask A) added 2026-07-08; regenerate via `request(ops="verbs()")` before editing this line). ### KG pack verbs (17 — ADR-017, ADR-046, ADR-089) diff --git a/crates/khive-pack-git/Cargo.toml b/crates/khive-pack-git/Cargo.toml index 47e480b9..4817b8a2 100644 --- a/crates/khive-pack-git/Cargo.toml +++ b/crates/khive-pack-git/Cargo.toml @@ -24,6 +24,7 @@ tracing = { workspace = true } tokio = { workspace = true } anyhow = { workspace = true } clap = { workspace = true } +blake3 = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["full"] } diff --git a/crates/khive-pack-git/src/cache.rs b/crates/khive-pack-git/src/cache.rs new file mode 100644 index 00000000..9e521c24 --- /dev/null +++ b/crates/khive-pack-git/src/cache.rs @@ -0,0 +1,484 @@ +//! Scratch-clone cache for `git.digest`'s remote-URL mode (ADR-088 +//! Amendment 1). +//! +//! Clones/fetches into `~/.khive/scratch/git-digest//`, keyed by +//! canonical URL (`crate::source::cache_key`). An LRU cap evicts the +//! least-recently-used clone (by a `.khive-last-used` marker file's mtime, +//! touched on every successful `ensure_clone`) once the cache exceeds +//! `digest_cache_max_repos` entries or `digest_cache_max_bytes` total size -- +//! eviction is safe because ingest cursors live in the database, not the +//! clone (ADR-088 Amendment 1 §Remote-URL mode). Eviction only ever removes +//! entries it can *prove* it owns (`is_owned_entry`: a 16-hex cache-key +//! directory name containing both a `.git` dir and the `.khive-last-used` +//! marker) -- a `KHIVE_GIT_DIGEST_SCRATCH_ROOT` override pointed at a broader +//! or pre-existing directory must never lose unrelated operator data. +//! +//! A per-clone size cap (`digest_cache_clone_max_bytes`) rejects a clone/ +//! fetch that grows past its own budget *before* it ever enters the +//! addressable cache slot: `ensure_clone` clones/fetches into a staging +//! directory outside the cache root, measures it, and only moves it into +//! `//` when it is under the cap. A too-large clone is +//! deleted from staging and never touches `evict_lru`'s bookkeeping or the +//! cache slot. This guarantees the cap is enforced before the clone enters +//! the cache -- it does NOT bound the transient disk usage of the clone/ +//! fetch child process itself while it runs in staging (`git` has no +//! reliable pre-flight or mid-transfer size check for a partial +//! `--filter=blob:none` clone); a single oversized `git clone` can still +//! transiently consume disk in the staging directory before this check +//! rejects and removes it. +//! +//! Config is env-var driven today (`KHIVE_GIT_DIGEST_CACHE_MAX_REPOS`, +//! `KHIVE_GIT_DIGEST_CACHE_MAX_BYTES`, `KHIVE_GIT_DIGEST_CLONE_MAX_BYTES`, +//! `KHIVE_GIT_DIGEST_SCRATCH_ROOT`) rather than a `[git]` TOML section -- +//! see the implementation report for why. + +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::time::SystemTime; + +use uuid::Uuid; + +use crate::source::cache_key; + +pub const DEFAULT_MAX_REPOS: usize = 5; +pub const DEFAULT_MAX_TOTAL_BYTES: u64 = 2 * 1024 * 1024 * 1024; +pub const DEFAULT_CLONE_MAX_BYTES: u64 = 1024 * 1024 * 1024; + +const MARKER_FILE: &str = ".khive-last-used"; + +#[derive(Debug)] +pub enum CacheError { + Io(std::io::Error), + Git(String), + CloneTooLarge { bytes: u64, cap: u64 }, +} + +impl std::fmt::Display for CacheError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + CacheError::Io(e) => write!(f, "scratch-cache I/O error: {e}"), + CacheError::Git(msg) => write!(f, "{msg}"), + CacheError::CloneTooLarge { bytes, cap } => write!( + f, + "clone exceeds the per-clone size cap ({bytes} bytes > {cap} bytes); \ + the clone was removed. Raise KHIVE_GIT_DIGEST_CLONE_MAX_BYTES if this \ + repository's history is legitimately this large." + ), + } + } +} + +impl std::error::Error for CacheError {} + +impl From for CacheError { + fn from(e: std::io::Error) -> Self { + CacheError::Io(e) + } +} + +fn scratch_root() -> PathBuf { + if let Ok(over) = std::env::var("KHIVE_GIT_DIGEST_SCRATCH_ROOT") { + if !over.is_empty() { + return PathBuf::from(over); + } + } + let home = std::env::var("HOME").unwrap_or_else(|_| ".".into()); + PathBuf::from(home) + .join(".khive") + .join("scratch") + .join("git-digest") +} + +fn env_u64(key: &str, default: u64) -> u64 { + std::env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn max_repos() -> usize { + std::env::var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(DEFAULT_MAX_REPOS) +} + +fn max_total_bytes() -> u64 { + env_u64("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES", DEFAULT_MAX_TOTAL_BYTES) +} + +fn clone_max_bytes() -> u64 { + env_u64("KHIVE_GIT_DIGEST_CLONE_MAX_BYTES", DEFAULT_CLONE_MAX_BYTES) +} + +/// Ensure a local clone of `canonical_url` exists and is up to date; returns +/// the repo's local path. +/// +/// An existing cache slot is updated in place (`git fetch --prune`) and +/// re-measured against the per-clone cap -- a repo that grew past the cap +/// since it was last fetched is evicted from the cache slot on the spot. A +/// fresh clone is written into a private staging directory first +/// (`git clone --filter=blob:none`), measured there, and only *moved* into +/// the addressable `//` slot once it is under the cap -- +/// an oversized clone never enters the cache slot, never participates in +/// `evict_lru`'s accounting, and is removed from staging immediately. +/// +/// Runs LRU eviction over the rest of the cache after a successful +/// clone/fetch (this clone is exempt from its own eviction pass). +pub fn ensure_clone(canonical_url: &str) -> Result { + let root = scratch_root(); + std::fs::create_dir_all(&root)?; + let key = cache_key(canonical_url); + let repo_dir = root.join(&key); + let cap = clone_max_bytes(); + + if repo_dir.join(".git").exists() { + fetch(&repo_dir)?; + let size = dir_size(&repo_dir)?; + if size > cap { + let _ = std::fs::remove_dir_all(&repo_dir); + return Err(CacheError::CloneTooLarge { bytes: size, cap }); + } + } else { + let staging_dir = root.join(format!(".staging-{}", Uuid::new_v4())); + clone(canonical_url, &staging_dir).inspect_err(|_| { + // `git clone` can create and partially populate the destination + // before failing (network drop, auth failure, bad ref) -- clean + // it up so a run of failures doesn't leave `.staging-*` litter + // under the scratch root. `evict_lru` deliberately never touches + // non-owned names (`is_owned_entry`), so nothing else would ever + // reclaim this on its own. + let _ = std::fs::remove_dir_all(&staging_dir); + })?; + let size = dir_size(&staging_dir).inspect_err(|_| { + let _ = std::fs::remove_dir_all(&staging_dir); + })?; + if size > cap { + let _ = std::fs::remove_dir_all(&staging_dir); + return Err(CacheError::CloneTooLarge { bytes: size, cap }); + } + std::fs::rename(&staging_dir, &repo_dir).map_err(|e| { + let _ = std::fs::remove_dir_all(&staging_dir); + CacheError::Io(e) + })?; + } + + touch(&repo_dir)?; + evict_lru(&root, &repo_dir)?; + Ok(repo_dir) +} + +fn clone(url: &str, dest: &Path) -> Result<(), CacheError> { + let status = Command::new("git") + .arg("-c") + .arg("core.hooksPath=/dev/null") + .arg("clone") + .arg("--filter=blob:none") + .arg(url) + .arg(dest) + .env("GIT_TERMINAL_PROMPT", "0") + .status() + .map_err(|e| CacheError::Git(format!("spawning git clone: {e}")))?; + if !status.success() { + return Err(CacheError::Git(format!( + "git clone {url} failed (exit {status})" + ))); + } + Ok(()) +} + +fn fetch(repo: &Path) -> Result<(), CacheError> { + let status = Command::new("git") + .arg("-c") + .arg("core.hooksPath=/dev/null") + .arg("-C") + .arg(repo) + .arg("fetch") + .arg("--prune") + .env("GIT_TERMINAL_PROMPT", "0") + .status() + .map_err(|e| CacheError::Git(format!("spawning git fetch: {e}")))?; + if !status.success() { + return Err(CacheError::Git(format!( + "git fetch in {} failed (exit {status})", + repo.display() + ))); + } + Ok(()) +} + +fn touch(repo_dir: &Path) -> Result<(), CacheError> { + std::fs::write(repo_dir.join(MARKER_FILE), b"")?; + Ok(()) +} + +/// Recursive directory size, following no symlinks (`symlink_metadata` +/// throughout, so a symlink itself is sized but never traversed -- clones +/// never legitimately contain symlinked directories pointing outside the +/// clone, and this avoids any possibility of a symlink loop). +fn dir_size(path: &Path) -> Result { + let mut total = 0u64; + let mut stack = vec![path.to_path_buf()]; + while let Some(p) = stack.pop() { + let md = std::fs::symlink_metadata(&p)?; + if md.is_dir() { + for entry in std::fs::read_dir(&p)? { + stack.push(entry?.path()); + } + } else { + total += md.len(); + } + } + Ok(total) +} + +/// Whether `path` is a directory `ensure_clone` could plausibly have +/// created: a 16-lowercase-hex `cache_key`-shaped directory name (never a +/// UUID staging dir, never an arbitrary operator directory) containing both +/// a `.git` entry and the `.khive-last-used` marker written by `touch`. +/// Eviction (and any future scratch-root cleanup) must only ever remove +/// entries that pass this check -- a `KHIVE_GIT_DIGEST_SCRATCH_ROOT` +/// override pointed at a broader or pre-existing directory must never lose +/// unrelated data sitting next to the cache slots. +fn is_owned_entry(path: &Path) -> bool { + let name = match path.file_name().and_then(|n| n.to_str()) { + Some(n) => n, + None => return false, + }; + if name.len() != 16 + || !name + .chars() + .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()) + { + return false; + } + path.join(".git").exists() && path.join(MARKER_FILE).exists() +} + +/// Evict least-recently-used clones under `root` (by `.khive-last-used` +/// mtime) until both the repo-count cap and the total-byte cap are +/// satisfied. `keep` (the clone `ensure_clone` just touched) is never +/// evicted. Only removes paths that are direct children of `root` AND pass +/// `is_owned_entry` -- eviction never touches user-owned or non-cache paths. +fn evict_lru(root: &Path, keep: &Path) -> Result<(), CacheError> { + let mut entries: Vec<(PathBuf, SystemTime, u64)> = Vec::new(); + for entry in std::fs::read_dir(root)? { + let entry = entry?; + let p = entry.path(); + if !p.is_dir() || p == keep || !is_owned_entry(&p) { + continue; + } + let mtime = std::fs::metadata(p.join(MARKER_FILE)) + .and_then(|m| m.modified()) + .unwrap_or(SystemTime::UNIX_EPOCH); + let size = dir_size(&p)?; + entries.push((p, mtime, size)); + } + entries.sort_by_key(|(_, mtime, _)| *mtime); + + let keep_size = dir_size(keep)?; + let mut total: u64 = entries.iter().map(|(_, _, s)| s).sum::() + keep_size; + let mut count = entries.len() + 1; + let cap_repos = max_repos(); + let cap_bytes = max_total_bytes(); + + for (path, _, size) in entries { + if count <= cap_repos && total <= cap_bytes { + break; + } + let _ = std::fs::remove_dir_all(&path); + count -= 1; + total = total.saturating_sub(size); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::LazyLock; + + /// `scratch_root()` reads process-global env vars; serialize tests that + /// touch it. + static ENV_MUTEX: LazyLock> = LazyLock::new(|| std::sync::Mutex::new(())); + + /// Build a directory shaped exactly like a real `ensure_clone` cache + /// slot: a 16-lowercase-hex name (a real `cache_key` output) containing + /// a `.git` dir and (optionally) the `.khive-last-used` marker. + fn make_owned_entry(root: &Path, key: &str, with_marker: bool) -> PathBuf { + assert_eq!(key.len(), 16, "test cache keys must be 16 hex chars"); + let p = root.join(key); + std::fs::create_dir_all(p.join(".git")).unwrap(); + if with_marker { + std::fs::write(p.join(MARKER_FILE), b"").unwrap(); + } + p + } + + /// ADR-088 Amendment 1 fix-round r2 Medium-1: a `git clone` failure (bad + /// source, no network needed -- a nonexistent local path fails + /// immediately) must not leave a `.staging-` directory behind. + /// `evict_lru` deliberately never touches non-owned names, so a leaked + /// staging dir would otherwise accumulate forever across repeated + /// failures. + #[test] + fn ensure_clone_cleans_up_staging_dir_on_clone_failure() { + let _guard = ENV_MUTEX.lock().unwrap(); + let dir = tempfile::tempdir().expect("tempdir"); + std::env::set_var("KHIVE_GIT_DIGEST_SCRATCH_ROOT", dir.path()); + + let bogus_source = dir.path().join("does-not-exist-as-a-repo"); + let result = ensure_clone(bogus_source.to_str().expect("utf8 path")); + assert!( + result.is_err(), + "cloning a nonexistent local path must fail: {result:?}" + ); + + let leftovers: Vec<_> = std::fs::read_dir(dir.path()) + .expect("read scratch root") + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .filter(|name| name.starts_with(".staging-")) + .collect(); + assert!( + leftovers.is_empty(), + "a failed clone must not leave .staging-* directories behind: {leftovers:?}" + ); + + std::env::remove_var("KHIVE_GIT_DIGEST_SCRATCH_ROOT"); + } + + #[test] + fn evict_lru_removes_oldest_past_repo_cap() { + let _guard = ENV_MUTEX.lock().unwrap(); + let dir = tempfile::tempdir().expect("tempdir"); + std::env::set_var("KHIVE_GIT_DIGEST_SCRATCH_ROOT", dir.path()); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS", "1"); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES", "1000000000"); + + let root = dir.path(); + let old = make_owned_entry(root, "1111111111111111", true); + // Ensure a real mtime gap. + std::thread::sleep(std::time::Duration::from_millis(20)); + let new = make_owned_entry(root, "2222222222222222", true); + + evict_lru(root, &new).expect("evict"); + + assert!(!old.exists(), "the older clone must be evicted"); + assert!(new.exists(), "the kept clone must survive"); + + std::env::remove_var("KHIVE_GIT_DIGEST_SCRATCH_ROOT"); + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS"); + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES"); + } + + #[test] + fn evict_lru_only_touches_children_of_root() { + let _guard = ENV_MUTEX.lock().unwrap(); + let dir = tempfile::tempdir().expect("tempdir"); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS", "5"); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES", "1000000000"); + + let root = dir.path().join("scratch-root"); + std::fs::create_dir_all(&root).unwrap(); + let kept = make_owned_entry(&root, "3333333333333333", true); + + evict_lru(&root, &kept).expect("evict"); + assert!(kept.exists()); + + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS"); + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES"); + } + + #[test] + fn evict_lru_never_removes_a_foreign_directory_under_root() { + let _guard = ENV_MUTEX.lock().unwrap(); + let dir = tempfile::tempdir().expect("tempdir"); + // Cap of 0 repos: without ownership filtering this would previously + // have wiped out every child of root, including operator data. + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS", "0"); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES", "0"); + + let root = dir.path().join("scratch-root"); + std::fs::create_dir_all(&root).unwrap(); + let foreign = root.join("not-a-cache-entry"); + std::fs::create_dir_all(&foreign).unwrap(); + std::fs::write(foreign.join("important.txt"), b"do not delete me").unwrap(); + let kept = make_owned_entry(&root, "4444444444444444", true); + + evict_lru(&root, &kept).expect("evict"); + + assert!( + foreign.exists(), + "a directory that doesn't look like a cache slot must survive eviction" + ); + assert!( + foreign.join("important.txt").exists(), + "foreign directory contents must be untouched" + ); + + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS"); + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES"); + } + + #[test] + fn evict_lru_never_removes_an_owned_looking_dir_missing_the_marker() { + let _guard = ENV_MUTEX.lock().unwrap(); + let dir = tempfile::tempdir().expect("tempdir"); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS", "0"); + std::env::set_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES", "0"); + + let root = dir.path().join("scratch-root"); + std::fs::create_dir_all(&root).unwrap(); + // Has a .git dir and a valid cache-key-shaped name, but no marker -- + // e.g. a clone that failed after `clone()` but before `touch()`. + let no_marker = make_owned_entry(&root, "5555555555555555", false); + let kept = make_owned_entry(&root, "6666666666666666", true); + + evict_lru(&root, &kept).expect("evict"); + + assert!( + no_marker.exists(), + "an owned-looking directory without the marker must survive eviction" + ); + + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_REPOS"); + std::env::remove_var("KHIVE_GIT_DIGEST_CACHE_MAX_BYTES"); + } + + #[test] + fn is_owned_entry_rejects_non_cache_shapes() { + let dir = tempfile::tempdir().expect("tempdir"); + let root = dir.path(); + + // Wrong length. + let short = root.join("abc123"); + std::fs::create_dir_all(short.join(".git")).unwrap(); + std::fs::write(short.join(MARKER_FILE), b"").unwrap(); + assert!(!is_owned_entry(&short)); + + // Uppercase hex (cache_key is always lowercase). + let upper = root.join("ABCDEF0123456789"); + std::fs::create_dir_all(upper.join(".git")).unwrap(); + std::fs::write(upper.join(MARKER_FILE), b"").unwrap(); + assert!(!is_owned_entry(&upper)); + + // Right shape but missing .git. + let no_git = root.join("7777777777777777"); + std::fs::create_dir_all(&no_git).unwrap(); + std::fs::write(no_git.join(MARKER_FILE), b"").unwrap(); + assert!(!is_owned_entry(&no_git)); + + let owned = make_owned_entry(root, "8888888888888888", true); + assert!(is_owned_entry(&owned)); + } + + #[test] + fn dir_size_sums_file_bytes_recursively() { + let dir = tempfile::tempdir().expect("tempdir"); + std::fs::write(dir.path().join("a.txt"), b"12345").unwrap(); + std::fs::create_dir_all(dir.path().join("sub")).unwrap(); + std::fs::write(dir.path().join("sub/b.txt"), b"1234567890").unwrap(); + assert_eq!(dir_size(dir.path()).unwrap(), 15); + } +} diff --git a/crates/khive-pack-git/src/handlers.rs b/crates/khive-pack-git/src/handlers.rs new file mode 100644 index 00000000..4ef4a90a --- /dev/null +++ b/crates/khive-pack-git/src/handlers.rs @@ -0,0 +1,223 @@ +//! `git.digest` verb handler (ADR-088 Amendment 1). +//! +//! Resolves the `source` argument (local path or `https://` URL, cloning/ +//! fetching remote sources into the scratch cache), resolves or auto-creates +//! the repo-anchor `project` entity, then drives the shared +//! `ingest::run_ingest` core with a bounded, cursor-resumable pass. + +use anyhow::anyhow; +use serde_json::{json, Value}; +use uuid::Uuid; + +use khive_runtime::{KhiveRuntime, NamespaceToken, RuntimeError, VerbRegistry}; +use khive_storage::types::{SqlStatement, SqlValue}; + +use crate::cache; +use crate::ingest::{resolve_project_id, run_ingest, IngestInclude, IngestOptions}; +use crate::source::{parse_source, repo_basename, DigestSource}; +use crate::GitPack; + +const DEFAULT_MAX_ITEMS: i64 = 500; +const MIN_MAX_ITEMS: i64 = 1; +const MAX_MAX_ITEMS: i64 = 2000; + +impl GitPack { + pub(crate) async fn handle_digest( + &self, + token: &NamespaceToken, + registry: &VerbRegistry, + params: Value, + ) -> Result { + let source_raw = params + .get("source") + .and_then(Value::as_str) + .ok_or_else(|| RuntimeError::InvalidInput("git.digest requires source".into()))?; + let source = + parse_source(source_raw).map_err(|e| RuntimeError::InvalidInput(e.to_string()))?; + + // Parsed as i64 (not u64) so an out-of-range negative value clamps to + // MIN_MAX_ITEMS instead of failing `as_u64` and silently falling + // through to the default -- a caller passing `-1` gets the smallest + // legal budget, not an unrequested 500-item pass. A non-integer + // value (string, float, bool, array, object) is rejected outright + // rather than silently defaulted. + let max_items = match params.get("max_items") { + None | Some(Value::Null) => DEFAULT_MAX_ITEMS, + Some(v) => v.as_i64().ok_or_else(|| { + RuntimeError::InvalidInput(format!("max_items must be an integer, got {v:?}")) + })?, + } + .clamp(MIN_MAX_ITEMS, MAX_MAX_ITEMS) as u64; + + let include = match params.get("include") { + None | Some(Value::Null) => IngestInclude::default(), + Some(v) => parse_include(v)?, + }; + + let mut warnings: Vec = Vec::new(); + + // Resolve a local repo path -- remote sources clone/fetch into the + // scratch cache first (ADR-088 Amendment 1 §Remote-URL mode). + let (repo_path, gh_capable) = match &source { + DigestSource::Local(p) => (p.clone(), true), + DigestSource::Remote { canonical, gh_slug } => { + let cloned = cache::ensure_clone(canonical).map_err(|e| { + RuntimeError::InvalidInput(format!( + "remote clone/fetch of {canonical:?} failed: {e}" + )) + })?; + if gh_slug.is_none() { + warnings.push(format!( + "host for {canonical:?} is not github.com; issue/pull_request \ + ingestion is skipped (commits-only degradation, ADR-088 Amendment 1)" + )); + } + (cloned, gh_slug.is_some()) + } + }; + + // Resolve or auto-create the repo-anchor `project` entity. + let (project_id, project_created) = match params.get("project").and_then(Value::as_str) { + Some(raw) => { + let id = resolve_project_id(self.runtime(), raw) + .await + .map_err(|e| RuntimeError::InvalidInput(e.to_string()))? + .ok_or_else(|| { + RuntimeError::InvalidInput(format!( + "project {raw:?} did not resolve to an entity" + )) + })?; + (id, false) + } + None => resolve_or_create_project(self.runtime(), registry, token, &source).await?, + }; + + let effective_include = IngestInclude { + commits: include.commits, + issues: include.issues && gh_capable, + pull_requests: include.pull_requests && gh_capable, + }; + + let mut report = run_ingest( + self.runtime(), + token, + registry, + IngestOptions { + repo: repo_path, + project: project_id.to_string(), + max_items: Some(max_items), + include: effective_include, + }, + ) + .await + .map_err(|e| RuntimeError::InvalidInput(e.to_string()))?; + + report.warnings.extend(warnings); + report.project_id = Some(project_id.to_string()); + report.project_created = project_created; + + serde_json::to_value(&report) + .map_err(|e| RuntimeError::InvalidInput(format!("serializing report: {e}"))) + } +} + +fn parse_include(v: &Value) -> Result { + let arr = v + .as_array() + .ok_or_else(|| RuntimeError::InvalidInput("include must be an array of strings".into()))?; + let mut include = IngestInclude { + commits: false, + issues: false, + pull_requests: false, + }; + for entry in arr { + let s = entry + .as_str() + .ok_or_else(|| RuntimeError::InvalidInput("include entries must be strings".into()))?; + match s { + "commits" => include.commits = true, + "issues" => include.issues = true, + "pull_requests" => include.pull_requests = true, + other => { + return Err(RuntimeError::InvalidInput(format!( + "unknown include kind {other:?}; valid: commits | issues | pull_requests" + ))) + } + } + } + Ok(include) +} + +/// Find an existing `project` entity whose `properties.repo_url` matches the +/// source's canonical URL/path, or whose `name` matches the repo basename; +/// create the anchor when none is found (ADR-088 Amendment 1 — auto-creation +/// is reported via `IngestReport.project_created`, never silent). +async fn resolve_or_create_project( + runtime: &KhiveRuntime, + registry: &VerbRegistry, + token: &NamespaceToken, + source: &DigestSource, +) -> Result<(Uuid, bool), RuntimeError> { + let repo_url = match source { + DigestSource::Local(p) => p.to_string_lossy().to_string(), + DigestSource::Remote { canonical, .. } => canonical.clone(), + }; + let name = repo_basename(source); + + if let Some(id) = find_project_by_repo(runtime, token, &repo_url, &name) + .await + .map_err(|e| RuntimeError::InvalidInput(e.to_string()))? + { + return Ok((id, false)); + } + + let resp = registry + .dispatch( + "create", + json!({ + "kind": "project", + "name": name, + "properties": { "repo_url": repo_url }, + }), + ) + .await?; + let id = resp + .get("id") + .and_then(Value::as_str) + .and_then(|s| Uuid::parse_str(s).ok()) + .ok_or_else(|| { + RuntimeError::InvalidInput("create(kind=project) did not return an id".into()) + })?; + Ok((id, true)) +} + +async fn find_project_by_repo( + runtime: &KhiveRuntime, + token: &NamespaceToken, + repo_url: &str, + name: &str, +) -> anyhow::Result> { + let sql = runtime.sql(); + let mut r = sql.reader().await.map_err(|e| anyhow!("{e}"))?; + let row = r + .query_row(SqlStatement { + sql: "SELECT id FROM entities WHERE kind='project' AND namespace=?1 \ + AND deleted_at IS NULL \ + AND (json_extract(properties,'$.repo_url')=?2 OR name=?3) \ + LIMIT 1" + .into(), + params: vec![ + SqlValue::Text(token.namespace().as_str().to_string()), + SqlValue::Text(repo_url.to_string()), + SqlValue::Text(name.to_string()), + ], + label: Some("git_digest_find_project_by_repo".into()), + }) + .await + .map_err(|e| anyhow!("{e}"))?; + Ok(row.and_then(|r| match r.get("id") { + Some(SqlValue::Uuid(u)) => Some(*u), + Some(SqlValue::Text(s)) => Uuid::parse_str(s).ok(), + _ => None, + })) +} diff --git a/crates/khive-pack-git/src/ingest.rs b/crates/khive-pack-git/src/ingest.rs index 8a396bcb..a1b74c25 100644 --- a/crates/khive-pack-git/src/ingest.rs +++ b/crates/khive-pack-git/src/ingest.rs @@ -21,6 +21,28 @@ use uuid::Uuid; use khive_runtime::{secret_gate, KhiveRuntime, NamespaceToken, VerbRegistry}; use khive_storage::types::{SqlStatement, SqlValue}; +use crate::refs; + +/// Which record kinds a `run_ingest` pass processes. `Default` selects all +/// three — the CLI's historical behavior and the `git.digest` verb's default +/// (ADR-088 Amendment 1). +#[derive(Debug, Clone, Copy)] +pub struct IngestInclude { + pub commits: bool, + pub issues: bool, + pub pull_requests: bool, +} + +impl Default for IngestInclude { + fn default() -> Self { + Self { + commits: true, + issues: true, + pull_requests: true, + } + } +} + /// Options for one ingest pass. #[derive(Debug, Clone)] pub struct IngestOptions { @@ -28,6 +50,59 @@ pub struct IngestOptions { pub repo: PathBuf, /// The repo-anchor `project` entity — full UUID or an 8+ hex prefix. pub project: String, + /// Bounded work per call, counted across commits + issues + PRs + /// (ADR-088 Amendment 1). `None` means unbounded — the CLI's historical + /// one-shot behavior. + pub max_items: Option, + /// Which record kinds to ingest this pass. + pub include: IngestInclude, +} + +impl IngestOptions { + /// Convenience constructor for callers that want the CLI's historical + /// unbounded, all-kinds behavior. + pub fn unbounded(repo: PathBuf, project: String) -> Self { + Self { + repo, + project, + max_items: None, + include: IngestInclude::default(), + } + } +} + +/// Bounds the number of new-record creation attempts across a `run_ingest` +/// pass (ADR-088 Amendment 1 `max_items`). Only creation attempts (success or +/// failure) consume budget — cheap natural-key "already exists" skips do not, +/// since they are not the work the bound exists to limit. +struct Budget { + remaining: Option, +} + +impl Budget { + fn try_consume(&mut self) -> bool { + match &mut self.remaining { + None => true, + Some(0) => false, + Some(n) => { + *n -= 1; + true + } + } + } + + fn exhausted(&self) -> bool { + matches!(self.remaining, Some(0)) + } +} + +/// A newly created note this pass, retained so the post-ingestion reference- +/// extraction sweep (`link_references`) can resolve cross-references between +/// records created in the *same* pass regardless of ingestion order (PRs and +/// issues are ingested before commits) without re-reading them from storage. +struct NewRecordForRef { + id: Uuid, + text: String, } /// Outcome of one ingest pass. Serializable so CLI callers can emit it as JSON. @@ -43,6 +118,29 @@ pub struct IngestReport { /// skipped but commits still ingested (ADR-088 §5 graceful-absence rule). pub gh_available: bool, pub warnings: Vec, + /// `false` when `max_items` was exhausted before this pass reached the + /// end of every included kind's history — callers loop until `true` + /// (ADR-088 Amendment 1). Always `true` for an unbounded + /// (`max_items: None`) pass. + pub done: bool, + /// The repo-anchor `project` entity id this pass resolved (or the + /// verb-level caller created). + pub project_id: Option, + /// `true` when the `git.digest` verb auto-created the `project` anchor + /// because none was found (ADR-088 Amendment 1) — never set by + /// `run_ingest` itself, only by the verb handler after it returns. + pub project_created: bool, + /// `annotates` edges created from a `Closes/Fixes/Resolves #N` or bare + /// `#N` reference in a commit message or issue/PR body to the referenced + /// issue/PR note (ADR-088 Amendment 1 ingest enrichment). + pub reference_edges_created: u64, + /// References that named a number this pass could not resolve to an + /// ingested issue/PR note within the same project — skipped, not an + /// error (fail-open). + pub reference_edges_unresolved: u64, + /// `precedes` edges created from a commit's `parents[]` to the commit + /// itself (ADR-088 Amendment 1 ingest enrichment). + pub parent_edges_created: u64, } /// Run one ingest pass: issues + PRs first (via `gh`, when available), then @@ -56,72 +154,108 @@ pub async fn run_ingest( registry: &VerbRegistry, opts: IngestOptions, ) -> Result { - let mut report = IngestReport::default(); + let mut report = IngestReport { + done: true, + ..IngestReport::default() + }; let project_id = resolve_id(runtime, token, &opts.project) .await? .ok_or_else(|| anyhow!("--project {:?} did not resolve to an entity", opts.project))?; + report.project_id = Some(project_id.to_string()); let mut merge_sha_to_pr: HashMap = HashMap::new(); let mut number_to_pr: HashMap = HashMap::new(); + let mut budget = Budget { + remaining: opts.max_items, + }; + let mut new_records: Vec = Vec::new(); // Graceful degradation covers both "gh is not on PATH" and "gh is present // but this repo has no usable GitHub remote" (e.g. a synthetic/local-only // repo) — either way, issues/PRs are skipped with a warning and commits // still ingest (ADR-088 §5). A hard `gh` failure must never abort the // whole pass. - if gh_available(&opts.repo) { - report.gh_available = true; - match ingest_prs( - runtime, - token, - registry, - &opts.repo, - project_id, - &mut report, - &mut merge_sha_to_pr, - &mut number_to_pr, - ) - .await - { - Ok(()) => {} - Err(e) => report - .warnings - .push(format!("gh pr list failed, skipping pull requests: {e}")), + if opts.include.issues || opts.include.pull_requests { + if gh_available(&opts.repo) { + report.gh_available = true; + if opts.include.pull_requests && !budget.exhausted() { + match ingest_prs( + runtime, + token, + registry, + &opts.repo, + project_id, + &mut report, + &mut merge_sha_to_pr, + &mut number_to_pr, + &mut budget, + &mut new_records, + ) + .await + { + Ok(()) => {} + Err(e) => report + .warnings + .push(format!("gh pr list failed, skipping pull requests: {e}")), + } + } + if opts.include.issues && !budget.exhausted() { + if let Err(e) = ingest_issues( + runtime, + token, + registry, + &opts.repo, + project_id, + &mut report, + &mut budget, + &mut new_records, + ) + .await + { + report + .warnings + .push(format!("gh issue list failed, skipping issues: {e}")); + } + } + } else { + report.gh_available = false; + report.warnings.push( + "gh CLI not found on PATH; skipped issues and pull requests — commits still ingest" + .to_string(), + ); } - if let Err(e) = ingest_issues( + } + + if opts.include.commits && !budget.exhausted() { + ingest_commits( runtime, token, registry, &opts.repo, project_id, + &merge_sha_to_pr, + &number_to_pr, &mut report, + &mut budget, + &mut new_records, ) - .await - { - report - .warnings - .push(format!("gh issue list failed, skipping issues: {e}")); - } - } else { - report.gh_available = false; - report.warnings.push( - "gh CLI not found on PATH; skipped issues and pull requests — commits still ingest" - .to_string(), - ); + .await?; + } + + if budget.exhausted() { + report.done = false; } - ingest_commits( + link_references( runtime, token, registry, - &opts.repo, project_id, - &merge_sha_to_pr, - &number_to_pr, + &new_records, &mut report, ) - .await?; + .await; Ok(report) } @@ -142,6 +276,99 @@ async fn resolve_id( .map_err(|e| anyhow!("{e}")) } +/// Public wrapper for the `git.digest` verb handler, which needs to resolve +/// (but never auto-create) an explicitly supplied `project` argument the same +/// way `run_ingest` does. +pub async fn resolve_project_id(runtime: &KhiveRuntime, raw: &str) -> Result> { + if let Ok(u) = Uuid::parse_str(raw) { + return Ok(Some(u)); + } + runtime + .resolve_prefix_unfiltered(raw) + .await + .map_err(|e| anyhow!("{e}")) +} + +/// Find an existing `issue` or `pull_request` note by `properties.number` +/// within `project_id` — GitHub numbers a repository's issues and PRs from +/// one shared sequence, so a `#N` reference can resolve to either kind. +async fn find_issue_or_pr_by_number( + runtime: &KhiveRuntime, + token: &NamespaceToken, + project_id: Uuid, + number: u64, +) -> Result> { + if let Some(id) = find_by_number(runtime, token, "issue", project_id, number).await? { + return Ok(Some(id)); + } + find_by_number(runtime, token, "pull_request", project_id, number).await +} + +/// Post-ingestion sweep (ADR-088 Amendment 1 ingest enrichment): extract +/// GitHub reference-grammar mentions from every note created *this pass* +/// (commits, issues, PRs — order-independent, since all three are already in +/// `new_records` by the time this runs) and materialize `annotates` edges to +/// the referenced issue/PR note, carrying `ref_kind` ("closes" | "mentions") +/// as edge metadata. Fail-open throughout: a malformed or unresolvable +/// reference is skipped and counted, never aborts the pass. +async fn link_references( + runtime: &KhiveRuntime, + token: &NamespaceToken, + registry: &VerbRegistry, + project_id: Uuid, + new_records: &[NewRecordForRef], + report: &mut IngestReport, +) { + for record in new_records { + let mentions = refs::dedupe_prefer_closes(refs::extract_references(&record.text)); + for mention in mentions { + let target = match find_issue_or_pr_by_number( + runtime, + token, + project_id, + mention.number, + ) + .await + { + Ok(Some(id)) => id, + Ok(None) => { + report.reference_edges_unresolved += 1; + continue; + } + Err(e) => { + report + .warnings + .push(format!("resolving reference #{}: {e}", mention.number)); + continue; + } + }; + if target == record.id { + // A note referencing its own number (rare, e.g. a PR body + // that quotes its own number) — not a real cross-reference. + continue; + } + match registry + .dispatch( + "link", + json!({ + "source_id": record.id.to_string(), + "target_id": target.to_string(), + "relation": "annotates", + "metadata": { "ref_kind": mention.kind.as_str() }, + }), + ) + .await + { + Ok(_) => report.reference_edges_created += 1, + Err(e) => report.warnings.push(format!( + "linking reference #{} from {}: {e}", + mention.number, record.id + )), + } + } + } +} + /// `true` when `gh` is on PATH and can run inside `repo` (a lightweight /// `gh --version` probe — cheap and does not require network access). fn gh_available(repo: &Path) -> bool { @@ -436,6 +663,11 @@ fn squash_merge_pr_number(subject: &str) -> Option { close[open + 2..].parse::().ok() } +/// Max characters for the `name` field the amendment's readable-names rider +/// sets on newly ingested notes (issues/PRs: `"# "`; commits: +/// `"<short_sha> <subject>"`). +const NAME_MAX_CHARS: usize = 120; + #[allow(clippy::too_many_arguments)] async fn ingest_commits( runtime: &KhiveRuntime, @@ -446,6 +678,8 @@ async fn ingest_commits( merge_sha_to_pr: &HashMap<String, Uuid>, number_to_pr: &HashMap<u64, Uuid>, report: &mut IngestReport, + budget: &mut Budget, + new_records: &mut Vec<NewRecordForRef>, ) -> Result<()> { let since = read_cursor(runtime, project_id, "commits").await?; let commits = walk_commits(repo, since.as_deref())?; @@ -465,8 +699,14 @@ async fn ingest_commits( // sha natural key), so a retried pass never double-creates them. let mut last_sha: Option<String> = since; let mut cursor_stalled = false; + // Parent SHA -> note id for commits created earlier THIS pass (walked + // oldest-first) — combined with `find_commit_by_sha`'s DB lookup below, + // this resolves parent edges regardless of which pass the parent landed + // in. + let mut local_sha_to_id: HashMap<String, Uuid> = HashMap::new(); for c in &commits { - if find_commit_by_sha(runtime, token, &c.sha).await?.is_some() { + if let Some(existing) = find_commit_by_sha(runtime, token, &c.sha).await? { + local_sha_to_id.insert(c.sha.clone(), existing); report.commits_skipped_existing += 1; if !cursor_stalled { last_sha = Some(c.sha.clone()); @@ -474,6 +714,10 @@ async fn ingest_commits( continue; } + if budget.exhausted() { + break; + } + let raw_content = if c.body.trim().is_empty() { c.subject.clone() } else { @@ -522,11 +766,15 @@ async fn ingest_commits( "parents": c.parents, }); + let name = refs::truncate_chars(&format!("{} {}", c.short_sha, c.subject), NAME_MAX_CHARS); + + budget.try_consume(); match registry .dispatch( "create", json!({ "kind": "commit", + "name": name, "content": content, "properties": properties, "annotates": annotates, @@ -534,11 +782,55 @@ async fn ingest_commits( ) .await { - Ok(_) => { + Ok(v) => { report.commits_ingested += 1; if !cursor_stalled { last_sha = Some(c.sha.clone()); } + if let Some(id) = v + .get("id") + .and_then(|v| v.as_str()) + .and_then(|s| Uuid::parse_str(s).ok()) + { + local_sha_to_id.insert(c.sha.clone(), id); + new_records.push(NewRecordForRef { + id, + text: content.clone(), + }); + // Parent -> child `precedes` edges (ADR-088 Amendment 1 + // ingest enrichment). Fail-open: an unresolved or + // failing parent link is skipped/warned, never aborts + // the pass. + for parent_sha in &c.parents { + let parent_id = match local_sha_to_id.get(parent_sha).copied() { + Some(pid) => Some(pid), + None => find_commit_by_sha(runtime, token, parent_sha).await?, + }; + let Some(parent_id) = parent_id else { + continue; + }; + if parent_id == id { + continue; + } + match registry + .dispatch( + "link", + json!({ + "source_id": parent_id.to_string(), + "target_id": id.to_string(), + "relation": "precedes", + }), + ) + .await + { + Ok(_) => report.parent_edges_created += 1, + Err(e) => report.warnings.push(format!( + "linking parent {parent_sha} -> {} precedes: {e}", + c.sha + )), + } + } + } } Err(e) => { report @@ -628,18 +920,79 @@ fn gh_json(repo: &Path, args: &[&str]) -> Result<String> { Ok(String::from_utf8_lossy(&output.stdout).into_owned()) } -#[allow(clippy::too_many_arguments)] -async fn ingest_prs( - runtime: &KhiveRuntime, - token: &NamespaceToken, - registry: &VerbRegistry, - repo: &Path, - project_id: Uuid, - report: &mut IngestReport, - merge_sha_to_pr: &mut HashMap<String, Uuid>, - number_to_pr: &mut HashMap<u64, Uuid>, -) -> Result<()> { - let since = read_cursor(runtime, project_id, "prs").await?; +/// Per-page fetch cap for both PR and issue paging (ADR-088 Amendment 1 fix +/// round, Issue High-1). `gh {pr,issue} list --search` is backed by GitHub's +/// search API, which never returns more than this many results for a single +/// query regardless of `--limit` — paging works around that ceiling by +/// advancing an `updated:>=` floor between calls, not by requesting more +/// than one page can hold. +const PAGE_LIMIT: usize = 1000; + +/// What a paging loop should do after processing one fetched page. Pure and +/// unit-testable independent of `gh`, the database, or async machinery — the +/// entire "was the remote window proven exhausted" decision lives here +/// (ADR-088 Amendment 1 fix-round High-1: a single hard-coded `--limit 1000` +/// fetch could previously report `done: true` while a repo's remaining +/// PRs/issues past position 1000 were never seen). +#[derive(Debug, Clone, PartialEq, Eq)] +enum PageOutcome { + /// The page held fewer than `PAGE_LIMIT` items: the remote window is + /// proven exhausted regardless of local budget state. + WindowComplete, + /// The page was full (`PAGE_LIMIT` items) and the local budget is + /// exhausted: stop paging, but the window is NOT proven exhausted. + StopBudgetExhausted, + /// The page was full and the last item's `updated_at` did not advance + /// past the current floor (more than `PAGE_LIMIT` records share one + /// timestamp — an unresolvable pathological case): stop paging rather + /// than loop forever. The window is NOT proven exhausted. + StopFloorStalled, + /// The page was full, the budget is not exhausted, and the floor + /// advanced: fetch the next page starting at this floor. + Continue(String), +} + +fn decide_page_outcome( + page_len: usize, + current_floor: Option<&str>, + last_updated_at: Option<&str>, + budget_exhausted: bool, +) -> PageOutcome { + if page_len < PAGE_LIMIT { + return PageOutcome::WindowComplete; + } + if budget_exhausted { + return PageOutcome::StopBudgetExhausted; + } + match last_updated_at { + Some(next) if Some(next) != current_floor => PageOutcome::Continue(next.to_string()), + _ => PageOutcome::StopFloorStalled, + } +} + +/// `PageOutcome::WindowComplete` is the only outcome under which `done` can +/// stay `true` on the local-budget question alone; every other outcome means +/// more remote records may exist past the last fetched page. Test-only +/// helper — production code matches on `PageOutcome` directly (see +/// `ingest_prs`/`ingest_issues`'s paging loops). +#[cfg(test)] +fn page_outcome_proves_window_complete(outcome: PageOutcome) -> bool { + matches!(outcome, PageOutcome::WindowComplete) +} + +fn search_query(floor: Option<&str>) -> String { + match floor { + Some(f) => format!("sort:updated-asc updated:>={f}"), + None => "sort:updated-asc".to_string(), + } +} + +const PR_FIELDS: &str = "number,title,author,createdAt,mergedAt,closedAt,updatedAt,baseRefName,headRefName,mergeCommit,body"; +const ISSUE_FIELDS: &str = + "number,title,author,createdAt,closedAt,updatedAt,labels,stateReason,body"; + +fn fetch_pr_page(repo: &Path, floor: Option<&str>) -> Result<Vec<GhPr>> { + let search = search_query(floor); let raw = gh_json( repo, &[ @@ -647,34 +1000,51 @@ async fn ingest_prs( "list", "--state", "all", + "--search", + search.as_str(), "--limit", "1000", "--json", - "number,title,author,createdAt,mergedAt,closedAt,updatedAt,baseRefName,headRefName,mergeCommit,body", + PR_FIELDS, ], )?; - let mut prs: Vec<GhPr> = serde_json::from_str(&raw).context("parsing gh pr list --json")?; - // `gh pr list` makes no ordering guarantee. The frozen-cursor retry - // guarantee below (cursor freezes at the last contiguous success, so a - // failed record is retried next pass) only holds if records are walked - // in nondecreasing updated_at order; otherwise a newer record ahead of - // an older failing one in the raw list would push the cursor past the - // failure and it would never be retried. Sort ascending first (stable — - // ties keep gh's original relative order). - // - // Sorting alone does not cover a TIE: if a successful record and a - // failing record share the exact same `updated_at`, the success still - // advances the cursor to that timestamp, and an exclusive `updated > - // cursor` retry check would then see the failed record's `updated_at == - // cursor` and treat it as not-new on the next pass — stranding it - // forever, sort or no sort. `is_new` below is therefore inclusive - // (`updated >= cursor`): every record at the cursor timestamp is - // re-examined on every subsequent pass until the cursor moves past it. - // That reprocessing is bounded (only records sharing the exact cursor - // value) and free (the natural-key lookup below turns every - // already-created one into a cheap no-op), so it converges the moment - // the last tied record either succeeds or the cursor advances. - prs.sort_by(|a, b| a.updated_at.cmp(&b.updated_at)); + serde_json::from_str(&raw).context("parsing gh pr list --json") +} + +fn fetch_issue_page(repo: &Path, floor: Option<&str>) -> Result<Vec<GhIssue>> { + let search = search_query(floor); + let raw = gh_json( + repo, + &[ + "issue", + "list", + "--state", + "all", + "--search", + search.as_str(), + "--limit", + "1000", + "--json", + ISSUE_FIELDS, + ], + )?; + serde_json::from_str(&raw).context("parsing gh issue list --json") +} + +#[allow(clippy::too_many_arguments)] +async fn ingest_prs( + runtime: &KhiveRuntime, + token: &NamespaceToken, + registry: &VerbRegistry, + repo: &Path, + project_id: Uuid, + report: &mut IngestReport, + merge_sha_to_pr: &mut HashMap<String, Uuid>, + number_to_pr: &mut HashMap<u64, Uuid>, + budget: &mut Budget, + new_records: &mut Vec<NewRecordForRef>, +) -> Result<()> { + let since = read_cursor(runtime, project_id, "prs").await?; // `cursor_stalled` mirrors `ingest_commits`: once one PR fails to create, // later PRs in this pass are still attempted (so every failure surfaces @@ -683,21 +1053,113 @@ async fn ingest_prs( // retries it, while already-landed PRs are no-ops via the natural key. let mut max_updated: Option<String> = since.clone(); let mut cursor_stalled = false; - for pr in prs { - let is_new = since - .as_deref() - .zip(pr.updated_at.as_deref()) - .map(|(cursor, updated)| updated >= cursor) - .unwrap_or(true); - - if let Some(existing) = - find_by_number(runtime, token, "pull_request", project_id, pr.number).await? - { - number_to_pr.insert(pr.number, existing); - if let Some(oid) = pr.merge_commit.as_ref().and_then(|m| m.oid.clone()) { - merge_sha_to_pr.insert(oid, existing); + let mut floor = since.clone(); + let mut window_complete = true; + + 'paging: loop { + let mut page = fetch_pr_page(repo, floor.as_deref())?; + let page_len = page.len(); + // Each page is already `sort:updated-asc` server-side, but `--search` + // makes no hard ordering guarantee across ties — re-sort defensively + // so the frozen-cursor invariant (records walked in nondecreasing + // `updated_at` order) holds regardless. `is_new` below is inclusive + // (`updated >= cursor`) for exactly the tie reason documented at + // length in the pre-pagination version of this function (a + // successful and a failing record sharing one `updated_at` must both + // be re-examined next pass until the cursor moves past that tie). + page.sort_by(|a, b| a.updated_at.cmp(&b.updated_at)); + let last_updated_at = page.last().and_then(|pr| pr.updated_at.clone()); + + for pr in page { + let is_new = since + .as_deref() + .zip(pr.updated_at.as_deref()) + .map(|(cursor, updated)| updated >= cursor) + .unwrap_or(true); + + if let Some(existing) = + find_by_number(runtime, token, "pull_request", project_id, pr.number).await? + { + number_to_pr.insert(pr.number, existing); + if let Some(oid) = pr.merge_commit.as_ref().and_then(|m| m.oid.clone()) { + merge_sha_to_pr.insert(oid, existing); + } + report.prs_skipped_existing += 1; + if !cursor_stalled { + if let Some(u) = &pr.updated_at { + if max_updated + .as_deref() + .map(|m| u.as_str() > m) + .unwrap_or(true) + { + max_updated = Some(u.clone()); + } + } + } + continue; + } + if !is_new { + continue; + } + if budget.exhausted() { + break; + } + + let raw_body = pr.body.unwrap_or_default(); + let content = secret_gate::mask_secrets(&raw_body).into_owned(); + let properties = json!({ + "number": pr.number, + "title": pr.title, + "author": pr.author.and_then(|a| a.login), + "created_at": pr.created_at, + "merged_at": pr.merged_at, + "closed_at": pr.closed_at, + "base_ref": pr.base_ref_name, + "head_ref": pr.head_ref_name, + "project_id": project_id.to_string(), + }); + let name = + refs::truncate_chars(&format!("#{} {}", pr.number, pr.title), NAME_MAX_CHARS); + + budget.try_consume(); + let result = match registry + .dispatch( + "create", + json!({ + "kind": "pull_request", + "name": name, + "content": content, + "properties": properties, + "annotates": [project_id.to_string()], + }), + ) + .await + { + Ok(v) => v, + Err(e) => { + report + .warnings + .push(format!("create pull_request #{}: {e}", pr.number)); + cursor_stalled = true; + continue; + } + }; + + if let Some(id) = result + .get("id") + .and_then(|v| v.as_str()) + .and_then(|s| Uuid::parse_str(s).ok()) + { + number_to_pr.insert(pr.number, id); + if let Some(oid) = pr.merge_commit.and_then(|m| m.oid) { + merge_sha_to_pr.insert(oid, id); + } + new_records.push(NewRecordForRef { + id, + text: content.clone(), + }); } - report.prs_skipped_existing += 1; + report.prs_ingested += 1; if !cursor_stalled { if let Some(u) = &pr.updated_at { if max_updated @@ -709,78 +1171,38 @@ async fn ingest_prs( } } } - continue; - } - if !is_new { - continue; } - let raw_body = pr.body.unwrap_or_default(); - let content = secret_gate::mask_secrets(&raw_body).into_owned(); - let properties = json!({ - "number": pr.number, - "title": pr.title, - "author": pr.author.and_then(|a| a.login), - "created_at": pr.created_at, - "merged_at": pr.merged_at, - "closed_at": pr.closed_at, - "base_ref": pr.base_ref_name, - "head_ref": pr.head_ref_name, - "project_id": project_id.to_string(), - }); - - let result = match registry - .dispatch( - "create", - json!({ - "kind": "pull_request", - "content": content, - "properties": properties, - "annotates": [project_id.to_string()], - }), - ) - .await - { - Ok(v) => v, - Err(e) => { - report - .warnings - .push(format!("create pull_request #{}: {e}", pr.number)); - cursor_stalled = true; - continue; - } - }; - - if let Some(id) = result - .get("id") - .and_then(|v| v.as_str()) - .and_then(|s| Uuid::parse_str(s).ok()) - { - number_to_pr.insert(pr.number, id); - if let Some(oid) = pr.merge_commit.and_then(|m| m.oid) { - merge_sha_to_pr.insert(oid, id); - } - } - report.prs_ingested += 1; - if !cursor_stalled { - if let Some(u) = &pr.updated_at { - if max_updated - .as_deref() - .map(|m| u.as_str() > m) - .unwrap_or(true) - { - max_updated = Some(u.clone()); - } + match decide_page_outcome( + page_len, + floor.as_deref(), + last_updated_at.as_deref(), + budget.exhausted(), + ) { + PageOutcome::WindowComplete => break 'paging, + PageOutcome::StopBudgetExhausted | PageOutcome::StopFloorStalled => { + window_complete = false; + break 'paging; } + PageOutcome::Continue(next_floor) => floor = Some(next_floor), } } + if !window_complete { + // The remote window may hold more PRs than this pass ever fetched + // (ADR-088 Amendment 1 fix-round High-1) — the local budget alone is + // not a complete signal; report `done = false` regardless of budget + // state so the caller's resume loop keeps going. + report.done = false; + } + if let Some(cursor) = max_updated { write_cursor(runtime, project_id, "prs", &cursor).await?; } Ok(()) } +#[allow(clippy::too_many_arguments)] async fn ingest_issues( runtime: &KhiveRuntime, token: &NamespaceToken, @@ -788,32 +1210,10 @@ async fn ingest_issues( repo: &Path, project_id: Uuid, report: &mut IngestReport, + budget: &mut Budget, + new_records: &mut Vec<NewRecordForRef>, ) -> Result<()> { let since = read_cursor(runtime, project_id, "issues").await?; - let raw = gh_json( - repo, - &[ - "issue", - "list", - "--state", - "all", - "--limit", - "1000", - "--json", - "number,title,author,createdAt,closedAt,updatedAt,labels,stateReason,body", - ], - )?; - let mut issues: Vec<GhIssue> = - serde_json::from_str(&raw).context("parsing gh issue list --json")?; - // See `ingest_prs`: the frozen-cursor retry guarantee requires walking - // records in nondecreasing updated_at order, which `gh issue list` does - // not itself guarantee. Sort ascending first (stable — ties keep gh's - // original relative order). Sorting doesn't cover a TIE between a - // successful and a failing record sharing the same `updated_at` — see - // `ingest_prs`'s comment on why `is_new` below is an inclusive - // (`updated >= cursor`) check, and why the resulting reprocessing of - // every record at the cursor timestamp is bounded and converges. - issues.sort_by(|a, b| a.updated_at.cmp(&b.updated_at)); // `cursor_stalled` mirrors `ingest_commits`/`ingest_prs`: a per-record // create failure is aggregated as a warning and later records in this @@ -822,18 +1222,115 @@ async fn ingest_issues( // forever; already-landed records are no-ops via the natural key. let mut max_updated: Option<String> = since.clone(); let mut cursor_stalled = false; - for issue in issues { - let is_new = since - .as_deref() - .zip(issue.updated_at.as_deref()) - .map(|(cursor, updated)| updated >= cursor) - .unwrap_or(true); - - if find_by_number(runtime, token, "issue", project_id, issue.number) - .await? - .is_some() - { - report.issues_skipped_existing += 1; + let mut floor = since.clone(); + let mut window_complete = true; + + 'paging: loop { + let mut page = fetch_issue_page(repo, floor.as_deref())?; + let page_len = page.len(); + // See `ingest_prs`: the frozen-cursor retry guarantee requires + // walking records in nondecreasing updated_at order, which `--search + // sort:updated-asc` does not itself guarantee across ties — sort + // defensively. + page.sort_by(|a, b| a.updated_at.cmp(&b.updated_at)); + let last_updated_at = page.last().and_then(|i| i.updated_at.clone()); + + for issue in page { + let is_new = since + .as_deref() + .zip(issue.updated_at.as_deref()) + .map(|(cursor, updated)| updated >= cursor) + .unwrap_or(true); + + if find_by_number(runtime, token, "issue", project_id, issue.number) + .await? + .is_some() + { + report.issues_skipped_existing += 1; + if !cursor_stalled { + if let Some(u) = &issue.updated_at { + if max_updated + .as_deref() + .map(|m| u.as_str() > m) + .unwrap_or(true) + { + max_updated = Some(u.clone()); + } + } + } + continue; + } + if !is_new { + continue; + } + if budget.exhausted() { + break; + } + + let raw_body = issue.body.unwrap_or_default(); + let content = secret_gate::mask_secrets(&raw_body).into_owned(); + let labels: Vec<String> = issue + .labels + .unwrap_or_default() + .into_iter() + .map(|l| l.name) + .collect(); + let mut properties = json!({ + "number": issue.number, + "title": issue.title, + "author": issue.author.and_then(|a| a.login), + "created_at": issue.created_at, + "closed_at": issue.closed_at, + "labels": labels, + "project_id": project_id.to_string(), + }); + // gh reports stateReason as "" for open issues and UPPERCASE enum values + // (NOT_PLANNED) for closed ones; the kind hook governs any PRESENT value + // against the lowercase GitHub stateReason enum, so normalize case and encode + // "open / no reason" as absent. + if let Some(reason) = issue.state_reason.as_deref().filter(|r| !r.is_empty()) { + properties["state_reason"] = json!(reason.to_ascii_lowercase()); + } + let name = refs::truncate_chars( + &format!("#{} {}", issue.number, issue.title), + NAME_MAX_CHARS, + ); + + budget.try_consume(); + let result = match registry + .dispatch( + "create", + json!({ + "kind": "issue", + "name": name, + "content": content, + "properties": properties, + "annotates": [project_id.to_string()], + }), + ) + .await + { + Ok(v) => v, + Err(e) => { + report + .warnings + .push(format!("create issue #{}: {e}", issue.number)); + cursor_stalled = true; + continue; + } + }; + if let Some(id) = result + .get("id") + .and_then(|v| v.as_str()) + .and_then(|s| Uuid::parse_str(s).ok()) + { + new_records.push(NewRecordForRef { + id, + text: content.clone(), + }); + } + + report.issues_ingested += 1; if !cursor_stalled { if let Some(u) = &issue.updated_at { if max_updated @@ -845,72 +1342,93 @@ async fn ingest_issues( } } } - continue; - } - if !is_new { - continue; - } - - let raw_body = issue.body.unwrap_or_default(); - let content = secret_gate::mask_secrets(&raw_body).into_owned(); - let labels: Vec<String> = issue - .labels - .unwrap_or_default() - .into_iter() - .map(|l| l.name) - .collect(); - let mut properties = json!({ - "number": issue.number, - "title": issue.title, - "author": issue.author.and_then(|a| a.login), - "created_at": issue.created_at, - "closed_at": issue.closed_at, - "labels": labels, - "project_id": project_id.to_string(), - }); - // gh reports stateReason as "" for open issues and UPPERCASE enum values - // (NOT_PLANNED) for closed ones; the kind hook governs any PRESENT value - // against the lowercase GitHub stateReason enum, so normalize case and encode - // "open / no reason" as absent. - if let Some(reason) = issue.state_reason.as_deref().filter(|r| !r.is_empty()) { - properties["state_reason"] = json!(reason.to_ascii_lowercase()); } - if let Err(e) = registry - .dispatch( - "create", - json!({ - "kind": "issue", - "content": content, - "properties": properties, - "annotates": [project_id.to_string()], - }), - ) - .await - { - report - .warnings - .push(format!("create issue #{}: {e}", issue.number)); - cursor_stalled = true; - continue; - } - - report.issues_ingested += 1; - if !cursor_stalled { - if let Some(u) = &issue.updated_at { - if max_updated - .as_deref() - .map(|m| u.as_str() > m) - .unwrap_or(true) - { - max_updated = Some(u.clone()); - } + match decide_page_outcome( + page_len, + floor.as_deref(), + last_updated_at.as_deref(), + budget.exhausted(), + ) { + PageOutcome::WindowComplete => break 'paging, + PageOutcome::StopBudgetExhausted | PageOutcome::StopFloorStalled => { + window_complete = false; + break 'paging; } + PageOutcome::Continue(next_floor) => floor = Some(next_floor), } } + if !window_complete { + report.done = false; + } + if let Some(cursor) = max_updated { write_cursor(runtime, project_id, "issues", &cursor).await?; } Ok(()) } + +#[cfg(test)] +mod paging_tests { + use super::*; + + #[test] + fn search_query_omits_updated_qualifier_with_no_floor() { + assert_eq!(search_query(None), "sort:updated-asc"); + } + + #[test] + fn search_query_includes_inclusive_updated_floor() { + assert_eq!( + search_query(Some("2024-01-01T00:00:00Z")), + "sort:updated-asc updated:>=2024-01-01T00:00:00Z" + ); + } + + #[test] + fn short_page_proves_window_complete_regardless_of_budget() { + let outcome = decide_page_outcome(42, None, Some("2024-01-01T00:00:00Z"), false); + assert_eq!(outcome, PageOutcome::WindowComplete); + assert!(page_outcome_proves_window_complete(outcome)); + + // Even a page that runs out of budget mid-way is still a proof of + // completeness if the page itself was short — the loop always + // finishes sorting/processing the whole (short) page first. + let outcome = decide_page_outcome(0, None, None, true); + assert_eq!(outcome, PageOutcome::WindowComplete); + } + + /// This is the exact ADR-088 Amendment 1 fix-round High-1 scenario: a + /// full (`PAGE_LIMIT`-sized) page came back, but the local budget was + /// NOT exhausted (e.g. every record in the page already existed and + /// consumed no budget) and paging is still forced to stop because the + /// floor didn't move. `done` must be false here — the remote window is + /// not proven exhausted just because the local budget wasn't hit. + #[test] + fn full_page_with_stalled_floor_is_not_window_complete_even_with_budget_left() { + let outcome = decide_page_outcome(PAGE_LIMIT, Some("X"), Some("X"), false); + assert_eq!(outcome, PageOutcome::StopFloorStalled); + assert!(!page_outcome_proves_window_complete(outcome)); + } + + #[test] + fn full_page_with_advancing_floor_and_budget_left_continues() { + let outcome = decide_page_outcome(PAGE_LIMIT, Some("A"), Some("B"), false); + assert_eq!(outcome, PageOutcome::Continue("B".to_string())); + assert!(!page_outcome_proves_window_complete(outcome)); + } + + #[test] + fn full_page_with_exhausted_budget_stops_without_proving_completeness() { + let outcome = decide_page_outcome(PAGE_LIMIT, Some("A"), Some("B"), true); + assert_eq!(outcome, PageOutcome::StopBudgetExhausted); + assert!(!page_outcome_proves_window_complete(outcome)); + } + + #[test] + fn full_page_with_no_updated_at_stalls_rather_than_looping_forever() { + let outcome = decide_page_outcome(PAGE_LIMIT, Some("A"), None, false); + assert_eq!(outcome, PageOutcome::StopFloorStalled); + } +} diff --git a/crates/khive-pack-git/src/lib.rs b/crates/khive-pack-git/src/lib.rs index 383aff06..15a9d19a 100644 --- a/crates/khive-pack-git/src/lib.rs +++ b/crates/khive-pack-git/src/lib.rs @@ -1,13 +1,29 @@ -//! `khive-pack-git` — the git-lifecycle pack (ADR-088). +//! `khive-pack-git` — the git-lifecycle pack (ADR-088, amended by ADR-088 +//! Amendment 1). //! //! Contributes three note kinds (`commit`, `issue`, `pull_request`) that make -//! repository provenance queryable through the KG graph, populated by a -//! batch, cursor-based ingester (`ingest`) rather than any new agent-facing -//! verb. See `docs/adr/ADR-088-git-lifecycle-pack.md`. +//! repository provenance queryable through the KG graph, and one +//! agent-facing verb, `git.digest(source, project?, max_items?, include?)` +//! (`handlers`), that drives the batch, cursor-based ingester (`ingest`) +//! against either a local path or a remote `https://` URL (cloned/fetched +//! into a daemon-owned scratch cache, `cache`). See +//! `docs/adr/ADR-088-git-lifecycle-pack.md` and +//! `docs/adr/ADR-088-amendment-1-git-digest.md`. +//! +//! | Verb | Args | What it does | +//! | ---- | ---- | ------------ | +//! | `git.digest` | `source`, `project?`, `max_items?`, `include?` | Ingest commit/issue/PR provenance from a local path or `https://` URL, bounded and cursor-resumable | +//! +//! `kkernel git-ingest` remains the unbounded, all-kinds admin CLI path over +//! the same shared `ingest::run_ingest` core. +pub mod cache; +pub mod handlers; pub mod hook; pub mod ingest; mod pack; +pub mod refs; +pub mod source; pub(crate) mod vocab; pub use pack::GitPack; diff --git a/crates/khive-pack-git/src/pack.rs b/crates/khive-pack-git/src/pack.rs index 3d415de1..3c019915 100644 --- a/crates/khive-pack-git/src/pack.rs +++ b/crates/khive-pack-git/src/pack.rs @@ -15,12 +15,14 @@ use khive_types::{EdgeEndpointRule, HandlerDef, Pack}; use crate::hook::{CommitHook, IssueLikeHook}; use crate::vocab::{GIT_NOTE_KIND_SPECS, GIT_SCHEMA_PLAN_STMTS}; -/// Git-lifecycle pack (ADR-088) — registers `commit` / `issue` / `pull_request` -/// note kinds populated by the batch ingester in `src/ingest.rs`. Contributes -/// no new verbs and no edge endpoint rules: provenance edges use the base -/// `annotates` contract only. +/// Git-lifecycle pack (ADR-088, amended by ADR-088 Amendment 1) — registers +/// `commit` / `issue` / `pull_request` note kinds populated by the batch +/// ingester in `src/ingest.rs`, and one agent-facing verb, `git.digest` +/// (`src/handlers.rs`). Extends the base edge contract with `precedes` +/// commit→commit (parent→child lineage, ADR-088 Amendment 1 ingest +/// enrichment) — the only new endpoint rule this pack contributes; +/// everything else uses the base `annotates` contract. pub struct GitPack { - #[allow(dead_code)] runtime: KhiveRuntime, } @@ -28,7 +30,8 @@ impl Pack for GitPack { const NAME: &'static str = "git"; const NOTE_KINDS: &'static [&'static str] = &["commit", "issue", "pull_request"]; const ENTITY_KINDS: &'static [&'static str] = &[]; - const HANDLERS: &'static [HandlerDef] = &[]; + const HANDLERS: &'static [HandlerDef] = &crate::vocab::GIT_HANDLERS; + const EDGE_RULES: &'static [EdgeEndpointRule] = &crate::vocab::GIT_EDGE_RULES; const REQUIRES: &'static [&'static str] = &["kg"]; const NOTE_KIND_SPECS: &'static [NoteKindSpec] = &GIT_NOTE_KIND_SPECS; const SCHEMA_PLAN: Option<PackSchemaPlan> = Some(PackSchemaPlan { @@ -42,6 +45,13 @@ impl GitPack { pub fn new(runtime: KhiveRuntime) -> Self { Self { runtime } } + + /// Accessor for `src/handlers.rs`, which lives in a sibling module and + /// so cannot reach the private `runtime` field directly (mirrors + /// `khive-pack-gtd`'s identical `GtdPack::runtime()` accessor). + pub(crate) fn runtime(&self) -> &KhiveRuntime { + &self.runtime + } } // -- inventory self-registration -------------------------------------------- @@ -115,12 +125,15 @@ impl PackRuntime for GitPack { async fn dispatch( &self, verb: &str, - _params: Value, - _registry: &VerbRegistry, - _token: &NamespaceToken, + params: Value, + registry: &VerbRegistry, + token: &NamespaceToken, ) -> Result<Value, RuntimeError> { - Err(RuntimeError::InvalidInput(format!( - "git pack does not handle verb {verb:?}" - ))) + match verb { + "git.digest" => self.handle_digest(token, registry, params).await, + _ => Err(RuntimeError::InvalidInput(format!( + "git pack does not handle verb {verb:?}" + ))), + } } } diff --git a/crates/khive-pack-git/src/refs.rs b/crates/khive-pack-git/src/refs.rs new file mode 100644 index 00000000..ee9e769e --- /dev/null +++ b/crates/khive-pack-git/src/refs.rs @@ -0,0 +1,257 @@ +//! GitHub reference-grammar extraction (ADR-088 Amendment 1, ingest +//! enrichment riders): `Closes/Fixes/Resolves #N` and bare `#N` mentions in +//! commit messages and issue/PR bodies. +//! +//! Extraction never panics or errors -- fail-open per the amendment. A +//! malformed shape (`#54abc`, a `#` with no digits after it) is simply not +//! matched, not reported as an error. + +/// The two reference kinds materialized as `annotates` edge metadata +/// (`ref_kind` on the edge -- see `crates/khive-pack-git/src/handlers.rs`). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RefKind { + Closes, + Mentions, +} + +impl RefKind { + pub fn as_str(&self) -> &'static str { + match self { + RefKind::Closes => "closes", + RefKind::Mentions => "mentions", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RefMention { + pub number: u64, + pub kind: RefKind, +} + +/// GitHub's closing-keyword grammar (case-insensitive): `Closes`, `Fixes`, +/// `Resolves` and their inflections, immediately preceding `#N` (whitespace +/// and/or a colon allowed in between). +const CLOSING_KEYWORDS: &[&str] = &[ + "close", "closes", "closed", "fix", "fixes", "fixed", "resolve", "resolves", "resolved", +]; + +/// Extract every GitHub-style issue/PR reference from `text`. +/// +/// A `#N` immediately preceded (skipping whitespace/`:`) by a closing +/// keyword is `RefKind::Closes`; every other `#N` is `RefKind::Mentions`. A +/// `#` not immediately followed by at least one digit, or a digit run +/// immediately followed by another alphanumeric character (e.g. `#54abc`, +/// not a clean reference), is skipped. May return duplicate `(number, kind)` +/// pairs for text with repeated references -- callers that materialize one +/// edge per referenced number should dedupe (see `dedupe_prefer_closes`). +pub fn extract_references(text: &str) -> Vec<RefMention> { + let bytes = text.as_bytes(); + let mut mentions = Vec::new(); + let mut i = 0usize; + while i < bytes.len() { + if bytes[i] != b'#' { + i += 1; + continue; + } + let hash_idx = i; + let Some((number, next)) = parse_number(text, hash_idx + 1) else { + i += 1; + continue; + }; + if next < bytes.len() && bytes[next].is_ascii_alphanumeric() { + // "#54abc" -- not a clean reference. + i = next; + continue; + } + let kind = if preceded_by_closing_keyword(text, hash_idx) { + RefKind::Closes + } else { + RefKind::Mentions + }; + mentions.push(RefMention { number, kind }); + i = next; + } + mentions +} + +/// Collapse duplicate `number`s to a single mention, preferring `Closes` +/// over `Mentions` when both occur for the same number (a closing reference +/// is strictly more informative). +pub fn dedupe_prefer_closes(mentions: Vec<RefMention>) -> Vec<RefMention> { + let mut by_number: std::collections::BTreeMap<u64, RefKind> = std::collections::BTreeMap::new(); + for m in mentions { + by_number + .entry(m.number) + .and_modify(|k| { + if m.kind == RefKind::Closes { + *k = RefKind::Closes; + } + }) + .or_insert(m.kind); + } + by_number + .into_iter() + .map(|(number, kind)| RefMention { number, kind }) + .collect() +} + +fn parse_number(text: &str, start: usize) -> Option<(u64, usize)> { + let bytes = text.as_bytes(); + let mut end = start; + while end < bytes.len() && bytes[end].is_ascii_digit() { + end += 1; + } + if end == start { + return None; + } + let n: u64 = text[start..end].parse().ok()?; + Some((n, end)) +} + +/// `true` when the `#` at `hash_idx` is immediately preceded (allowing +/// trailing whitespace/`:` in between) by one of [`CLOSING_KEYWORDS`], +/// case-insensitively, with a non-alphanumeric (or start-of-string) +/// boundary before the keyword. +fn preceded_by_closing_keyword(text: &str, hash_idx: usize) -> bool { + let before = &text[..hash_idx]; + let trimmed = before.trim_end_matches(|c: char| c.is_whitespace() || c == ':'); + let lower = trimmed.to_ascii_lowercase(); + for kw in CLOSING_KEYWORDS { + if let Some(boundary_idx) = lower.len().checked_sub(kw.len()) { + if &lower[boundary_idx..] == *kw { + let prev_char = lower[..boundary_idx].chars().next_back(); + if prev_char + .map(|c| !c.is_ascii_alphanumeric()) + .unwrap_or(true) + { + return true; + } + } + } + } + false +} + +/// Truncate `s` to at most `max_chars` characters (char-boundary safe, no +/// ellipsis -- the amendment specifies "truncated", not a marker suffix). +pub fn truncate_chars(s: &str, max_chars: usize) -> String { + if s.chars().count() <= max_chars { + return s.to_string(); + } + s.chars().take(max_chars).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn numbers_of(kind: RefKind, mentions: &[RefMention]) -> Vec<u64> { + mentions + .iter() + .filter(|m| m.kind == kind) + .map(|m| m.number) + .collect() + } + + #[test] + fn closes_keyword_variants_detected() { + for kw in [ + "Closes", + "closes", + "Close", + "Fixes", + "fix", + "Resolved", + "resolves:", + ] { + let text = format!("{kw} #42"); + let mentions = extract_references(&text); + assert_eq!( + mentions, + vec![RefMention { + number: 42, + kind: RefKind::Closes + }], + "{text:?}" + ); + } + } + + #[test] + fn bare_hash_number_is_mention() { + let mentions = extract_references("see #7 for context"); + assert_eq!( + mentions, + vec![RefMention { + number: 7, + kind: RefKind::Mentions + }] + ); + } + + #[test] + fn multiple_references_in_one_text() { + let mentions = extract_references("Fixes #12, Resolves #34, see also #56"); + assert_eq!(numbers_of(RefKind::Closes, &mentions), vec![12, 34]); + assert_eq!(numbers_of(RefKind::Mentions, &mentions), vec![56]); + } + + #[test] + fn no_false_positive_on_number_immediately_followed_by_letters() { + let mentions = extract_references("see #54abc for the hex code, not an issue"); + assert!(mentions.is_empty(), "{mentions:?}"); + } + + #[test] + fn no_false_positive_on_bare_hash_with_no_digits() { + let mentions = extract_references("a #hashtag not a number, and a lone # too"); + assert!(mentions.is_empty(), "{mentions:?}"); + } + + #[test] + fn double_hash_still_extracts_the_digit_run() { + let mentions = extract_references("##54"); + assert_eq!( + mentions, + vec![RefMention { + number: 54, + kind: RefKind::Mentions + }] + ); + } + + #[test] + fn dedupe_prefers_closes_over_mentions_for_same_number() { + let mentions = vec![ + RefMention { + number: 9, + kind: RefKind::Mentions, + }, + RefMention { + number: 9, + kind: RefKind::Closes, + }, + ]; + let deduped = dedupe_prefer_closes(mentions); + assert_eq!( + deduped, + vec![RefMention { + number: 9, + kind: RefKind::Closes + }] + ); + } + + #[test] + fn truncate_chars_leaves_short_strings_untouched() { + assert_eq!(truncate_chars("short", 120), "short"); + } + + #[test] + fn truncate_chars_cuts_long_strings_at_char_boundary() { + let long = "a".repeat(200); + let truncated = truncate_chars(&long, 120); + assert_eq!(truncated.chars().count(), 120); + } +} diff --git a/crates/khive-pack-git/src/source.rs b/crates/khive-pack-git/src/source.rs new file mode 100644 index 00000000..945a2977 --- /dev/null +++ b/crates/khive-pack-git/src/source.rs @@ -0,0 +1,262 @@ +//! `git.digest` source resolution (ADR-088 Amendment 1): local paths vs. +//! `https://` remote URLs, canonicalization, and `github.com` owner/repo +//! slug derivation for `gh`-based issue/PR ingestion. + +use std::path::PathBuf; + +/// A digest source, resolved from the `git.digest` verb's `source` argument. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DigestSource { + /// An absolute local path known to contain a `.git` entry (directory or, + /// for worktrees, a `gitdir:` pointer file). + Local(PathBuf), + /// A remote `https://` URL to clone/fetch into the scratch cache. + Remote { + /// Canonical form used as the cache key (trailing `/` and `.git` + /// suffix stripped) — same URL always maps to the same cache slot. + canonical: String, + /// `Some((owner, repo))` when the host is `github.com` — the only + /// host `gh` can serve issues/PRs for. `None` for any other https + /// host: the amendment's commits-only degradation applies. + gh_slug: Option<(String, String)>, + }, +} + +/// Parse and validate the `source` argument. +/// +/// - `https://` URLs are accepted for any host. +/// - `ssh://`, `git://`, `http://`, and scp-like `user@host:path` shorthand +/// are rejected outright — no interactive auth in the daemon (ADR-088 +/// Amendment 1, security posture). +/// - Anything else is treated as a local path: it must be absolute and must +/// contain a `.git` entry; arbitrary directory walking is not performed. +pub fn parse_source(raw: &str) -> Result<DigestSource, String> { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return Err("source must not be empty".to_string()); + } + + if let Some(rest) = trimmed.strip_prefix("https://") { + if rest.is_empty() { + return Err(format!("source {trimmed:?} is not a valid https:// URL")); + } + let canonical = canonicalize_https_url(trimmed); + let gh_slug = github_slug(&canonical); + return Ok(DigestSource::Remote { canonical, gh_slug }); + } + + for (scheme, hint) in [ + ( + "ssh://", + "SSH URLs are rejected in v1 (no interactive auth in the daemon)", + ), + ( + "git://", + "the git:// protocol is rejected in v1 -- use an https:// URL", + ), + ("http://", "plain http:// URLs are rejected -- use https://"), + ] { + if trimmed.starts_with(scheme) { + return Err(format!("source {trimmed:?}: {hint}")); + } + } + if is_scp_shorthand(trimmed) { + return Err(format!( + "source {trimmed:?}: SSH shorthand URLs are rejected in v1 (no interactive auth in the daemon)" + )); + } + + if !trimmed.starts_with('/') { + return Err(format!( + "source {trimmed:?} is neither an https:// URL nor an absolute local path (relative paths are rejected)" + )); + } + let path = PathBuf::from(trimmed); + if !path.join(".git").exists() { + return Err(format!( + "local path {trimmed:?} does not contain a .git entry" + )); + } + Ok(DigestSource::Local(path)) +} + +/// `true` for scp-like SSH shorthand (`user@host:path`, e.g. +/// `git@github.com:org/repo.git`) — recognized by an `@` before the first +/// `:` and no leading `/`. +fn is_scp_shorthand(s: &str) -> bool { + if s.starts_with('/') { + return false; + } + let Some(at) = s.find('@') else { + return false; + }; + let Some(colon) = s.find(':') else { + return false; + }; + if colon <= at { + return false; + } + let user = &s[..at]; + !user.is_empty() + && user + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.') +} + +fn canonicalize_https_url(url: &str) -> String { + let mut s = url.trim_end_matches('/').to_string(); + if let Some(stripped) = s.strip_suffix(".git") { + s = stripped.to_string(); + } + s +} + +/// Derive `(owner, repo)` from a canonicalized `https://github.com/...` URL. +/// Returns `None` for any other host, or a github.com URL with fewer than +/// two path segments. +fn github_slug(canonical: &str) -> Option<(String, String)> { + let rest = canonical.strip_prefix("https://")?; + let mut parts = rest.splitn(2, '/'); + let host = parts.next()?; + if !matches!(host, "github.com" | "www.github.com") { + return None; + } + let path = parts.next()?; + let mut segs = path.split('/').filter(|s| !s.is_empty()); + let owner = segs.next()?; + let repo = segs.next()?; + Some((owner.to_string(), repo.to_string())) +} + +/// Basename used as the default `project` entity name and as a fallback +/// scratch-directory label. +pub fn repo_basename(source: &DigestSource) -> String { + match source { + DigestSource::Local(p) => p + .file_name() + .and_then(|f| f.to_str()) + .unwrap_or("repo") + .to_string(), + DigestSource::Remote { canonical, .. } => canonical + .rsplit('/') + .next() + .filter(|s| !s.is_empty()) + .unwrap_or("repo") + .to_string(), + } +} + +/// Deterministic cache-key for the scratch clone directory: the first 16 hex +/// characters of `blake3(canonical_url)` -- short enough for a filesystem +/// path component, long enough that collisions are not a practical concern +/// for a handful of cached repos. +pub fn cache_key(canonical_url: &str) -> String { + blake3::hash(canonical_url.as_bytes()).to_hex()[..16].to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn https_github_url_derives_slug_and_canonical_form() { + let src = parse_source("https://github.com/ohdearquant/khive.git").unwrap(); + match src { + DigestSource::Remote { canonical, gh_slug } => { + assert_eq!(canonical, "https://github.com/ohdearquant/khive"); + assert_eq!( + gh_slug, + Some(("ohdearquant".to_string(), "khive".to_string())) + ); + } + other => panic!("expected Remote, got {other:?}"), + } + } + + #[test] + fn https_github_url_trailing_slash_canonicalizes_same_as_no_slash() { + let a = parse_source("https://github.com/org/repo/").unwrap(); + let b = parse_source("https://github.com/org/repo").unwrap(); + assert_eq!(a, b); + } + + #[test] + fn https_non_github_host_has_no_gh_slug() { + let src = parse_source("https://gitlab.com/org/repo").unwrap(); + match src { + DigestSource::Remote { gh_slug, .. } => assert_eq!(gh_slug, None), + other => panic!("expected Remote, got {other:?}"), + } + } + + #[test] + fn ssh_url_rejected() { + let err = parse_source("ssh://git@github.com/org/repo.git").unwrap_err(); + assert!(err.contains("SSH"), "{err}"); + } + + #[test] + fn scp_shorthand_rejected() { + let err = parse_source("git@github.com:org/repo.git").unwrap_err(); + assert!(err.contains("SSH"), "{err}"); + } + + #[test] + fn git_protocol_rejected() { + let err = parse_source("git://github.com/org/repo.git").unwrap_err(); + assert!(err.contains("git://"), "{err}"); + } + + #[test] + fn plain_http_rejected() { + let err = parse_source("http://github.com/org/repo").unwrap_err(); + assert!(err.contains("http://"), "{err}"); + } + + #[test] + fn relative_local_path_rejected() { + let err = parse_source("relative/path/repo").unwrap_err(); + assert!(err.contains("absolute"), "{err}"); + } + + #[test] + fn absolute_local_path_without_git_dir_rejected() { + let dir = tempfile::tempdir().expect("tempdir"); + let err = parse_source(dir.path().to_str().unwrap()).unwrap_err(); + assert!(err.contains(".git"), "{err}"); + } + + #[test] + fn absolute_local_path_with_git_dir_accepted() { + let dir = tempfile::tempdir().expect("tempdir"); + std::fs::create_dir_all(dir.path().join(".git")).unwrap(); + let src = parse_source(dir.path().to_str().unwrap()).unwrap(); + assert_eq!(src, DigestSource::Local(dir.path().to_path_buf())); + } + + #[test] + fn repo_basename_local_uses_dir_name() { + let src = DigestSource::Local(PathBuf::from("/home/x/my-repo")); + assert_eq!(repo_basename(&src), "my-repo"); + } + + #[test] + fn repo_basename_remote_uses_last_path_segment() { + let src = DigestSource::Remote { + canonical: "https://github.com/org/my-repo".to_string(), + gh_slug: Some(("org".to_string(), "my-repo".to_string())), + }; + assert_eq!(repo_basename(&src), "my-repo"); + } + + #[test] + fn cache_key_is_deterministic_and_16_hex_chars() { + let a = cache_key("https://github.com/org/repo"); + let b = cache_key("https://github.com/org/repo"); + let c = cache_key("https://github.com/org/other"); + assert_eq!(a, b); + assert_ne!(a, c); + assert_eq!(a.len(), 16); + assert!(a.chars().all(|ch| ch.is_ascii_hexdigit())); + } +} diff --git a/crates/khive-pack-git/src/vocab.rs b/crates/khive-pack-git/src/vocab.rs index d3ca38df..e5d556f7 100644 --- a/crates/khive-pack-git/src/vocab.rs +++ b/crates/khive-pack-git/src/vocab.rs @@ -1,11 +1,20 @@ -//! Git pack vocabulary: note kind specs and the pack-auxiliary cursor schema. +//! Git pack vocabulary: note kind specs, the `git.digest` handler +//! declaration, the `precedes` commit→commit edge extension, and the +//! pack-auxiliary cursor schema. //! -//! No `HANDLERS` and no `EDGE_RULES` are declared here (unlike gtd): this pack -//! introduces zero new verbs and relies exclusively on the base `annotates` -//! contract (note -> any substrate) for provenance edges — no endpoint -//! extension is needed. See `crates/khive-pack-git/src/pack.rs`. +//! ADR-088 v0 shipped with no `HANDLERS` and no `EDGE_RULES`: zero new verbs, +//! relying exclusively on the base `annotates` contract (note -> any +//! substrate) for provenance edges. ADR-088 Amendment 1 adds exactly one +//! verb (`git.digest`) and one endpoint extension (`precedes` commit→commit, +//! for parent→child commit lineage — the base contract only allows +//! `precedes` between five entity kinds, never between notes; this pack +//! extends it the same additive way `khive-pack-gtd` extends `depends_on` to +//! task→task). See `crates/khive-pack-git/src/pack.rs`. use khive_runtime::{NoteKindSpec, NoteLifecycleSpec}; +use khive_types::{ + EdgeEndpointRule, EdgeRelation, EndpointKind, HandlerDef, ParamDef, VerbCategory, Visibility, +}; /// Lifecycle declaration shared by `issue` and `pull_request` — both track an /// open/closed state with the same posture as ADR-088's `finding` precedent: @@ -54,3 +63,62 @@ pub(crate) static GIT_SCHEMA_PLAN_STMTS: [&str; 2] = [ "CREATE INDEX IF NOT EXISTS idx_git_mirror_cursor_updated \ ON git_mirror_cursor(updated_at DESC)", ]; + +/// ADR-088 Amendment 1 ingest enrichment: parent→child commit lineage as +/// `precedes` edges. The base endpoint contract only allows `precedes` +/// between five entity kinds (`document`, `dataset`, `artifact`, `service`, +/// `project` — see `khive-runtime::operations::BASE_ENTITY_ENDPOINT_RULES`); +/// it has no note→note case at all. This is the same additive-extension +/// mechanism `khive-pack-gtd` uses for `depends_on` task→task. +pub(crate) static GIT_EDGE_RULES: [EdgeEndpointRule; 1] = [EdgeEndpointRule { + relation: EdgeRelation::Precedes, + source: EndpointKind::NoteOfKind("commit"), + target: EndpointKind::NoteOfKind("commit"), +}]; + +/// Illocutionary classification (Searle 1976): `git.digest` commits data to +/// the graph (ingests notes and edges), so it is `Commissive` — the same +/// category `create`/`link`/`remember` use. +pub(crate) static GIT_HANDLERS: [HandlerDef; 1] = [HandlerDef { + name: "git.digest", + description: "Ingest commit/issue/pull_request provenance from a local git repo path or an \ + https:// URL into the graph. Bounded and cursor-resumable: call repeatedly \ + until the response's `done` field is true.", + visibility: Visibility::Verb, + category: VerbCategory::Commissive, + params: &[ + ParamDef { + name: "source", + param_type: "string", + required: true, + description: "Absolute local path to a git repository (must contain a .git entry), \ + or an https:// URL. Any https host is accepted; non-github.com hosts \ + degrade to commits-only (gh cannot serve their issues/PRs). ssh://, \ + git://, http://, and scp-shorthand (user@host:path) sources are \ + rejected.", + }, + ParamDef { + name: "project", + param_type: "string", + required: false, + description: "UUID or 8+ hex prefix of the repo-anchor project entity. When absent, \ + resolved by matching properties.repo_url or name, or created if none \ + is found (see the response's project_id and project_created).", + }, + ParamDef { + name: "max_items", + param_type: "integer", + required: false, + description: "Bounded work for this call, counted across commits + issues + PRs \ + (default 500, clamped to 1..=2000). Cursor-resumable: call again \ + while the response's done field is false.", + }, + ParamDef { + name: "include", + param_type: "array of string", + required: false, + description: "Which record kinds to ingest this call: any of commits | issues | \ + pull_requests (default: all three).", + }, + ], +}]; diff --git a/crates/khive-pack-git/tests/acceptance.rs b/crates/khive-pack-git/tests/acceptance.rs index 88424389..cfd3af46 100644 --- a/crates/khive-pack-git/tests/acceptance.rs +++ b/crates/khive-pack-git/tests/acceptance.rs @@ -219,10 +219,7 @@ async fn ingest_links_commits_to_document_and_pr_by_provenance_query() { &rt, &token, ®istry, - IngestOptions { - repo: repo.to_path_buf(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.to_path_buf(), project_id.to_string()), ) .await .expect("ingest ok"); @@ -331,10 +328,7 @@ async fn ingest_masks_secrets_in_commit_message() { &rt, &token, ®istry, - IngestOptions { - repo: repo.to_path_buf(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.to_path_buf(), project_id.to_string()), ) .await .expect("ingest ok"); @@ -485,10 +479,7 @@ async fn issue_and_pr_idempotency_is_scoped_per_project() { &rt, &token, ®istry, - IngestOptions { - repo: repo_b.clone(), - project: project_b.to_string(), - }, + IngestOptions::unbounded(repo_b.clone(), project_b.to_string()), ) .await .expect("ingest ok (pass 1)"); @@ -543,10 +534,7 @@ async fn issue_and_pr_idempotency_is_scoped_per_project() { &rt, &token, ®istry, - IngestOptions { - repo: repo_b.clone(), - project: project_b.to_string(), - }, + IngestOptions::unbounded(repo_b.clone(), project_b.to_string()), ) .await .expect("ingest ok (pass 2)"); @@ -646,10 +634,7 @@ async fn gh_boundary_contract_and_partial_ingest_failure() { &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 1)"); @@ -680,6 +665,24 @@ async fn gh_boundary_contract_and_partial_ingest_failure() { !args_log.contains("-C"), "gh must never receive an unsupported -C flag: {args_log}" ); + // ADR-088 Amendment 1 fix-round r2 High-1: the paging rewrite (--search + // "sort:updated-asc ...") must not silently drop --state all -- gh + // defaults to open-only listing, which would make closed issues and + // closed/merged PRs vanish from every ingest. + let pr_and_issue_invocations: Vec<&str> = args_log + .lines() + .filter(|l| l.starts_with("pr ") || l.starts_with("issue ")) + .collect(); + assert!( + !pr_and_issue_invocations.is_empty(), + "expected at least one gh pr/issue list invocation: {args_log}" + ); + for line in &pr_and_issue_invocations { + assert!( + line.contains("--state all"), + "every gh pr/issue list invocation must request --state all: {line:?}" + ); + } let cwd_log = std::fs::read_to_string(log_dir.join("cwd.log")).expect("read cwd.log"); let cwd_lines: Vec<&str> = cwd_log.lines().filter(|l| !l.is_empty()).collect(); assert!( @@ -756,10 +759,7 @@ async fn gh_boundary_contract_and_partial_ingest_failure() { &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 2)"); @@ -862,10 +862,7 @@ async fn issue_ingest_sorts_by_updated_at_so_frozen_cursor_survives_out_of_order &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 1)"); @@ -907,10 +904,7 @@ async fn issue_ingest_sorts_by_updated_at_so_frozen_cursor_survives_out_of_order &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 2)"); @@ -1005,10 +999,7 @@ async fn pr_ingest_sorts_by_updated_at_so_frozen_cursor_survives_out_of_order_li &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 1)"); @@ -1050,10 +1041,7 @@ async fn pr_ingest_sorts_by_updated_at_so_frozen_cursor_survives_out_of_order_li &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 2)"); @@ -1154,10 +1142,7 @@ async fn issue_ingest_retries_tie_at_cursor_timestamp() { &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 1)"); @@ -1196,10 +1181,7 @@ async fn issue_ingest_retries_tie_at_cursor_timestamp() { &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 2)"); @@ -1283,10 +1265,7 @@ async fn pr_ingest_retries_tie_at_cursor_timestamp() { &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 1)"); @@ -1325,10 +1304,7 @@ async fn pr_ingest_retries_tie_at_cursor_timestamp() { &rt, &token, ®istry, - IngestOptions { - repo: repo.clone(), - project: project_id.to_string(), - }, + IngestOptions::unbounded(repo.clone(), project_id.to_string()), ) .await .expect("ingest ok (pass 2)"); @@ -1368,3 +1344,316 @@ async fn pr_ingest_retries_tie_at_cursor_timestamp() { "exactly #5, #20 — no duplicates: {numbers:?}" ); } + +// ── `git.digest` verb (ADR-088 Amendment 1) ───────────────────────────────── + +/// End-to-end over the `git.digest` verb itself (not `run_ingest` directly): +/// no `project` argument auto-creates the repo-anchor entity (reported via +/// `project_created`), a `Closes #N` commit message materializes an +/// `annotates` edge (with `ref_kind: "closes"`) from the commit to a +/// pre-existing `issue` note, a second commit's `parents[]` materializes a +/// `precedes` edge from parent to child, and both commit and issue notes get +/// the amendment's readable `name`. +#[tokio::test] +async fn digest_verb_auto_creates_project_and_enriches_references() { + let _guard = ENV_MUTEX.lock().await; + let (_rt, _token, registry) = fixture().await; + + let dir = tempfile::tempdir().expect("tempdir"); + let repo = dir.path(); + init_repo(repo); + write(repo, "README.md", "hello\n"); + commit(repo, &["README.md"], "Initial commit"); + + let first = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "max_items": 10}), + ) + .await + .expect("digest ok (pass 1)"); + + assert_eq!(first["done"], true, "{first}"); + assert_eq!(first["project_created"], true, "{first}"); + assert_eq!(first["commits_ingested"], 1, "{first}"); + let project_id = first["project_id"] + .as_str() + .expect("project_id present") + .to_string(); + + // Second digest call for the same local path resolves the SAME project + // (no duplicate anchor entity). + let second = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "max_items": 10}), + ) + .await + .expect("digest ok (pass 2, no new commits)"); + assert_eq!(second["project_created"], false, "{second}"); + assert_eq!(second["project_id"], project_id, "{second}"); + assert_eq!( + second["commits_ingested"], 0, + "no new commits since pass 1: {second}" + ); + + // Pre-create an issue #42 the project already tracks (as if ingested by + // an earlier `gh`-backed pass), then a commit whose message closes it. + let issue_id = create( + ®istry, + json!({ + "kind": "issue", + "content": "", + "properties": {"number": 42, "title": "Some bug", "project_id": project_id}, + "annotates": [project_id], + }), + ) + .await; + + write(repo, "src/lib.rs", "// fix\n"); + commit(repo, &["src/lib.rs"], "Fix the bug\n\nCloses #42"); + let commit2_sha = head_sha(repo); + + let third = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "project": project_id, "max_items": 10}), + ) + .await + .expect("digest ok (pass 3)"); + assert_eq!(third["commits_ingested"], 1, "{third}"); + assert_eq!( + third["reference_edges_created"], 1, + "the Closes #42 reference resolves: {third}" + ); + assert_eq!( + third["parent_edges_created"], 1, + "the second commit's parent link to the first: {third}" + ); + + // The issue has exactly one incoming `annotates` edge, from the closing + // commit, carrying ref_kind=closes. + let issue_neighbors = registry + .dispatch( + "neighbors", + json!({"id": issue_id.to_string(), "direction": "incoming", "relations": ["annotates"]}), + ) + .await + .expect("neighbors ok"); + let hits = issue_neighbors.as_array().expect("array"); + assert_eq!(hits.len(), 1, "{hits:?}"); + assert_eq!(hits[0]["kind"], "commit"); + + // The closing commit's own record carries the readable name and its sha. + let commit_note = registry + .dispatch("list", json!({"kind": "commit", "limit": 10})) + .await + .expect("list ok"); + let items = commit_note.as_array().expect("array"); + let closing = items + .iter() + .find(|c| c["properties"]["sha"] == commit2_sha) + .expect("closing commit note present"); + let name = closing["name"].as_str().expect("name present"); + assert!( + name.contains("Fix the bug"), + "commit name must carry the subject: {name:?}" + ); + + // Parent -> child `precedes` edge: the first commit precedes the second. + let first_commit = items + .iter() + .find(|c| c["properties"]["sha"] != commit2_sha) + .expect("first commit note present"); + let first_commit_id = first_commit["id"].as_str().expect("id present"); + let precedes_neighbors = registry + .dispatch( + "neighbors", + json!({"id": first_commit_id, "direction": "outgoing", "relations": ["precedes"]}), + ) + .await + .expect("neighbors ok"); + let precedes_hits = precedes_neighbors.as_array().expect("array"); + assert_eq!( + precedes_hits.len(), + 1, + "the first commit precedes exactly the second: {precedes_hits:?}" + ); + assert_eq!( + precedes_hits[0]["id"].as_str().unwrap(), + closing["id"].as_str().unwrap() + ); +} + +/// `max_items` bounds work per call and the response is cursor-resumable: +/// looping `git.digest` calls until `done` eventually ingests every commit +/// with no duplicates. +#[tokio::test] +async fn digest_verb_max_items_is_bounded_and_resumable() { + let _guard = ENV_MUTEX.lock().await; + let (_rt, _token, registry) = fixture().await; + + let dir = tempfile::tempdir().expect("tempdir"); + let repo = dir.path(); + init_repo(repo); + for i in 0..3 { + write(repo, "f.txt", &format!("v{i}\n")); + commit(repo, &["f.txt"], &format!("commit {i}")); + } + + let mut total_ingested = 0u64; + let mut project_id: Option<String> = None; + let mut calls = 0; + loop { + calls += 1; + assert!(calls <= 10, "must converge well within 10 calls"); + let mut args = + json!({"source": repo.to_str().unwrap(), "max_items": 1, "include": ["commits"]}); + if let Some(p) = &project_id { + args["project"] = json!(p); + } + let resp = registry + .dispatch("git.digest", args) + .await + .expect("digest ok"); + project_id = Some(resp["project_id"].as_str().unwrap().to_string()); + total_ingested += resp["commits_ingested"].as_u64().unwrap(); + if resp["done"].as_bool().unwrap() { + break; + } + } + assert_eq!(total_ingested, 3, "all three commits eventually ingest"); + + let list = registry + .dispatch("list", json!({"kind": "commit", "limit": 10})) + .await + .expect("list ok"); + assert_eq!( + list.as_array().expect("array").len(), + 3, + "no duplicates across the bounded/resumed calls" + ); +} + +/// Source validation surfaces as a normal verb error (ssh:// rejected, +/// ADR-088 Amendment 1 security posture) rather than panicking or silently +/// no-op'ing. +#[tokio::test] +async fn digest_verb_rejects_ssh_source() { + let (_rt, _token, registry) = fixture().await; + let err = registry + .dispatch( + "git.digest", + json!({"source": "ssh://git@github.com/org/repo.git"}), + ) + .await + .expect_err("ssh source must be rejected"); + assert!(format!("{err}").contains("SSH"), "{err}"); +} + +/// `max_items` boundary handling (ADR-088 Amendment 1 fix-round Medium-2): +/// a negative value must clamp to the lower bound (1), NOT silently fall +/// through `as_u64`'s failure into the 500 default. `0` clamps to 1 too; +/// values above 2000 clamp to 2000; a non-integer value is a hard error. +#[tokio::test] +async fn digest_verb_max_items_negative_and_zero_clamp_to_one() { + let _guard = ENV_MUTEX.lock().await; + + for requested in [-1, 0] { + let (_rt, _token, registry) = fixture().await; + let dir = tempfile::tempdir().expect("tempdir"); + let repo = dir.path(); + init_repo(repo); + for i in 0..3 { + write(repo, "f.txt", &format!("v{i}\n")); + commit(repo, &["f.txt"], &format!("commit {i}")); + } + + let resp = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "max_items": requested, "include": ["commits"]}), + ) + .await + .unwrap_or_else(|e| panic!("digest ok for max_items={requested}: {e}")); + assert_eq!( + resp["commits_ingested"].as_u64().unwrap(), + 1, + "max_items={requested} must clamp to the lower bound (1 item this call): {resp:?}" + ); + assert!( + !resp["done"].as_bool().unwrap(), + "2 commits remain after a 1-item pass: {resp:?}" + ); + } +} + +#[tokio::test] +async fn digest_verb_max_items_above_cap_clamps_to_two_thousand() { + let (_rt, _token, registry) = fixture().await; + let dir = tempfile::tempdir().expect("tempdir"); + let repo = dir.path(); + init_repo(repo); + write(repo, "f.txt", "v\n"); + commit(repo, &["f.txt"], "only commit"); + + let resp = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "max_items": 2001, "include": ["commits"]}), + ) + .await + .expect("digest ok"); + assert_eq!(resp["commits_ingested"].as_u64().unwrap(), 1); + assert!( + resp["done"].as_bool().unwrap(), + "a single-commit repo finishes in one call however large max_items clamps to: {resp:?}" + ); +} + +#[tokio::test] +async fn digest_verb_max_items_at_boundary_values() { + for (requested, expected_ingested) in [(1i64, 1u64), (2000i64, 1u64)] { + let (_rt, _token, registry) = fixture().await; + let dir = tempfile::tempdir().expect("tempdir"); + let repo = dir.path(); + init_repo(repo); + write(repo, "f.txt", "v\n"); + commit(repo, &["f.txt"], "only commit"); + + let resp = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "max_items": requested, "include": ["commits"]}), + ) + .await + .unwrap_or_else(|e| panic!("digest ok for max_items={requested}: {e}")); + assert_eq!( + resp["commits_ingested"].as_u64().unwrap(), + expected_ingested, + "max_items={requested}: {resp:?}" + ); + } +} + +#[tokio::test] +async fn digest_verb_rejects_non_integer_max_items() { + let (_rt, _token, registry) = fixture().await; + let dir = tempfile::tempdir().expect("tempdir"); + let repo = dir.path(); + init_repo(repo); + write(repo, "f.txt", "v\n"); + commit(repo, &["f.txt"], "only commit"); + + let err = registry + .dispatch( + "git.digest", + json!({"source": repo.to_str().unwrap(), "max_items": "lots"}), + ) + .await + .expect_err("a non-integer max_items must be a hard error, not a silent default"); + assert!( + format!("{err}").contains("max_items"), + "error must name the offending field: {err}" + ); +} diff --git a/crates/kkernel/src/git_ingest.rs b/crates/kkernel/src/git_ingest.rs index da1bfb44..8b1fef49 100644 --- a/crates/kkernel/src/git_ingest.rs +++ b/crates/kkernel/src/git_ingest.rs @@ -83,10 +83,7 @@ pub async fn run_git_ingest(args: GitIngestArgs) -> Result<()> { &runtime, &token, ®istry, - IngestOptions { - repo: args.repo, - project: args.project, - }, + IngestOptions::unbounded(args.repo, args.project), ) .await?; diff --git a/docs/guide/api-reference.md b/docs/guide/api-reference.md index d8d72582..36650d97 100644 --- a/docs/guide/api-reference.md +++ b/docs/guide/api-reference.md @@ -1,13 +1,13 @@ # API Reference -khive exposes exactly one MCP tool, `request`. Everything else — 75 verbs across 9 +khive exposes exactly one MCP tool, `request`. Everything else — 77 verbs across 9 production packs — is dispatched through that single tool via a small request DSL. This page documents the DSL grammar, the response envelope, and every verb's full parameter contract, so an agent can call khive correctly without reading Rust source. This page is verified against the live registry (`request(ops="verbs()")`, run -2026-07-07) and the pack source (`crates/khive-pack-*/src/*.rs` `HandlerDef`/`ParamDef` -struct literals). Verb count: **75**, matching both the live registry `total` field and +2026-07-09) and the pack source (`crates/khive-pack-*/src/*.rs` `HandlerDef`/`ParamDef` +struct literals). Verb count: **76**, matching both the live registry `total` field and the sum of the 9 pack counts below. If your server reports a different total, your `KHIVE_PACKS` configuration loads a different pack set than the default — run `request(ops="verbs()")` against your own server to get the authoritative list. @@ -24,19 +24,19 @@ An always-machine-readable copy of this page is at | `kg` | 17 | `KHIVE_PACKS=kg` (default) | No — base substrate | | `gtd` | 5 | `KHIVE_PACKS=kg,gtd` | Yes | | `memory` | 5 | `KHIVE_PACKS=kg,memory` | Yes | -| `brain` | 14 | `KHIVE_PACKS=kg,brain` | Yes | +| `brain` | 15 | `KHIVE_PACKS=kg,brain` | Yes | | `comm` | 7 | `KHIVE_PACKS=kg,comm` | Yes | | `schedule` | 4 | `KHIVE_PACKS=kg,schedule` | Yes | | `knowledge` | 19 | `KHIVE_PACKS=kg,knowledge` | Yes | | `session` | 4 | `KHIVE_PACKS=kg,session` | Yes | -| `git` | 0 | `KHIVE_PACKS=kg,git` | Yes | +| `git` | 1 | `KHIVE_PACKS=kg,git` | Yes | -`git` contributes zero verbs — it registers the `commit` / `issue` / `pull_request` note -kinds and a batch ingester (`crates/khive-pack-git/src/ingest.rs`), consumed outside the -`request` DSL, not new MCP-callable verbs. +`git` also registers the `commit` / `issue` / `pull_request` note kinds and the shared +`run_ingest` core (`crates/khive-pack-git/src/ingest.rs`) that both `git.digest` and the +`kkernel git-ingest` CLI drive. The default binary (no `KHIVE_PACKS`/`--pack` override) loads all 9 packs: 17 + 5 + 5 + -14 + 7 + 4 + 19 + 4 + 0 = **75 verbs**. +15 + 7 + 4 + 19 + 4 + 1 = **77 verbs**. Verb names in the `kg` pack are bare (`create`, `search`, `link`, …). Every other pack namespaces its verbs with a `pack.` prefix (`gtd.assign`, `memory.recall`, @@ -1350,6 +1350,35 @@ request(ops="session.export(id=\"<session-id>\", format=\"markdown\")") --- +## `git` pack — 1 verb + +Batch, cursor-based git-history ingester (ADR-088, ADR-088 Amendment 1). Optional; load +with `KHIVE_PACKS=kg,git`. Also registers the `commit` / `issue` / `pull_request` note +kinds, used by `git.digest` below and by the `kkernel git-ingest` CLI (both drive the +same underlying ingest core, so ingest enrichment — readable `name`s, `Closes #N` +reference edges, parent→child commit `precedes` edges — applies identically either way). + +### `git.digest` — Commissive + +Walk a local repository path or clone/fetch a remote `https://` URL, then ingest commits +and (when the source is a github.com repo and the `gh` CLI is available) issues and pull +requests as provenance notes, resolving or auto-creating the repo-anchor `project` entity. +Bounded and cursor-resumable: call again with the same `source`/`project` while the +response's `done` field is `false`. + +| Param | Type | Required | Notes | +| ----------- | --------------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `source` | string | yes | A local filesystem path (must contain `.git`) or an `https://` URL. Any `https` host is accepted; non-github.com hosts degrade to commits-only. `ssh://`, `git://`, `http://`, and scp-shorthand (`user@host:path`) sources are rejected. | +| `project` | string | no | UUID or 8+ hex prefix of the repo-anchor `project` entity. When absent, resolved by matching `properties.repo_url` or `name`, or created if none is found (see the response's `project_id` and `project_created`). | +| `max_items` | integer | no | Bounded work for this call, counted across commits + issues + PRs (default 500, clamped to 1..=2000). Cursor-resumable: call again while the response's `done` field is `false`. | +| `include` | array\<string\> | no | Which record kinds to ingest this call: any of `commits` \| `issues` \| `pull_requests` (default: all three). | + +``` +request(ops="git.digest(source=\"https://github.com/org/repo\", max_items=500)") +``` + +--- + ## Further reading - [Getting Started](getting-started.html): install and connect an MCP client. diff --git a/tests/smoke_test.py b/tests/smoke_test.py index a442f7a2..9e01d4ad 100644 --- a/tests/smoke_test.py +++ b/tests/smoke_test.py @@ -197,7 +197,7 @@ def main(): assert isinstance(verbs_result["verbs"], list), f"verbs must be a list: {verbs_result}" # Surface-contract tripwire: the default config (no --pack, KHIVE_PACKS # unset) loads 9 production packs (kg, gtd, memory, brain, comm, schedule, - # knowledge, session, git), so verbs() returns exactly 76 user-facing + # knowledge, session, git), so verbs() returns exactly 77 user-facing # MCP-callable verbs (count what verbs() returns, not internal dispatch # arms). The session pack contributes 4 agent-facing T1 verbs # (store/list/resume/export), promoted from internal subhandlers to @@ -206,15 +206,15 @@ def main(): # verified live 2026-07-04), comm.probe (#644 read-only inbound # poll), and brain.event_counts (#724, ADR-103 Stage 1 windowed event # read) are included in the count; git contributes - # 0 verbs (note kinds + ingester only). Update this number when the + # git.digest (ADR-088 Amendment 1). Update this number when the # pack set or verb surface changes; a silent drift here is the bug # this assertion exists to catch. - assert verbs_result["total"] == 76, ( - f"expected 76 user-facing verbs from the 9 default packs " + assert verbs_result["total"] == 77, ( + f"expected 77 user-facing verbs from the 9 default packs " f"(session contributes 4 T1 verbs promoted to Visibility::Verb per " f"ADR-083; context is the 17th kg-substrate bare verb per ADR-089; " f"comm.health is #606; comm.probe is #644; brain.event_counts is " - f"#724/ADR-103; git contributes 0), " + f"#724/ADR-103; git contributes git.digest), " f"got {verbs_result['total']}: {verbs_result}" ) verb_names = [v["verb"] for v in verbs_result["verbs"]]