Desk door part A: the unix-socket door, peer uid checked before a byte is read, joined to the one serve loop

desk_door.rs owns who may speak, the socket's life and the frame bounds; Daemon::handle_desk owns which Mind method answers (Answer via desk_answer, the replay flag gating index.committed; Forks and Fork refused Unavailable until desk-door-reads). startup returns Started; run takes the door; its five callers pass it or None. desk_wire::answer_ref is the one spelling of a stored answer's ref.

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

Cargo.lock

@@ -935,6 +935,7 @@ dependencies = [
  "cultnet-rs 0.1.0",
  "huginn-mind",
  "rmp-serde",
+ "rustix",
  "serde",
  "serde_json",
  "sha2",

crates/eureka-state/tests/common/mod.rs

@@ -164,7 +164,7 @@ pub fn serve<I: IndexSink<OwnedRedbMessagePackBackingStore> + Send + 'static>(
     let loop_stopping = Arc::clone(&stopping);
     let registry = schema_registry().unwrap();
     let thread = std::thread::spawn(move || {
-        run(&mut daemon, &mut hub, &registry, &loop_stopping, &ServeOptions::default()).unwrap();
+        run(&mut daemon, &mut hub, &registry, None, &loop_stopping, &ServeOptions::default()).unwrap();
     });
     Server { addr, stopping, thread: Some(thread), ended: Arc::default(), _root: Some(root) }
 }

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

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<S: MindStore, I: IndexSink<S>> {
     mind: Mind<S>,
@@ -231,6 +235,30 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
         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<Utc>) -> 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<String>, Vec<bool>)>;
+    pub(crate) type Seen = Vec<(Vec<String>, Vec<bool>)>;
 
     #[derive(Clone, Default)]
-    struct RecordingIndex {
-        seen: Arc<Mutex<Seen>>,
+    pub(crate) struct RecordingIndex {
+        pub(crate) seen: Arc<Mutex<Seen>>,
     }
 
     impl<S: MindStore> IndexSink<S> for RecordingIndex {

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<CultNetSchemaRegistry> {
+    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<DeskDoor> {
+    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<S: MindStore, I: IndexSink<S>>(&mut self, daemon: &mut Daemon<S, I>, now: DateTime<Utc>) -> 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<St, S, I>(
+    stream: St,
+    peer_uid: u32,
+    daemon: &mut Daemon<S, I>,
+    registry: &CultNetSchemaRegistry,
+    now: DateTime<Utc>,
+    ready: impl FnOnce(&St) -> io::Result<()>,
+) -> Served
+where
+    St: Read + Write + AsFd,
+    S: MindStore,
+    I: IndexSink<S>,
+{
+    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<S: MindStore, I: IndexSink<S>>(
+    daemon: &mut Daemon<S, I>,
+    registry: &CultNetSchemaRegistry,
+    message: CultNetMessage,
+    now: DateTime<Utc>,
+) -> 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:#}")))
+}

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<String>) {
     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, &registry, 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");
     }

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};

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, &registry, &stopping, &ServeOptions::default())
+    if let Some(door) = &desk {
+        eprintln!("huginn-daemon serving the desk door on {}", door.path().display());
+    }
+    run(&mut daemon, &mut hub, &registry, desk.as_mut(), &stopping, &ServeOptions::default())
 }

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, &registry, &loop_stopping, &options));
+        let serving = std::thread::spawn(move || run(&mut daemon, &mut hub, &registry, None, &loop_stopping, &options));
         Self { endpoint, stopping, serving: Some(serving), embedder, index, _root: root }
     }
 

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<String>,
+    /// 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<DeskOptions>,
+}
+
+/// 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<Item = String>) -> Result<Options> {
     let mut args = args;
@@ -132,7 +143,7 @@ pub fn parse_options(args: impl Iterator<Item = String>) -> Result<Options> {
     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<Item = String>) -> Result<Options> {
         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<Item = String>) -> Result<Options> {
 /// 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<E, V>(
-    options: &Options,
-    embedder: E,
-    index: V,
-    backoff: Backoff,
-) -> Result<(Daemon<OwnedRedbMessagePackBackingStore, WorkerSink>, CultNetRudpServerHub, CultNetSchemaRegistry)>
+pub fn startup<E, V>(options: &Options, embedder: E, index: V, backoff: Backoff) -> Result<Started>
 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<OwnedRedbMessagePackBackingStore, WorkerSink>,
+    pub hub: CultNetRudpServerHub,
+    pub registry: CultNetSchemaRegistry,
+    pub desk: Option<DeskDoor>,
 }
 
 /// 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<S: MindStore, I: IndexSink<S>>(
     daemon: &mut Daemon<S, I>,
     hub: &mut CultNetRudpServerHub,
     registry: &CultNetSchemaRegistry,
+    mut desk: Option<&mut DeskDoor>,
     stopping: &AtomicBool,
     options: &ServeOptions,
 ) -> Result<()> {
@@ -523,6 +551,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
             }
             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<S: MindStore, I: IndexSink<S>>(
             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, &registry, &loop_stopping, &ServeOptions::default());
+            let result = run(&mut daemon, &mut hub, &registry, 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, &registry, &loop_stopping, &ServeOptions::default());
+            let result = run(&mut daemon, &mut hub, &registry, 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, &registry, &loop_stopping, &ServeOptions::default());
+            let result = run(&mut daemon, &mut hub, &registry, 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.

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<seq>` form, which no stored answer has.
+pub fn answer_ref(answer: &ForkDeskAnswer) -> Option<PipelineRef> {
+    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<u32> {
     parse_n(key.strip_prefix(prefix(fork).as_str())?)