furnace-server 0.15.0

//! Building a fabric: neurons, receptors, effectors, and the pathways between them

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::Instant;

use super::nerve::{Lane, Nerve};
use super::pathway::{Pathway, PathwayKind, Summation};
use super::summing::Summing;
use super::{Cell, Counters, Fabric, Inner};
use crate::nerve::effector::Effector;
use crate::nerve::latency::LatencyEstimate;
use crate::nerve::motor::{ActionSpec, Motion, MotionError, Motor, Taps};
use crate::nerve::neuron::{Gate, Mechanisms, Neuron, NeuronKind};
use crate::nerve::receptor::{Receptor, Stop};
use crate::nerve::signal::{NeuronId, PathwayId};
use amate::Engine;

/// Why a part of the nervous system cannot be built
#[derive(Debug)]
#[non_exhaustive]
pub enum BuildError {
    /// Two neurons have the same name; effectors and motor neurons are addressed by name
    DuplicateName(String),
    /// An action was declared on a neuron that takes no commands: only motor neurons and
    /// effectors do
    NotAnAction(String),
    /// A pathway names a neuron that was not added to this builder
    UnknownNeuron {
        /// The pathway's source
        from: NeuronId,
        /// The pathway's target
        to: NeuronId,
    },
    /// A URL the nervous system's HTTP client cannot call: not `http://host[:port]/path`
    /// (HTTPS among them)
    Url {
        /// The effector or receptor
        name: String,
        /// What is wrong with the URL
        reason: String,
    },
    /// A runtime of the fabric could not be started
    Runtime(&'static str, std::io::Error),
}

impl std::fmt::Display for BuildError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::DuplicateName(name) => write!(f, "two neurons are named '{name}'"),
            Self::NotAnAction(name) => write!(
                f,
                "'{name}' is declared an action, but only motor neurons and effectors take commands"
            ),
            Self::UnknownNeuron { from, to } => {
                write!(
                    f,
                    "pathway {from} -> {to} names a neuron that was not added"
                )
            }
            Self::Url { name, reason } => write!(f, "{name}: {reason}"),
            Self::Runtime(which, e) => write!(f, "the {which} runtime: {e}"),
        }
    }
}

impl std::error::Error for BuildError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Runtime(_, e) => Some(e),
            _ => None,
        }
    }
}

/// Builds a [`Fabric`]: add neurons, connect them, build
pub struct FabricBuilder {
    neurons: Vec<(Arc<dyn Neuron>, Mechanisms)>,
    receptors: Vec<(NeuronId, Arc<dyn Receptor>)>,
    pathways: Vec<(NeuronId, NeuronId, PathwayKind, f32, Option<Summation>)>,
    actions: Vec<(NeuronId, String, serde_json::Value)>,
    nerves: Vec<Nerve>,
    /// The nerve each neuron was added in, by the neuron's index
    nerve_of: Vec<Option<usize>>,
    /// The nerve whose closure is running
    current_nerve: Option<usize>,
    efferent_threads: usize,
    efferent: Option<tokio::runtime::Handle>,
}

impl Default for FabricBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl FabricBuilder {
    /// An empty builder
    pub fn new() -> Self {
        Self {
            neurons: Vec::new(),
            receptors: Vec::new(),
            pathways: Vec::new(),
            actions: Vec::new(),
            nerves: Vec::new(),
            nerve_of: Vec::new(),
            current_nerve: None,
            efferent_threads: 2,
            efferent: None,
        }
    }

    /// Threads of the motor side's runtime (2 by default)
    pub fn efferent_threads(mut self, threads: usize) -> Self {
        self.efferent_threads = threads.max(1);
        self
    }

    /// Run the motor side (effectors' calls, retries, backoff) on a runtime of your own, the
    /// server's say, instead of one the fabric starts. The runtime must outlive the fabric;
    /// [`efferent_threads`](Self::efferent_threads) no longer applies.
    pub fn efferent(mut self, runtime: tokio::runtime::Handle) -> Self {
        self.efferent = Some(runtime);
        self
    }

    /// Add a neuron, with no mechanisms in front of it ([`gate`](Self::gate) adds them).
    /// Names must be unique, since effectors are addressed by name in commands:
    /// [`build`](Self::build) refuses two of the same name.
    pub fn neuron(&mut self, neuron: impl Neuron) -> NeuronId {
        self.add(Arc::new(neuron))
    }

