Merge Cut 13a: envelope to huginn-mind, eureka-state client core

Soul pass found no blocker; its four findings (whole-call deadline,
answer correlation, Rejected vs Unavailable, disconnect) are fixed at
3dc50ee with mutation-killed tests.

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

AGENTS.md

@@ -18,12 +18,14 @@ Studio's job, not Huginn's.
   status included, over `cultcache-rs` and `epiphany-pipeline` pinned by git
   rev. `huginn-daemon` serves one mind over CultNet RUDP through `cultnet-rs`
   at the same rev, and reaches the leaf and the store type through
-  `huginn-mind` so one crate pins each revision. `eureka-state` is a stub.
+  `huginn-mind` so one crate pins each revision. `eureka-state` is the client
+  core: one call to a daemon over CultNet, holding no state.
 - Owned: an instance's memory documents, their admission and the read side
   that derives their status, in `huginn-mind`,
   over a redb CultCache store at `<state_root>/minds/<instance>/mind.redb`;
   and their CultNet surface in `huginn-daemon`, which owns the socket, the
-  sessions, the process and the operation envelope, and no rule. The two wire
+  sessions and the process, and no rule; the operation envelope is
+  `huginn-mind`'s. The two wire
   schemas are published from `schemas/cultnet/`. There is no index yet.
 - To depend on: Qdrant, directly, once retrieval exists. Unreachable Qdrant is
   to be a loud refusal, never a fallback store. No crate connects to it yet.

Cargo.lock

@@ -335,6 +335,15 @@ dependencies = [
 [[package]]
 name = "eureka-state"
 version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "chrono",
+ "cultnet-rs",
+ "huginn-daemon",
+ "huginn-mind",
+ "rmp-serde",
+ "tempfile",
+]
 
 [[package]]
 name = "fastrand"
