tests for the mutation survivors: char bound and newline accounting, catalog paging, batch bound, a source search through the daemon, the shared queue bound and abandon by ticket

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/huginn-daemon/src/daemon.rs

@@ -551,6 +551,83 @@ pub(crate) mod tests {
         fn abandon(&mut self, _ticket: SearchTicket) {}
     }
 
+    /// An index that answers source searches from a script and counts how often
+    /// it was asked.
+    struct SourcesIndex {
+        asked: Arc<Mutex<u32>>,
+        hits: SourceHits,
+    }
+
+    impl<S: MindStore> IndexSink<S> for SourcesIndex {
+        fn committed(&mut self, _mind: &Mind<S>, _writes: &[PipelineRef]) -> anyhow::Result<()> {
+            Ok(())
+        }
+
+        fn status(&self) -> IndexStatus {
+            IndexStatus::Current
+        }
+
+        fn search(&mut self, _text: &str, _top_k: u32, _among: Vec<String>) -> Result<SearchTicket, String> {
+            Err("this index cannot search the mind".into())
+        }
+
+        fn search_sources(&mut self, _query: &SourceQuery) -> Result<SearchTicket, String> {
+            *self.asked.lock().unwrap() += 1;
+            Ok(SearchTicket(7))
+        }
+
+        fn sources_searched(&mut self) -> Vec<(SearchTicket, Result<SourceHits, String>)> {
+            vec![(SearchTicket(7), Ok(self.hits.clone()))]
+        }
+
+        fn search_deadline(&self) -> std::time::Duration {
+            std::time::Duration::ZERO
+        }
+
+        fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+            Vec::new()
+        }
+
+        fn abandon(&mut self, _ticket: SearchTicket) {}
+    }
+
+    /// A source search is checked at the door before the index is asked, is
+    /// answered with the hits the index found and nothing joined from the mind,
+    /// turns a failure into `Unavailable`, and is refused for another mind.
+    #[test]
+    fn a_source_search_reaches_the_index_only_when_well_formed_and_answers_its_hits() {
+        use huginn_mind::{SourceHit, SourceModel};
+        let hits = SourceHits {
+            indexed_under: SourceModel { name: "m".into(), digest: "d".into() },
+            hits: vec![SourceHit { repo: "R".into(), path: "p.rs".into(), commit: "c".into(), line_start: 1, line_end: 2, score: 0.5, text: "t".into() }],
+        };
+        let asked = Arc::new(Mutex::new(0));
+        let root = tempfile::tempdir().unwrap();
+        let mind = Mind::open(root.path(), &slug(INSTANCE)).unwrap();
+        let mut daemon = Daemon::new(mind, SourcesIndex { asked: Arc::clone(&asked), hits: hits.clone() });
+        let request = |instance: &str, top_k: u32| HuginnMindRequest::SearchSources {
+            instance: slug(instance),
+            query: SourceQuery { text: huginn_mind::eureka_pipeline::Line("find".into()), top_k, repo: None },
+        };
+
+        let Handled::Searching(search) = daemon.handle(request(INSTANCE, 5), now()) else { panic!("a well-formed source search is a search") };
+        assert_eq!((search.ticket, *asked.lock().unwrap()), (SearchTicket(7), 1));
+        let mut finished = daemon.searched();
+        let (ticket, found) = finished.remove(0);
+        assert_eq!(ticket, SearchTicket(7));
+        assert_eq!(daemon.finish(search, found), HuginnMindResponse::Sources(hits));
+
+        let Handled::Searching(search) = daemon.handle(request(INSTANCE, 5), now()) else { panic!() };
+        let failed = daemon.finish(search, Found::Failed("the index fell over".into()));
+        assert!(matches!(failed, HuginnMindResponse::Refused(MindRefusal::Unavailable { ref detail }) if detail.contains("fell over")), "{failed:?}");
+
+        for refused in [request(INSTANCE, 0), request(INSTANCE, 51), request(OTHER, 5)] {
+            let answer = daemon.handle(refused, now()).answered();
+            assert!(matches!(answer, HuginnMindResponse::Refused(_)), "{answer:?}");
+        }
+        assert_eq!(*asked.lock().unwrap(), 2, "a refused query never reached the index");
+    }
+
     /// A daemon over the instance's mind under `state_root`, with no index.
     pub(crate) fn open_unindexed(
         state_root: &Path,

crates/huginn-daemon/src/index/tests.rs

@@ -2057,3 +2057,30 @@ fn a_worker_stuck_on_the_minds_collection_still_answers_a_source_search() {
     assert_eq!((answered.hits[0].repo.as_str(), answered.hits[0].path.as_str()), ("Huginn", "a.rs"));
     assert_eq!(answered.indexed_under.digest, "d1");
 }
+
+/// Source searches take their own tickets, share the queue's bound with the
+/// mind's searches, and an abandon removes the one ticket named from either.
+#[test]
+fn source_searches_have_distinct_tickets_share_the_queue_bound_and_abandon_by_ticket() {
+    let shared = shared_with(Vec::new(), None, Vec::new(), false);
+    let mut sink = WorkerSink { shared, identified: true, deadline: SEARCH_DEADLINE };
+    let first = ask_sources(&mut sink, "a").unwrap();
+    let second = ask_sources(&mut sink, "b").unwrap();
+    let mind = ask_sink(&mut sink, "m", 3).unwrap();
+    assert!(first != second && second != mind && first != mind, "{first:?} {second:?} {mind:?}");
+
+    abandon(&mut sink, first);
+    {
+        let guard = lock(&sink.shared);
+        let left: Vec<SearchTicket> = guard.source_searches.iter().map(|queued| queued.ticket).collect();
+        assert_eq!((left, guard.searches.len()), (vec![second], 1), "only the named ticket went");
+    }
+
+    for _ in 0..SEARCHES_QUEUED_MAX - 2 {
+        ask_sources(&mut sink, "fill").unwrap();
+    }
+    assert!(ask_sources(&mut sink, "over").is_err(), "the bound counts both kinds");
+    assert!(ask_sink(&mut sink, "over", 3).is_err(), "and refuses both kinds");
+    abandon(&mut sink, second);
+    assert!(ask_sink(&mut sink, "now", 3).is_ok(), "a freed slot is a slot");
+}

crates/huginn-daemon/src/sources/chunk.rs

@@ -131,4 +131,23 @@ mod tests {
         let held = Chunk { line_start: 3, line_end: 4, text: "fn a() {}".into() };
         assert_eq!(embedded_text("Huginn", "src/a.rs", &held), "Huginn/src/a.rs\nfn a() {}");
     }
+
+    /// The character bound is inclusive: a line of exactly the bound is kept
+    /// whole, one more character is cut; and the running count of a window
+    /// includes each joining newline.
+    #[test]
+    fn the_character_bound_is_inclusive_and_counts_the_joining_newlines() {
+        let exact = "z".repeat(CHUNK_CHARS_MAX);
+        let kept = chunk(&format!("{exact}\n"));
+        assert_eq!((kept.len(), kept[0].text.chars().count()), (1, CHUNK_CHARS_MAX));
+        let over = chunk(&format!("{exact}z\nnext\n"));
+        let shown: Vec<(u32, u32, usize)> = over.iter().map(|held| (held.line_start, held.line_end, held.text.chars().count())).collect();
+        assert_eq!(shown, [(1, 1, CHUNK_CHARS_MAX), (2, 2, 4)]);
+
+        // 13 lines of 100 characters are 1,312 with their newlines; the 14th would make 1,413.
+        let hundred = "h".repeat(100);
+        let text: String = (0..20).map(|_| format!("{hundred}\n")).collect();
+        let ranges: Vec<(u32, u32)> = chunk(&text).iter().map(|held| (held.line_start, held.line_end)).collect();
+        assert_eq!(ranges, [(1, 13), (14, 20)]);
+    }
 }

