From 8041d82fe1d65fbdbb05d9080be217c5f2e56108 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 00:36:18 +0200 Subject: [PATCH] eureka-state tests: the in-process daemon fixtures move to tests/common The daemon half of the client's fixtures (an in-process daemon over a temporary mind on a loopback port, and the scripted server) becomes tests/common/mod.rs, so the MCP binary's tests (Cut 13b) drive the same daemon instead of a second copy of it. No behaviour change; client.rs shrinks by the moved lines. Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/client.rs | 153 +-------------------- crates/eureka-state/tests/common/mod.rs | 174 ++++++++++++++++++++++++ 2 files changed, 177 insertions(+), 150 deletions(-) create mode 100644 crates/eureka-state/tests/common/mod.rs diff --git a/crates/eureka-state/tests/client.rs b/crates/eureka-state/tests/client.rs index 07f458a..1ef33e8 100644 --- a/crates/eureka-state/tests/client.rs +++ b/crates/eureka-state/tests/client.rs @@ -28,29 +28,12 @@ use huginn_mind::{ }; use tempfile::TempDir; -const INSTANCE: &str = "eureka"; +mod common; +use common::*; + const OTHER: &str = "thought-cage"; -const CAMPAIGN: &str = "eureka-state"; -const REPO: &str = "GameCult/Epiphany"; const TIMEOUT: Duration = Duration::from_secs(20); -fn slug(value: &str) -> Slug { - Slug(value.into()) -} - -fn sha() -> Sha { - Sha("5f98228d".into()) -} - -fn identity(name: &str) -> PipelineDocument { - PipelineDocument::Instance(PipelineInstance { - instance: slug(name), - display_name: Short(format!("{name} mind")), - created_at: Date("2026-09-29".into()), - host: Short(name.into()), - }) -} - fn stewardship() -> PipelineDocument { PipelineDocument::Stewardship(PipelineStewardship { instance: slug(INSTANCE), @@ -61,29 +44,6 @@ fn stewardship() -> PipelineDocument { }) } -fn campaign() -> PipelineDocument { - PipelineDocument::Campaign(PipelineCampaign { - slug: slug(CAMPAIGN), - title: Title("Eureka pipeline state".into()), - repos: vec![OrgRepo(REPO.into())], - working_branch: Short("hands/cut13a".into()), - target_doc: DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() }, - }) -} - -fn batch(instance: &str, documents: Vec) -> PipelineAdmissionBatch { - PipelineAdmissionBatch { - instance: slug(instance), - provenance: PipelineProvenance { - faculty: Faculty::Hands, - agent: Short("claude".into()), - session: Short("session-1".into()), - tool: Short("admit".into()), - }, - documents, - } -} - /// A `Line` and a `Short` at the leaf's own bound. fn line(n: usize) -> Line { Line(format!("{n:0>1000}")) @@ -151,108 +111,6 @@ fn cut_ref(cut: &str) -> PipelineRef { PipelineRef { kind: PipelineKind::CutSpec, id: Short(format!("{CAMPAIGN}:cut_spec:cut-{cut}.r1")) } } -/// No index: the client's subject is the transport, not the projection. -struct Inert; - -impl IndexSink for Inert { - fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { - Ok(()) - } - fn status(&self) -> IndexStatus { - IndexStatus::Current - } - fn search(&mut self, _text: &str, _top_k: u32) -> Result { - Err("this index cannot search".into()) - } - fn search_deadline(&self) -> Duration { - Duration::ZERO - } - fn searched(&mut self) -> Vec<(SearchTicket, Result)> { - Vec::new() - } - fn abandon(&mut self, _ticket: SearchTicket) {} -} - -type TestDaemon = Daemon; - -fn answered(handled: Handled) -> HuginnMindResponse { - match handled { - Handled::Answered(response) => response, - Handled::Searching(_) => panic!("the inert index cannot search"), - } -} - -/// A daemon over a temporary mind whose identity is admitted, and whatever -/// else `seed` admits. -fn mind(seed: Vec>) -> (TempDir, TestDaemon) { - let root = tempfile::tempdir().unwrap(); - let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), Inert); - for documents in std::iter::once(vec![identity(INSTANCE)]).chain(seed) { - let outcome = answered(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), Utc::now())); - assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}"); - } - (root, daemon) -} - -/// A server on a loopback port that stops when dropped. -struct Server { - addr: SocketAddr, - stopping: Arc, - thread: Option>, - /// The sessions a scripted server has seen end: what the client's - /// disconnect looks like from the daemon's side. - ended: Arc, - _root: Option, -} - -impl Drop for Server { - fn drop(&mut self) { - self.stopping.store(true, Ordering::Relaxed); - if let Some(thread) = self.thread.take() { - let _ = thread.join(); - } - } -} - -fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { - let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); - let addr = hub.local_addr().unwrap(); - let stopping = Arc::new(AtomicBool::new(false)); - let loop_stopping = Arc::clone(&stopping); - let registry = schema_registry().unwrap(); - let thread = std::thread::spawn(move || { - run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()).unwrap(); - }); - Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) } -} - -/// A server that answers each schema message with whatever `script` says. -fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + 'static) -> Server { - let mut hub = bind("127.0.0.1:0".parse().unwrap(), "scripted").unwrap(); - let addr = hub.local_addr().unwrap(); - let stopping = Arc::new(AtomicBool::new(false)); - let loop_stopping = Arc::clone(&stopping); - let ended = Arc::new(AtomicUsize::new(0)); - let loop_ended = Arc::clone(&ended); - let thread = std::thread::spawn(move || { - while !loop_stopping.load(Ordering::Relaxed) { - hub.poll_resends().unwrap(); - while let Some(event) = hub.receive_event_once().unwrap() { - if matches!(event, CultNetRudpServerEvent::Disconnected { .. }) { - loop_ended.fetch_add(1, Ordering::Relaxed); - } - let CultNetRudpServerEvent::Frame { session, frame } = event else { continue }; - let message = - decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0).unwrap(); - hub.send_schema_message(&session, &script(&message)).unwrap(); - } - std::thread::sleep(Duration::from_millis(2)); - } - }); - Server { addr, stopping, thread: Some(thread), ended, _root: None } -} - -/// A scripted server that answers any operation with a deferral of `body` and /// serves its chunks faithfully. fn deferring(body: Vec) -> Server { deferring_in_chunks(body, 256 * 1024, Duration::ZERO) @@ -298,11 +156,6 @@ fn rejected(error: ClientError) -> (SocketAddr, String, String) { } } -/// A scripted server whose every answer is `reply`. -fn answering(reply: CultNetMessage) -> Server { - scripted(move |_| reply.clone()) -} - fn nothing_found(message_id: &str, operation: &str) -> CultNetMessage { encode_response(message_id, operation, &HuginnMindResponse::View(None), "scripted").unwrap() } diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs new file mode 100644 index 0000000..5729e68 --- /dev/null +++ b/crates/eureka-state/tests/common/mod.rs @@ -0,0 +1,174 @@ +//! The daemon half of the fixtures: an in-process Huginn daemon over a +//! temporary mind, on a loopback port, shared by the client and the server +//! tests. +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::thread::JoinHandle; +use std::time::Duration; + +use chrono::Utc; +use cultnet_rs::{CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, decode_cultnet_message_from_slice}; +use huginn_daemon::serve::{bind, run, schema_registry}; +use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions}; +use huginn_mind::epiphany_pipeline::{ + Date, DocRef, OrgRepo, PipelineCampaign, PipelineDocument, PipelineInstance, Sha, Short, Slug, Title, +}; +use huginn_mind::{ + Faculty, HuginnMindRequest, HuginnMindResponse, IndexStatus, Mind, MindStore, OwnedRedbMessagePackBackingStore, + PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance, +}; +use tempfile::TempDir; + +pub const INSTANCE: &str = "eureka"; + +pub fn slug(value: &str) -> Slug { + Slug(value.into()) +} + +pub fn identity(name: &str) -> PipelineDocument { + PipelineDocument::Instance(PipelineInstance { + instance: slug(name), + display_name: Short(format!("{name} mind")), + created_at: Date("2026-09-29".into()), + host: Short(name.into()), + }) +} + +pub fn batch(instance: &str, documents: Vec) -> PipelineAdmissionBatch { + PipelineAdmissionBatch { + instance: slug(instance), + provenance: PipelineProvenance { + faculty: Faculty::Hands, + agent: Short("claude".into()), + session: Short("session-1".into()), + tool: Short("admit".into()), + }, + documents, + } +} + +/// No index: the client's subject is the transport, not the projection. +pub struct Inert; + +impl IndexSink for Inert { + fn committed(&mut self, _mind: &Mind, _writes: &[PipelineRef]) -> anyhow::Result<()> { + Ok(()) + } + fn status(&self) -> IndexStatus { + IndexStatus::Current + } + fn search(&mut self, _text: &str, _top_k: u32) -> Result { + Err("this index cannot search".into()) + } + fn search_deadline(&self) -> Duration { + Duration::ZERO + } + fn searched(&mut self) -> Vec<(SearchTicket, Result)> { + Vec::new() + } + fn abandon(&mut self, _ticket: SearchTicket) {} +} + +pub type TestDaemon = Daemon; + +pub fn answered(handled: Handled) -> HuginnMindResponse { + match handled { + Handled::Answered(response) => response, + Handled::Searching(_) => panic!("the inert index cannot search"), + } +} + +/// A daemon over a temporary mind whose identity is admitted, and whatever +/// else `seed` admits. +pub fn mind(seed: Vec>) -> (TempDir, TestDaemon) { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::new(Mind::open(root.path(), &slug(INSTANCE)).unwrap(), Inert); + for documents in std::iter::once(vec![identity(INSTANCE)]).chain(seed) { + let outcome = answered(daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), Utc::now())); + assert!(matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), "{outcome:?}"); + } + (root, daemon) +} + +/// A server on a loopback port that stops when dropped. +pub struct Server { + pub addr: SocketAddr, + pub stopping: Arc, + pub thread: Option>, + /// The sessions a scripted server has seen end: what the client's + /// disconnect looks like from the daemon's side. + pub ended: Arc, + pub _root: Option, +} + +impl Drop for Server { + fn drop(&mut self) { + self.stopping.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +pub fn serve(root: TempDir, mut daemon: TestDaemon) -> Server { + let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); + let addr = hub.local_addr().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let registry = schema_registry().unwrap(); + let thread = std::thread::spawn(move || { + run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()).unwrap(); + }); + Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) } +} + +/// A server that answers each schema message with whatever `script` says. +pub fn scripted(mut script: impl FnMut(&CultNetMessage) -> CultNetMessage + Send + 'static) -> Server { + let mut hub = bind("127.0.0.1:0".parse().unwrap(), "scripted").unwrap(); + let addr = hub.local_addr().unwrap(); + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let ended = Arc::new(AtomicUsize::new(0)); + let loop_ended = Arc::clone(&ended); + let thread = std::thread::spawn(move || { + while !loop_stopping.load(Ordering::Relaxed) { + hub.poll_resends().unwrap(); + while let Some(event) = hub.receive_event_once().unwrap() { + if matches!(event, CultNetRudpServerEvent::Disconnected { .. }) { + loop_ended.fetch_add(1, Ordering::Relaxed); + } + let CultNetRudpServerEvent::Frame { session, frame } = event else { continue }; + let message = + decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0).unwrap(); + hub.send_schema_message(&session, &script(&message)).unwrap(); + } + std::thread::sleep(Duration::from_millis(2)); + } + }); + Server { addr, stopping, thread: Some(thread), ended, _root: None } +} + +/// A scripted server whose every answer is `reply`. +pub fn answering(reply: CultNetMessage) -> Server { + scripted(move |_| reply.clone()) +} + +pub const CAMPAIGN: &str = "eureka-state"; +pub const REPO: &str = "GameCult/Epiphany"; + +pub fn sha() -> Sha { + Sha("5f98228d".into()) +} + +pub fn campaign() -> PipelineDocument { + PipelineDocument::Campaign(PipelineCampaign { + slug: slug(CAMPAIGN), + title: Title("Eureka pipeline state".into()), + repos: vec![OrgRepo(REPO.into())], + working_branch: Short("hands/cut13a".into()), + target_doc: DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() }, + }) +}