@@ -477,7 +486,6 @@ dependencies = [
  "cultnet-rs",
  "huginn-mind",
  "rmp-serde",
- "schemars",
  "serde",
  "serde_json",
  "sha2",
@@ -491,6 +499,7 @@ name = "huginn-mind"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "base64",
  "chrono",
  "cultcache-rs",
  "cultnet-rs",

README.md

@@ -42,10 +42,11 @@ A Cargo workspace of three crates:
 
 - `crates/huginn-mind`: storage, identity, admission, and queries and derived
   status over memory documents.
-- `crates/huginn-daemon`: the CultNet surface. The socket, the sessions, the
-  process and the operation envelope, and no rule.
-- `crates/eureka-state`: a stub. It will carry typed state for the Eureka
-  pipeline; today it holds none.
+- `crates/huginn-daemon`: the CultNet surface. The socket, the sessions and
+  the process, and no rule. The operation envelope both sides speak is
+  `huginn-mind`'s.
+- `crates/eureka-state`: the client core, `HuginnClient::call`: one request to a
+  daemon over CultNet RUDP, a deferred answer resolved, and no state.
 
 `huginn-mind` is live: it opens one instance's store (an owned redb CultCache
 at `<state_root>/minds/<instance>/mind.redb`, locked for the mind's lifetime),
@@ -71,7 +72,7 @@ failure and reaches no mind. The two schemas it publishes live in
 `schemas/cultnet/` and are pinned to their derivation by a test. There is no
 index yet.
 
-`eureka-state` is a stub; the campaign's cut map in
+`eureka-state` has no MCP surface yet; the campaign's cut map in
 `Epiphany/notes/eureka-pipeline-state-cut.md` owns what each crate must do
 next.
 

crates/eureka-state/Cargo.toml

@@ -5,4 +5,19 @@ edition.workspace = true
 license.workspace = true
 publish.workspace = true
 
+# The client core: one call over CultNet to a Huginn daemon. It owns the
+# transport and nothing of the mind. The operation envelope, the wire
+# vocabulary and the deferred-body bound are `huginn-mind`'s, so the client and
+# the daemon cannot spell them differently; the content fetch and its
+# verification are `cultnet-rs`'s.
 [dependencies]
+anyhow = "1"
+cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" }
+huginn-mind = { path = "../huginn-mind" }
+rmp-serde = "1"
+
+# Tests run a real daemon in-process against a temporary mind.
+[dev-dependencies]
+chrono = "0.4.44"
+huginn-daemon = { path = "../huginn-daemon" }
+tempfile = "3"

crates/eureka-state/src/lib.rs

@@ -0,0 +1,185 @@
+//! `eureka-state`, the client core: one call to a Huginn daemon.
+//!
+//! `HuginnClient::call` sends one `HuginnMindRequest` in CultNet's operation
+//! envelope over RUDP, waits for the answer, and returns it as the mind's own
+//! `HuginnMindResponse`. An answer too large for one send arrives as a
+//! `Deferred` manifest; the client fetches the body on the same session,
+//! verifies it, and decodes it, so the caller sees the answer the mind gave
+//! and never the deferral.
+//!
+//! The client owns the transport and no state. A refusal is an answer, so it
+//! comes back as `Ok(HuginnMindResponse::Refused(..))`. `ClientError` is the
+//! only error, and it means the daemon could not be reached or did not answer
+//! in the shape the protocol promises.
+//!
+//! Every call is its own session: it connects, asks, fetches what was
+//! deferred, and disconnects, whether the call succeeded or failed. Nothing is
+//! kept between calls.
+
+use std::fmt;
+use std::net::SocketAddr;
+use std::time::{Duration, Instant};
+
+use anyhow::{Context, Result, anyhow, bail};
+use cultnet_rs::{
+    CULTNET_OPERATION_CONNECTION_ID, CultMesh, CultMeshRudpSocketOptions, CultNetMessage,
+    CultNetRudpSocketTransportConnection, fetch_content,
+};
+use huginn_mind::envelope::{OperationFailure, decode_response, encode_request};
+use huginn_mind::epiphany_pipeline::Slug;
+use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES};
+
+/// The correlation key of the one request a session carries.
+const MESSAGE_ID: &str = "eureka-state-call";
+const POLL: Duration = Duration::from_millis(2);
+
+/// Why a call produced no answer. `Unavailable` is the transport's failure and
+/// the only one worth retrying: the daemon could not be reached, went quiet
+/// past the call's deadline, or answered out of protocol. `Rejected` is the
+/// daemon refusing the envelope itself, which a retry of the same call cannot
+/// change.
+#[derive(Debug)]
+pub enum ClientError {
+    Unavailable { endpoint: SocketAddr, detail: String },
+    Rejected { endpoint: SocketAddr, code: String, detail: String },
+}
+
+impl fmt::Display for ClientError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Unavailable { endpoint, detail } => {
+                write!(f, "the Huginn daemon at rudp://{endpoint} is unavailable: {detail}")
+            }
+            Self::Rejected { endpoint, code, detail } => {
+                write!(f, "the Huginn daemon at rudp://{endpoint} rejected the request: {code}: {detail}")
+            }
+        }
+    }
+}
+
+impl std::error::Error for ClientError {}
+
+/// What `exchange` can fail with, before the endpoint is attached.
+enum Failure {
+    Transport(anyhow::Error),
+    Rejected(OperationFailure),
+}
+
+impl From<anyhow::Error> for Failure {
+    fn from(error: anyhow::Error) -> Self {
+        Self::Transport(error)
+    }
+}
+
+/// A daemon's address, the instance the caller acts for, and the time one whole
+/// call may take: the connection, the answer, and every chunk of a deferred
+/// body together must finish within `timeout` of the call starting.
+#[derive(Clone, Debug)]
+pub struct HuginnClient {
+    endpoint: SocketAddr,
+    instance: Slug,
+    timeout: Duration,
+}
+
+impl HuginnClient {
+    pub fn new(endpoint: SocketAddr, instance: Slug, timeout: Duration) -> Self {
+        Self { endpoint, instance, timeout }
+    }
+
+    /// The instance this client acts for. Requests name their own instance;
+    /// this is the one the caller's tools fill in.
+    pub fn instance(&self) -> &Slug {
+        &self.instance
+    }
+
+    pub fn call(&self, request: HuginnMindRequest) -> Result<HuginnMindResponse, ClientError> {
+        let deadline = Instant::now() + self.timeout;
+        let unavailable =
+            |error: anyhow::Error| ClientError::Unavailable { endpoint: self.endpoint, detail: format!("{error:#}") };
+        let mut session = self.connect(deadline).map_err(unavailable)?;
+        let outcome = self.exchange(&mut session, &request, deadline);
+        // Best effort: the daemon drops a session it never hears from, only later.
+        let _ = session.disconnect(Vec::new());
+        outcome.map_err(|failure| match failure {
+            Failure::Transport(error) => unavailable(error),
+            Failure::Rejected(rejection) => ClientError::Rejected {
+                endpoint: self.endpoint,
+                code: rejection.code,
+                detail: rejection.message,
+            },
+        })
+    }
+
+    fn exchange(
+        &self,
+        session: &mut CultNetRudpSocketTransportConnection,
+        request: &HuginnMindRequest,
+        deadline: Instant,
+    ) -> Result<HuginnMindResponse, Failure> {
+        let reply = self.ask(session, &encode_request(MESSAGE_ID, request, None)?, deadline)?;
+        let CultNetMessage::OperationResponse { operation, .. } = &reply else {
+            return Err(anyhow!("the daemon answered with something other than an operation response").into());
+        };
+        if operation != request.operation() {
+            return Err(anyhow!("the daemon answered {operation}, not the {} that was asked", request.operation()).into());
+        }
+        let (message_id, answer) = decode_response(&reply)?;
+        if message_id != MESSAGE_ID {
+            return Err(anyhow!("the daemon answered request {message_id}, not {MESSAGE_ID}").into());
+        }
+        match answer.map_err(Failure::Rejected)? {
+            HuginnMindResponse::Deferred(deferred) => {
+                let body = fetch_content(&deferred.manifest, MAX_DEFERRED_BODY_BYTES, |chunk| {
+                    self.ask(session, &chunk, deadline)
+                })?;
+                match rmp_serde::from_slice::<HuginnMindResponse>(&body).context("the deferred body is not a response")? {
+                    HuginnMindResponse::Deferred(_) => Err(anyhow!("a deferred body is itself deferred").into()),
+                    answered => Ok(answered),
+                }
+            }
+            answered => Ok(answered),
+        }
+    }
+
+    fn connect(&self, deadline: Instant) -> Result<CultNetRudpSocketTransportConnection> {
+        let mut session = CultMesh::create_rudp_client_for_endpoint(
+            format!("eureka-state-{}", self.instance.0),
+            CULTNET_OPERATION_CONNECTION_ID,
+            &format!("rudp://{}", self.endpoint),
+            CultMeshRudpSocketOptions::default(),
+        )?;
+        session.connect(Vec::new())?;
+        loop {
+            session.poll_resends()?;
+            let _ = session.receive_once()?;
+            if session.connected() {
+                return Ok(session);
+            }
+            if Instant::now() >= deadline {
+                bail!("no accept before the call's {:?} deadline", self.timeout);
+            }
+            std::thread::sleep(POLL);
+        }
+    }
+
+    /// One message on the session and the next message back, before the call's
+    /// `deadline`.
+    fn ask(
+        &self,
+        session: &mut CultNetRudpSocketTransportConnection,
+        message: &CultNetMessage,
+        deadline: Instant,
+    ) -> Result<CultNetMessage> {
+        session.send_schema_message(message)?;
+        loop {
+            session.poll_resends()?;
+            if let Some(reply) = session.receive_schema_message_once()? {
+                return Ok(reply);
+            }
+            if Instant::now() >= deadline {
+                bail!("no answer before the call's {:?} deadline", self.timeout);
+            }
+            std::thread::sleep(POLL);
+        }
+    }
+}

