admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(memory): repo-grouped GitHub raw archive + entities & priority (#3047)

Browse Source
This commit is contained in:
Steven Enamakel
2026-05-30 21:35:45 -07:00
committed by GitHub
parent daa0a13f2e
commit 850d275841
19 changed files with 602 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/memory_sources/readers/composio.rs
+3
View File
@@ -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,
}
}
src/openhuman/memory_sources/readers/folder.rs
+3
View File
@@ -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,
}
}
src/openhuman/memory_sources/readers/github.rs
+269 -41
View File
@@ -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/<owner>/<repo>`, which slugifies (via
/// `slugify_source_id`) to `github-com-<owner>-<repo>` so a source's
/// commits/issues/PRs land under
/// `raw/github-com-<owner>-<repo>/{commits,issues,prs}/`.
pub(crate) fn repo_archive_source_id(url: &str) -> Option<String> {
let (owner, repo) = parse_github_url(url).ok()?;
Some(format!("github.com/{owner}/{repo}"))
}
/// Chunk-store source id for a single repo item.
///
/// `github:<owner>/<repo>:<item_id>` keeps the per-item dedup key (each
/// commit/issue/PR is an immutable document) while producing a clean
/// `chunks/document/github-<owner>-<repo>-…` scope instead of the opaque
/// `mem_src:src_<uuid>:…` form.
pub(crate) fn chunk_source_id(url: &str, item_id: &str) -> Option<String> {
let (owner, repo) = parse_github_url(url).ok()?;
Some(format!("github:{owner}/{repo}:{item_id}"))
}
/// Map a [`SourceItem`] id (`commit:<sha>`, `issue:<n>`, `pr:<n>`) 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<String, String> {
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<GhUser>,
}
#[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<String, String> {
// ── List helpers ────────────────────────────────────────────────────
async fn list_commits(owner: &str, repo: &str, use_gh: bool) -> Result<Vec<SourceItem>, 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<T: serde::de::DeserializeOwned>(
owner: &str,
repo: &str,
resource: &str,
extra_query: &str,
max: u32,
use_gh: bool,
) -> Result<Vec<T>, String> {
let mut out: Vec<T> = Vec::new();
let mut page = 1u32;
// Try parsing as array of commit objects
let commits: Vec<GhCommit> =
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<T> = 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<Vec<SourceItem>, String> {
let commits: Vec<GhCommit> = 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<Vec<Sourc
.collect())
}
async fn list_issues(owner: &str, repo: &str, use_gh: bool) -> Result<Vec<SourceItem>, 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<Vec<SourceItem>, 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<SourceItem> = Vec::new();
let mut page = 1u32;
let issues: Vec<GhIssue> =
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<GhIssue> = 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<Vec<SourceItem>, String> {
let json_str = fetch_github(
&format!("repos/{owner}/{repo}/pulls?per_page=30&state=all"),
use_gh,
)
.await?;
let prs: Vec<GhPr> = 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<Vec<SourceItem>, String> {
let prs: Vec<GhPr> = 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<i64> {
.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<Item = &'a str>) -> String {
let mut seen = std::collections::HashSet::new();
let mut out: Vec<String> = 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/<owner>/<repo>` → slugify → `github-com-<owner>-<repo>`.
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());
}
}
src/openhuman/memory_sources/readers/twitter.rs
+3
View File
@@ -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,
}
}
src/openhuman/memory_sources/registry.rs
+3
View File
@@ -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,
src/openhuman/memory_sources/rpc.rs
+12
View File
@@ -68,6 +68,12 @@ pub struct AddRequest {
#[serde(default)]
pub paths: Vec<String>,
#[serde(default)]
pub max_commits: Option<u32>,
#[serde(default)]
pub max_issues: Option<u32>,
#[serde(default)]
pub max_prs: Option<u32>,
#[serde(default)]
pub query: Option<String>,
#[serde(default)]
pub since_days: Option<u32>,
@@ -105,6 +111,12 @@ pub async fn add_rpc(req: AddRequest) -> Result<RpcOutcome<AddResponse>, 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,
src/openhuman/memory_sources/status.rs
+3
View File
@@ -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:%");
src/openhuman/memory_sources/sync.rs
+180 -4
View File
@@ -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<u
source_ref: Some(format!("{}:{}", source.id, item.id)),
};
let composite_source_id = format!("mem_src:{}:{}", source.id, item.id);
let tags = vec![
// GitHub items use a clean, repo-scoped chunk source id
// (`github:<owner>/<repo>:<item_id>`) instead of the opaque
// `mem_src:src_<uuid>:…` 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-<owner>-<repo>/{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<u
Ok(ingested)
}
/// Whether a GitHub item should be treated as high-priority source
/// material when building the memory tree.
///
/// Commit messages are always high-priority; issues and PRs are
/// high-priority once **closed** (issues/PRs) or **merged** (PRs) — a
/// resolved thread carries the decision/outcome, which is the part worth
/// remembering. Open items stay at the default priority.
fn github_item_is_high_priority(item_id: &str, content: &SourceContent) -> 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-<owner>-<repo>/{commits,issues,prs}/<ts>_<uid>.md`,
/// where `<uid>` 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));
}
}
src/openhuman/memory_sources/types.rs
+12
View File
@@ -65,6 +65,15 @@ pub struct MemorySourceEntry {
pub branch: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub paths: Vec<String>,
/// Max commits to pull per sync (default 2000 when absent).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_commits: Option<u32>,
/// Max issues to pull per sync (default 2000 when absent).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_issues: Option<u32>,
/// Max pull requests to pull per sync (default 2000 when absent).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_prs: Option<u32>,
// ── 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,
src/openhuman/memory_store/content/raw.rs
+34
View File
@@ -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/<owner>/<repo>` slugifies to `github-com-<owner>-<repo>`.
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"
);
}
}
src/openhuman/memory_tree/score/mod.rs
+36 -4
View File
@@ -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<ScoreResu
// dragging the total down. Fall back to `combine_cheap_only` which
// excludes that term from both numerator and denominator, so the cheap
// signals alone produce the total.
let total = if llm_consulted {
let mut total = if llm_consulted {
self::signals::combine(&signals, &cfg.weights)
} else {
self::signals::combine_cheap_only(&signals, &cfg.weights)
};
// 4b. Priority boost. Chunks tagged PRIORITY_TAG at ingest (GitHub
// commit messages, closed/merged issues & PRs) are higher-signal than
// ambient activity. Nudge their total up (clamped) so they clear the
// admission gate more readily and rank higher when the memory tree is
// built from chunk scores.
let priority = chunk.metadata.tags.iter().any(|t| t == PRIORITY_TAG);
if priority {
let boosted = (total + PRIORITY_BOOST).min(1.0);
if boosted > 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
src/openhuman/memory_tree/summarise.rs
+14 -1
View File
@@ -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;
tests/app_credentials_threads_memory_sources_raw_coverage_e2e.rs
+3
View File
@@ -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,
}
}
tests/memory_raw_coverage_e2e.rs
+3
View File
@@ -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,
}
}
tests/memory_sources_closure_round23_raw_coverage_e2e.rs
+3
View File
@@ -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,
}
}
tests/memory_sources_readers_round21_raw_coverage_e2e.rs
+3
View File
@@ -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,
}
}
tests/memory_sync_sources_raw_coverage_e2e.rs
+3
View File
@@ -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,
}
}
tests/memory_threads_raw_coverage_e2e.rs
+12
View File
@@ -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
tests/near90_closure_raw_coverage_e2e.rs
+3
View File
@@ -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,
}
}