From 850d275841a858c03bd3619e5d8a98991f821822 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sat, 30 May 2026 21:35:45 -0700 Subject: [PATCH] feat(memory): repo-grouped GitHub raw archive + entities & priority (#3047) --- .../memory_sources/readers/composio.rs | 3 + .../memory_sources/readers/folder.rs | 3 + .../memory_sources/readers/github.rs | 310 +++++++++++++++--- .../memory_sources/readers/twitter.rs | 3 + src/openhuman/memory_sources/registry.rs | 3 + src/openhuman/memory_sources/rpc.rs | 12 + src/openhuman/memory_sources/status.rs | 3 + src/openhuman/memory_sources/sync.rs | 184 ++++++++++- src/openhuman/memory_sources/types.rs | 12 + src/openhuman/memory_store/content/raw.rs | 34 ++ src/openhuman/memory_tree/score/mod.rs | 40 ++- src/openhuman/memory_tree/summarise.rs | 15 +- ...threads_memory_sources_raw_coverage_e2e.rs | 3 + tests/memory_raw_coverage_e2e.rs | 3 + ...ources_closure_round23_raw_coverage_e2e.rs | 3 + ...ources_readers_round21_raw_coverage_e2e.rs | 3 + tests/memory_sync_sources_raw_coverage_e2e.rs | 3 + tests/memory_threads_raw_coverage_e2e.rs | 12 + tests/near90_closure_raw_coverage_e2e.rs | 3 + 19 files changed, 602 insertions(+), 50 deletions(-) diff --git a/src/openhuman/memory_sources/readers/composio.rs b/src/openhuman/memory_sources/readers/composio.rs index 107f2c44b..d39c0a6c5 100644 --- a/src/openhuman/memory_sources/readers/composio.rs +++ b/src/openhuman/memory_sources/readers/composio.rs @@ -86,6 +86,9 @@ mod tests { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/src/openhuman/memory_sources/readers/folder.rs b/src/openhuman/memory_sources/readers/folder.rs index 2f6020282..2e29e45a4 100644 --- a/src/openhuman/memory_sources/readers/folder.rs +++ b/src/openhuman/memory_sources/readers/folder.rs @@ -162,6 +162,9 @@ mod tests { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/src/openhuman/memory_sources/readers/github.rs b/src/openhuman/memory_sources/readers/github.rs index 491064857..55e75dfee 100644 --- a/src/openhuman/memory_sources/readers/github.rs +++ b/src/openhuman/memory_sources/readers/github.rs @@ -13,11 +13,24 @@ use crate::openhuman::config::Config; use crate::openhuman::memory_sources::types::{ ContentType, MemorySourceEntry, SourceContent, SourceItem, SourceKind, }; +use crate::openhuman::memory_store::content::raw::RawKind; use super::SourceReader; const DEFAULT_BRANCH: &str = "main"; +/// Default number of items of **each** type (commits, issues, PRs) to pull +/// when the source entry doesn't override it. Tunable per-source via +/// `max_commits` / `max_issues` / `max_prs` on [`MemorySourceEntry`]. +pub(crate) const DEFAULT_GITHUB_ITEM_LIMIT: u32 = 2000; + +/// GitHub REST API maximum page size (`per_page`). +const GH_PAGE_SIZE: u32 = 100; + +/// Hard ceiling on pagination loops so a misbehaving API (always returning a +/// full page) can never spin forever even if `max` is enormous. +const GH_MAX_PAGES: u32 = 1000; + pub struct GithubReader; /// Parse `owner` and `repo` from a GitHub URL. @@ -26,7 +39,7 @@ pub struct GithubReader; /// shape — extra segments like `/tree/main` or `/blob/...` are rejected /// so callers can't accidentally derive the wrong owner/repo from a /// deep link. -fn parse_github_url(url: &str) -> Result<(String, String), String> { +pub(crate) fn parse_github_url(url: &str) -> Result<(String, String), String> { let trimmed = url.trim(); let rest = trimmed .strip_prefix("https://github.com/") @@ -84,6 +97,42 @@ impl ItemKind { } } +// ── Raw-archive coordinates ───────────────────────────────────────── + +/// Slugifiable raw-archive source id for a repo URL. +/// +/// Returns `github.com//`, which slugifies (via +/// `slugify_source_id`) to `github-com--` so a source's +/// commits/issues/PRs land under +/// `raw/github-com--/{commits,issues,prs}/`. +pub(crate) fn repo_archive_source_id(url: &str) -> Option { + let (owner, repo) = parse_github_url(url).ok()?; + Some(format!("github.com/{owner}/{repo}")) +} + +/// Chunk-store source id for a single repo item. +/// +/// `github:/:` keeps the per-item dedup key (each +/// commit/issue/PR is an immutable document) while producing a clean +/// `chunks/document/github---…` scope instead of the opaque +/// `mem_src:src_:…` form. +pub(crate) fn chunk_source_id(url: &str, item_id: &str) -> Option { + let (owner, repo) = parse_github_url(url).ok()?; + Some(format!("github:{owner}/{repo}:{item_id}")) +} + +/// Map a [`SourceItem`] id (`commit:`, `issue:`, `pr:`) to its +/// raw-archive [`RawKind`] and the clean uid used as the filename suffix. +pub(crate) fn raw_archive_coords(item_id: &str) -> Option<(RawKind, String)> { + let (kind, rest) = ItemKind::from_id(item_id)?; + let raw_kind = match kind { + ItemKind::Commit => RawKind::Commit, + ItemKind::Issue => RawKind::Issue, + ItemKind::PullRequest => RawKind::PullRequest, + }; + Some((raw_kind, rest.to_string())) +} + // ── gh CLI helpers ────────────────────────────────────────────────── const GH_CLI_TIMEOUT: Duration = Duration::from_secs(30); @@ -138,6 +187,11 @@ async fn api_get(path: &str) -> Result { struct GhCommit { sha: String, commit: GhCommitInner, + /// Top-level GitHub user that authored the commit (distinct from the + /// embedded git author identity). Present when the commit author maps + /// to a GitHub account; absent for unlinked email-only authors. + #[serde(default)] + author: Option, } #[derive(Debug, Deserialize)] @@ -212,18 +266,27 @@ impl SourceReader for GithubReader { let (owner, repo) = parse_github_url(url)?; let use_gh = gh_available(); + // Per-type sync caps. Default to the last DEFAULT_GITHUB_ITEM_LIMIT + // of each type; a source entry may override any of them. + let max_commits = source.max_commits.unwrap_or(DEFAULT_GITHUB_ITEM_LIMIT); + let max_issues = source.max_issues.unwrap_or(DEFAULT_GITHUB_ITEM_LIMIT); + let max_prs = source.max_prs.unwrap_or(DEFAULT_GITHUB_ITEM_LIMIT); + tracing::debug!( owner = %owner, repo = %repo, use_gh = use_gh, + max_commits, + max_issues, + max_prs, "[memory_sources:github] listing items" ); let mut items = Vec::new(); let mut errors = Vec::new(); - // Commits (last 30) - match list_commits(&owner, &repo, use_gh).await { + // Commits (most recent `max_commits`) + match list_commits(&owner, &repo, max_commits, use_gh).await { Ok(commits) => items.extend(commits), Err(e) => { tracing::warn!(error = %e, "[memory_sources:github] failed to list commits"); @@ -231,8 +294,8 @@ impl SourceReader for GithubReader { } } - // Issues (last 30 open + recently closed) - match list_issues(&owner, &repo, use_gh).await { + // Issues (most recent `max_issues`, open + closed) + match list_issues(&owner, &repo, max_issues, use_gh).await { Ok(issues) => items.extend(issues), Err(e) => { tracing::warn!(error = %e, "[memory_sources:github] failed to list issues"); @@ -240,8 +303,8 @@ impl SourceReader for GithubReader { } } - // Pull requests (last 30) - match list_prs(&owner, &repo, use_gh).await { + // Pull requests (most recent `max_prs`, open + closed) + match list_prs(&owner, &repo, max_prs, use_gh).await { Ok(prs) => items.extend(prs), Err(e) => { tracing::warn!(error = %e, "[memory_sources:github] failed to list PRs"); @@ -319,13 +382,55 @@ async fn fetch_github(api_path: &str, use_gh: bool) -> Result { // ── List helpers ──────────────────────────────────────────────────── -async fn list_commits(owner: &str, repo: &str, use_gh: bool) -> Result, String> { - let json_str = - fetch_github(&format!("repos/{owner}/{repo}/commits?per_page=30"), use_gh).await?; +/// Fetch up to `max` rows from a paginated GitHub list endpoint. +/// +/// Walks `?per_page=100&page=N` until `max` rows are collected or the API +/// returns a short page (the last page). `extra_query` is appended verbatim +/// (e.g. `"state=all"`). The result is truncated to exactly `max`. +async fn fetch_all_pages( + owner: &str, + repo: &str, + resource: &str, + extra_query: &str, + max: u32, + use_gh: bool, +) -> Result, String> { + let mut out: Vec = Vec::new(); + let mut page = 1u32; - // Try parsing as array of commit objects - let commits: Vec = - serde_json::from_str(&json_str).map_err(|e| format!("parse commits: {e}"))?; + while (out.len() as u32) < max && page <= GH_MAX_PAGES { + let remaining = max - out.len() as u32; + let per_page = remaining.min(GH_PAGE_SIZE); + let mut path = format!("repos/{owner}/{repo}/{resource}?per_page={per_page}&page={page}"); + if !extra_query.is_empty() { + path.push('&'); + path.push_str(extra_query); + } + + let json_str = fetch_github(&path, use_gh).await?; + let batch: Vec = serde_json::from_str(&json_str) + .map_err(|e| format!("parse {resource} page {page}: {e}"))?; + let got = batch.len(); + out.extend(batch); + + // Short page ⇒ no more rows upstream. + if got < per_page as usize { + break; + } + page += 1; + } + + out.truncate(max as usize); + Ok(out) +} + +async fn list_commits( + owner: &str, + repo: &str, + max: u32, + use_gh: bool, +) -> Result, String> { + let commits: Vec = fetch_all_pages(owner, repo, "commits", "", max, use_gh).await?; Ok(commits .into_iter() @@ -346,38 +451,61 @@ async fn list_commits(owner: &str, repo: &str, use_gh: bool) -> Result Result, String> { - let json_str = fetch_github( - &format!("repos/{owner}/{repo}/issues?per_page=30&state=all"), - use_gh, - ) - .await?; +async fn list_issues( + owner: &str, + repo: &str, + max: u32, + use_gh: bool, +) -> Result, String> { + // The `/issues` endpoint interleaves PRs. Page through FULL pages + // (per_page=100) and accumulate only non-PR items until we've collected + // `max` actual issues or the endpoint is exhausted — otherwise a repo with + // many interleaved PRs would yield fewer than `max` issues. We can't reuse + // `fetch_all_pages` here because it shrinks `per_page` to the raw-row + // remainder and counts PRs against the budget. + let mut out: Vec = Vec::new(); + let mut page = 1u32; - let issues: Vec = - serde_json::from_str(&json_str).map_err(|e| format!("parse issues: {e}"))?; + while (out.len() as u32) < max && page <= GH_MAX_PAGES { + let path = + format!("repos/{owner}/{repo}/issues?per_page={GH_PAGE_SIZE}&page={page}&state=all"); + let json_str = fetch_github(&path, use_gh).await?; + let batch: Vec = serde_json::from_str(&json_str) + .map_err(|e| format!("parse issues page {page}: {e}"))?; + let got = batch.len(); - Ok(issues - .into_iter() - .filter(|i| i.pull_request.is_none()) // filter out PRs from issues endpoint - .map(|i| { + for i in batch { + if i.pull_request.is_some() { + continue; // drop PRs surfaced by the issues endpoint + } let ts = i.updated_at.as_deref().and_then(parse_iso_ts); - SourceItem { + out.push(SourceItem { id: format!("issue:{}", i.number), title: format!("#{} {}", i.number, i.title), updated_at_ms: ts, + }); + if out.len() as u32 >= max { + break; } - }) - .collect()) + } + + // Short page ⇒ no more rows upstream. + if got < GH_PAGE_SIZE as usize { + break; + } + page += 1; + } + + Ok(out) } -async fn list_prs(owner: &str, repo: &str, use_gh: bool) -> Result, String> { - let json_str = fetch_github( - &format!("repos/{owner}/{repo}/pulls?per_page=30&state=all"), - use_gh, - ) - .await?; - - let prs: Vec = serde_json::from_str(&json_str).map_err(|e| format!("parse PRs: {e}"))?; +async fn list_prs( + owner: &str, + repo: &str, + max: u32, + use_gh: bool, +) -> Result, String> { + let prs: Vec = fetch_all_pages(owner, repo, "pulls", "state=all", max, use_gh).await?; Ok(prs .into_iter() @@ -418,6 +546,15 @@ async fn read_commit( }) .unwrap_or_default(); + // GitHub login of the committer, rendered as an `@handle` so the + // entity extractor registers it as a `handle:` entity in the memory + // tree (unique committers become first-class entities). + let handle = commit + .author + .as_ref() + .map(|u| format!("@{}", u.login)) + .unwrap_or_default(); + let date = commit .commit .committer @@ -433,10 +570,16 @@ async fn read_commit( .unwrap_or("") .to_string(); + let author_line = if handle.is_empty() { + author.clone() + } else { + format!("{author} ({handle})") + }; + let body = format!( "# Commit: {title}\n\n\ **SHA:** {sha}\n\ - **Author:** {author}\n\ + **Author:** {author_line}\n\ **Date:** {date}\n\n\ ## Message\n\n\ {}", @@ -453,6 +596,7 @@ async fn read_commit( "repo": repo, "sha": sha, "author": author, + "author_handle": commit.author.as_ref().map(|u| u.login.clone()), }), }) } @@ -479,10 +623,16 @@ async fn read_issue( // Fetch comments let comments = fetch_issue_comments(owner, repo, number, use_gh).await; + // Unique participants (author + commenters), rendered as `@handle`s so + // each becomes a `handle:` entity in the memory tree. + let participants = + unique_handles(std::iter::once(author).chain(comments.iter().map(|c| c.user.as_str()))); + let mut body = format!( "# Issue #{number}: {title}\n\n\ **State:** {state}\n\ - **Author:** {author}\n\ + **Author:** @{author}\n\ + **Participants:** {participants}\n\ **Labels:** {label_str}\n\ **Created:** {created}\n\ **Updated:** {updated}\n\n\ @@ -503,7 +653,7 @@ async fn read_issue( body.push_str("\n\n## Comments\n"); for comment in &comments { body.push_str(&format!( - "\n### {} ({})\n\n{}\n", + "\n### @{} ({})\n\n{}\n", comment.user, comment.created_at, comment.body )); } @@ -550,10 +700,16 @@ async fn read_pr( // Fetch review comments let comments = fetch_issue_comments(owner, repo, number, use_gh).await; + // Unique participants (author + commenters), rendered as `@handle`s so + // each becomes a `handle:` entity in the memory tree. + let participants = + unique_handles(std::iter::once(author).chain(comments.iter().map(|c| c.user.as_str()))); + let mut body = format!( "# PR #{number}: {title}\n\n\ **State:** {state} ({merged})\n\ - **Author:** {author}\n\ + **Author:** @{author}\n\ + **Participants:** {participants}\n\ **Labels:** {label_str}\n\ **Created:** {created}\n\ **Updated:** {updated}\n\n\ @@ -575,7 +731,7 @@ async fn read_pr( body.push_str("\n\n## Comments\n"); for comment in &comments { body.push_str(&format!( - "\n### {} ({})\n\n{}\n", + "\n### @{} ({})\n\n{}\n", comment.user, comment.created_at, comment.body )); } @@ -652,6 +808,29 @@ fn parse_iso_ts(s: &str) -> Option { .map(|dt| dt.timestamp_millis()) } +/// Render GitHub logins as a deduped, order-preserving, space-separated +/// list of `@handle`s. Empty / `unknown` logins are skipped; an empty +/// result renders as `none`. Used so unique committers/commenters surface +/// as `handle:` entities in the memory tree. +fn unique_handles<'a>(logins: impl Iterator) -> String { + let mut seen = std::collections::HashSet::new(); + let mut out: Vec = Vec::new(); + for login in logins { + let l = login.trim(); + if l.is_empty() || l == "unknown" { + continue; + } + if seen.insert(l.to_string()) { + out.push(format!("@{l}")); + } + } + if out.is_empty() { + "none".to_string() + } else { + out.join(" ") + } +} + #[cfg(test)] mod tests { use super::*; @@ -699,4 +878,53 @@ mod tests { assert!(ItemKind::from_id("unknown:123").is_none()); assert!(ItemKind::from_id("noprefix").is_none()); } + + #[test] + fn repo_archive_source_id_slugs_to_repo_folder() { + // `github.com//` → slugify → `github-com--`. + assert_eq!( + repo_archive_source_id("https://github.com/tinyhumansai/openhuman").as_deref(), + Some("github.com/tinyhumansai/openhuman") + ); + assert!(repo_archive_source_id("not-a-url").is_none()); + } + + #[test] + fn chunk_source_id_is_clean_and_per_item() { + assert_eq!( + chunk_source_id("https://github.com/org/repo", "commit:abc123").as_deref(), + Some("github:org/repo:commit:abc123") + ); + assert_eq!( + chunk_source_id("https://github.com/org/repo", "pr:42").as_deref(), + Some("github:org/repo:pr:42") + ); + } + + #[test] + fn unique_handles_dedups_and_skips_unknown() { + assert_eq!( + unique_handles(["alice", "bob", "alice", "unknown", ""].into_iter()), + "@alice @bob" + ); + assert_eq!(unique_handles(["unknown", ""].into_iter()), "none"); + assert_eq!(unique_handles(std::iter::empty()), "none"); + } + + #[test] + fn raw_archive_coords_maps_kind_and_uid() { + assert_eq!( + raw_archive_coords("commit:deadbeef"), + Some((RawKind::Commit, "deadbeef".to_string())) + ); + assert_eq!( + raw_archive_coords("issue:7"), + Some((RawKind::Issue, "7".to_string())) + ); + assert_eq!( + raw_archive_coords("pr:99"), + Some((RawKind::PullRequest, "99".to_string())) + ); + assert!(raw_archive_coords("bogus:1").is_none()); + } } diff --git a/src/openhuman/memory_sources/readers/twitter.rs b/src/openhuman/memory_sources/readers/twitter.rs index 98ed5c34d..469ab8170 100644 --- a/src/openhuman/memory_sources/readers/twitter.rs +++ b/src/openhuman/memory_sources/readers/twitter.rs @@ -87,6 +87,9 @@ mod tests { query: Some("AI safety".into()), since_days: Some(3), max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/src/openhuman/memory_sources/registry.rs b/src/openhuman/memory_sources/registry.rs index 97f676ced..7929d22ca 100644 --- a/src/openhuman/memory_sources/registry.rs +++ b/src/openhuman/memory_sources/registry.rs @@ -176,6 +176,9 @@ pub async fn upsert_composio_source( url: None, branch: None, paths: Vec::new(), + max_commits: None, + max_issues: None, + max_prs: None, query: None, since_days: None, max_items: None, diff --git a/src/openhuman/memory_sources/rpc.rs b/src/openhuman/memory_sources/rpc.rs index 6871dc7df..eb8cfbfd4 100644 --- a/src/openhuman/memory_sources/rpc.rs +++ b/src/openhuman/memory_sources/rpc.rs @@ -68,6 +68,12 @@ pub struct AddRequest { #[serde(default)] pub paths: Vec, #[serde(default)] + pub max_commits: Option, + #[serde(default)] + pub max_issues: Option, + #[serde(default)] + pub max_prs: Option, + #[serde(default)] pub query: Option, #[serde(default)] pub since_days: Option, @@ -105,6 +111,12 @@ pub async fn add_rpc(req: AddRequest) -> Result, String> url: req.url, branch: req.branch, paths: req.paths, + // Per-type GitHub sync caps default to DEFAULT_GITHUB_ITEM_LIMIT + // (2000) in the reader when None. Overrides are applied via the + // update/patch path (`MemorySourcePatch`). + max_commits: req.max_commits, + max_issues: req.max_issues, + max_prs: req.max_prs, query: req.query, since_days: req.since_days, max_items: req.max_items, diff --git a/src/openhuman/memory_sources/status.rs b/src/openhuman/memory_sources/status.rs index 39ef16eba..1f1f8cda5 100644 --- a/src/openhuman/memory_sources/status.rs +++ b/src/openhuman/memory_sources/status.rs @@ -175,6 +175,9 @@ mod tests { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, }; assert_eq!(source_id_prefix(&entry), "mem_src:src_abc:%"); diff --git a/src/openhuman/memory_sources/sync.rs b/src/openhuman/memory_sources/sync.rs index 881c8c749..c54d43ac2 100644 --- a/src/openhuman/memory_sources/sync.rs +++ b/src/openhuman/memory_sources/sync.rs @@ -11,10 +11,15 @@ //! events on the global bus and UI subscribers stream them per source id. use crate::openhuman::config::Config; -use crate::openhuman::memory::ingest_pipeline::ingest_document; +use crate::openhuman::memory::ingest_pipeline::{ingest_document, IngestResult}; use crate::openhuman::memory::sync::{emit_sync_stage, MemorySyncStage, MemorySyncTrigger}; use crate::openhuman::memory_sources::readers; -use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind}; +use crate::openhuman::memory_sources::readers::github; +use crate::openhuman::memory_sources::types::{ + MemorySourceEntry, SourceContent, SourceItem, SourceKind, +}; +use crate::openhuman::memory_store::chunks::store::{set_chunk_raw_refs, RawRef}; +use crate::openhuman::memory_store::content::raw::{self as raw_store, raw_rel_path, RawItem}; use crate::openhuman::memory_sync::canonicalize::document::DocumentInput; use crate::openhuman::memory_sync::composio::{self, SyncReason}; @@ -189,17 +194,43 @@ async fn sync_via_reader(source: &MemorySourceEntry, config: Config) -> Result/:`) instead of the opaque + // `mem_src:src_:…` form. Other reader kinds keep the + // generic composite id. + let composite_source_id = if source.kind == SourceKind::GithubRepo { + source + .url + .as_deref() + .and_then(|url| github::chunk_source_id(url, &item.id)) + .unwrap_or_else(|| format!("mem_src:{}:{}", source.id, item.id)) + } else { + format!("mem_src:{}:{}", source.id, item.id) + }; + let mut tags = vec![ "memory_sources".to_string(), source.kind.as_str().to_string(), ]; + // Prioritise GitHub commit messages and closed/merged issues & PRs + // when building the memory tree — the scorer boosts PRIORITY_TAG + // chunks (see `memory_tree::score`). + if source.kind == SourceKind::GithubRepo && github_item_is_high_priority(&item.id, &content) + { + tags.push(crate::openhuman::memory_tree::score::PRIORITY_TAG.to_string()); + } match ingest_document(&config, &composite_source_id, "user", tags, doc).await { Ok(result) => { if !result.already_ingested { ingested += 1; } + // Mirror GitHub items into a browsable, repo-grouped raw + // archive (`raw/github-com--/{commits,issues,prs}/`) + // and point the chunks at it. Best-effort: archiving never + // fails the sync. + if source.kind == SourceKind::GithubRepo { + archive_github_raw(&config, source, item, &content, &result); + } } Err(e) => { tracing::warn!( @@ -224,3 +255,148 @@ async fn sync_via_reader(source: &MemorySourceEntry, config: Config) -> Result bool { + if item_id.starts_with("commit:") { + return true; + } + let state_closed = content + .metadata + .get("state") + .and_then(|v| v.as_str()) + .map(|s| s.eq_ignore_ascii_case("closed")) + .unwrap_or(false); + let merged = content + .metadata + .get("merged") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + state_closed || merged +} + +/// Mirror a freshly-ingested GitHub item into the on-disk raw archive and +/// tag its chunks with a `RawRef` pointing at the archive file. +/// +/// Layout: `raw/github-com--/{commits,issues,prs}/_.md`, +/// where `` is the commit SHA / issue number / PR number. The body +/// written is the reader's already-rendered markdown (`content.body`), which +/// carries the commit message / issue conversation + metadata / PR body + +/// metadata the memory tree is built from. +/// +/// Best-effort: any failure here is logged and swallowed — the chunked +/// content has already been persisted by `ingest_document`. +fn archive_github_raw( + config: &Config, + source: &MemorySourceEntry, + item: &SourceItem, + content: &SourceContent, + result: &IngestResult, +) { + let Some(url) = source.url.as_deref() else { + return; + }; + let Some(raw_source_id) = github::repo_archive_source_id(url) else { + tracing::warn!( + source_id = %source.id, + "[memory_sources:github] could not derive raw archive id from url" + ); + return; + }; + let Some((kind, uid)) = github::raw_archive_coords(&item.id) else { + return; + }; + let created_at_ms = item.updated_at_ms.unwrap_or(0); + let content_root = config.memory_tree_content_root(); + + let raw_item = RawItem { + uid: &uid, + created_at_ms, + markdown: &content.body, + kind, + }; + if let Err(e) = raw_store::write_raw_items( + &content_root, + &raw_source_id, + std::slice::from_ref(&raw_item), + ) { + tracing::warn!( + item_id = %item.id, + error = %e, + "[memory_sources:github] raw archive write failed" + ); + return; + } + + let rel_path = raw_rel_path(&raw_source_id, kind, created_at_ms, &uid); + let refs = vec![RawRef { + path: rel_path, + start: 0, + end: None, + }]; + for chunk_id in &result.chunk_ids { + if let Err(e) = set_chunk_raw_refs(config, chunk_id, &refs) { + tracing::warn!( + chunk_id = %chunk_id, + error = %format!("{e:#}"), + "[memory_sources:github] set raw ref failed" + ); + } + } + tracing::debug!( + item_id = %item.id, + archive = %raw_source_id, + chunks = result.chunk_ids.len(), + "[memory_sources:github] archived raw item" + ); +} + +#[cfg(test)] +mod tests { + use super::github_item_is_high_priority; + use crate::openhuman::memory_sources::types::{ContentType, SourceContent}; + + fn content_with(meta: serde_json::Value) -> SourceContent { + SourceContent { + id: "x".into(), + title: "t".into(), + body: "b".into(), + content_type: ContentType::Markdown, + metadata: meta, + } + } + + #[test] + fn commits_are_always_high_priority() { + let c = content_with(serde_json::json!({})); + assert!(github_item_is_high_priority("commit:abc123", &c)); + } + + #[test] + fn closed_issue_is_high_priority_open_is_not() { + let closed = content_with(serde_json::json!({ "state": "closed" })); + assert!(github_item_is_high_priority("issue:1", &closed)); + let open = content_with(serde_json::json!({ "state": "open" })); + assert!(!github_item_is_high_priority("issue:1", &open)); + } + + #[test] + fn merged_pr_is_high_priority_even_when_open_state() { + let merged = content_with(serde_json::json!({ "state": "open", "merged": true })); + assert!(github_item_is_high_priority("pr:7", &merged)); + let unmerged = content_with(serde_json::json!({ "state": "open", "merged": false })); + assert!(!github_item_is_high_priority("pr:7", &unmerged)); + } + + #[test] + fn missing_metadata_defaults_to_low_priority() { + let c = content_with(serde_json::json!({})); + assert!(!github_item_is_high_priority("issue:9", &c)); + } +} diff --git a/src/openhuman/memory_sources/types.rs b/src/openhuman/memory_sources/types.rs index 82c067709..f23364041 100644 --- a/src/openhuman/memory_sources/types.rs +++ b/src/openhuman/memory_sources/types.rs @@ -65,6 +65,15 @@ pub struct MemorySourceEntry { pub branch: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub paths: Vec, + /// Max commits to pull per sync (default 2000 when absent). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_commits: Option, + /// Max issues to pull per sync (default 2000 when absent). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_issues: Option, + /// Max pull requests to pull per sync (default 2000 when absent). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_prs: Option, // ── TwitterQuery ── #[serde(default, skip_serializing_if = "Option::is_none")] @@ -246,6 +255,9 @@ mod tests { url: None, branch: None, paths: Vec::new(), + max_commits: None, + max_issues: None, + max_prs: None, query: None, since_days: None, max_items: None, diff --git a/src/openhuman/memory_store/content/raw.rs b/src/openhuman/memory_store/content/raw.rs index caa842ccc..c135dd59a 100644 --- a/src/openhuman/memory_store/content/raw.rs +++ b/src/openhuman/memory_store/content/raw.rs @@ -52,6 +52,12 @@ pub enum RawKind { Contact, /// Long-form posts — LinkedIn posts, tweets, blog entries. Post, + /// Git commits (one file per commit) — GitHub repo sources. + Commit, + /// Issues with their conversation + metadata — GitHub repo sources. + Issue, + /// Pull requests with their body + metadata — GitHub repo sources. + PullRequest, } impl RawKind { @@ -64,6 +70,9 @@ impl RawKind { Self::Document => "documents", Self::Contact => "contacts", Self::Post => "posts", + Self::Commit => "commits", + Self::Issue => "issues", + Self::PullRequest => "prs", } } } @@ -438,4 +447,29 @@ mod tests { "raw/slack-team/chats/42_msg-with-bad.md" ); } + + #[test] + fn github_raw_kinds_use_repo_grouped_subdirs() { + assert_eq!(RawKind::Commit.as_dir(), "commits"); + assert_eq!(RawKind::Issue.as_dir(), "issues"); + assert_eq!(RawKind::PullRequest.as_dir(), "prs"); + // `github.com//` slugifies to `github-com--`. + assert_eq!( + raw_rel_path( + "github.com/tinyhumansai/openhuman", + RawKind::Commit, + 1_700_000_000_000, + "2a958e87" + ), + "raw/github-com-tinyhumansai-openhuman/commits/1700000000000_2a958e87.md" + ); + assert_eq!( + raw_rel_path("github.com/org/repo", RawKind::Issue, 0, "42"), + "raw/github-com-org-repo/issues/0_42.md" + ); + assert_eq!( + raw_rel_path("github.com/org/repo", RawKind::PullRequest, 0, "99"), + "raw/github-com-org-repo/prs/0_99.md" + ); + } } diff --git a/src/openhuman/memory_tree/score/mod.rs b/src/openhuman/memory_tree/score/mod.rs index 0b8895a8e..00525db52 100644 --- a/src/openhuman/memory_tree/score/mod.rs +++ b/src/openhuman/memory_tree/score/mod.rs @@ -40,6 +40,17 @@ pub const DEFAULT_DEFINITE_KEEP: f32 = 0.85; /// without consulting the LLM extractor. Catches obvious noise cheaply. pub const DEFAULT_DEFINITE_DROP: f32 = 0.15; +/// Metadata tag marking a chunk as high-priority source material. The +/// ingest path stamps this on GitHub commit messages and closed/merged +/// issues & PRs (see `memory_sources::sync`), which are higher-signal than +/// ambient activity and should be preferred when building the memory tree. +pub const PRIORITY_TAG: &str = "priority_high"; + +/// Additive score nudge applied to chunks carrying [`PRIORITY_TAG`]. Pushes +/// them above the drop threshold and higher in retrieval / summary ordering +/// without fully overriding the content signals. Clamped to `1.0`. +pub const PRIORITY_BOOST: f32 = 0.25; + /// Whole outcome of [`score_chunk`]. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ScoreResult { @@ -236,17 +247,38 @@ pub async fn score_chunk(chunk: &Chunk, cfg: &ScoringConfig) -> Result total { + log::debug!( + "[memory::score] priority boost chunk_id={} {:.3} -> {:.3}", + chunk.id, + total, + boosted + ); + total = boosted; + } + } + // 5. Admission gate. Source and interaction priors are deliberately // non-zero, so guard against very short entity-free chatter being kept by - // metadata alone. - let tiny_entity_free = - scoring_token_count < self::signals::token_count::TOKEN_MIN && extracted.is_empty(); + // metadata alone. Priority-tagged chunks bypass the tiny/entity-free + // drop since the tag itself is a strong relevance signal. + let tiny_entity_free = !priority + && scoring_token_count < self::signals::token_count::TOKEN_MIN + && extracted.is_empty(); let kept = !tiny_entity_free && total >= cfg.drop_threshold; let drop_reason = if kept { None diff --git a/src/openhuman/memory_tree/summarise.rs b/src/openhuman/memory_tree/summarise.rs index 11a0657c9..4ccf38367 100644 --- a/src/openhuman/memory_tree/summarise.rs +++ b/src/openhuman/memory_tree/summarise.rs @@ -169,8 +169,21 @@ pub fn fallback_summary(inputs: &[SummaryInput], budget: u32) -> SummaryOutput { } fn build_user_prompt(inputs: &[SummaryInput], per_input_cap_tokens: u32) -> String { + // Higher-priority inputs (by score) lead the prompt so the most + // important source material — e.g. commit messages and closed/merged + // issues & PRs, which carry a priority boost at ingest — is summarised + // first and is least likely to be truncated under budget pressure. + // `sort_by` is stable, so chronological order is preserved among + // equal-score inputs. + let mut order: Vec<&SummaryInput> = inputs.iter().collect(); + order.sort_by(|a, b| { + b.score + .partial_cmp(&a.score) + .unwrap_or(std::cmp::Ordering::Equal) + }); + let mut out = String::new(); - for inp in inputs { + for inp in order { let trimmed = inp.content.trim(); if trimmed.is_empty() { continue; diff --git a/tests/app_credentials_threads_memory_sources_raw_coverage_e2e.rs b/tests/app_credentials_threads_memory_sources_raw_coverage_e2e.rs index bb0b23551..b8b222187 100644 --- a/tests/app_credentials_threads_memory_sources_raw_coverage_e2e.rs +++ b/tests/app_credentials_threads_memory_sources_raw_coverage_e2e.rs @@ -186,6 +186,9 @@ fn source_entry(id: &str, kind: SourceKind, label: &str) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/tests/memory_raw_coverage_e2e.rs b/tests/memory_raw_coverage_e2e.rs index 118554517..6d4352ccd 100644 --- a/tests/memory_raw_coverage_e2e.rs +++ b/tests/memory_raw_coverage_e2e.rs @@ -67,6 +67,9 @@ fn source_entry(kind: SourceKind, id: &str) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/tests/memory_sources_closure_round23_raw_coverage_e2e.rs b/tests/memory_sources_closure_round23_raw_coverage_e2e.rs index 780eccf0c..c808b6c0c 100644 --- a/tests/memory_sources_closure_round23_raw_coverage_e2e.rs +++ b/tests/memory_sources_closure_round23_raw_coverage_e2e.rs @@ -131,6 +131,9 @@ fn source_entry(id: &str, kind: SourceKind) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/tests/memory_sources_readers_round21_raw_coverage_e2e.rs b/tests/memory_sources_readers_round21_raw_coverage_e2e.rs index b7fb6211f..283005354 100644 --- a/tests/memory_sources_readers_round21_raw_coverage_e2e.rs +++ b/tests/memory_sources_readers_round21_raw_coverage_e2e.rs @@ -69,6 +69,9 @@ fn source_entry(id: &str, kind: SourceKind) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/tests/memory_sync_sources_raw_coverage_e2e.rs b/tests/memory_sync_sources_raw_coverage_e2e.rs index 5d5c54a50..734cd59e8 100644 --- a/tests/memory_sync_sources_raw_coverage_e2e.rs +++ b/tests/memory_sync_sources_raw_coverage_e2e.rs @@ -112,6 +112,9 @@ fn source(kind: SourceKind, id: &str) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } diff --git a/tests/memory_threads_raw_coverage_e2e.rs b/tests/memory_threads_raw_coverage_e2e.rs index 3db3b7dc6..35c8b6e6e 100644 --- a/tests/memory_threads_raw_coverage_e2e.rs +++ b/tests/memory_threads_raw_coverage_e2e.rs @@ -217,6 +217,9 @@ fn source(kind: SourceKind, id: &str) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } } @@ -3832,6 +3835,9 @@ async fn memory_sources_registry_rpc_and_schema_handlers_cover_crud_edges() { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, }) .await @@ -3852,6 +3858,9 @@ async fn memory_sources_registry_rpc_and_schema_handlers_cover_crud_edges() { query: None, since_days: None, max_items: Some(4), + max_commits: None, + max_issues: None, + max_prs: None, selector: None, }) .await @@ -3873,6 +3882,9 @@ async fn memory_sources_registry_rpc_and_schema_handlers_cover_crud_edges() { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, }) .await diff --git a/tests/near90_closure_raw_coverage_e2e.rs b/tests/near90_closure_raw_coverage_e2e.rs index a17bb8647..b0f3a4a7b 100644 --- a/tests/near90_closure_raw_coverage_e2e.rs +++ b/tests/near90_closure_raw_coverage_e2e.rs @@ -193,6 +193,9 @@ fn source_entry(id: &str, kind: SourceKind) -> MemorySourceEntry { query: None, since_days: None, max_items: None, + max_commits: None, + max_issues: None, + max_prs: None, selector: None, } }