crates/eureka-state/tests/client.rs

@@ -0,0 +1,546 @@
+//! The client against a real daemon, in-process on a loopback port over a
+//! temporary mind, and against scripted servers where the daemon cannot be
+//! made to misbehave.
+
+use std::net::{SocketAddr, UdpSocket};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+use chrono::Utc;
+use cultnet_rs::{
+    CultNetMessage, CultNetRudpServerEvent, CultNetWireContract, Selection, answer_content_chunk_request,
+    decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, pack_content,
+};
+use eureka_state::{ClientError, HuginnClient};
+use huginn_daemon::serve::{MAX_RESPONSE_BYTES, bind, run, schema_registry};
+use huginn_daemon::{Daemon, Handled, Hits, IndexSink, SearchTicket, ServeOptions};
+use huginn_mind::envelope::{OperationFailure, encode_failure, encode_response};
+use huginn_mind::epiphany_pipeline::{
+    AuthorityMap, CodeLocation, CutDelete, CutVerification, Date, DocRef, FileChange, Line, NegativeCheck, OrgRepo,
+    PipelineCampaign, PipelineCutSpec, PipelineDocument, PipelineInstance, PipelineKind, PipelineRef,
+    PipelineStewardship, Sha, Short, Slug, StructuralDelta, Title, VerificationTest,
+};
+use huginn_mind::{
+    DeferredAnswer, Faculty, HuginnMindRequest, HuginnMindResponse, IndexStatus, Mind, MindRefusal, MindStore,
+    OwnedRedbMessagePackBackingStore, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineProvenance,
+};
+use tempfile::TempDir;
+
+const INSTANCE: &str = "eureka";
+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),
+        repo: OrgRepo(REPO.into()),
+        sequence: 1,
+        assigned_on: Date("2026-09-29".into()),
+        note: "assigned".into(),
+    })
+}
+
+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<PipelineDocument>) -> 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}"))
+}
+
+fn short(n: usize) -> Short {
+    Short(format!("{n:0>200}"))
+}
+
+/// A cut spec with every list the leaf bounds filled: at 256 file changes a
+/// megabyte of field content, more than one send carries.
+fn cut_spec(cut: &str, file_changes: usize) -> PipelineDocument {
+    let lines = |count: usize| (0..count).map(line).collect::<Vec<_>>();
+    let shorts = |count: usize| (0..count).map(short).collect::<Vec<_>>();
+    PipelineDocument::CutSpec(PipelineCutSpec {
+        campaign: slug(CAMPAIGN),
+        cut: cut.into(),
+        revision: 1,
+        title: Title(short(0).0),
+        repo: OrgRepo(REPO.into()),
+        branch: short(1),
+        base: sha(),
+        depends_on: vec![],
+        first: lines(16),
+        deletes: (0..64).map(|n| CutDelete { path: short(n), lines: 9, note: line(n) }).collect(),
+        keeps_moves: lines(64),
+        adds: lines(64),
+        file_changes: (0..file_changes)
+            .map(|n| FileChange {
+                location: CodeLocation { path: short(n), line: 1, end_line: Some(9) },
+                change: line(n),
+            })
+            .collect(),
+        authority_map: Some(AuthorityMap {
+            owner: line(0),
+            inputs: lines(16),
+            outputs: lines(16),
+            derived_state: lines(16),
+            forbidden_writers: lines(16),
+            shared_paths: lines(16),
+            deletion_line: line(1),
+        }),
+        verification: CutVerification {
+            builds: lines(64),
+            tests: (0..64).map(|n| VerificationTest { name: short(n), pins: line(n) }).collect(),
+            negative: (0..64).map(|n| NegativeCheck { pattern: short(n), scope: line(n) }).collect(),
+            operator: lines(64),
+        },
+        estimate: StructuralDelta {
+            lines_added: 0,
+            lines_removed: 0,
+            dependencies_added: shorts(64),
+            dependencies_removed: shorts(64),
+            formats_added: shorts(64),
+            formats_removed: shorts(64),
+            targets_added: shorts(64),
+            targets_removed: shorts(64),
+        },
+        rulings: vec![],
+        questions: vec![],
+    })
+}
+
+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<S: MindStore> IndexSink<S> for Inert {
+    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) -> Result<SearchTicket, String> {
+        Err("this index cannot search".into())
+    }
+    fn search_deadline(&self) -> Duration {
+        Duration::ZERO
+    }
+    fn searched(&mut self) -> Vec<(SearchTicket, Result<Hits, String>)> {
+        Vec::new()
+    }
+    fn abandon(&mut self, _ticket: SearchTicket) {}
+}
+
+type TestDaemon = Daemon<OwnedRedbMessagePackBackingStore, Inert>;
+
+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<Vec<PipelineDocument>>) -> (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<AtomicBool>,
+    thread: Option<JoinHandle<()>>,
+    /// The sessions a scripted server has seen end: what the client's
+    /// disconnect looks like from the daemon's side.
+    ended: Arc<AtomicUsize>,
+    _root: Option<TempDir>,
+}
+
+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, &registry, &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<u8>) -> Server {
+    deferring_in_chunks(body, 256 * 1024, Duration::ZERO)
+}
+
+/// `deferring`, with `chunk_bytes` per chunk and each chunk answered only after
+/// `drip`.
+fn deferring_in_chunks(body: Vec<u8>, chunk_bytes: usize, drip: Duration) -> Server {
+    let (manifest, chunks) = pack_content("art", "package", "", "", "", &body, chunk_bytes).expect("a body packs");
+    let deferred = encode_response(
+        "eureka-state-call",
+        "whoami",
+        &HuginnMindResponse::Deferred(DeferredAnswer { manifest }),
+        "scripted",
+    )
+    .unwrap();
+    scripted(move |message| match message {
+        CultNetMessage::OperationRequest { .. } => deferred.clone(),
+        chunk => {
+            std::thread::sleep(drip);
+            answer_content_chunk_request(chunk, |hash| {
+                chunks.iter().find(|c| c.chunk_hash == hash).map(|c| c.payload.as_slice())
+            })
+        }
+    })
+}
+
+fn client(addr: SocketAddr) -> HuginnClient {
+    HuginnClient::new(addr, slug(INSTANCE), TIMEOUT)
+}
+
+fn unavailable(error: ClientError) -> (SocketAddr, String) {
+    match error {
+        ClientError::Unavailable { endpoint, detail } => (endpoint, detail),
+        rejected @ ClientError::Rejected { .. } => panic!("expected Unavailable, got {rejected}"),
+    }
+}
+
+fn rejected(error: ClientError) -> (SocketAddr, String, String) {
+    match error {
+        ClientError::Rejected { endpoint, code, detail } => (endpoint, code, detail),
+        unavailable @ ClientError::Unavailable { .. } => panic!("expected Rejected, got {unavailable}"),
+    }
+}
+
+/// 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()
+}
+
+/// Wait for a scripted server to have seen `count` sessions end.
+fn ended_within(server: &Server, count: usize, wait: Duration) -> bool {
+    let until = Instant::now() + wait;
+    while server.ended.load(Ordering::Relaxed) < count {
+        if Instant::now() >= until {
+            return false;
+        }
+        std::thread::sleep(Duration::from_millis(10));
+    }
+    true
+}
+
+fn document_query(instance: &str) -> HuginnMindRequest {
+    HuginnMindRequest::Query {
+        instance: slug(instance),
+        selection: Selection { projection: "document".into(), ..Selection::default() },
+        semantic: None,
+    }
+}
+
+/// Every operation the wire has, asked of a live daemon over the socket, and
+/// each answered with the mind's own answer.
+#[test]
+fn every_operation_round_trips_against_a_live_daemon() {
+    let (root, daemon) = mind(vec![]);
+    let server = serve(root, daemon);
+    let client = client(server.addr);
+    assert_eq!(client.instance(), &slug(INSTANCE), "the client reports the instance it was configured for");
+
+    let HuginnMindResponse::Whoami(status) = client.call(HuginnMindRequest::Whoami).unwrap() else {
+        panic!("whoami is answered with a status");
+    };
+    assert_eq!((status.instance, status.documents), (slug(INSTANCE), 1));
+
+    let admitted = client.call(HuginnMindRequest::Admit(batch(INSTANCE, vec![stewardship(), campaign()]))).unwrap();
+    let HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { writes, .. }) = admitted else {
+        panic!("the batch is committed, got {admitted:?}");
+    };
+    assert_eq!(writes.len(), 2);
+
+    let HuginnMindResponse::Query(page) = client.call(document_query(INSTANCE)).unwrap() else {
+        panic!("a query is answered with a page");
+    };
+    assert_eq!(page.matched, 3, "the identity and the two admitted documents");
+
+    let campaign_ref = PipelineRef { kind: PipelineKind::Campaign, id: Short(format!("{CAMPAIGN}:campaign:self")) };
+    let viewed = client.call(HuginnMindRequest::View { instance: slug(INSTANCE), id: campaign_ref.clone() }).unwrap();
+    let HuginnMindResponse::View(Some(view)) = viewed else { panic!("the campaign is viewable, got {viewed:?}") };
+    assert_eq!(view.id, campaign_ref);
+}
+
+/// A refusal is data. A read that names another mind, a write that declares
+/// another mind, and a document the mind does not hold each come back as `Ok`.
+#[test]
+fn a_foreign_instance_is_an_answer_not_an_error() {
+    let (root, daemon) = mind(vec![]);
+    let server = serve(root, daemon);
+    let client = client(server.addr);
+    let foreign = MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() };
+
+    assert_eq!(client.call(document_query(OTHER)).unwrap(), HuginnMindResponse::Refused(foreign.clone()));
+    assert_eq!(
+        client.call(HuginnMindRequest::Admit(batch(OTHER, vec![identity(OTHER)]))).unwrap(),
+        HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(foreign)),
+    );
+    let absent = PipelineRef { kind: PipelineKind::CutSpec, id: Short(format!("{CAMPAIGN}:cut_spec:cut-none.r1")) };
+    assert_eq!(
+        client.call(HuginnMindRequest::View { instance: slug(INSTANCE), id: absent }).unwrap(),
+        HuginnMindResponse::View(None),
+    );
+}
+
+/// The wide cut spec's answer is over one send and arrives deferred; the
+/// narrow one arrives whole. Both come back equal to what the mind gave.
+#[test]
+fn a_deferred_answer_resolves_to_the_same_page_as_a_direct_one() {
+    let (root, mut daemon) =
+        mind(vec![vec![stewardship(), campaign()], vec![cut_spec("wide", 256), cut_spec("fits", 180)]]);
+    let measured = |request: &HuginnMindRequest, daemon: &mut TestDaemon| {
+        let answer = answered(daemon.handle(request.clone(), Utc::now()));
+        let envelope = encode_response("m", request.operation(), &answer, "huginn-eureka").unwrap();
+        let size = encode_cultnet_message_to_vec(&envelope, CultNetWireContract::CultNetSchemaV0).unwrap().len();
+        (answer, size as u64)
+    };
+    let wide = HuginnMindRequest::View { instance: slug(INSTANCE), id: cut_ref("wide") };
+    let fits = HuginnMindRequest::View { instance: slug(INSTANCE), id: cut_ref("fits") };
+    let (wide_answer, wide_size) = measured(&wide, &mut daemon);
+    let (fits_answer, fits_size) = measured(&fits, &mut daemon);
+    assert!(matches!(wide_answer, HuginnMindResponse::View(Some(_))), "{wide_answer:?}");
+    assert!(wide_size > MAX_RESPONSE_BYTES, "the wide answer must be over one send, or nothing was deferred");
+    assert!(fits_size <= MAX_RESPONSE_BYTES, "the narrow answer must fit one send");
+
+    let server = serve(root, daemon);
+    let client = client(server.addr);
+    assert_eq!(client.call(wide).unwrap(), wide_answer);
+    assert_eq!(client.call(fits).unwrap(), fits_answer);
+}
+
+/// Nothing listens on the port. The call fails naming the endpoint, and it
+/// fails when the configured timeout says: two timeouts, two waits.
+#[test]
+fn a_closed_port_is_unavailable_naming_the_endpoint_within_the_timeout() {
+    let closed = UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap();
+    let mut waited = Vec::new();
+    for timeout in [Duration::from_millis(300), Duration::from_millis(1200)] {
+        let started = Instant::now();
+        let error = HuginnClient::new(closed, slug(INSTANCE), timeout).call(HuginnMindRequest::Whoami).unwrap_err();
+        let elapsed = started.elapsed();
+        assert!(elapsed >= timeout, "gave up after {elapsed:?}, before its {timeout:?}");
+        assert!(elapsed < timeout + Duration::from_millis(800), "gave up after {elapsed:?}, long past its {timeout:?}");
+        assert!(error.to_string().contains(&closed.to_string()), "{error}");
+        let (endpoint, detail) = unavailable(error);
+        assert_eq!(endpoint, closed);
+        assert!(detail.contains("no accept") && detail.contains("deadline"), "{detail}");
+        waited.push(elapsed);
+    }
+    assert!(waited[1] > waited[0] + Duration::from_millis(500), "{waited:?}");
+}
+
+/// A daemon that accepts and then says nothing is a timeout on the answer,
+/// not on the connection.
+#[test]
+fn a_daemon_that_never_answers_is_unavailable_within_the_timeout() {
+    let server = scripted(|_| {
+        std::thread::sleep(Duration::from_secs(3));
+        CultNetMessage::Error { error: "late".into(), code: None, details: None }
+    });
+    let started = Instant::now();
+    let error = HuginnClient::new(server.addr, slug(INSTANCE), Duration::from_millis(400))
+        .call(HuginnMindRequest::Whoami)
+        .unwrap_err();
+    let elapsed = started.elapsed();
+    assert!(elapsed >= Duration::from_millis(400) && elapsed < Duration::from_secs(2), "{elapsed:?}");
+    let (endpoint, detail) = unavailable(error);
+    assert_eq!(endpoint, server.addr);
+    assert!(detail.contains("no answer"), "{detail}");
+}
+
+/// An envelope failure means the request never reached a mind. It is the
+/// daemon refusing, not a transport failure, so a caller that retries on
+/// `Unavailable` does not loop on it; the code and the message arrive intact.
+#[test]
+fn an_envelope_failure_is_rejected_and_carries_its_code() {
+    let server = answering(encode_failure(
+        "eureka-state-call",
+        "whoami",
+        &OperationFailure { code: "wrong-service".into(), message: "not huginn.mind".into() },
+        "scripted",
+    ));
+    let error = client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err();
+    assert!(error.to_string().contains("wrong-service"), "{error}");
+    let (endpoint, code, detail) = rejected(error);
+    assert_eq!((endpoint, code.as_str(), detail.as_str()), (server.addr, "wrong-service", "not huginn.mind"));
+}
+
+/// An answer to another request, or to another operation, is not this call's
+/// answer, whatever it says.
+#[test]
+fn an_answer_that_is_not_to_this_request_is_refused() {
+    let foreign = answering(nothing_found("someone-elses-call", "whoami"));
+    let (_, detail) = unavailable(client(foreign.addr).call(HuginnMindRequest::Whoami).unwrap_err());
+    assert!(detail.contains("someone-elses-call"), "{detail}");
+
+    let other_operation = answering(nothing_found("eureka-state-call", "query"));
+    let (_, detail) = unavailable(client(other_operation.addr).call(HuginnMindRequest::Whoami).unwrap_err());
+    assert!(detail.contains("query") && detail.contains("whoami"), "{detail}");
+
+    let matching = answering(nothing_found("eureka-state-call", "whoami"));
+    assert_eq!(client(matching.addr).call(HuginnMindRequest::Whoami).unwrap(), HuginnMindResponse::View(None));
+}
+
+/// The timeout bounds the whole call. Seven chunks each answered after 300 ms
+/// take two seconds; no single wait exceeds the 500 ms timeout, and the call
+/// still gives up at it.
+#[test]
+fn the_timeout_bounds_the_whole_call_not_each_wait() {
+    let server = deferring_in_chunks(vec![7_u8; 7 * 1024], 1024, Duration::from_millis(300));
+    let timeout = Duration::from_millis(500);
+    let started = Instant::now();
+    let error = HuginnClient::new(server.addr, slug(INSTANCE), timeout).call(HuginnMindRequest::Whoami).unwrap_err();
+    let elapsed = started.elapsed();
+    assert!(elapsed >= timeout && elapsed < Duration::from_millis(1100), "{elapsed:?}");
+    let (_, detail) = unavailable(error);
+    assert!(detail.contains("deadline"), "{detail}");
+}
+
+/// Every call ends its session, whether it answered, was rejected, or gave up.
+#[test]
+fn every_call_disconnects_its_session() {
+    let answered = answering(nothing_found("eureka-state-call", "whoami"));
+    client(answered.addr).call(HuginnMindRequest::Whoami).unwrap();
+    assert!(ended_within(&answered, 1, Duration::from_secs(5)), "an answered call left its session open");
+
+    let refused = answering(encode_failure(
+        "eureka-state-call",
+        "whoami",
+        &OperationFailure { code: "wrong-service".into(), message: "no".into() },
+        "scripted",
+    ));
+    rejected(client(refused.addr).call(HuginnMindRequest::Whoami).unwrap_err());
+    assert!(ended_within(&refused, 1, Duration::from_secs(5)), "a rejected call left its session open");
+
+    let silent = scripted(|_| {
+        std::thread::sleep(Duration::from_millis(800));
+        CultNetMessage::Error { error: "late".into(), code: None, details: None }
+    });
+    let error = HuginnClient::new(silent.addr, slug(INSTANCE), Duration::from_millis(300))
+        .call(HuginnMindRequest::Whoami)
+        .unwrap_err();
+    unavailable(error);
+    assert!(ended_within(&silent, 1, Duration::from_secs(5)), "a timed-out call left its session open");
+}
+
+/// A deferred body that is not a response, and one that is another deferral,
+/// are protocol violations named as such.
+#[test]
+fn a_deferred_body_must_be_a_response_and_not_another_deferral() {
+    let garbage = deferring(vec![0xc1; 40_000]);
+    let (_, detail) = unavailable(client(garbage.addr).call(HuginnMindRequest::Whoami).unwrap_err());
+    assert!(detail.contains("not a response"), "{detail}");
+
+    let (inner, _) = pack_content("inner", "package", "", "", "", b"x", 16).unwrap();
+    let again = rmp_serde::to_vec_named(&HuginnMindResponse::Deferred(DeferredAnswer { manifest: inner })).unwrap();
+    let nested = deferring(again);
+    let (_, detail) = unavailable(client(nested.addr).call(HuginnMindRequest::Whoami).unwrap_err());
+    assert!(detail.contains("itself deferred"), "{detail}");
+}
+
+/// A body over the deferral cap is refused before its first chunk is asked
+/// for, by the number the daemon defers under.
+#[test]
+fn a_body_over_the_deferral_cap_is_refused() {
+    let over = huginn_mind::MAX_DEFERRED_BODY_BYTES as usize + 1;
+    let server = deferring(vec![0_u8; over]);
+    let (_, detail) = unavailable(client(server.addr).call(HuginnMindRequest::Whoami).unwrap_err());
+    assert!(detail.contains("exceeds"), "{detail}");
+}

