diff --git a/Cargo.lock b/Cargo.lock index d887bf8..e563a9a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -935,6 +935,7 @@ dependencies = [ "cultnet-rs 0.1.0", "huginn-mind", "rmp-serde", + "rustix", "serde", "serde_json", "sha2", diff --git a/crates/eureka-state/tests/common/mod.rs b/crates/eureka-state/tests/common/mod.rs index bf96c83..4dd1367 100644 --- a/crates/eureka-state/tests/common/mod.rs +++ b/crates/eureka-state/tests/common/mod.rs @@ -164,7 +164,7 @@ pub fn serve + Send + 'static>( 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(); + run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default()).unwrap(); }); Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) } } diff --git a/crates/huginn-daemon/Cargo.toml b/crates/huginn-daemon/Cargo.toml index 59e4270..93ffc47 100644 --- a/crates/huginn-daemon/Cargo.toml +++ b/crates/huginn-daemon/Cargo.toml @@ -30,6 +30,8 @@ chrono = "0.4.44" cultnet-rs.workspace = true huginn-mind = { path = "../huginn-mind" } rmp-serde = "1" +# The desk door asks the kernel who is on the other end of a unix socket; mind-body's line. +rustix = { version = "1", default-features = false, features = ["std", "net"] } serde = { version = "1", features = ["derive"] } # The vector store's and the embedder's HTTP APIs are the xenos boundary, and # JSON exists in `index/ollama.rs` and `index/qdrant.rs` only. `ureq` 3 with diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index 6311f50..f5ed4ff 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -5,6 +5,7 @@ use chrono::{DateTime, Utc}; use cultnet_rs::Selection; use crate::sources::forge::HttpForge; +use huginn_mind::desk_wire::{DeskRequest, DeskResponse, answer_ref}; use huginn_mind::eureka_pipeline::{PipelineRef, Slug}; use huginn_mind::wire::{ACTIVITY_REPOS_MAX, HuginnMindRequest, HuginnMindResponse, IndexStatus}; use huginn_mind::{Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery}; @@ -118,6 +119,9 @@ pub fn runtime_id(instance: &Slug) -> String { format!("huginn-{}", instance.0) } +/// What a desk read is told until follow-up desk-door-reads builds the reads and deletes this. +const DESK_READS_UNBUILT: &str = "desk reads not built"; + /// One mind and whatever indexes what lands in it. pub struct Daemon> { mind: Mind, @@ -231,6 +235,30 @@ impl> Daemon { Handled::Answered(answer) } + /// One desk request, answered by the Mind method it names. The Mind is the instance's own + /// (`mind` is checked against it); `deciders` is this call's list, handed to the Mind and + /// dropped. A fresh answer reaches the index as `Admit`'s writes do; a replay wrote nothing. + /// The reads wait on follow-up desk-door-reads and are refused until they exist. + pub fn handle_desk(&mut self, request: DeskRequest, now: DateTime) -> DeskResponse { + if let Err(refusal) = self.mind.require_instance(request.mind()) { + return DeskResponse::Refused(refusal); + } + match request { + DeskRequest::Answer { subject, authority, deciders, request, .. } => { + let (response, replayed) = self.mind.desk_answer(&deciders, subject, authority, request, now); + if !replayed + && let DeskResponse::Answered { answer, .. } = &response + && let Some(stored) = answer_ref(answer) + && let Err(error) = self.index.committed(&self.mind, std::slice::from_ref(&stored)) + { + eprintln!("huginn: the index refused the landed write [{}]: {error:#}", stored.id.0); + } + response + } + DeskRequest::Forks { .. } | DeskRequest::Fork { .. } => DeskResponse::Refused(MindRefusal::Unavailable { detail: DESK_READS_UNBUILT.into() }), + } + } + /// A semantic query starts only if the mind accepts it, and the mind /// decides its members before the index is asked anything. A selection /// with no members is answered at once: nothing is embedded and the index @@ -665,11 +693,11 @@ pub(crate) mod tests { Ok(Daemon::new(Mind::open(state_root, instance)?, NoIndex)) } - type Seen = Vec<(Vec, Vec)>; + pub(crate) type Seen = Vec<(Vec, Vec)>; #[derive(Clone, Default)] - struct RecordingIndex { - seen: Arc>, + pub(crate) struct RecordingIndex { + pub(crate) seen: Arc>, } impl IndexSink for RecordingIndex { diff --git a/crates/huginn-daemon/src/desk_door.rs b/crates/huginn-daemon/src/desk_door.rs new file mode 100644 index 0000000..b967351 --- /dev/null +++ b/crates/huginn-daemon/src/desk_door.rs @@ -0,0 +1,275 @@ +//! The desk door: the second door onto the one mind, a unix socket one peer may speak on. +//! +//! This module owns three things. Who may speak: the kernel's report of the connecting +//! process's uid must equal the configured one, and it is asked before a byte is read, so a +//! refused peer has nothing read and nothing written. The socket's life: bind, replace a stale +//! one, refuse a live one or a file that is not a socket, mode 0660, unlink on drop. And the +//! bounds: one request per connection, one reply, a header over `DESK_MAX_REQUEST_BYTES` refused +//! before anything is allocated, `DESK_IO_TIMEOUT` on every read and write, a reply past +//! `DESK_MAX_RESPONSE_BYTES` refused by name. +//! +//! It owns no rule. Which Mind method answers is `Daemon::handle_desk`'s; when the door is +//! served is `serve::run`'s, once per pass in the thread that owns the Mind. A peer that is +//! slow costs the loop at most `DESK_ACCEPTS_PER_PASS` connections of `2 * DESK_IO_TIMEOUT` +//! each, and only the configured peer can reach that. +//! +//! The door keeps no request id: a retry after a lost reply is a new connection with the same +//! request, and the stored answer is what replays it, across a restart as well. + +use std::fmt; +use std::fs; +use std::io::{self, Read, Write}; +use std::os::fd::AsFd; +use std::os::unix::fs::{FileTypeExt, PermissionsExt}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Context, Result, bail, ensure}; +use chrono::{DateTime, Utc}; +use cultnet_rs::{ + CultNetMessage, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry, CultNetTransportProfile, CultNetWireContract, + TcpFramedTransportConnection, TcpFramedTransportProfileOptions, create_tcp_framed_transport_profile, decode_cultnet_message_from_slice, + encode_cultnet_message_to_vec, +}; +use huginn_mind::desk_wire::{ + DESK_MAX_REQUEST_BYTES, DESK_MAX_RESPONSE_BYTES, DESK_REQUEST_SCHEMA, DESK_REQUEST_SCHEMA_JSON, DESK_RESPONSE_SCHEMA, + DESK_RESPONSE_SCHEMA_JSON, DESK_SERVICE_ID, DeskResponse, +}; +use huginn_mind::envelope::{OperationFailure, decode_desk_request, encode_desk_failure, encode_desk_response}; +use huginn_mind::{MindRefusal, MindStore}; + +use crate::daemon::{Daemon, IndexSink}; +use crate::serve::{MAX_ECHOED_FIELD_BYTES, overlong_echo}; + +/// How long the door waits on one read or one write of the configured peer. +pub const DESK_IO_TIMEOUT: Duration = Duration::from_millis(500); + +/// Connections one loop pass serves, so a peer that connects in a stream cannot hold the RUDP door off. +pub const DESK_ACCEPTS_PER_PASS: u32 = 4; + +/// The listening socket, and the one uid that may speak on it. +pub struct DeskDoor { + listener: UnixListener, + path: PathBuf, + peer_uid: u32, + registry: CultNetSchemaRegistry, +} + +/// What became of one connection. `Display` is the log line for the ones that are not answered. +#[derive(Debug, PartialEq, Eq)] +pub enum Served { + Answered, + /// The peer's uid is not the configured one: nothing was read or written. + Refused { uid: u32, expected: u32 }, + /// The connection ended without a reply, and why. + Dropped(String), +} + +impl fmt::Display for Served { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Answered => formatter.write_str("desk door answered a peer"), + Self::Refused { uid, expected } => write!(formatter, "desk door refused a peer with uid {uid} (expects {expected})"), + Self::Dropped(why) => write!(formatter, "desk door dropped a connection: {why}"), + } + } +} + +/// The two desk contracts, as the door's catalog advertises them and nothing else. +pub fn desk_registry() -> Result { + let mut registry = CultNetSchemaRegistry::new(); + for (schema_id, title, schema_json) in [ + (DESK_REQUEST_SCHEMA, "Huginn desk request", DESK_REQUEST_SCHEMA_JSON), + (DESK_RESPONSE_SCHEMA, "Huginn desk response", DESK_RESPONSE_SCHEMA_JSON), + ] { + registry.register(CultNetSchemaRegistration { + schema_id: schema_id.to_string(), + kind: CultNetSchemaKind::WireMessage, + wire_contracts: vec![CultNetWireContract::CultNetSchemaV0], + schema_version: Some(schema_id.to_string()), + document_type: None, + title: Some(title.to_string()), + schema_json: Some(schema_json.to_string()), + })?; + } + Ok(registry) +} + +/// Binds the desk socket. A path that exists and is not a socket is refused; a socket that +/// accepts a connection is a live server and is refused; a socket that refuses the connection is +/// stale and is replaced (`Mind::open`'s exclusive lock has already shown no live daemon of this +/// instance). The socket is then mode 0660 and non-blocking. +pub fn bind_desk(path: &Path, peer_uid: u32) -> Result { + ensure!(path.is_absolute(), "the desk socket path must be absolute, got {}", path.display()); + ensure!(peer_uid != 0, "the desk peer uid must not be 0"); + match fs::symlink_metadata(path) { + Ok(found) => { + ensure!(found.file_type().is_socket(), "{} exists and is not a socket", path.display()); + match UnixStream::connect(path) { + Ok(_) => bail!("{} is a live socket; another server holds it", path.display()), + Err(error) if error.kind() == io::ErrorKind::ConnectionRefused => { + fs::remove_file(path).with_context(|| format!("removing the stale socket {}", path.display()))?; + } + Err(error) => return Err(error).with_context(|| format!("probing the socket {}", path.display())), + } + } + Err(error) if error.kind() == io::ErrorKind::NotFound => {} + Err(error) => return Err(error).with_context(|| format!("reading {}", path.display())), + } + let listener = UnixListener::bind(path).with_context(|| format!("binding {}", path.display()))?; + // The door owns the path from here, so a failure below still unlinks it on drop. + let door = DeskDoor { listener, path: path.to_path_buf(), peer_uid, registry: desk_registry()? }; + fs::set_permissions(path, fs::Permissions::from_mode(0o660)).with_context(|| format!("setting the mode of {}", path.display()))?; + door.listener.set_nonblocking(true)?; + Ok(door) +} + +impl DeskDoor { + pub fn path(&self) -> &Path { + &self.path + } + + /// Serves at most `DESK_ACCEPTS_PER_PASS` waiting connections, each to completion, and says + /// how many it took. A refusal or a dropped connection is logged and counted: the loop was + /// busy with it. + pub fn serve_ready>(&mut self, daemon: &mut Daemon, now: DateTime) -> u32 { + let mut taken = 0; + while taken < DESK_ACCEPTS_PER_PASS { + let stream = match self.listener.accept() { + Ok((stream, _)) => stream, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => break, + Err(error) => { + eprintln!("huginn: the desk socket could not accept: {error}"); + break; + } + }; + taken += 1; + let served = serve_connection(stream, self.peer_uid, daemon, &self.registry, now, |stream| { + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(DESK_IO_TIMEOUT))?; + stream.set_write_timeout(Some(DESK_IO_TIMEOUT)) + }); + if served != Served::Answered { + eprintln!("huginn: {served}"); + } + } + taken + } +} + +impl Drop for DeskDoor { + fn drop(&mut self) { + let _ = fs::remove_file(&self.path); + } +} + +fn profile() -> CultNetTransportProfile { + create_tcp_framed_transport_profile( + "huginn-desk", + TcpFramedTransportProfileOptions { max_payload_bytes: Some(DESK_MAX_REQUEST_BYTES as u32), ..Default::default() }, + ) +} + +/// One connection: the peer first, before anything is read, then one frame in, one frame out. +/// `ready` readies the stream for blocking, bounded I/O and runs only after the peer is known to +/// be the configured one. +pub fn serve_connection( + stream: St, + peer_uid: u32, + daemon: &mut Daemon, + registry: &CultNetSchemaRegistry, + now: DateTime, + ready: impl FnOnce(&St) -> io::Result<()>, +) -> Served +where + St: Read + Write + AsFd, + S: MindStore, + I: IndexSink, +{ + let uid = match rustix::net::sockopt::socket_peercred(&stream) { + Ok(credentials) => credentials.uid.as_raw(), + Err(error) => return Served::Dropped(format!("the peer's credentials could not be read: {error}")), + }; + if uid != peer_uid { + return Served::Refused { uid, expected: peer_uid }; + } + if let Err(error) = ready(&stream) { + return Served::Dropped(format!("the stream could not be readied: {error}")); + } + let mut transport = TcpFramedTransportConnection::new(stream, profile()); + let frame = match transport.receive() { + Ok(frame) => frame, + Err(error) => return Served::Dropped(format!("no whole frame arrived: {error:#}")), + }; + let message = match decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) { + Ok(message) => message, + Err(error) => return Served::Dropped(format!("a frame was not a CultNet message: {error:#}")), + }; + let reply = answer_desk_message(daemon, registry, message, now); + let bytes = match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) { + Ok(bytes) => bytes, + Err(error) => return Served::Dropped(format!("the reply could not be encoded: {error:#}")), + }; + match transport.send("schema", &bytes) { + Ok(()) => Served::Answered, + Err(error) => Served::Dropped(format!("the reply was not delivered: {error:#}")), + } +} + +/// One message in, one message out. Nothing here reaches a mind except through +/// `Daemon::handle_desk`, and an envelope that does not decode never gets that far. +pub fn answer_desk_message>( + daemon: &mut Daemon, + registry: &CultNetSchemaRegistry, + message: CultNetMessage, + now: DateTime, +) -> CultNetMessage { + let runtime_id = daemon.runtime_id(); + if let Some(field) = overlong_echo(&message) { + return CultNetMessage::Error { + error: format!("{DESK_SERVICE_ID} refuses a {field} over {MAX_ECHOED_FIELD_BYTES} bytes"), + code: None, + details: None, + }; + } + match &message { + CultNetMessage::OperationRequest { message_id, operation, .. } => match decode_desk_request(&message) { + Ok((message_id, request)) => { + let operation = request.operation(); + let response = daemon.handle_desk(request, now); + respond(&response, &message_id, operation, &runtime_id, DESK_MAX_RESPONSE_BYTES) + } + Err(failure) => encode_desk_failure(message_id, operation, &failure, &runtime_id), + }, + CultNetMessage::SchemaCatalogRequest { .. } => match registry.create_catalog_response(&message) { + Ok(response) => response, + Err(error) => CultNetMessage::Error { error: format!("{error:#}"), code: None, details: None }, + }, + _ => CultNetMessage::Error { + error: format!("{DESK_SERVICE_ID} answers cultnet.operation_request.v0 and cultnet.schema_catalog_request.v0"), + code: None, + details: None, + }, + } +} + +/// One answer as its message, or, past `limit` encoded, the refusal that names the size. +pub(crate) fn respond(response: &DeskResponse, message_id: &str, operation: &str, runtime_id: &str, limit: usize) -> CultNetMessage { + let not_encodable = |detail: String| { + encode_desk_failure(message_id, operation, &OperationFailure { code: "response-not-encodable".into(), message: detail }, runtime_id) + }; + let reply = match encode_desk_response(message_id, operation, response, runtime_id) { + Ok(reply) => reply, + Err(error) => return not_encodable(format!("{error:#}")), + }; + let size = match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) { + Ok(bytes) => bytes.len(), + Err(error) => return not_encodable(format!("{error:#}")), + }; + if size <= limit { + return reply; + } + let refusal = DeskResponse::Refused(MindRefusal::ResponseTooLarge { bytes: size as u64, limit: limit as u64 }); + encode_desk_response(message_id, operation, &refusal, runtime_id).unwrap_or_else(|error| not_encodable(format!("{error:#}"))) +} diff --git a/crates/huginn-daemon/src/desk_door_tests.rs b/crates/huginn-daemon/src/desk_door_tests.rs index 50dc557..d4e691f 100644 --- a/crates/huginn-daemon/src/desk_door_tests.rs +++ b/crates/huginn-daemon/src/desk_door_tests.rs @@ -79,7 +79,7 @@ fn wire(hash: &str, request_id: &str) -> ForkAnswerRequest { answer: Some(ForkAnswer { choice: Some("A".into()), note: None }), retracted: false, retracts: None, - answered_at: "2026-11-10T12:00:00Z".into(), + answered_at: "2026-09-16T11:00:00Z".into(), request_id: request_id.into(), } } @@ -137,7 +137,7 @@ fn failure_of(path: &Path, bytes: &[u8]) -> OperationFailure { fn counted(response: DeskResponse) -> (String, Option) { match response { DeskResponse::Answered { answer, counted_answer_id, .. } => { - assert_eq!(answer.outcome, "counted", "{answer:?}"); + assert_eq!(answer.outcome, "Counted", "{answer:?}"); (answer.answer_id, counted_answer_id) } other => panic!("expected an answer, got {other:?}"), @@ -298,11 +298,9 @@ mod desk_door { let registry = desk_registry().unwrap(); let catalog = CultNetMessage::SchemaCatalogRequest { message_id: "c-1".into(), - include_schema_json: false, + include_schema_json: Some(true), schema_ids: None, kinds: None, - source_runtime_id: None, - target_runtime_id: None, }; let reply = answer_desk_message(&mut daemon, ®istry, catalog, now()); let text = format!("{reply:?}"); @@ -329,7 +327,7 @@ mod door_smoke { let hash = rig.hash.clone(); let response = rig.serving(|path| reply_of(path, "m-1", &answer("x", DeskAuthority::Operator, &["x", "y"], wire(&hash, R1)))); let DeskResponse::Answered { answer, counted_answer_id, .. } = &response else { panic!("{response:?}") }; - assert_eq!((answer.outcome.as_str(), counted_answer_id.as_deref(), answer.authority), ("counted", Some("n1"), DeskAuthority::Operator)); + assert_eq!((answer.outcome.as_str(), counted_answer_id.as_deref(), answer.authority), ("Counted", Some("n1"), DeskAuthority::Operator)); let seen = rig.index.seen.lock().unwrap().clone(); assert_eq!(seen.len(), 1, "the fresh answer reached the index once"); let stored = huginn_mind::eureka_pipeline::PipelineRef { @@ -490,8 +488,8 @@ mod door_smoke { options_hash: Sha256Hex(hash.clone()), choice: Some(short("A")), note: None, - answered_at: short("2026-11-10T12:00:00Z"), - received_at: short("2026-11-10T12:00:00Z"), + answered_at: short("2026-09-16T11:00:00Z"), + received_at: short("2026-09-16T11:00:00Z"), claim_in_future: false, request_id: RequestId(R1.into()), body_sha256: Sha256Hex("0".repeat(64)), @@ -520,7 +518,7 @@ mod door_smoke { let response = rig.serving(|path| reply_of(path, "m-1", &answer("x", DeskAuthority::Operator, &["x"], wire(&"0".repeat(64), R1)))); let DeskResponse::Answered { answer, counted_answer_id, .. } = response else { panic!("a stale answer is stored") }; assert_eq!(answer.code, Some(ForkAnswerCode::StaleFork)); - assert_eq!(answer.outcome, "refused"); + assert_eq!(answer.outcome, "Refused"); assert_eq!(counted_answer_id, None); // The reply type has no field a card could ride in; the encoded reply holds nothing but the answer. assert_eq!(answer.options_hash, "0".repeat(64)); @@ -545,11 +543,11 @@ mod door_smoke { }); let (first, other_list, again, empty) = out; let DeskResponse::Answered { answer, .. } = &first else { panic!("{first:?}") }; - assert_eq!((answer.outcome.as_str(), answer.authority), ("counted", DeskAuthority::Operator)); + assert_eq!((answer.outcome.as_str(), answer.authority), ("Counted", DeskAuthority::Operator)); let DeskResponse::Answered { answer, .. } = &other_list else { panic!("{other_list:?}") }; assert_eq!(answer.code, Some(ForkAnswerCode::NotADecider), "the second list does not name x"); let DeskResponse::Answered { answer, .. } = &again else { panic!("{again:?}") }; - assert_eq!(answer.outcome, "revision", "nothing was kept between calls: the first list counts again"); + assert_eq!(answer.outcome, "Revision", "nothing was kept between calls: the first list counts again"); assert_eq!((empty.code.as_str(), empty.message.contains("deciders")), ("payload-not-a-request", true)); assert_eq!(rig.state().0, before.0 + 3, "three answers landed: two counted, one refused"); } diff --git a/crates/huginn-daemon/src/lib.rs b/crates/huginn-daemon/src/lib.rs index 294f239..8e69882 100644 --- a/crates/huginn-daemon/src/lib.rs +++ b/crates/huginn-daemon/src/lib.rs @@ -15,12 +15,17 @@ pub mod bodies; pub mod daemon; +pub mod desk_door; pub mod index; pub mod serve; pub mod sources; +#[cfg(test)] +mod desk_door_tests; + #[cfg(test)] mod semantic_tests; pub use daemon::{Daemon, Found, Handled, Hits, IndexSink, Search, SearchTicket, runtime_id}; -pub use serve::{Options, ServeOptions, parse_options, run, startup}; +pub use desk_door::DeskDoor; +pub use serve::{Options, ServeOptions, Started, parse_options, run, startup}; diff --git a/crates/huginn-daemon/src/main.rs b/crates/huginn-daemon/src/main.rs index 91c49d8..8b27ff0 100644 --- a/crates/huginn-daemon/src/main.rs +++ b/crates/huginn-daemon/src/main.rs @@ -9,13 +9,13 @@ use anyhow::{Context, Result}; use huginn_daemon::index::Backoff; use huginn_daemon::index::ollama::pool; use huginn_daemon::index::qdrant::QdrantIndex; -use huginn_daemon::serve::{ServeOptions, parse_options, run, startup}; +use huginn_daemon::serve::{ServeOptions, Started, parse_options, run, startup}; fn main() -> Result<()> { let options = parse_options(std::env::args().skip(1))?; let embedder = pool(&options.ollama_urls, &options.reserve_ollama_urls, &options.embedding_model); let index = QdrantIndex::new(&options.qdrant_url); - let (mut daemon, mut hub, registry) = match startup(&options, embedder, index, Backoff::default()) { + let Started { mut daemon, mut hub, registry, mut desk } = match startup(&options, embedder, index, Backoff::default()) { Ok(started) => started, Err(error) => { eprintln!("huginn-daemon refuses to serve {}: {error:#}", options.instance.0); @@ -33,5 +33,8 @@ fn main() -> Result<()> { } eprintln!("huginn-daemon serving {} on {}", options.instance.0, hub.local_addr()?); - run(&mut daemon, &mut hub, ®istry, &stopping, &ServeOptions::default()) + if let Some(door) = &desk { + eprintln!("huginn-daemon serving the desk door on {}", door.path().display()); + } + run(&mut daemon, &mut hub, ®istry, desk.as_mut(), &stopping, &ServeOptions::default()) } diff --git a/crates/huginn-daemon/src/semantic_tests.rs b/crates/huginn-daemon/src/semantic_tests.rs index 4d46b90..321b446 100644 --- a/crates/huginn-daemon/src/semantic_tests.rs +++ b/crates/huginn-daemon/src/semantic_tests.rs @@ -176,7 +176,7 @@ impl Harness { let stopping = Arc::new(AtomicBool::new(false)); let loop_stopping = Arc::clone(&stopping); let options = ServeOptions::default(); - let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, ®istry, &loop_stopping, &options)); + let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &options)); Self { endpoint, stopping, serving: Some(serving), embedder, index, _root: root } } diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index e5e9a9f..22c27df 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -79,6 +79,7 @@ use crate::bodies::{ DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, }; use crate::daemon::{Daemon, Handled, IndexSink, Search, SearchTicket, runtime_id}; +use crate::desk_door::{DeskDoor, bind_desk}; use huginn_mind::envelope::{OperationFailure, decode_request, encode_failure, encode_response}; use huginn_mind::MAX_DEFERRED_BODY_BYTES; use crate::index::{Backoff, Embedder, VectorIndex, WorkerSink}; @@ -118,12 +119,22 @@ pub struct Options { pub embedding_model: String, /// The forge a persona's activity read goes to; absent, that read is refused. pub forge_url: Option, + /// The desk door: the unix socket Bifrost's receiver answers on, and the one uid that may + /// speak on it. Absent, the daemon serves the RUDP door alone. + pub desk: Option, +} + +/// Where the desk door listens and who may speak there. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DeskOptions { + pub socket: PathBuf, + pub peer_uid: u32, } /// Exactly `--state-root`, `--instance`, `--bind`, `--qdrant-url`, /// `--ollama-url` and `--embedding-model`, each required, each once, except /// `--ollama-url`, which repeats for each GPU host, and the optional -/// `--reserve-ollama-url`, which repeats for each CPU host, and the optional `--forge-url`, once. The URLs are plain +/// `--reserve-ollama-url`, which repeats for each CPU host, and the optional `--forge-url`, once, and the optional pair `--desk-socket` (an absolute path) and `--desk-peer-uid` (a uid other than 0), both or neither, each once. The URLs are plain /// `http://`: this build carries no TLS. pub fn parse_options(args: impl Iterator) -> Result { let mut args = args; @@ -132,7 +143,7 @@ pub fn parse_options(args: impl Iterator) -> Result { while let Some(name) = args.next() { let name = name.strip_prefix("--").with_context(|| format!("expected --option, got {name:?}"))?; ensure!( - matches!(name, "state-root" | "instance" | "bind" | "qdrant-url" | "ollama-url" | "reserve-ollama-url" | "embedding-model" | "forge-url"), + matches!(name, "state-root" | "instance" | "bind" | "qdrant-url" | "ollama-url" | "reserve-ollama-url" | "embedding-model" | "forge-url" | "desk-socket" | "desk-peer-uid"), "unsupported Huginn option --{name}" ); let value = args.next().with_context(|| format!("missing value for --{name}"))?; @@ -168,6 +179,17 @@ pub fn parse_options(args: impl Iterator) -> Result { reserve_ollama_urls, embedding_model, forge_url: values.contains_key("forge-url").then(|| url("forge-url")).transpose()?, + desk: match (values.get("desk-socket"), values.get("desk-peer-uid")) { + (None, None) => None, + (Some(socket), Some(uid)) => { + let socket = PathBuf::from(socket); + ensure!(socket.is_absolute(), "--desk-socket must be an absolute path"); + let peer_uid: u32 = uid.parse().with_context(|| format!("--desk-peer-uid must be a uid, got {uid:?}"))?; + ensure!(peer_uid != 0, "--desk-peer-uid must not be 0"); + Some(DeskOptions { socket, peer_uid }) + } + _ => anyhow::bail!("--desk-socket and --desk-peer-uid are given together or not at all"), + }, }) } @@ -177,18 +199,14 @@ pub fn parse_options(args: impl Iterator) -> Result { /// of the refusal and nothing reaches the vector store for a process that /// never listens. The worker's first act is to reconcile the collection /// against every indexable document the mind holds now. -pub fn startup( - options: &Options, - embedder: E, - index: V, - backoff: Backoff, -) -> Result<(Daemon, CultNetRudpServerHub, CultNetSchemaRegistry)> +pub fn startup(options: &Options, embedder: E, index: V, backoff: Backoff) -> Result where E: Embedder + Send + 'static, V: VectorIndex + crate::sources::SourceIndex + Send + 'static, { let mind = Mind::open(&options.state_root, &options.instance)?; let hub = bind(options.bind, &runtime_id(&options.instance))?; + let desk = options.desk.as_ref().map(|desk| bind_desk(&desk.socket, desk.peer_uid)).transpose()?; let startup_entries = mind.index_entries(None)?; let sink = WorkerSink::spawn(embedder, index, &options.instance, mind.genesis_receipt_id()?, startup_entries, backoff); let daemon = Daemon::new(mind, sink); @@ -196,7 +214,16 @@ where Some(url) => daemon.with_forge(crate::sources::forge::HttpForge::new(url)), None => daemon, }; - Ok((daemon, hub, schema_registry()?)) + Ok(Started { daemon, hub, registry: schema_registry()?, desk }) +} + +/// What `startup` hands the loop: the daemon over the opened mind, the bound RUDP hub, the +/// catalog the RUDP door advertises, and the desk door when one was asked for. +pub struct Started { + pub daemon: Daemon, + pub hub: CultNetRudpServerHub, + pub registry: CultNetSchemaRegistry, + pub desk: Option, } /// The bytes one reliable packet carries, and the packets one session may hold @@ -255,7 +282,7 @@ pub const MAX_ECHOED_FIELD_BYTES: usize = 256; /// The first field of a request that a reply would echo and that is over /// `MAX_ECHOED_FIELD_BYTES`. -fn overlong_echo(message: &CultNetMessage) -> Option<&'static str> { +pub(crate) fn overlong_echo(message: &CultNetMessage) -> Option<&'static str> { let echoed: Vec<(&'static str, &str)> = match message { CultNetMessage::OperationRequest { message_id, service_id, operation, payload_schema, .. } => { vec![ @@ -479,6 +506,7 @@ pub fn run>( daemon: &mut Daemon, hub: &mut CultNetRudpServerHub, registry: &CultNetSchemaRegistry, + mut desk: Option<&mut DeskDoor>, stopping: &AtomicBool, options: &ServeOptions, ) -> Result<()> { @@ -523,6 +551,7 @@ pub fn run>( } served += 1; } + let desk_served = desk.as_deref_mut().map_or(0, |door| door.serve_ready(daemon, Utc::now())); let finished = collect_searches(daemon, &waiting); let answered = !finished.is_empty(); for (ticket, found) in finished { @@ -531,7 +560,7 @@ pub fn run>( let reply = respond(&response, &asked.message_id, asked.operation, &daemon.runtime_id(), &mut bodies, Utc::now()); send(hub, &asked.session, &reply); } - if served == 0 && !answered { + if served == 0 && desk_served == 0 && !answered { std::thread::sleep(options.idle_sleep); } } @@ -756,7 +785,7 @@ mod tests { let loop_stopping = Arc::clone(&stopping); let serving = std::thread::spawn(move || { - let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()); + let result = run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default()); (result, daemon) }); @@ -863,7 +892,7 @@ mod tests { let stopping = Arc::new(AtomicBool::new(false)); let loop_stopping = Arc::clone(&stopping); let serving = std::thread::spawn(move || { - let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()); + let result = run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default()); (result, daemon) }); @@ -1104,7 +1133,7 @@ mod tests { let stopping = Arc::new(AtomicBool::new(false)); let loop_stopping = Arc::clone(&stopping); let serving = std::thread::spawn(move || { - let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()); + let result = run(&mut daemon, &mut hub, ®istry, None, &loop_stopping, &ServeOptions::default()); (result, daemon) }); @@ -1541,6 +1570,7 @@ mod tests { reserve_ollama_urls: Vec::new(), embedding_model: "a-model".into(), forge_url: None, + desk: None, }; let started = |options: &Options| { let (embedder, index) = crate::index::fakes::pair(); @@ -1556,8 +1586,8 @@ mod tests { let error = format!("{:#}", started(&options).err().expect("a held port refuses")); assert!(error.contains("binding"), "{error}"); drop(occupied); - let (_daemon, hub, _registry) = started(&options).expect("both gates open"); - assert_eq!(hub.local_addr().unwrap(), port); + let started = started(&options).expect("both gates open"); + assert_eq!(started.hub.local_addr().unwrap(), port); } #[test] @@ -1579,6 +1609,7 @@ mod tests { reserve_ollama_urls: Vec::new(), embedding_model: "qwen3-embedding:0.6b".into(), forge_url: None, + desk: None, } ); // Each of the six is required: there is no index-less mode. diff --git a/crates/huginn-mind/src/desk_wire.rs b/crates/huginn-mind/src/desk_wire.rs index 028f424..05db840 100644 --- a/crates/huginn-mind/src/desk_wire.rs +++ b/crates/huginn-mind/src/desk_wire.rs @@ -15,7 +15,7 @@ use chrono::{DateTime, Utc}; use eureka_pipeline::{ - DeskAuthority, DeskOutcome, DeskRefusal, PipelineDeskAnswer, PipelineDocument, PipelineKind, PipelineQuestion, RequestId, Sha256Hex, Short, Slug, Words, + DeskAuthority, DeskOutcome, DeskRefusal, PipelineDeskAnswer, PipelineDocument, PipelineKind, PipelineQuestion, PipelineRef, RequestId, Sha256Hex, Short, Slug, Words, }; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -330,6 +330,12 @@ fn answer_key(fork: &str, n: u32) -> String { format!("{}{}", prefix(fork), n_text(n)) } +/// The record a projected answer was stored under: the ref the index is told about when the +/// answer is fresh. `None` for an id outside the `n` form, which no stored answer has. +pub fn answer_ref(answer: &ForkDeskAnswer) -> Option { + Some(PipelineRef { kind: PipelineKind::DeskAnswer, id: Short(answer_key(&answer.fork_id, parse_n(&answer.answer_id)?)) }) +} + /// The sequence a key of `fork`'s answers names; `None` for another fork's key or any other text. fn answer_seq(fork: &str, key: &str) -> Option { parse_n(key.strip_prefix(prefix(fork).as_str())?)