    /// [`FabricBuilder::neuron`] for a neuron already shared
    pub(crate) fn add(&mut self, neuron: Arc<dyn Neuron>) -> NeuronId {
        let id = NeuronId(u32::try_from(self.neurons.len()).expect("fewer than 4G neurons"));
        self.neurons.push((neuron, Mechanisms::NONE));
        self.nerve_of.push(self.current_nerve);
        id
    }

    /// A nerve: the trunk to one part of the body. Every neuron `parts` adds (motor
    /// neurons, effectors, receptors, reflexes) belongs to it. What enters the nerve from
    /// outside takes its lane, a few at a time and the most urgent first, so a part that is
    /// flooded holds up nothing else; what stays inside (the hand commanding its fingers, a
    /// fingertip's pressure reaching the hand) never waits there. `nerve` is a [`Nerve`] or
    /// just its name.
    ///
    /// ```no_run
    /// # use furnace::nerve::prelude::*;
    /// # use serde_json::json;
    /// # let mut b = Fabric::builder();
    /// let grasp = b.nerve("right_arm", |b| {
    ///     b.effector(Webhook::new("grip", "http://127.0.0.1:9000/grip")?);
    ///     Ok::<_, BuildError>(b.motor("grasp", |m| async move {
    ///         m.command("grip", json!({"force": 0.7})).await
    ///     }))
    /// })?;
    /// # Ok::<(), BuildError>(())
    /// ```
    pub fn nerve<R>(&mut self, nerve: impl Into<Nerve>, parts: impl FnOnce(&mut Self) -> R) -> R {
        self.nerves.push(nerve.into());
        let outer = self.current_nerve.replace(self.nerves.len() - 1);
        let parts = parts(self);
        self.current_nerve = outer;
        parts
    }