crates/huginn-daemon/Cargo.toml

@@ -24,7 +24,6 @@ chrono = "0.4.44"
 cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" }
 huginn-mind = { path = "../huginn-mind" }
 rmp-serde = "1"
-schemars = "1"
 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/bodies.rs

@@ -31,9 +31,6 @@ use cultnet_rs::{CultMeshCdnArtifactManifest, CultMeshCdnChunk};
 /// The size of one chunk of a deferred body. Four fit one send with room for a
 /// control-plane answer beside them on the same session.
 pub const DEFERRED_CHUNK_BYTES: usize = 256 * 1024;
-/// The largest answer that is deferred rather than refused, measured on the
-/// encoded payload.
-pub const MAX_DEFERRED_BODY_BYTES: u64 = 64 * 1024 * 1024;
 /// What the daemon keeps across all deferred bodies.
 pub const DEFERRED_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
 
@@ -163,6 +160,7 @@ impl DeferredBodies {
 mod tests {
     use super::*;
     use cultnet_rs::pack_content;
+    use huginn_mind::MAX_DEFERRED_BODY_BYTES;
 
     /// The three numbers a client author is told, spelled out: a chunk of
     /// 256 KiB, a largest deferred body of 64 MiB, a budget of 256 MiB that

crates/huginn-daemon/src/lib.rs

@@ -1,14 +1,12 @@
 //! `huginn-daemon`: one process that serves one mind over CultNet.
 //!
-//! The crate owns five things and no rule. `startup` owns "may this process
+//! The crate owns four things and no rule. `startup` owns "may this process
 //! serve this mind", by delegating to `Mind::open` and refusing before
 //! anything listens. `index` owns when the mind's documents are embedded and
 //! written to the vector store, off the serving path. `Daemon::handle` owns "which `Mind` method answers this
 //! request": it dispatches, it asks `Mind::require_instance` for every read
-//! that names an instance, and it compares nothing itself. `envelope` owns "is
-//! this a request at all": a message it cannot decode is answered with a typed
-//! failure and touches no mind. `serve` owns "which session gets which reply"
-//! and whether an answer is delivered whole, deferred or refused; `bodies`
+//! that names an instance, and it compares nothing itself. `serve` owns "which
+//! session gets which reply" and whether an answer is delivered whole, deferred or refused; `bodies`
 //! owns how long and how much of a deferred answer is held.
 //!
 //! Every admission rule, the instance check, the receipt, the derived status
@@ -17,7 +15,6 @@
 
 pub mod bodies;
 pub mod daemon;
-pub mod envelope;
 pub mod index;
 pub mod serve;
 
@@ -25,5 +22,4 @@ pub mod serve;
 mod semantic_tests;
 
 pub use daemon::{Daemon, Handled, Hits, IndexSink, Search, SearchTicket, runtime_id};
-pub use envelope::{FAILURE_SCHEMA, OperationFailure};
 pub use serve::{Options, ServeOptions, parse_options, run, startup};

crates/huginn-daemon/src/semantic_tests.rs

@@ -20,7 +20,7 @@ use tempfile::TempDir;
 
 use crate::daemon::Daemon;
 use crate::daemon::tests::{CAMPAIGN, INSTANCE, batch, campaign_seed, now, slug};
-use crate::envelope::{decode_response, encode_request};
+use huginn_mind::envelope::{decode_response, encode_request};
 use crate::index::fakes::{FakeEmbedder, FakeIndex, Gate};
 use crate::index::{Backoff, Described, Hit, WorkerSink, collection_name};
 use crate::serve::{ServeOptions, bind, run, schema_registry};

crates/huginn-daemon/src/serve.rs

@@ -76,10 +76,11 @@ use huginn_mind::wire::{
 use huginn_mind::{Mind, MindRefusal, MindStore, OwnedRedbMessagePackBackingStore};
 
 use crate::bodies::{
-    DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, MAX_DEFERRED_BODY_BYTES,
+    DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies,
 };
 use crate::daemon::{Daemon, Handled, IndexSink, Search, SearchTicket, runtime_id};
-use crate::envelope::{OperationFailure, decode_request, encode_failure, encode_response};
+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};
 
 /// What the loop does when nothing is waiting. A deferred body untouched for
@@ -345,7 +346,7 @@ fn encode_or_fail(
         Err(error) => encode_failure(
             message_id,
             operation,
-            &crate::envelope::OperationFailure {
+            &OperationFailure {
                 code: "response-not-encodable".into(),
                 message: format!("{error:#}"),
             },
@@ -548,7 +549,7 @@ mod tests {
         open_unindexed, seeded_wide, slug,
     };
     use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef};
-    use crate::envelope::{FAILURE_SCHEMA, decode_response, encode_request};
+    use huginn_mind::envelope::{FAILURE_SCHEMA, decode_response, encode_request};
     use crate::index::Backoff;
     use base64::Engine;
     use base64::engine::general_purpose::STANDARD;
@@ -632,7 +633,7 @@ mod tests {
             assert_eq!(rejected(&reply).1, code);
         }
         // The sixth code, `not-an-operation-request`, is the codec's and is
-        // pinned in `envelope::tests`: `answer` never produces it, because a
+        // pinned in `huginn_mind::envelope::tests`: `answer` never produces it, because a
         // message of another family is answered with `Error` instead.
         let reply = answer(&mut daemon, &registry, &mut fresh_bodies(), CultNetMessage::Error { error: "hello".into(), code: None, details: None }, now()).reply();
         let CultNetMessage::Error { error, .. } = &reply else { panic!("expected an error, got {reply:?}") };

crates/huginn-mind/Cargo.toml

@@ -15,6 +15,7 @@ publish.workspace = true
 # daemon's (Cut 10, Cut 11).
 [dependencies]
 anyhow = "1"
+base64 = "0.22"
 chrono = "0.4.44"
 cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultcache-rs" }
 cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "6251cbcd37c12ef3f0c1cc50063a03369225b7ac", package = "cultnet-rs" }

crates/huginn-mind/src/envelope.rs

@@ -14,7 +14,7 @@ use anyhow::Result;
 use base64::Engine;
 use base64::engine::general_purpose::STANDARD;
 use cultnet_rs::CultNetMessage;
-use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_RESPONSE_SCHEMA, MIND_SERVICE_ID};
+use crate::wire::{HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_RESPONSE_SCHEMA, MIND_SERVICE_ID};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
@@ -151,9 +151,9 @@ pub fn decode_response(message: &CultNetMessage) -> Result<(String, Result<Hugin
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::daemon::tests::{INSTANCE, OTHER, batch, identity, slug};
-    use huginn_mind::wire::MindStatus;
-    use huginn_mind::{MindRefusal, PipelineAdmissionOutcome};
+    use crate::fixtures::{INSTANCE, OTHER_INSTANCE as OTHER, instance, provenance, slug};
+    use crate::wire::MindStatus;
+    use crate::{Faculty, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome};
 
     fn refusal() -> MindRefusal {
         MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }
@@ -183,7 +183,7 @@ mod tests {
                     schema_epoch: "epiphany.pipeline.epoch.v2".into(),
                     documents: 0,
                     receipts: 0,
-                    index: huginn_mind::wire::IndexStatus::Current,
+                    index: crate::wire::IndexStatus::Current,
                 }),
                 "accepted",
             ),
@@ -214,7 +214,11 @@ mod tests {
     /// `Error` instead.
     #[test]
     fn a_request_round_trips_and_a_foreign_message_is_not_a_request() {
-        let request = HuginnMindRequest::Admit(batch(INSTANCE, vec![identity(INSTANCE)]));
+        let request = HuginnMindRequest::Admit(PipelineAdmissionBatch {
+            instance: slug(INSTANCE),
+            provenance: provenance(Faculty::Hands),
+            documents: vec![instance(INSTANCE)],
+        });
         let message = encode_request("m-3", &request, Some("eureka-state".into())).unwrap();
         assert_eq!(decode_request(&message).unwrap(), ("m-3".into(), request));
 

crates/huginn-mind/src/lib.rs

@@ -14,7 +14,8 @@
 //! selection (CultNet's, over rows this crate supplies) through the same
 //! `docs` the rules use; nothing is stored for it.
 //!
-//! The wire vocabulary is here too, so the daemon and the client share one set
+//! The wire vocabulary and its operation envelope (`envelope`) are here too,
+//! so the daemon and the client share one set
 //! of types.
 //!
 //! Document shape, bounds, formats and keys are `epiphany-pipeline`'s; the
@@ -26,6 +27,7 @@
 
 pub mod admission;
 mod docs;
+pub mod envelope;
 pub mod index;
 pub mod mind;
 mod rows;
@@ -50,7 +52,7 @@ pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvena
 pub use refusal::MindRefusal;
 pub use store::{MindStore, OwnedRedbMessagePackBackingStore};
 pub use wire::{
-    DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON,
+    DeferredAnswer, HuginnMindRequest, HuginnMindResponse, IndexStatus, MAX_DEFERRED_BODY_BYTES, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON,
     MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus,
 };
 

crates/huginn-mind/src/wire.rs

@@ -23,6 +23,11 @@ pub const MIND_SERVICE_ID: &str = "huginn.mind";
 pub const MIND_REQUEST_SCHEMA: &str = "huginn.mind_request.v1";
 pub const MIND_RESPONSE_SCHEMA: &str = "huginn.mind_response.v1";
 
+/// The largest answer that is deferred rather than refused, measured on the
+/// encoded payload. The daemon refuses past it and a client caps its fetch at
+/// it, so both sides read one number.
+pub const MAX_DEFERRED_BODY_BYTES: u64 = 64 * 1024 * 1024;
+
 /// The published JSON schema of each wire type, as the catalog advertises it.
 /// The files are derived from the types below and pinned byte for byte by
 /// `published_wire_schemas_match_derivation`, so the catalog cannot advertise