use crate::app::{App, RecallRequest, WriteRequest};
use cyberbrain_core::{Error, NoteKind, Result};
use cyberbrain_policy::Actor;
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
pub const SOCKET: &str = "daemon.sock";
const PROTOCOL: u32 = 4;
#[cfg_attr(not(unix), allow(dead_code))]
const OLDEST_PROTOCOL: u32 = 3;
const MAX_SOCKET_PATH: usize = 100;
const MAX_REQUEST: usize = 1 << 20;
#[derive(Debug, Serialize, Deserialize)]
struct Request {
v: u32,
op: Op,
actor: String,
json: bool,
}
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "kebab-case")]
enum Op {
Recall {
query: String,
n: Option<usize>,
ring: Option<u8>,
bereich: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
at: Option<jiff::Timestamp>,
},
Write(WriteArgs),
Scan {
full: bool,
dry_run: bool,
},
}
#[derive(Debug, Serialize, Deserialize)]
struct WriteArgs {
ring: u8,
kind: NoteKind,
name: String,
body: String,
tags: Vec<String>,
bereich: Option<String>,
retention: Option<String>,
force: bool,
dry_run: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
supersedes: Option<Vec<String>>,
#[serde(
default,
skip_serializing_if = "Option::is_none",
deserialize_with = "present"
)]
valid_from: Option<Option<jiff::Timestamp>>,
#[serde(
default,
skip_serializing_if = "Option::is_none",
deserialize_with = "present"
)]
invalid_at: Option<Option<jiff::Timestamp>>,
}
fn present<'de, D, T>(d: D) -> std::result::Result<Option<Option<T>>, D::Error>
where
D: serde::Deserializer<'de>,
T: Deserialize<'de>,
{
Option::<T>::deserialize(d).map(Some)
}
#[derive(Debug, Default, Serialize, Deserialize)]
#[cfg_attr(not(unix), allow(dead_code))] struct Response {
code: Option<i32>,
stdout: Option<String>,
stderr: Option<String>,
reason: Option<String>,
}
pub enum Answer {
Done {
stdout: Option<String>,
stderr: Option<String>,
code: i32,
},
Local,
}
fn socket_path(root: &Path) -> Option<PathBuf> {
let p = root.join(SOCKET);
(p.as_os_str().len() <= MAX_SOCKET_PATH).then_some(p)
}
fn disabled() -> bool {
std::env::var_os("CYBERBRAIN_NO_DAEMON").is_some_and(|v| !v.is_empty() && v != "0")
}
#[cfg_attr(not(unix), allow(dead_code))]
fn actor_from(s: &str) -> Actor {
match s {
"operator" => Actor::Operator,
other => Actor::Agent(other.strip_prefix("agent:").unwrap_or(other).to_string()),
}
}
#[cfg_attr(not(unix), allow(dead_code))]
fn rendered<T: Serialize>(
value: &T,
json: bool,
human: impl FnOnce(&T) -> String,
) -> Result<String> {
if json {
serde_json::to_string_pretty(value)
.map_err(|e| Error::Index(format!("report does not serialise: {e}")))
} else {
Ok(human(value))
}
}
pub fn recall(
root: &Path,
actor: &Actor,
query: &str,
req: &RecallRequest,
json: bool,
) -> Option<String> {
let op = Op::Recall {
query: query.to_string(),
n: req.n,
ring: req.ring.map(|r| r.as_u8()),
bereich: req.bereich.clone(),
at: req.at,
};
match send(root, actor, op, json)? {
Sent::Answered(Response {
code: Some(0),
stdout,
..
}) => stdout,
_ => None,
}
}
pub fn write(root: &Path, actor: &Actor, req: &WriteRequest, json: bool) -> Result<Answer> {
let what = format!("the write of {}", req.name);
if req.body.len() > MAX_REQUEST / 2
|| req.choice.is_some()
|| req.expected_updated.is_some()
|| req.arriving.is_some()
{
return Ok(Answer::Local);
}
let op = Op::Write(WriteArgs {
ring: req.ring.as_u8(),
kind: req.kind,
name: req.name.clone(),
body: req.body.clone(),
tags: req.tags.clone(),
bereich: req.bereich.clone().flatten(),
retention: req.retention.clone().flatten(),
force: req.force,
dry_run: req.dry_run,
supersedes: req.supersedes.clone(),
valid_from: req.valid_from,
invalid_at: req.invalid_at,
});
once(
send(root, actor, op, json),
&what,
"Check with `cyberbrain recall` before writing it again",
)
}
pub fn scan(root: &Path, actor: &Actor, full: bool, dry_run: bool, json: bool) -> Result<Answer> {
once(
send(root, actor, Op::Scan { full, dry_run }, json),
"the scan",
"`cyberbrain doctor` says whether the index still differs from the files",
)
}
fn once(sent: Option<Sent>, what: &str, check: &str) -> Result<Answer> {
match sent {
None => Ok(Answer::Local),
Some(Sent::Answered(Response {
code: Some(code),
stdout,
stderr,
..
})) => Ok(Answer::Done {
stdout,
stderr,
code,
}),
Some(Sent::Answered(_)) => Ok(Answer::Local),
Some(Sent::Lost) => Err(Error::Index(format!(
"{what} went to the background daemon, which did not answer; it may or may not \
have happened. {check}"
))),
}
}
#[cfg_attr(not(unix), allow(dead_code))] enum Sent {
Answered(Response),
Lost,
}
fn send(root: &Path, actor: &Actor, op: Op, json: bool) -> Option<Sent> {
if disabled() {
return None;
}
let path = socket_path(root)?;
client::ask(
root,
&path,
&Request {
v: PROTOCOL,
op,
actor: actor.to_string(),
json,
},
)
}
#[cfg(unix)]
mod client {
use super::*;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::UnixStream;
use std::os::unix::process::CommandExt;
use std::time::Duration;
pub(super) fn ask(root: &Path, path: &Path, req: &Request) -> Option<Sent> {
let Ok(mut stream) = UnixStream::connect(path) else {
start(root);
return None;
};
let wait = match req.op {
Op::Scan { .. } => 600,
_ => 60,
};
let _ = stream.set_read_timeout(Some(Duration::from_secs(wait)));
let _ = stream.set_write_timeout(Some(Duration::from_secs(5)));
let mut line = serde_json::to_string(req).ok()?;
line.push('\n');
stream.write_all(line.as_bytes()).ok()?;
let mut answer = String::new();
if BufReader::new(stream).read_line(&mut answer).is_err() {
return Some(Sent::Lost);
}
Some(
serde_json::from_str::<Response>(&answer)
.map(Sent::Answered)
.unwrap_or(Sent::Lost),
)
}
fn start(root: &Path) {
let Ok(exe) = std::env::current_exe() else {
return;
};
let _ = std::process::Command::new(exe)
.arg("--store")
.arg(root)
.arg("daemon")
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.process_group(0)
.spawn();
}
}
#[cfg(not(unix))]
mod client {
use super::*;
pub(super) fn ask(_: &Path, _: &Path, _: &Request) -> Option<Sent> {
None
}
}
#[cfg_attr(not(unix), allow(dead_code))]
fn accepts(v: u32) -> bool {
(OLDEST_PROTOCOL..=PROTOCOL).contains(&v)
}
#[cfg(not(unix))]
pub fn serve(_base: App, _idle_secs: u64) -> Result<i32> {
Err(Error::Config(
"the daemon needs Unix sockets; this platform answers every recall locally".into(),
))
}
#[cfg(unix)]
pub fn serve(base: App, idle_secs: u64) -> Result<i32> {
server::run(base, idle_secs)
}
#[cfg(unix)]
mod server {
use super::*;
use std::collections::HashMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
fn stamps(files: &[PathBuf]) -> Vec<Option<(u64, SystemTime, u64)>> {
files
.iter()
.map(|f| {
let m = std::fs::metadata(f).ok()?;
Some((
m.len(),
m.modified().ok()?,
std::os::unix::fs::MetadataExt::ino(&m),
))
})
.collect()
}
struct Shared {
base: Arc<App>,
apps: Mutex<HashMap<String, Arc<App>>>,
watched: Vec<PathBuf>,
at_start: Vec<Option<(u64, SystemTime, u64)>>,
socket: PathBuf,
last: AtomicU64,
}
impl Shared {
fn stale(&self) -> bool {
stamps(&self.watched) != self.at_start
}
fn app_for(&self, actor: &str) -> Result<Arc<App>> {
let mut apps = self
.apps
.lock()
.map_err(|_| Error::Index("daemon lock".into()))?;
if let Some(a) = apps.get(actor) {
return Ok(a.clone());
}
let app = App::open(Some(self.base.root()), actor_from(actor))?;
app.share_model_with(&self.base);
let app = Arc::new(app);
apps.insert(actor.to_string(), app.clone());
Ok(app)
}
fn leave(&self) -> ! {
let _ = std::fs::remove_file(&self.socket);
std::process::exit(0)
}
}
pub(super) fn run(base: App, idle_secs: u64) -> Result<i32> {
let Some(socket) = socket_path(base.root()) else {
return Err(Error::Config(format!(
"the store path is too long for a socket ({} > {MAX_SOCKET_PATH} bytes); \
recall stays in its own process",
base.root().join(SOCKET).as_os_str().len()
)));
};
let listener = match UnixListener::bind(&socket) {
Ok(l) => l,
Err(e) if e.kind() == std::io::ErrorKind::AddrInUse => {
if UnixStream::connect(&socket).is_ok() {
return Ok(0); }
let _ = std::fs::remove_file(&socket);
UnixListener::bind(&socket).map_err(|source| Error::Io {
path: socket.clone(),
source,
})?
}
Err(source) => {
return Err(Error::Io {
path: socket.clone(),
source,
});
}
};
let _ = std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o600));
let mut watched = base.watched_files();
if let Ok(exe) = std::env::current_exe() {
watched.push(exe);
}
let at_start = stamps(&watched);
let shared = Arc::new(Shared {
base: Arc::new(base),
apps: Mutex::new(HashMap::new()),
watched,
at_start,
socket,
last: AtomicU64::new(now_secs()),
});
{
let s = shared.clone();
std::thread::spawn(move || s.base.preload_model());
}
{
let s = shared.clone();
std::thread::spawn(move || {
loop {
std::thread::sleep(Duration::from_secs(idle_secs.clamp(1, 30)));
let idle = now_secs().saturating_sub(s.last.load(Ordering::Relaxed));
if idle >= idle_secs || s.stale() {
s.leave();
}
}
});
}
for stream in listener.incoming() {
let Ok(stream) = stream else { continue };
let s = shared.clone();
std::thread::spawn(move || handle(&s, stream));
}
Ok(0)
}
fn handle(s: &Shared, stream: UnixStream) {
s.last.store(now_secs(), Ordering::Relaxed);
let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
let mut line = String::new();
let Ok(reader) = stream.try_clone() else {
return;
};
if BufReader::new(reader.take(MAX_REQUEST as u64))
.read_line(&mut line)
.is_err()
{
return;
}
let stale = s.stale();
let response = if stale {
not_done("stale: the binary, the configuration or the model changed")
} else if !line.ends_with('\n') {
not_done("incomplete request")
} else {
answer(s, &line)
};
let mut out = serde_json::to_string(&response).unwrap_or_else(|_| "{}".into());
out.push('\n');
let mut w = stream;
let _ = w.write_all(out.as_bytes());
let _ = w.flush();
s.last.store(now_secs(), Ordering::Relaxed);
if stale {
s.leave();
}
}
fn not_done(reason: &str) -> Response {
Response {
reason: Some(reason.to_string()),
..Response::default()
}
}
fn answer(s: &Shared, line: &str) -> Response {
let req: Request = match serde_json::from_str(line) {
Ok(r) => r,
Err(e) => return not_done(&format!("not a daemon request: {e}")),
};
if !accepts(req.v) {
return not_done(&format!(
"protocol {} is not one of {OLDEST_PROTOCOL}..={PROTOCOL}",
req.v
));
}
let json = req.json;
match carry_out(s, req) {
Ok((stdout, code)) => Response {
code: Some(code),
stdout: Some(stdout),
..Response::default()
},
Err(Failed { stdout, error }) => Response {
code: Some(error.exit_code()),
stdout,
stderr: Some(crate::error_text(&error, json)),
..Response::default()
},
}
}
struct Failed {
stdout: Option<String>,
error: Error,
}
impl From<Error> for Failed {
fn from(error: Error) -> Self {
Failed {
stdout: None,
error,
}
}
}
fn carry_out(s: &Shared, req: Request) -> std::result::Result<(String, i32), Failed> {
let app = s.app_for(&req.actor)?;
match req.op {
Op::Recall {
query,
n,
ring,
bereich,
at,
} => {
let ring = ring.map(cyberbrain_core::Ring::try_from).transpose()?;
let rr = RecallRequest {
n,
ring,
bereich,
at,
};
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| Error::Index(format!("cannot start the async runtime: {e}")))?;
let result = rt.block_on(app.recall(&query, &rr))?;
Ok((rendered(&result, req.json, crate::render::recall)?, 0))
}
Op::Write(w) => {
let wr = WriteRequest {
ring: cyberbrain_core::Ring::try_from(w.ring)?,
kind: w.kind,
name: w.name,
body: w.body,
tags: w.tags,
bereich: w.bereich.map(Some),
retention: w.retention.map(Some),
force: w.force,
choice: None,
expected_updated: None,
supersedes: w.supersedes,
valid_from: w.valid_from,
invalid_at: w.invalid_at,
arriving: None,
dry_run: w.dry_run,
};
let outcome = app.write(wr)?;
let stdout = rendered(&outcome, req.json, crate::render_write)?;
match crate::write_exit(&outcome) {
Ok(code) => Ok((stdout, code)),
Err(error) => Err(Failed {
stdout: Some(stdout),
error,
}),
}
}
Op::Scan { full, dry_run } => {
let report = app.scan(crate::app::ScanOptions { full, dry_run })?;
Ok((rendered(&report, req.json, crate::render::scan)?, 0))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_actor_round_trips_through_its_name() {
for a in [Actor::Operator, Actor::Agent("claude-code:8c734a31".into())] {
assert_eq!(actor_from(&a.to_string()).to_string(), a.to_string());
}
}
#[test]
fn a_protocol_3_request_is_still_understood() {
let line = r#"{"v":3,"op":{"op":"write","ring":2,"kind":"knowledge","name":"x","body":"y","tags":[],"bereich":null,"retention":null,"force":false,"dry_run":false},"actor":"operator","json":true}"#;
let req: Request = serde_json::from_str(line).unwrap();
assert!(accepts(req.v));
let Op::Write(w) = req.op else {
panic!("a write")
};
assert_eq!(
(w.supersedes, w.valid_from, w.invalid_at),
(None, None, None),
"absent keeps what the note has"
);
let line = r#"{"v":3,"op":{"op":"recall","query":"q","n":null,"ring":null,"bereich":null},"actor":"operator","json":false}"#;
let req: Request = serde_json::from_str(line).unwrap();
assert!(matches!(req.op, Op::Recall { at: None, .. }));
assert!(!accepts(2) && !accepts(PROTOCOL + 1));
}
#[test]
fn validity_keeps_its_three_states_on_the_wire() {
let t: jiff::Timestamp = "2026-09-01T00:00:00Z".parse().unwrap();
for (from, to) in [
(None, None),
(Some(None), Some(Some(t))),
(Some(Some(t)), Some(None)),
] {
let args = WriteArgs {
ring: 2,
kind: NoteKind::Knowledge,
name: "x".into(),
body: "y".into(),
tags: Vec::new(),
bereich: None,
retention: None,
force: false,
dry_run: false,
supersedes: Some(vec!["alt".into()]),
valid_from: from,
invalid_at: to,
};
let line = serde_json::to_string(&Op::Write(args)).unwrap();
let Op::Write(back) = serde_json::from_str(&line).unwrap() else {
panic!("a write")
};
assert_eq!((back.valid_from, back.invalid_at), (from, to), "{line}");
assert_eq!(back.supersedes, Some(vec!["alt".to_string()]));
}
}
#[test]
fn a_path_too_long_for_a_socket_gets_no_daemon() {
assert!(socket_path(Path::new("/s")).is_some());
assert!(socket_path(&PathBuf::from("/".to_string() + &"x".repeat(120))).is_none());
}
}