    /// The spine of `brain`: the tracts between it and the body. Ascending, what a receptor
    /// or a reflex should tell the brain ([`Spine::ascend`]); descending, the actions it may
    /// call ([`Spine::action`]), whose outcomes and questions ascend to it too.
    pub fn spine(&mut self, brain: NeuronId) -> Spine<'_> {
        Spine { b: self, brain }
    }

    /// Add a receptor: a neuron that is also started with the fabric and may be mounted as a
    /// route. Habituation belongs here ([`gate`](Self::gate)).
    pub fn receptor<R: Receptor>(&mut self, receptor: R) -> NeuronId {
        let receptor: Arc<R> = Arc::new(receptor);
        let neuron: Arc<dyn Neuron> = receptor.clone();
        let id = self.add(neuron);
        self.receptors.push((id, receptor));
        id
    }

    /// Add an effector: an [`Act`](crate::nerve::effector::Act) with the default local loop
    /// (`b.effector(Webhook::new(..)?)`), or an [`Effector`] configured first. Commands
    /// reach it by its name from any neuron, with or without a pathway to it.
    pub fn effector(&mut self, effector: impl Into<Effector>) -> NeuronId {
        self.neuron(effector.into())
    }

    /// Add a motor neuron: commands addressed to `name` start `program` on the motor side,
    /// which carries them out by commanding the motor neurons and effectors below it (see
    /// [`Motor`]). Commands reach it by its name from any neuron, as they reach effectors.
    pub fn motor<F, Fut>(&mut self, name: &str, program: F) -> NeuronId
    where
        F: Fn(Motion) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = Result<serde_json::Value, MotionError>> + Send + 'static,
    {
        self.neuron(Motor::new(name, program))
    }

    /// Offer the motor neuron or effector `id` to brains as an action: what it does and the
    /// shape of its arguments (a JSON Schema, or an example), for the prompt. A brain sees
    /// these and nothing else, so the parts below an action (the fingers under a hand) stay
    /// out of its sight. [`build`](Self::build) refuses a neuron that takes no commands.
    ///
    /// # Panics
    ///
    /// For an id this builder did not hand out.
    pub fn action(
        &mut self,
        id: NeuronId,
        description: &str,
        args: serde_json::Value,
    ) -> &mut Self {
        assert!(
            id.index() < self.neurons.len(),
            "{id} is not in this builder"
        );
        self.actions.push((id, description.to_string(), args));
        self
    }

    /// Put `mechanisms` (threshold, habituation, refractory period) in front of neuron `id`
    ///
    /// # Panics
    ///
    /// For an id this builder did not hand out.
    pub fn gate(&mut self, id: NeuronId, mechanisms: Mechanisms) -> &mut Self {
        self.neurons[id.index()].1 = mechanisms;
        self
    }

    /// The id the next pathway added will have
    pub(crate) fn next_pathway(&self) -> PathwayId {
        PathwayId(u32::try_from(self.pathways.len()).expect("fewer than 4G pathways"))
    }

    /// A reflex pathway: `to` is called directly when `from` fires. Weight 1 unless
    /// [`Link::weight`] says otherwise.
    pub fn reflex(&mut self, from: NeuronId, to: NeuronId) -> Link<'_> {
        self.link(from, to, PathwayKind::Reflex, REFLEX_WEIGHT, None)
    }

    /// A deliberative pathway: signals from `from` queue to a worker that sums them into
    /// firings of `to`. Weight 2, so it wins over a reflex pathway whenever the signal's
    /// deadline allows it, and the default [`Summation`]; [`Link`] changes either.
    pub fn deliberative(&mut self, from: NeuronId, to: NeuronId) -> Link<'_> {
        self.link(
            from,
            to,
            PathwayKind::Deliberative,
            DELIBERATIVE_WEIGHT,
            Some(Summation::default()),
        )
    }

    fn link(
        &mut self,
        from: NeuronId,
        to: NeuronId,
        kind: PathwayKind,
        weight: f32,
        summation: Option<Summation>,
    ) -> Link<'_> {
        self.pathways.push((from, to, kind, weight, summation));
        Link {
            pathway: self.pathways.last_mut().expect("just pushed"),
        }
    }

    /// Wire the fabric and start its runtimes and its deliberative workers. An error for two
    /// neurons of the same name, a pathway to a neuron of another builder, an action that
    /// takes no commands, or a runtime that cannot be started.
    pub fn build(self) -> Result<Fabric, BuildError> {
        let count = self.neurons.len();
        let mut names = std::collections::HashSet::new();
        for (neuron, _) in &self.neurons {
            if !names.insert(neuron.name()) {
                return Err(BuildError::DuplicateName(neuron.name().to_string()));
            }
        }
        drop(names);
        for &(from, to, ..) in &self.pathways {
            if from.index() >= count || to.index() >= count {
                return Err(BuildError::UnknownNeuron { from, to });
            }
        }
        let mut actions = Vec::with_capacity(self.actions.len());
        let mut is_action = vec![false; count];
        for (id, description, args) in self.actions {
            is_action[id.index()] = true;
            let neuron = &self.neurons[id.index()].0;
            if !matches!(neuron.kind(), NeuronKind::Motor | NeuronKind::Effector) {
                return Err(BuildError::NotAnAction(neuron.name().to_string()));
            }
            actions.push(ActionSpec {
                name: neuron.name().to_string(),
                description,
                args,
            });
        }
        let (efferent, efferent_handle) = match self.efferent {
            Some(handle) => (None, handle),
            None => {
                let runtime = tokio::runtime::Builder::new_multi_thread()
                    .worker_threads(self.efferent_threads)
                    .thread_name("furnace-efferent")
                    .enable_all()
                    .build()
                    .map_err(|e| BuildError::Runtime("efferent", e))?;
                let handle = runtime.handle().clone();
                (Some(runtime), handle)
            }
        };
        let afferent = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(1)
            .thread_name("furnace-afferent")
            .enable_all()
            .build()
            .map_err(|e| BuildError::Runtime("afferent", e))?;
        let mut by_name = HashMap::new();
        let mut cells: Vec<Cell> = self
            .neurons
            .into_iter()
            .enumerate()
            .map(|(i, (neuron, mechanisms))| {
                by_name.insert(neuron.name().to_string(), NeuronId(i as u32));
                let kind = neuron.kind();
                Cell {
                    arcs: if kind == NeuronKind::Effector {
                        (0..count).map(|_| OnceLock::new()).collect()
                    } else {
                        Vec::new()
                    },
                    neuron,
                    kind,
                    gate: Gate::new(mechanisms),
                    counters: Counters::default(),
                    out: Vec::new(),
                    last_prepared: AtomicU64::new(0),
                    taps: Taps::default(),
                    nerve: self.nerve_of[i],
                    action: is_action[i],
                }
            })
            .collect();
        let inner = Arc::new_cyclic(|weak: &Weak<Inner>| {
            let pathways: Vec<Pathway> = self
                .pathways
                .into_iter()
                .enumerate()
                .map(|(i, (from, to, kind, weight, summation))| {
                    let id = PathwayId(i as u32);
                    cells[from.index()].out.push(id);
                    let queue = summation.map(|s| {
                        Engine::start(
                            Summing {
                                fabric: weak.clone(),
                                pathway: id,
                            },
                            amate::Config {
                                max_batch: s.max_batch,
                                workers: s.workers,
                                capacity: Some(s.capacity),
                                waits_unannounced: !s.announced_only,
                            },
                            move |_| format!("furnace-nerve-{i}"),
                        )
                    });
                    Pathway {
                        id,
                        from,
                        to,
                        kind,
                        weight,
                        latency: LatencyEstimate::new(),
                        signals: AtomicU64::new(0),
                        ineligible: AtomicU64::new(0),
                        overflow: AtomicU64::new(0),
                        summation,
                        queue,
                    }
                })
                .collect();
            Inner {
                epoch: Instant::now(),
                cells,
                pathways,
                by_name,
                receptors: self.receptors,
                actions,
                nerves: self.nerves.into_iter().map(Lane::new).collect(),
                started: AtomicBool::new(false),
                received: AtomicU64::new(0),
                reached_brain: AtomicU64::new(0),
                unrouted: AtomicU64::new(0),
                no_eligible: AtomicU64::new(0),
                loops: AtomicU64::new(0),
                commands_direct: AtomicU64::new(0),
                stop: Arc::new(Stop::default()),
                efferent_handle,
                efferent: Mutex::new(efferent),
                afferent_handle: afferent.handle().clone(),
                afferent: Mutex::new(Some(afferent)),
            }
        });
        Ok(Fabric(inner))
    }
}

