GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Repo activity: a repo the anonymous read cannot reach is Unseen and the others still read; the read says only what it knows

Browse source 
A feed that answers 404 (private) or a redirect (renamed) is a fact about that repo, unseen, on first reads and later ones; the redirect is not followed and, when it names a repo, the Mind is told where it points. A network failure or the budget still fails the read whole. The fake forge now answers statuses and Location as the live forge does.

more:bool becomes capped: the repos whose feed the read stopped at its page limit, and the notice and the wear line say only that: the read stopped at its page limit in a named repo, older entries there were not read.

The cut notice reserves its tail from the entry count's digits and the capped clause, so its 300-char bound holds at any count.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-03 03:35:43 +01:00
commit dcd8f2116b
8 changed files with 888 additions and 635 deletions
Download patch file Download diff file Expand all files Collapse all files

211
crates/huginn-daemon/src/sources/forge.rs
View file

@ -7,7 +7,7 @@ use std::time::{Duration, Instant};
use anyhow::{Context, Result, ensure};
use chrono::{DateTime, Utc};
use huginn_mind::eureka_pipeline::{OrgRepo, Short, Slug};
use huginn_mind::{OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent};
use huginn_mind::{OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved};
use serde_json::Value;
use super::{Catalog, Forge, SourceRepo};
@ -100,42 +100,66 @@ impl Forge for HttpForge {
/// The whole of one activity read, from its first call to its last. A call is given what is left of it, so a forge that
/// answers slowly, and not only one that never answers, cannot hold the daemon's one thread longer than this.
const ACTIVITY_CALL_TIMEOUT: Duration = Duration::from_secs(2);
/// How many pages of one repo's feed a read takes: 200 entries, of anyone. A repo with more inside the read's reach sets `more`.
/// How many pages of one repo's feed a read takes: 200 entries, of anyone. A repo with more inside the read's reach is named in `capped`.
const REPO_FEED_PAGES: usize = 4;
/// How far back a read with no cursor looks.
const FIRST_WINDOW_MS: i64 = 72 * 3600 * 1000;
impl HttpForge {
/// One GET inside the read's budget, anonymous. An error names no address: the forge's answer is the reader's to know about,
/// and nothing of it reaches a refusal.
fn within(&self, path: &str, deadline: Instant) -> Result<Value> {
/// One GET inside the read's budget, anonymous, and no redirect followed: a redirect names a path on the public address, which the
/// loopback does not serve. An error names no address: the forge's answer is the reader's to know about, and nothing of it reaches
/// a refusal.
fn answer(&self, path: &str, deadline: Instant) -> Result<ureq::http::Response<ureq::Body>> {
let left = deadline.checked_duration_since(Instant::now()).filter(|left| !left.is_zero()).context("the activity read ran out of time")?;
let mut response = self
.agent
self.agent
.get(format!("{}{path}", self.base))
.config()
.max_redirects(0)
.max_redirects_will_error(false)
.timeout_global(Some(left))
.build()
.call()
.map_err(|_| anyhow::anyhow!("the forge did not answer"))?;
ensure!(response.status().as_u16() == 200, "the forge refused a read");
let body = response.body_mut().read_to_string().map_err(|_| anyhow::anyhow!("the forge's answer was cut off"))?;
serde_json::from_str(&body).map_err(|_| anyhow::anyhow!("the forge's answer is not JSON"))
.map_err(|_| anyhow::anyhow!("the forge did not answer"))
}
/// A GET that must answer 200 with JSON, or the read fails.
fn within(&self, path: &str, deadline: Instant) -> Result<Value> {
json_of(self.answer(path, deadline)?)?.context("the forge refused a read")
}
/// One page of one repo's feed. A repo the anonymous read cannot reach answers 404 (private) or a redirect (renamed): that is a fact
/// about the repo, `Dark`, and the read goes on. Any other refusal fails the read.
fn feed(&self, repo: &OrgRepo, page: usize, deadline: Instant) -> Result<Feed> {
let response = self.answer(&format!("/api/v1/repos/{}/activities/feeds?limit={PAGE}&page={page}", repo.0), deadline)?;
let status = response.status().as_u16();
if status == 404 || (300..400).contains(&status) {
let to = response.headers().get("location").and_then(|location| location.to_str().ok()).and_then(redirect_target).filter(|to| !to.0.eq_ignore_ascii_case(&repo.0));
return Ok(Feed::Dark(to));
}
Ok(Feed::Entries(json_of(response)?.context("the forge refused a read")?))
}
/// What others did in `repos` on the forge, and on a read with no cursor what is open and what the forge cannot show. Each repo's
/// own feed is read, newest first, down to the cursor, or on a read with no cursor down to 72 hours ago, so other repos' traffic
/// cannot crowd hers out. `more` is set when a repo's feed had more to give past the page cap. One read, one budget; the first
/// call that fails, or that the budget cannot reach, fails the read whole: an answer is complete or says `more`, never a silent
/// cannot crowd hers out. `capped` names a repo whose feed the read stopped at the page cap. A repo whose feed answers 404 or a redirect (private, renamed) is a fact about that repo, never a failure of the read: it is `unseen`, and a redirect that names a repo is `moved` and is not followed. A network failure or the budget still fails the read whole. One read, one budget; the first
/// call that fails, or that the budget cannot reach, fails the read whole: an answer is complete or names `capped`, never a silent
/// part. Only ids, instants, logins, repo names, numbers and counts leave: a title, a comment and a commit message are never read.
pub fn activity(&self, repos: &[OrgRepo], instance: &Slug, after: Option<u64>, now: DateTime<Utc>) -> Result<RepoActivity> {
let deadline = Instant::now() + ACTIVITY_CALL_TIMEOUT;
let window_start = if after.is_some() { i64::MIN } else { now.timestamp_millis() - FIRST_WINDOW_MS };
let (mut head, mut events, mut more) = (0_u64, Vec::new(), false);
let (mut head, mut events, mut capped, mut dark, mut moved) = (0_u64, Vec::new(), Vec::new(), Vec::new(), Vec::new());
for repo in repos {
let mut reached = false;
for page in 1..=REPO_FEED_PAGES {
let feed = self.within(&format!("/api/v1/repos/{}/activities/feeds?limit={PAGE}&page={page}", repo.0), deadline)?;
let feed = match self.feed(repo, page, deadline)? {
Feed::Entries(feed) => feed,
Feed::Dark(to) => {
dark.push(repo.clone());
moved.extend(to.map(|to| RepoMoved { from: repo.clone(), to }));
reached = true;
break;
}
};
let entries = feed.as_array().context("the forge's feed is not a list")?;
for entry in entries {
let Some(id) = entry["id"].as_u64() else { continue };
@ -152,12 +176,15 @@ impl HttpForge {
break;
}
}
more |= !reached;
if !reached {
capped.push(repo.clone());
}
}
events.sort_by(|a, b| b.id.cmp(&a.id));
// The cursor never moves backwards: a read that saw nothing new keeps the one it was given.
head = head.max(after.unwrap_or(0));
let (mut open_prs, mut unseen) = (Vec::new(), Vec::new());
let moved = if after.is_none() { moved } else { Vec::new() };
if after.is_none() {
// The open PRs and the org listing are paged to the short page, never to a cap: the budget ends a read that runs long, and
// ends it whole, so each list is complete or the read fails. A cap would say "nothing else is open" past it, falsely.
@ -178,12 +205,36 @@ impl HttpForge {
break;
}
}
unseen = repos.iter().filter(|repo| !listed.contains(&repo.identity())).cloned().collect();
unseen = repos.iter().filter(|repo| !listed.contains(&repo.identity()) || dark.contains(repo)).cloned().collect();
}
Ok(RepoActivity { head, events, more, open_prs, unseen })
Ok(RepoActivity { head, events, capped, open_prs, unseen, moved })
}
}
/// What one page of a repo's feed answered.
enum Feed {
Entries(Value),
/// 404 or a redirect; the repo the redirect names, when it names one.
Dark(Option<OrgRepo>),
}
/// The body of a response as JSON when it answered 200, else `None`.
fn json_of(mut response: ureq::http::Response<ureq::Body>) -> Result<Option<Value>> {
if response.status().as_u16() != 200 {
return Ok(None);
}
let body = response.body_mut().read_to_string().map_err(|_| anyhow::anyhow!("the forge's answer was cut off"))?;
serde_json::from_str(&body).map(Some).map_err(|_| anyhow::anyhow!("the forge's answer is not JSON"))
}
/// The repo a redirect's `Location` names: the two segments after `/repos/`, owner and name, wherever the path sits and whatever follows.
fn redirect_target(location: &str) -> Option<OrgRepo> {
let rest = location.split_once("/repos/")?.1.split(['?', '#']).next()?;
let mut segments = rest.split('/');
let (owner, name) = (segments.next().filter(|owner| !owner.is_empty())?, segments.next().filter(|name| !name.is_empty())?);
Some(OrgRepo(format!("{owner}/{name}")))
}
fn instant_ms(text: &str) -> Option<i64> {
DateTime::parse_from_rfc3339(text).ok().map(|at| at.timestamp_millis())
}
@ -243,7 +294,7 @@ fn open_pr_of(pr: &Value, repos: &[OrgRepo], instance: &Slug) -> Option<OpenPr>
#[cfg(test)]
mod tests {
use huginn_mind::{PrAction, RepoChange};
use huginn_mind::{PrAction, RepoChange, RepoMoved};
use serde_json::json;
use super::*;
@ -325,9 +376,34 @@ mod tests {
assert!(HttpForge::new(&unstated).catalog().unwrap().shortfall.unwrap().contains("X-Total-Count"));
}
/// What a fake forge answers: a status, the `Location` header when there is one, and a body.
struct Reply {
status: u16,
location: Option<String>,
body: String,
}
impl From<String> for Reply {
fn from(body: String) -> Self {
Self { status: 200, location: None, body }
}
}
impl Reply {
/// What the live forge answers an anonymous reader for a private repo.
fn not_found() -> Self {
Self { status: 404, location: None, body: json!({ "message": "The target couldn't be found." }).to_string() }
}
/// What the live forge answers for a renamed repo: a 301 to its public address, `/forge` prefix and all.
fn moved_to(location: &str) -> Self {
Self { status: 301, location: Some(location.to_owned()), body: String::new() }
}
}
/// Answers each request by its path from `routes`; a path it has no answer for is held open and never answered, as a hung forge
/// does. Every path asked is recorded in order.
fn forge_of(routes: impl Fn(&str) -> Option<String> + Send + 'static) -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
fn forge_of<R: Into<Reply>>(routes: impl Fn(&str) -> Option<R> + Send + 'static) -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
use std::io::{Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
@ -342,9 +418,10 @@ mod tests {
let request = String::from_utf8_lossy(&buffer[..read]).into_owned();
let path = request.split_whitespace().nth(1).unwrap_or_default().to_owned();
record.lock().unwrap().push(path.clone());
match routes(&path) {
Some(body) => {
write!(stream, "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).unwrap();
match routes(&path).map(Into::into) {
Some(Reply { status, location, body }) => {
let location = location.map(|location| format!("location: {location}\r\n")).unwrap_or_default();
write!(stream, "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\n{location}content-length: {}\r\nconnection: close\r\n\r\n{body}", body.len()).unwrap();
}
None => held.push(stream),
}
@ -419,7 +496,7 @@ mod tests {
]
);
assert_eq!(read.events[0].at, at("2026-11-02T18:40:00Z").timestamp_millis() as u64);
assert!(read.open_prs.is_empty() && read.unseen.is_empty() && !read.more, "a cursor asks for neither list");
assert!(read.open_prs.is_empty() && read.unseen.is_empty() && read.capped.is_empty() && read.moved.is_empty(), "a cursor asks for neither list");
let wire = serde_json::to_string(&read).unwrap();
assert!(!wire.contains("CANARY") && !wire.contains("obey") && !wire.contains("Message"), "no forge text leaves the read: {wire}");
let asked = asked.lock().unwrap().clone();
@ -452,13 +529,13 @@ mod tests {
let (base, asked) = forge_of(feed("nibu"));
let far = HttpForge::new(&base).activity(&wanted, &me, Some(5), now).unwrap();
assert!(far.more, "stopped at the cap with the cursor not reached");
assert!(far.capped == repos(&["GameCult/Huginn"]), "stopped at the cap with the cursor not reached");
assert_eq!((far.events.len(), far.head), (200, 2500));
assert_eq!(asked.lock().unwrap().len(), 4, "four pages and no more");
let (base, asked) = forge_of(feed("nibu"));
let near = HttpForge::new(&base).activity(&wanted, &me, Some(2300), now).unwrap();
assert!(!near.more);
assert!(near.capped.is_empty());
assert_eq!((near.events.len(), near.head), (20, 2500), "ids above the cursor only");
assert_eq!(asked.lock().unwrap().len(), 1);
@ -481,7 +558,7 @@ mod tests {
let (base, asked) = forge_of(move |path| if path.contains("/orgs/GameCult/activities") { Some(busy.clone()) } else { repo_feed(&hers, path).or_else(|| Some("[]".into())) });
let read = HttpForge::new(&base).activity(&repos(&["GameCult/Ratatoskr"]), &me(), None, at(NOW)).unwrap();
assert_eq!(read.events.iter().map(|event| event.id).collect::<Vec<_>>(), [300]);
assert!(!read.more);
assert!(read.capped.is_empty());
assert!(asked.lock().unwrap().iter().all(|path| !path.contains("/orgs/GameCult/activities")));
}
@ -498,32 +575,32 @@ mod tests {
let wanted = repos(&["GameCult/Huginn"]);
let (base, _) = forge_of(move |path| if is_feed(path) { repo_feed(&entries, path) } else { Some("[]".into()) });
let busy = HttpForge::new(&base).activity(&wanted, &me(), None, at(NOW)).unwrap();
assert_eq!((busy.events.iter().map(|event| event.id).collect::<Vec<_>>(), busy.head, busy.more), (vec![300, 200], 300, false));
assert_eq!((busy.events.iter().map(|event| event.id).collect::<Vec<_>>(), busy.head, busy.capped.len()), (vec![300, 200], 300, 0));
let quiet_entries = vec![push_at(50, "GameCult/Huginn", "Nibu", 100), push_at(40, "GameCult/Huginn", "Nibu", 120)];
let (base, _) = forge_of(move |path| if is_feed(path) { repo_feed(&quiet_entries, path) } else { Some("[]".into()) });
let forge = HttpForge::new(&base);
let first = forge.activity(&wanted, &me(), None, at(NOW)).unwrap();
assert_eq!((first.events.len(), first.head, first.more), (0, 50, false), "nothing in the window, and a cursor all the same");
assert_eq!((first.events.len(), first.head, first.capped.len()), (0, 50, 0), "nothing in the window, and a cursor all the same");
let next = forge.activity(&wanted, &me(), Some(first.head), at(NOW)).unwrap();
assert_eq!((next.events.len(), next.head), (0, 50), "the history is not told as news");
}
/// A feed of exactly one full page goes on to the next page, and a short page ends it; a repo whose feed runs past the page cap
/// says `more` instead of ending quietly.
/// is named in `capped` instead of ending quietly.
#[test]
fn a_full_feed_page_asks_for_the_next_and_a_repo_past_the_cap_says_more() {
let wanted = repos(&["GameCult/Huginn"]);
let entries: Vec<Value> = (0..53).map(|n| push_at(1000 - n, "GameCult/Huginn", "Nibu", 5)).collect();
let (base, asked) = forge_of(move |path| if is_feed(path) { repo_feed(&entries, path) } else { Some("[]".into()) });
let read = HttpForge::new(&base).activity(&wanted, &me(), None, at(NOW)).unwrap();
assert_eq!((read.events.len(), read.more), (53, false));
assert_eq!((read.events.len(), read.capped.len()), (53, 0));
assert_eq!(asked.lock().unwrap().iter().filter(|path| is_feed(path)).count(), 2, "page 2 was asked, page 3 was not");
let entries: Vec<Value> = (0..400).map(|n| push_at(5000 - n, "GameCult/Huginn", "Nibu", 5)).collect();
let (base, asked) = forge_of(move |path| if is_feed(path) { repo_feed(&entries, path) } else { Some("[]".into()) });
let read = HttpForge::new(&base).activity(&wanted, &me(), None, at(NOW)).unwrap();
assert_eq!((read.events.len(), read.more), (REPO_FEED_PAGES * PAGE, true), "200 read and the rest disclosed");
assert_eq!((read.events.len(), read.capped), (REPO_FEED_PAGES * PAGE, repos(&["GameCult/Huginn"])), "200 read and the rest disclosed");
assert_eq!(asked.lock().unwrap().iter().filter(|path| is_feed(path)).count(), REPO_FEED_PAGES);
}
@ -540,12 +617,14 @@ mod tests {
let listing = vec![Value::Array(names[..PAGE].to_vec()).to_string(), Value::Array(names[PAGE..].to_vec()).to_string()];
let sample = sample();
let (base, asked) = forge_of(move |path| {
if is_feed(path) {
repo_feed(&sample, path)
if is_feed(path) && path.contains("/GameCult/Ghost/") {
Some(Reply::not_found())
} else if is_feed(path) {
repo_feed(&sample, path).map(Reply::from)
} else if path.contains("/issues/search") {
search.get(page_number(path)? - 1).cloned()
search.get(page_number(path)? - 1).cloned().map(Reply::from)
} else if path.contains("/orgs/GameCult/repos") {
listing.get(page_number(path)? - 1).cloned()
listing.get(page_number(path)? - 1).cloned().map(Reply::from)
} else {
None
}
@ -593,6 +672,64 @@ mod tests {
assert_eq!(asked.lock().unwrap().iter().filter(|path| path.contains("/issues/search")).count(), 4, "to the short page");
}
/// The forge as it answers an anonymous reader: public repos give their feed (`[]` when quiet), a private repo answers 404, a
/// renamed repo answers 301 to a path under `/forge` that this loopback does not serve (asking it gives 404).
fn live_forge(entries: Vec<Value>, public: &'static [&'static str], listing: String) -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
forge_of(move |path| {
if path.starts_with("/forge/") {
return Some(Reply::not_found());
}
if is_feed(path) {
let repo = path.strip_prefix("/api/v1/repos/")?.split("/activities").next()?;
if repo.eq_ignore_ascii_case("GameCult/cinny") {
return Some(Reply::moved_to(&format!("/forge/api/v1/repos/GameCult/Ratatoskr/activities/feeds?limit={PAGE}&page=1")));
}
return if public.iter().any(|name| name.eq_ignore_ascii_case(repo)) { repo_feed(&entries, path).map(Reply::from) } else { Some(Reply::not_found()) };
}
if path.contains("/issues/search") {
return Some("[]".to_owned().into());
}
path.contains("/orgs/GameCult/repos").then(|| listing.clone().into())
})
}
/// A repo the anonymous read cannot reach is a fact about that repo, not the end of the read: a private repo (404) and a renamed one
/// (301, whose target is not followed) are `unseen` and the others still read, on a first read and on a later one; the redirect's
/// target is told; a network failure still fails the read whole.
#[test]
fn a_repo_the_anonymous_read_cannot_reach_is_unseen_and_the_others_still_read() {
let entries = vec![push_at(300, "GameCult/Huginn", "Nibu", 10), push_at(200, "GameCult/Ratatoskr", "void", 11)];
let listing = json!([{ "full_name": "GameCult/Huginn" }, { "full_name": "GameCult/Ratatoskr" }]).to_string();
let (base, asked) = live_forge(entries, &["GameCult/Huginn", "GameCult/Ratatoskr"], listing);
let wanted = repos(&["GameCult/gamecult-ops", "GameCult/Huginn", "GameCult/cinny", "GameCult/Ratatoskr"]);
let forge = HttpForge::new(&base);
let first = forge.activity(&wanted, &me(), None, at(NOW)).unwrap();
assert_eq!(first.events.iter().map(|event| event.id).collect::<Vec<_>>(), [300, 200], "the readable repos were read");
assert_eq!(first.unseen, repos(&["GameCult/gamecult-ops", "GameCult/cinny"]));
assert_eq!(first.moved, [RepoMoved { from: "GameCult/cinny".into(), to: "GameCult/Ratatoskr".into() }]);
assert!(first.capped.is_empty());
assert!(asked.lock().unwrap().iter().all(|path| !path.starts_with("/forge/")), "a redirect is not followed: {:?}", asked.lock().unwrap());
let later = forge.activity(&wanted, &me(), Some(250), at(NOW)).unwrap();
assert_eq!((later.events.iter().map(|event| event.id).collect::<Vec<_>>(), later.head), (vec![300], 300), "a later read goes on past the dark repos");
assert!(later.unseen.is_empty() && later.moved.is_empty(), "a cursor asks for neither list");
let (base, _) = forge_of(|_| None::<String>);
assert!(HttpForge::new(&base).activity(&wanted, &me(), None, at(NOW)).is_err(), "no answer at all still fails the read");
let (base, _) = forge_of(|path| is_feed(path).then(|| Reply { status: 500, location: None, body: String::new() }));
assert!(HttpForge::new(&base).activity(&wanted, &me(), None, at(NOW)).is_err(), "a server error is not a fact about a repo");
}
/// A redirect names its repo in its path wherever that sits, and a redirect that names none tells nothing.
#[test]
fn a_redirect_names_its_repo_only_when_it_does() {
let target = |text: &str| redirect_target(text).map(|repo| repo.0);
assert_eq!(target("/forge/api/v1/repos/GameCult/Ratatoskr"), Some("GameCult/Ratatoskr".into()));
assert_eq!(target("https://eureka.gamecult.org/forge/api/v1/repos/GameCult/Ratatoskr/activities/feeds?limit=50&page=1"), Some("GameCult/Ratatoskr".into()));
assert_eq!((target("/user/login"), target("/forge/api/v1/repos/GameCult"), target("/api/v1/repos//x")), (None, None, None));
}
/// Every PR op the forge writes becomes its event with the number in element 0, and no other op does; the forge's source writes
/// all eight the same way, though only four have been seen in its live feed.
#[test]
@ -625,7 +762,7 @@ mod tests {
/// stalls fails it too: the bound is on the whole read.
#[test]
fn a_hung_forge_fails_the_read_within_its_bound() {
let (base, _) = forge_of(|_| None);
let (base, _) = forge_of(|_| None::<String>);
let started = Instant::now();
let read = HttpForge::new(&base).activity(&repos(&["GameCult/Huginn"]), &me(), None, at("2026-11-02T23:00:00Z"));
assert!(read.is_err());

2
crates/huginn-mind/src/lib.rs
View file

@ -56,7 +56,7 @@ pub use store::{MindStore, OwnedRedbMessagePackBackingStore};
pub use wire::{
DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MAX_DEFERRED_BODY_BYTES, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON,
MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus, PersonaStatus,
ACTIVITY_REPOS_MAX, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery,
ACTIVITY_REPOS_MAX, OpenPr, PrAction, RepoActivity, RepoChange, RepoEvent, RepoMoved, SOURCE_TOP_K_MAX, SourceHit, SourceHits, SourceModel, SourceQuery,
};
/// The leaf, whole, through the one crate that pins its rev: the daemon and

1116
crates/huginn-mind/src/wire.rs
View file

File diff suppressed because it is too large Load diff

66
crates/mind-body/src/persona.rs
View file

@ -2051,8 +2051,20 @@ fn notice_line(run: &str, outcome: &ResolutionOutcome, at_unix_ms: u64) -> Strin
const REPO_READ_EVERY_S: u64 = 300;
/// The most of a repo notice's text.
const REPO_NOTICE_MAX_CHARS: usize = 300;
/// What a cut notice's tail may take, so the cut still fits: " and 999 more", " and changes beyond what was read" and ".".
const REPO_NOTICE_TAIL_RESERVE: usize = 47;
/// The most of a repo name the notice and the wear section print, so the clause about a capped repo has a bound whatever it is told.
const NAME_MAX_CHARS: usize = 100;
/// What the read knows when it stopped at a repo's page limit, and no more than that: it read no further back there. Names the first
/// capped repo, and counts the others.
fn capped_clause(capped: &[OrgRepo]) -> Option<String> {
let first = plain(&capped.first()?.0).chars().take(NAME_MAX_CHARS).collect::<String>();
let others = match capped.len() - 1 {
0 => String::new(),
1 => " and 1 other repo".to_owned(),
n => format!(" and {n} other repos"),
};
Some(format!("the read stopped at its page limit in {first}{others}; older entries there were not read"))
}
/// What each actor did in each repo, as (repo, actor, words), repos and then actors in the order the read gave them.
fn repo_entries(events: &[RepoEvent]) -> Vec<(String, String, String)> {
@ -2111,18 +2123,22 @@ fn pr_action_word(action: PrAction) -> &'static str {
/// logins only, its text quoted and cut at an entry. Nothing a person wrote on the forge is in the read, so none of it can be here.
/// `None` when the read found nothing to say.
fn repo_notice_line(activity: &RepoActivity, at_unix_ms: u64) -> Option<String> {
if activity.events.is_empty() && !activity.more {
if activity.events.is_empty() && activity.capped.is_empty() {
return None;
}
let entries = repo_entries(&activity.events);
let (mut text, mut last_repo, mut shown) = (String::from("Changes in your repos: "), None::<&str>, 0usize);
let capped = capped_clause(&activity.capped);
let (mut text, mut last_repo, mut shown) = (String::from(if entries.is_empty() { "Your repos' feeds: " } else { "Changes in your repos: " }), None::<&str>, 0usize);
// What follows the entries, at its worst for any count: " and N more" with as many digits as there are entries, the full stop,
// and the capped clause as its own sentence. The bound is on the text, not on a count the tail happens to need.
let tail_room = format!(" and {} more", "9".repeat(entries.len().to_string().len())).chars().count() + 1 + capped.as_ref().map_or(0, |clause| clause.chars().count() + 3);
for (repo, _, what) in &entries {
let piece = if last_repo == Some(repo.as_str()) {
format!(", {what}")
} else {
format!("{}{repo}: {what}", if last_repo.is_some() { "; " } else { "" })
};
if text.chars().count() + piece.chars().count() > REPO_NOTICE_MAX_CHARS - REPO_NOTICE_TAIL_RESERVE {
if text.chars().count() + piece.chars().count() + tail_room > REPO_NOTICE_MAX_CHARS {
break;
}
text.push_str(&piece);
@ -2133,11 +2149,20 @@ fn repo_notice_line(activity: &RepoActivity, at_unix_ms: u64) -> Option<String>
if left > 0 {
text.push_str(&format!(" and {left} more"));
}
if activity.more {
text.push_str(if entries.is_empty() { "more than the read reached" } else { " and changes beyond what was read" });
if !entries.is_empty() {
text.push('.');
}
text.push('.');
Some(format!("notice role=organ at={} text={}\n", at_text(at_unix_ms), quoted(&text)))
if let Some(clause) = capped {
if !entries.is_empty() {
text.push(' ');
}
let mut letters = clause.chars();
text.extend(letters.next().map(|first| first.to_ascii_uppercase()));
text.push_str(letters.as_str());
text.push('.');
}
Some(format!("notice role=organ at={} text={}
", at_text(at_unix_ms), quoted(&text)))
}
/// The most the wear section may take of a first line.
@ -2148,10 +2173,11 @@ const WEAR_MAX_BYTES: usize = 1536;
enum WearFact {
/// The forge read failed: the other repo facts are absent, not zero.
Missing,
/// The read stopped short of the window: others' changes older than what it reached are not shown.
Beyond,
/// The read stopped at a repo's page limit: the older entries there were not read.
Capped(String),
Awaiting { repo: String, number: u32, author: String, at: u64 },
Unseen { repo: String },
/// A repo the anonymous read cannot see, and the repo the forge's redirect names for it when it names one.
Unseen { repo: String, redirect: Option<String> },
Changed { repo: String, words: String, at: u64 },
Yours { repo: String, number: u32, at: u64 },
}
@ -2160,7 +2186,7 @@ impl WearFact {
/// The priority order a full section drops from the end of.
fn rank(&self) -> u8 {
match self {
Self::Missing | Self::Beyond => 0,
Self::Missing | Self::Capped(_) => 0,
Self::Awaiting { .. } => 1,
Self::Unseen { .. } => 2,
Self::Changed { .. } => 3,
@ -2188,7 +2214,10 @@ fn wear_facts(activity: &RepoActivity, repos: &[OrgRepo], instance: &Slug, now_m
facts.push(WearFact::Awaiting { repo, number: pr.number, author: plain(&pr.author.0), at: pr.updated_at });
}
}
facts.extend(activity.unseen.iter().map(|repo| WearFact::Unseen { repo: plain(&repo.0) }));
facts.extend(activity.unseen.iter().map(|repo| WearFact::Unseen {
repo: plain(&repo.0),
redirect: activity.moved.iter().find(|moved| &moved.from == repo).map(|moved| plain(&moved.to.0)),
}));
// The read's events are others' changes of the last three days: the daemon's forge read is the one place that window and her own
// login are decided (`RepoActivity`), so nothing is filtered again here.
let recent: Vec<RepoEvent> = activity
@ -2200,9 +2229,7 @@ fn wear_facts(activity: &RepoActivity, repos: &[OrgRepo], instance: &Slug, now_m
let at = recent.iter().filter(|event| event.repo.0 == repo && event.actor.0 == actor).map(|event| event.at).max().unwrap_or(0);
facts.push(WearFact::Changed { repo, words, at });
}
if activity.more {
facts.push(WearFact::Beyond);
}
facts.extend(capped_clause(&activity.capped).map(WearFact::Capped));
facts
}
@ -2225,9 +2252,10 @@ fn render_wear(now_ms: u64, facts: &[WearFact]) -> String {
.iter()
.map(|fact| match fact {
WearFact::Missing => "- repo facts are missing: the read did not answer.\n".to_owned(),
WearFact::Beyond => "- more changes by others lie further back than the read went and are not shown.\n".to_owned(),
WearFact::Capped(clause) => format!("- {clause}.\n"),
WearFact::Awaiting { repo, number, author, at } => format!("- PR awaiting you: {repo}#{number} by {author} {}\n", age_text(now_ms, *at)),
WearFact::Unseen { repo } => format!("- a repo the forge read cannot see: {repo}\n"),
WearFact::Unseen { repo, redirect: None } => format!("- a repo the forge read cannot see: {repo}\n"),
WearFact::Unseen { repo, redirect: Some(to) } => format!("- a repo the forge read cannot see: {repo} (the forge redirects it to {to})\n"),
WearFact::Changed { repo, words, at } => format!("- changed by others: {repo}: {words} {}\n", age_text(now_ms, *at)),
WearFact::Yours { repo, number, at } => format!("- your open PR: {repo}#{number} {}\n", age_text(now_ms, *at)),
})

85
crates/mind-body/src/persona/tests.rs
View file

@ -15,6 +15,7 @@ use std::time::Duration as StdDuration;
use eureka_pipeline::{MemoryOrigin, PipelineDocument, PipelineKind as K, PipelineMemory, ResolutionOutcome, RunOperator, RunTurn, Short, Slug};
use huginn_mind::PipelineAdmissionOutcome as AdmissionOutcome;
use huginn_mind::RepoMoved;
use serde_json::json;
use tempfile::TempDir;
@ -4582,13 +4583,13 @@ impl MindPort for Forged<'_> {
Ok(activity)
}
Some(None) => Err(anyhow!("the forge could not be read")),
None => Ok(RepoActivity { head: self.head.get(), events: vec![], more: false, open_prs: vec![], unseen: vec![] }),
None => Ok(RepoActivity { head: self.head.get(), events: vec![], capped: vec![], open_prs: vec![], unseen: vec![], moved: vec![] }),
}
}
}
fn activity(head: u64, events: Vec<RepoEvent>) -> Option<RepoActivity> {
Some(RepoActivity { head, events, more: false, open_prs: vec![], unseen: vec![] })
Some(RepoActivity { head, events, capped: vec![], open_prs: vec![], unseen: vec![], moved: vec![] })
}
fn event(id: u64, repo: &str, actor: &str, change: RepoChange) -> RepoEvent {
@ -4651,13 +4652,13 @@ fn a_repo_change_reaches_a_live_session_as_one_notice() {
fn a_burst_is_one_line() {
let rig = stewarding_rig(3);
let burst: Vec<RepoEvent> = (0..50).map(|n| event(1000 - n, "GameCult/Huginn", &format!("contributor-{n}"), RepoChange::Pushed { commits: 2 })).collect();
let forged = Forged::over(&rig.mind, vec![activity(100, vec![]), Some(RepoActivity { more: true, ..activity(1000, burst).unwrap() })]);
let forged = Forged::over(&rig.mind, vec![activity(100, vec![]), Some(RepoActivity { capped: vec![OrgRepo::from("GameCult/Huginn")], ..activity(1000, burst).unwrap() })]);
let lines = live_session(&rig, &forged, PAST_THE_REPO_READ_MS, false, 2);
let notices = all_notices(&lines);
assert_eq!(notices.len(), 1, "{lines:?}");
let text = text_of(notices[0]);
assert!(text.chars().count() <= REPO_NOTICE_MAX_CHARS, "{} chars: {text}", text.chars().count());
assert!(text.contains(" more") && text.contains("beyond what was read") && text.starts_with("Changes in your repos: GameCult/Huginn: 2 commits pushed by contributor-0"), "{text}");
assert!(text.contains(" more") && text.ends_with("The read stopped at its page limit in GameCult/Huginn; older entries there were not read.") && text.starts_with("Changes in your repos: GameCult/Huginn: 2 commits pushed by contributor-0"), "{text}");
}
/// Polling every few seconds for a stretch shorter than the interval asks the forge nothing past the first line's read.
@ -4710,9 +4711,10 @@ fn a_first_line_shows_wear_after_her_responsibilities() {
let read = RepoActivity {
head: 100,
events: vec![RepoEvent { at: now - 5 * 3_600_000, ..event(100, "GameCult/Huginn", "nibu", RepoChange::Pushed { commits: 9 }) }],
more: false,
capped: vec![],
open_prs: vec![pr(8, "nibu", 5), pr(9, Slug(INSTANCE.into()).0.as_str(), 50)],
unseen: vec!["GameCult/Ghost".into()],
moved: vec![],
};
let forged = Forged::over(&rig.mind, vec![Some(read)]);
live_session(&rig, &forged, 1000, false, 1);
@ -4736,21 +4738,22 @@ fn a_read_that_stopped_short_says_so_on_the_first_line_and_in_a_notice() {
let read = RepoActivity {
head: 100,
events: vec![RepoEvent { at: now - 5 * 24 * 3_600_000, ..event(100, "GameCult/Huginn", "nibu", RepoChange::Pushed { commits: 2 }) }],
more: true,
capped: vec![OrgRepo::from("GameCult/Huginn")],
open_prs: vec![],
unseen: vec![],
moved: vec![],
};
let forged = Forged::over(&rig.mind, vec![Some(read)]);
live_session(&rig, &forged, 1000, false, 1);
let input = input_of(&rig, 0);
let wear = input.split_once("Where your machine shows wear, with ages (facts, not orders):\n").expect("the section").1.split("\n\n").next().unwrap();
assert!(wear.contains("changed by others: GameCult/Huginn: 2 commits pushed by nibu (5 d)"), "{wear}");
assert!(wear.contains("further back than the read went"), "{wear}");
assert!(wear.contains("- the read stopped at its page limit in GameCult/Huginn; older entries there were not read."), "{wear}");
let nothing_seen = RepoActivity { head: 9, events: vec![], more: true, open_prs: vec![], unseen: vec![] };
let nothing_seen = RepoActivity { head: 9, events: vec![], capped: vec![OrgRepo::from("GameCult/Huginn")], open_prs: vec![], unseen: vec![], moved: vec![] };
let notice = repo_notice_line(&nothing_seen, 1_700_000_000_000).expect("a truncated read is told");
assert!(text_of(&notice).contains("more than the read reached"), "{notice}");
assert!(repo_notice_line(&RepoActivity { more: false, ..nothing_seen }, 1_700_000_000_000).is_none());
assert_eq!(text_of(&notice), "Your repos' feeds: The read stopped at its page limit in GameCult/Huginn; older entries there were not read.");
assert!(repo_notice_line(&RepoActivity { capped: vec![], ..nothing_seen }, 1_700_000_000_000).is_none());
}
/// The repos the forge is asked about are her stewardships in force and nothing else: a Mind with none asks about none (and so
@ -4790,9 +4793,10 @@ fn forge_text_cannot_enter_an_organ_line() {
let read = RepoActivity {
head: 9,
events: vec![event(9, "GameCult/Huginn", hostile, RepoChange::Pushed { commits: 2 })],
more: false,
capped: vec![],
open_prs: vec![huginn_mind::OpenPr { repo: "GameCult/Huginn".into(), number: 1, author: Short(hostile.into()), updated_at: 1_700_000_000_000 }],
unseen: vec![],
moved: vec![],
};
let line = repo_notice_line(&read, 1_700_000_000_000).unwrap();
assert_eq!(line.lines().count(), 1, "{line}");
@ -4808,9 +4812,10 @@ fn forge_text_cannot_enter_an_organ_line() {
let read = RepoActivity {
head: 9,
events: vec![event(9, name, "nibu", RepoChange::Pushed { commits: 2 })],
more: false,
capped: vec![],
open_prs: vec![huginn_mind::OpenPr { repo: name.into(), number: 1, author: Short(INSTANCE.into()), updated_at: 1_700_000_000_000 }],
unseen: vec![name.into()],
moved: vec![],
};
let wear = render_wear(1_700_000_000_000, &wear_facts(&read, &["GameCult/Huginn".into()], &instance, 1_700_000_000_000));
assert!(wear.lines().all(|line| line.starts_with("- ") || line.starts_with("Where your machine")), "{wear}");
@ -4828,16 +4833,63 @@ fn a_cut_notice_never_passes_its_bound_at_any_entry_length() {
for len in 1..=80 {
for commits in [1_u32, 2, 9, 10, 99, 100, 12345] {
let burst: Vec<RepoEvent> = (0..events).map(|n| event(1000 - n, "GameCult/Huginn", &format!("{}{n}", "x".repeat(len)), RepoChange::Pushed { commits })).collect();
let read = RepoActivity { head: 1000, events: burst, more: true, open_prs: vec![], unseen: vec![] };
let read = RepoActivity { head: 1000, events: burst, capped: vec![OrgRepo::from("GameCult/Huginn")], open_prs: vec![], unseen: vec![], moved: vec![] };
let text = text_of(&repo_notice_line(&read, 1_700_000_000_000).unwrap());
let chars = text.chars().count();
assert!(chars <= REPO_NOTICE_MAX_CHARS, "{chars} chars with {events} events of {len} and {commits} commits: {text}");
assert!(text.contains(" more") && text.ends_with("beyond what was read."), "a cut says what it left out: {text}");
assert!(text.contains(" more") && text.ends_with("The read stopped at its page limit in GameCult/Huginn; older entries there were not read."), "a cut says what it left out: {text}");
widest = widest.max(chars);
}
}
}
assert!(widest + 2 >= REPO_NOTICE_MAX_CHARS, "the sweep reaches the edge it guards: widest {widest}");
assert!(widest + 4 >= REPO_NOTICE_MAX_CHARS, "the sweep reaches the edge it guards: widest {widest}");
}
/// The bound holds at a thousand leftover entries and past it, where the count itself is four digits: the promise is "never over 300
/// chars", not "to 999".
#[test]
fn a_cut_notice_holds_its_bound_with_a_four_digit_count_left() {
for events in [1_000_u64, 1100, 12_000] {
for len in [1_usize, 40, 80] {
let burst: Vec<RepoEvent> = (0..events).map(|n| event(100_000 - n, "GameCult/Huginn", &format!("{}{n}", "x".repeat(len)), RepoChange::Pushed { commits: 1 })).collect();
let read = RepoActivity { head: 100_000, events: burst, capped: vec![], open_prs: vec![], unseen: vec![], moved: vec![] };
let text = text_of(&repo_notice_line(&read, 1_700_000_000_000).unwrap());
let left: usize = text.split(" and ").last().unwrap().split(' ').next().unwrap().parse().expect("the notice counts what it left out");
assert!(left >= 900, "{left} left of {events}: {text}");
assert!(text.chars().count() <= REPO_NOTICE_MAX_CHARS, "{} chars with {events} events of {len}: {text}", text.chars().count());
}
}
}
/// A notice that cut and also stopped at the page limit in many repos with long names is still within the bound, and names only what
/// the read knows.
#[test]
fn the_capped_clause_claims_only_what_the_read_knows_and_stays_in_the_bound() {
let long = "r".repeat(200);
let capped: Vec<OrgRepo> = (0..30).map(|n| OrgRepo(format!("GameCult/{long}{n}"))).collect();
let burst: Vec<RepoEvent> = (0..60).map(|n| event(1000 - n, "GameCult/Huginn", &format!("contributor-{n}"), RepoChange::Pushed { commits: 2 })).collect();
let read = RepoActivity { head: 1000, events: burst, capped, open_prs: vec![], unseen: vec![], moved: vec![] };
let text = text_of(&repo_notice_line(&read, 1_700_000_000_000).unwrap());
assert!(text.chars().count() <= REPO_NOTICE_MAX_CHARS, "{} chars: {text}", text.chars().count());
assert!(text.ends_with("and 29 other repos; older entries there were not read."), "{text}");
assert!(!text.contains("further back") && !text.contains("beyond"), "it claims nothing about what it did not read: {text}");
}
/// A repo the forge redirects is told with where the redirect points, and one that is only dark is told without.
#[test]
fn a_redirected_unseen_repo_is_told_with_where_the_forge_points() {
let moved = RepoMoved { from: OrgRepo::from("GameCult/cinny"), to: OrgRepo::from("GameCult/Ratatoskr") };
let facts = wear_facts(
&RepoActivity { head: 1, events: vec![], capped: vec![], open_prs: vec![], unseen: vec![OrgRepo::from("GameCult/cinny"), OrgRepo::from("GameCult/gamecult-ops")], moved: vec![moved] },
&[],
&Slug("void".into()),
0,
);
let wear = render_wear(0, &facts);
assert!(wear.contains("- a repo the forge read cannot see: GameCult/cinny (the forge redirects it to GameCult/Ratatoskr)
"), "{wear}");
assert!(wear.contains("- a repo the forge read cannot see: GameCult/gamecult-ops
"), "{wear}");
}
/// The polls after the first interval are still one read per interval: over 750 s of polls at 5 s the forge is asked at the start
@ -4861,9 +4913,10 @@ fn each_changed_line_has_the_age_of_its_own_actor_and_repo() {
let read = RepoActivity {
head: 4,
events: vec![push(4, "GameCult/Huginn", "void", 1), push(3, "GameCult/Huginn", "nibu", 60), push(2, "GameCult/Ratatoskr", "nibu", 30)],
more: false,
capped: vec![],
open_prs: vec![],
unseen: vec![],
moved: vec![],
};
let wear = render_wear(now, &wear_facts(&read, &["GameCult/Huginn".into(), "GameCult/Ratatoskr".into()], &Slug(INSTANCE.into()), now));
assert!(wear.contains("- changed by others: GameCult/Huginn: 1 commit pushed by void (1 h)\n"), "{wear}");

2
crates/mind-body/src/queue.rs
View file

@ -555,7 +555,7 @@ mod tests {
#[test]
fn an_activity_read_goes_through_the_port_as_one_request() {
let seen = std::cell::RefCell::new(Vec::new());
let read = RepoActivity { head: 77, events: vec![], more: true, open_prs: vec![], unseen: vec![OrgRepo::from("GameCult/Ghost")] };
let read = RepoActivity { head: 77, events: vec![], capped: vec![OrgRepo::from("GameCult/Huginn")], open_prs: vec![], unseen: vec![OrgRepo::from("GameCult/Ghost")], moved: vec![] };
let answer = std::cell::RefCell::new(HuginnMindResponse::Activity(read.clone()));
let port = HuginnMind::over(Slug(INSTANCE.into()), Box::new(|request| {
seen.borrow_mut().push(request);

2
crates/mind-body/src/testkit.rs
View file

@ -204,7 +204,7 @@ impl MindPort for TestMind {
/// A forge with nothing new in it: the daemon with a healthy forge and a quiet org.
fn stewarded_activity(&self, _repos: &[huginn_mind::eureka_pipeline::OrgRepo], after: Option<u64>) -> Result<huginn_mind::RepoActivity> {
Ok(huginn_mind::RepoActivity { head: after.unwrap_or(0), events: vec![], more: false, open_prs: vec![], unseen: vec![] })
Ok(huginn_mind::RepoActivity { head: after.unwrap_or(0), events: vec![], capped: vec![], open_prs: vec![], unseen: vec![], moved: vec![] })
}
fn admit(&self, agent: &str, session: &str, documents: Vec<PipelineDocument>) -> Result<PipelineAdmissionOutcome> {

39
schemas/cultnet/huginn.mind_response.v1.schema.json
View file

@ -4338,9 +4338,15 @@
]
},
"RepoActivity": {
"description": "The answer to an activity read. `events` are other people's changes only,\nthe asker's own login already dropped, and on a read with no `after` only\nthose of the last 72 hours: the reader filters nothing again. `head` is the\nnewest feed id the read saw in those repos, whether or not any event\nsurvived the filter, an entry older than the window too: the cursor to ask\n`after` next. `more` says some repo's feed had more to give past the page\ncap, before it reached `after` (or the window start on a read with no\n`after`). `open_prs` and `unseen` come only on a read with no `after`.",
"description": "The answer to an activity read. `events` are other people's changes only,\nthe asker's own login already dropped, and on a read with no `after` only\nthose of the last 72 hours: the reader filters nothing again. `head` is the\nnewest feed id the read saw in those repos, whether or not any event\nsurvived the filter, an entry older than the window too: the cursor to ask\n`after` next. `capped` names the repos whose feed the read stopped reading at its page limit: their older entries were not read.\n`open_prs`, `unseen` and `moved` come only on a read with no `after`.",
"type": "object",
"properties": {
"capped": {
"type": "array",
"items": {
"$ref": "#/$defs/OrgRepo"
}
},
"events": {
"type": "array",
"items": {
@ -4352,8 +4358,12 @@
"format": "uint64",
"minimum": 0
},
"more": {
"type": "boolean"
"moved": {
"description": "Of those, the ones whose feed answered a redirect that names another repo. The redirect is not followed.",
"type": "array",
"items": {
"$ref": "#/$defs/RepoMoved"
}
},
"open_prs": {
"type": "array",
@ -4362,7 +4372,7 @@
}
},
"unseen": {
"description": "Repos asked about that the anonymous org listing does not hold.",
"description": "Repos asked about that the anonymous read cannot see: the org listing does not hold them, or their feed answered 404 or a redirect.",
"type": "array",
"items": {
"$ref": "#/$defs/OrgRepo"
@ -4372,9 +4382,10 @@
"required": [
"head",
"events",
"more",
"capped",
"open_prs",
"unseen"
"unseen",
"moved"
]
},
"RepoChange": {
@ -4466,6 +4477,22 @@
"change"
]
},
"RepoMoved": {
"description": "A redirect the forge answered for a repo's feed: it points at `to`.",
"type": "object",
"properties": {
"from": {
"$ref": "#/$defs/OrgRepo"
},
"to": {
"$ref": "#/$defs/OrgRepo"
}
},
"required": [
"from",
"to"
]
},
"ReportCommit": {
"type": "object",
"properties": {