crates/huginn-daemon/src/sources/forge.rs

@@ -90,4 +90,38 @@ mod tests {
         let names: Vec<String> = indexable(&page).unwrap().into_iter().map(|repo| repo.name).collect();
         assert_eq!(names, ["Open"]);
     }
+
+    /// Serves each page in `pages` once, to the request that names it.
+    fn serve(pages: Vec<String>) -> String {
+        use std::io::{Read, Write};
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        let address = listener.local_addr().unwrap();
+        std::thread::spawn(move || {
+            for _ in 0..pages.len() {
+                let (mut stream, _) = listener.accept().unwrap();
+                let mut buffer = [0_u8; 2048];
+                let read = stream.read(&mut buffer).unwrap();
+                let request = String::from_utf8_lossy(&buffer[..read]).into_owned();
+                let body = if request.contains("page=1&") { &pages[0] } else { &pages[1] };
+                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();
+            }
+        });
+        format!("http://{address}")
+    }
+
+    /// The catalog reads every page until one is short: a full first page
+    /// does not end it, and a short second page does; clone URLs name the org.
+    #[test]
+    fn the_catalog_reads_pages_until_a_short_one_and_clones_from_the_org() {
+        let page = |range: std::ops::Range<usize>| {
+            let repos: Vec<Value> = range.map(|n| json!({ "name": format!("r{n}"), "default_branch": "main", "private": false, "archived": false, "mirror": false, "empty": false })).collect();
+            Value::Array(repos).to_string()
+        };
+        let base = serve(vec![page(0..PAGE), page(PAGE..PAGE + 3)]);
+        let mut forge = HttpForge::new(&format!("{base}/"));
+        let names: Vec<String> = forge.catalog().unwrap().into_iter().map(|repo| repo.name).collect();
+        assert_eq!(names.len(), PAGE + 3);
+        assert_eq!((names[0].as_str(), names[PAGE + 2].as_str()), ("r0", "r52"));
+        assert_eq!(forge.clone_url("Demo"), format!("{base}/GameCult/Demo.git"));
+    }
 }

crates/huginn-daemon/src/sources/tests.rs

@@ -406,4 +406,18 @@ fn a_chunk_id_is_its_content_address() {
         assert_ne!(base, other);
     }
     assert_ne!(chunk_id("R", "pb", "lob", 0), chunk_id("R", "p", "blob", 0), "fields are separated, not concatenated");
+
+/// Chunks are embedded in batches of at most `BATCH`, however many a repo has.
+#[test]
+fn chunks_are_embedded_in_bounded_batches() {
+    let remote = Remote::new();
+    let many: Vec<(String, Vec<u8>)> = (0..70).map(|n| (format!("f{n}.txt"), format!("file {n}\n").into_bytes())).collect();
+    let files: Vec<(&str, &[u8])> = many.iter().map(|(path, bytes)| (path.as_str(), bytes.as_slice())).collect();
+    remote.commit(&files, &[]);
+    let mut rig = Rig::new();
+    rig.converged(&mut FakeForge::of(&[("Demo", &remote)]));
+    let sizes: Vec<usize> = rig.embedder.state.lock().unwrap().embedded.iter().map(Vec::len).collect();
+    assert_eq!(sizes, [BATCH, BATCH, 6]);
+    assert_eq!(rig.index.held().len(), 70);
+}
 }