/// The weight of a reflex pathway unless [`Link::weight`] changes it
const REFLEX_WEIGHT: f32 = 1.0;
/// The weight of a deliberative pathway unless [`Link::weight`] changes it: above a
/// reflex's, so the brain is chosen whenever there is time for it
const DELIBERATIVE_WEIGHT: f32 = 2.0;

/// A pathway just added, to change its weight or summation
pub struct Link<'a> {
    pathway: &'a mut (NeuronId, NeuronId, PathwayKind, f32, Option<Summation>),
}

impl Link<'_> {
    /// Priority among the eligible pathways from the same neuron: the heaviest wins
    pub fn weight(self, weight: f32) -> Self {
        self.pathway.3 = weight;
        self
    }

    /// How a deliberative pathway sums its queue into firings. On a reflex pathway it makes
    /// the pathway deliberative.
    pub fn summation(self, summation: Summation) -> Self {
        self.pathway.2 = PathwayKind::Deliberative;
        self.pathway.4 = Some(summation);
        self
    }
}

/// A brain's spine, from [`FabricBuilder::spine`]: the tracts between the brain and the
/// body
pub struct Spine<'a> {
    b: &'a mut FabricBuilder,
    brain: NeuronId,
}

impl Spine<'_> {
    /// The ascending tract from `from` (a receptor, a reflex): what passes its gate and its
    /// reflexes goes on to the brain over a deliberative pathway, summed into the brain's
    /// firings
    pub fn ascend(&mut self, from: NeuronId) -> &mut Self {
        self.b.deliberative(from, self.brain);
        self
    }

    /// The descending tract to `id`, a motor neuron or an effector: the brain may call it,
    /// and is told what it does and the shape of its arguments
    /// ([`FabricBuilder::action`]). A motor neuron's questions and how its programs ended
    /// ascend to the brain.
    pub fn action(
        &mut self,
        id: NeuronId,
        description: &str,
        args: serde_json::Value,
    ) -> &mut Self {
        self.b.action(id, description, args);
        if self.b.neurons[id.index()].0.kind() == NeuronKind::Motor {
            self.b.deliberative(id, self.brain);
        }
        self
    }
}