loopd: the memory block from memoryd, recall and memory as core tools
Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
//! The baseline: the fixed prefix of every request in an epoch — the system prompt, core memory
|
||||
//! and core tool schemas. It is snapshotted per session so an edit to `system.md` changes nothing
|
||||
//! The baseline: the fixed prefix of every request in an epoch — the system prompt, the memory
|
||||
//! block and core tool schemas. It is snapshotted per session so an edit to `system.md` changes nothing
|
||||
//! for a running session, and `messages` turns a session log back into the request's messages.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
@@ -25,7 +25,6 @@ pub enum BaselineError {
|
||||
Read(PathBuf, std::io::Error),
|
||||
Parse(PathBuf, serde_json::Error),
|
||||
Hash,
|
||||
Core(PathBuf, std::io::Error),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for BaselineError {
|
||||
@@ -34,11 +33,6 @@ impl std::fmt::Display for BaselineError {
|
||||
BaselineError::Read(path, err) => write!(f, "{}: {err}", path.display()),
|
||||
BaselineError::Parse(path, err) => write!(f, "{}: {err}", path.display()),
|
||||
BaselineError::Hash => write!(f, "hashing the baseline failed"),
|
||||
BaselineError::Core(path, err) => write!(
|
||||
f,
|
||||
"{}: {err}; see docs/runbook.md#core-memory-unreadable",
|
||||
path.display()
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -46,29 +40,26 @@ impl std::fmt::Display for BaselineError {
|
||||
impl std::error::Error for BaselineError {}
|
||||
|
||||
impl Baseline {
|
||||
/// The system prompt trimmed, plus core memory if it has content; the core tool schemas.
|
||||
/// The system prompt trimmed, with no memory: for tests and the self-test.
|
||||
pub fn assemble(cfg: &Config, registry: &Registry) -> Result<Baseline, BaselineError> {
|
||||
Baseline::assemble_with(cfg, registry, "")
|
||||
}
|
||||
|
||||
/// The system prompt trimmed, then a blank line and the memory block `memoryd` rendered (M5a
|
||||
/// spec, section 4), when there is one; the core tool schemas.
|
||||
pub fn assemble_with(
|
||||
cfg: &Config,
|
||||
registry: &Registry,
|
||||
memory: &str,
|
||||
) -> Result<Baseline, BaselineError> {
|
||||
let system = std::fs::read_to_string(&cfg.baseline.system)
|
||||
.map_err(|e| BaselineError::Read(cfg.baseline.system.clone(), e))?;
|
||||
let mut system = system.trim_end().to_string();
|
||||
|
||||
let core = cfg.paths.home.join("memory/core.md");
|
||||
match std::fs::read_to_string(&core) {
|
||||
Ok(text) => {
|
||||
let trimmed = text.trim();
|
||||
if !trimmed.is_empty() {
|
||||
system.push_str("\n\n");
|
||||
system.push_str(trimmed);
|
||||
}
|
||||
}
|
||||
// A missing file is fine; an unreadable one is reported so the owner is not given a
|
||||
// session without the memory they curated and no sign of it.
|
||||
Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
|
||||
return Err(BaselineError::Core(core.clone(), e));
|
||||
}
|
||||
Err(_) => {}
|
||||
let memory = memory.trim();
|
||||
if !memory.is_empty() {
|
||||
system.push_str("\n\n");
|
||||
system.push_str(memory);
|
||||
}
|
||||
|
||||
Ok(Baseline {
|
||||
system,
|
||||
tools: registry.core_schemas(),
|
||||
|
||||
@@ -10,6 +10,7 @@ use std::sync::{Arc, Mutex};
|
||||
use crate::baseline::Baseline;
|
||||
use crate::config::Config;
|
||||
use crate::llama::Client;
|
||||
use crate::memory::MemorySource;
|
||||
use crate::session::{Session, SessionError};
|
||||
use crate::tools::{Registry, ToolPort};
|
||||
use crate::turn::{Runtime, TurnError, run_turn};
|
||||
@@ -24,6 +25,8 @@ pub struct Context {
|
||||
pub client: Client,
|
||||
pub port: Box<dyn ToolPort>,
|
||||
pub registry: Registry,
|
||||
/// Where a new session's memory block comes from, and who hears that a session changed.
|
||||
pub memory: Box<dyn MemorySource>,
|
||||
busy: Mutex<HashSet<SessionId>>,
|
||||
}
|
||||
|
||||
@@ -33,12 +36,14 @@ impl Context {
|
||||
client: Client,
|
||||
port: Box<dyn ToolPort>,
|
||||
registry: Registry,
|
||||
memory: Box<dyn MemorySource>,
|
||||
) -> Context {
|
||||
Context {
|
||||
cfg,
|
||||
client,
|
||||
port,
|
||||
registry,
|
||||
memory,
|
||||
busy: Mutex::new(HashSet::new()),
|
||||
}
|
||||
}
|
||||
@@ -140,7 +145,11 @@ pub fn handle(stream: UnixStream, ctx: Arc<Context>) {
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let baseline = match Baseline::assemble(&ctx.cfg, &ctx.registry) {
|
||||
// A new session takes the memory block as it is now; no session starts without it.
|
||||
let assembled = ctx.memory.block().and_then(|block| {
|
||||
Baseline::assemble_with(&ctx.cfg, &ctx.registry, &block).map_err(|e| e.to_string())
|
||||
});
|
||||
let baseline = match assembled {
|
||||
Ok(baseline) => baseline,
|
||||
Err(e) => {
|
||||
drop(held);
|
||||
@@ -184,6 +193,17 @@ pub fn handle(stream: UnixStream, ctx: Arc<Context>) {
|
||||
// 5. Release the session before the final frame, so the next turn does not find it busy.
|
||||
drop(held);
|
||||
|
||||
// A finished turn's records are for `memoryd` to index; best effort, since `memoryd` catches
|
||||
// up at its next start.
|
||||
if outcome.is_ok()
|
||||
&& let Err(e) = ctx.memory.updated(&turn.session)
|
||||
{
|
||||
eprintln!(
|
||||
"loopd: memoryd was not told about {}: {e}",
|
||||
turn.session.as_str()
|
||||
);
|
||||
}
|
||||
|
||||
// 6. and 7. A turn done is success; a turn error maps to a code. A write failure means the
|
||||
// channel is gone, so we stop and finish quietly.
|
||||
match outcome {
|
||||
|
||||
@@ -173,6 +173,16 @@ pub struct Config {
|
||||
pub baseline: Baseline,
|
||||
#[serde(default)]
|
||||
pub broker: Broker,
|
||||
#[serde(default)]
|
||||
pub memory: Memory,
|
||||
}
|
||||
|
||||
/// `[memory]`: `memoryd`'s socket for `loopd` (M5a spec, section 9). An empty path means the
|
||||
/// default under `home`.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize)]
|
||||
#[serde(deny_unknown_fields, default)]
|
||||
pub struct Memory {
|
||||
pub socket: std::path::PathBuf,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -206,6 +216,15 @@ impl Config {
|
||||
}
|
||||
Ok(config)
|
||||
}
|
||||
/// `memoryd`'s socket for `loopd`.
|
||||
pub fn memory_socket(&self) -> std::path::PathBuf {
|
||||
if self.memory.socket.as_os_str().is_empty() {
|
||||
self.paths.home.join("run/loop-memory/memory.sock")
|
||||
} else {
|
||||
self.memory.socket.clone()
|
||||
}
|
||||
}
|
||||
|
||||
pub fn channel_socket(&self) -> std::path::PathBuf {
|
||||
if self.channel.socket.as_os_str().is_empty() {
|
||||
self.paths.home.join("run/loop/loop.sock")
|
||||
|
||||
@@ -6,6 +6,7 @@ pub mod channel;
|
||||
pub mod config;
|
||||
pub mod http;
|
||||
pub mod llama;
|
||||
pub mod memory;
|
||||
pub mod selftest;
|
||||
pub mod session;
|
||||
pub mod socket;
|
||||
|
||||
@@ -12,6 +12,7 @@ use loopd::broker_port::{BrokerPort, NoBroker, not_configured_line};
|
||||
use loopd::channel::{self, Context};
|
||||
use loopd::config::Config;
|
||||
use loopd::llama::Client;
|
||||
use loopd::memory::{MEMORY_WAIT, MemoryPort};
|
||||
use loopd::selftest::{SelfTestError, run};
|
||||
use loopd::tools::{Registry, ToolPort};
|
||||
|
||||
@@ -117,7 +118,8 @@ fn run_serve(path: &Path) -> ExitCode {
|
||||
Box::new(NoBroker)
|
||||
}
|
||||
};
|
||||
let ctx = Arc::new(Context::new(cfg, client, port, Registry::m3a()));
|
||||
let memory = Box::new(MemoryPort::new(cfg.memory_socket(), MEMORY_WAIT));
|
||||
let ctx = Arc::new(Context::new(cfg, client, port, Registry::m5a(), memory));
|
||||
if let Err(e) = channel::serve(listener, ctx) {
|
||||
eprintln!("loopd: the channel server stopped: {e}");
|
||||
return ExitCode::from(1);
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
//! Memory, as `loopd` sees it (M5a spec, section 9): the block `memoryd` renders for a new
|
||||
//! session's baseline, the schemas of the two core memory tools, and the notice after a turn.
|
||||
//! `loopd` keeps no memory itself; `brokerd` decides the tool calls and `memoryd` runs them.
|
||||
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use proto::{
|
||||
Empty, Envelope, Message, PROTOCOL_VERSION, SessionId, SessionUpdated, read_frame, write_frame,
|
||||
};
|
||||
|
||||
use crate::llama::ToolSchema;
|
||||
|
||||
/// Where a new session's baseline points when there is no memory block.
|
||||
pub const UNAVAILABLE: &str = "see docs/runbook.md#memory-unavailable";
|
||||
|
||||
/// How long `loopd` waits for `memoryd`.
|
||||
pub const MEMORY_WAIT: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Where the memory block comes from, and who hears that a session has new records.
|
||||
pub trait MemorySource: Send + Sync {
|
||||
/// The block for a new session. `Err` ends in `UNAVAILABLE`: no session starts without it.
|
||||
fn block(&self) -> Result<String, String>;
|
||||
/// The session has new records; best effort, a failure is only logged by the caller.
|
||||
fn updated(&self, session: &SessionId) -> Result<(), String>;
|
||||
}
|
||||
|
||||
/// No memory: for tests and the self-test, never for `loopd serve`.
|
||||
pub struct NoMemory;
|
||||
|
||||
impl MemorySource for NoMemory {
|
||||
fn block(&self) -> Result<String, String> {
|
||||
Ok(String::new())
|
||||
}
|
||||
fn updated(&self, _session: &SessionId) -> Result<(), String> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// A client of `memoryd`'s `loopd` socket. One connection per request.
|
||||
pub struct MemoryPort {
|
||||
socket: PathBuf,
|
||||
wait: Duration,
|
||||
}
|
||||
|
||||
impl MemoryPort {
|
||||
pub fn new(socket: PathBuf, wait: Duration) -> MemoryPort {
|
||||
MemoryPort { socket, wait }
|
||||
}
|
||||
|
||||
fn ask(&self, msg: Message) -> Result<Message, String> {
|
||||
let mut stream = match UnixStream::connect(&self.socket) {
|
||||
Ok(stream) => stream,
|
||||
Err(e) => return Err(self.line(&e)),
|
||||
};
|
||||
if let Err(e) = stream.set_write_timeout(Some(self.wait)) {
|
||||
return Err(self.line(&e));
|
||||
}
|
||||
if let Err(e) = stream.set_read_timeout(Some(self.wait)) {
|
||||
return Err(self.line(&e));
|
||||
}
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 1,
|
||||
r#final: true,
|
||||
msg,
|
||||
};
|
||||
if let Err(e) = write_frame(&mut stream, &env) {
|
||||
return Err(self.line(&e));
|
||||
}
|
||||
match read_frame(&mut stream) {
|
||||
Ok(env) => Ok(env.msg),
|
||||
Err(e) => Err(self.line(&e)),
|
||||
}
|
||||
}
|
||||
|
||||
fn line(&self, e: &dyn std::fmt::Display) -> String {
|
||||
format!("memoryd at {}: {e}; {UNAVAILABLE}", self.socket.display())
|
||||
}
|
||||
}
|
||||
|
||||
impl MemorySource for MemoryPort {
|
||||
fn block(&self) -> Result<String, String> {
|
||||
match self.ask(Message::MemoryBlock(Empty {}))? {
|
||||
Message::MemoryBlockReply(reply) => Ok(reply.text),
|
||||
Message::Error(e) => Err(format!("memoryd: {}; {UNAVAILABLE}", e.detail)),
|
||||
_ => Err(format!(
|
||||
"memoryd answered with the wrong message; {UNAVAILABLE}"
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn updated(&self, session: &SessionId) -> Result<(), String> {
|
||||
match self.ask(Message::SessionUpdated(SessionUpdated {
|
||||
session: session.clone(),
|
||||
}))? {
|
||||
Message::Ok(_) => Ok(()),
|
||||
Message::Error(e) => Err(format!("memoryd: {}", e.detail)),
|
||||
_ => Err("memoryd answered with the wrong message".to_string()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub const RECALL: &str = "recall";
|
||||
pub const MEMORY: &str = "memory";
|
||||
|
||||
pub fn recall_schema() -> ToolSchema {
|
||||
ToolSchema {
|
||||
name: RECALL.to_string(),
|
||||
description: "Search your memory: past conversations and notes. Each hit says where it \
|
||||
came from: identity, memory, or memory-tainted (from a session that read \
|
||||
untrusted content)."
|
||||
.to_string(),
|
||||
parameters: serde_json::json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": { "type": "string", "description": "Words to search for." },
|
||||
"k": { "type": "integer", "description": "How many hits, 1 to 10. Default 5." }
|
||||
},
|
||||
"required": ["query"]
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn memory_schema() -> ToolSchema {
|
||||
ToolSchema {
|
||||
name: MEMORY.to_string(),
|
||||
description: "Write memory. note: record something seen or told. add, replace, remove: \
|
||||
edit memory.md (your work) or user.md (your owner); entries are named by \
|
||||
their exact text. consolidate: rewrite a whole file, moving or dropping \
|
||||
entries to notes. Takes effect in the next session."
|
||||
.to_string(),
|
||||
parameters: serde_json::json!({
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"action": { "type": "string", "enum": ["note", "add", "replace", "remove", "consolidate"] },
|
||||
"file": { "type": "string", "enum": ["memory", "user"] },
|
||||
"text": { "type": "string", "description": "The new entry, or the note." },
|
||||
"old": { "type": "string", "description": "The exact text of the entry to replace or remove." },
|
||||
"entries": {
|
||||
"type": "array",
|
||||
"description": "consolidate: every current entry once, as {old, keep: true}, {old, text}, {old, move: true} or {old, drop: true}; new entries as {text}.",
|
||||
"items": { "type": "object" }
|
||||
}
|
||||
},
|
||||
"required": ["action"]
|
||||
}),
|
||||
}
|
||||
}
|
||||
@@ -82,6 +82,27 @@ impl Registry {
|
||||
])
|
||||
}
|
||||
|
||||
/// The tool set for the M5a epoch: `clock`, `recall` and `memory` (core), then the four tools
|
||||
/// `brokerd` decides on. `loopd serve` uses this one.
|
||||
pub fn m5a() -> Registry {
|
||||
let mut entries = Registry::m3a().entries;
|
||||
entries.insert(
|
||||
1,
|
||||
Entry {
|
||||
schema: crate::memory::recall_schema(),
|
||||
core: true,
|
||||
},
|
||||
);
|
||||
entries.insert(
|
||||
2,
|
||||
Entry {
|
||||
schema: crate::memory::memory_schema(),
|
||||
core: true,
|
||||
},
|
||||
);
|
||||
Registry::new(entries)
|
||||
}
|
||||
|
||||
/// The fixed tools array: core entries, then `find_tool`, then `call_tool`.
|
||||
pub fn core_schemas(&self) -> Vec<ToolSchema> {
|
||||
let mut schemas: Vec<ToolSchema> = self
|
||||
|
||||
@@ -31,21 +31,22 @@ fn assembles_system_prompt_and_core_schemas() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn core_memory_is_appended_when_present_and_only_then() {
|
||||
fn the_memory_block_is_appended_when_present_and_only_then() {
|
||||
let home = Home::new();
|
||||
let cfg = home.config(Path::new("/tmp/unused.sock"));
|
||||
home.write("memory/core.md", "The owner likes cork.\n\n");
|
||||
let b = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
|
||||
let block = "# soul.md\nYou are kind.\n\n# memory.md (0 of 2,200 characters)\n(empty)\n";
|
||||
let b = Baseline::assemble_with(&cfg, &Registry::m2b(), block).unwrap();
|
||||
assert_eq!(
|
||||
b.system,
|
||||
"You are Boxmaker, a test agent.\n\nThe owner likes cork."
|
||||
"You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind.\n\n# memory.md (0 of 2,200 characters)\n(empty)"
|
||||
);
|
||||
home.write("memory/core.md", " \n");
|
||||
let b = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
|
||||
home.write("memory/core.md", "An old core file is ignored.\n");
|
||||
let b = Baseline::assemble_with(&cfg, &Registry::m2b(), " \n").unwrap();
|
||||
assert_eq!(
|
||||
b.system, "You are Boxmaker, a test agent.",
|
||||
"an empty core file adds nothing"
|
||||
"an empty block adds nothing, and core.md is no longer read"
|
||||
);
|
||||
assert_eq!(b, Baseline::assemble(&cfg, &Registry::m2b()).unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -232,28 +233,3 @@ fn replay_of_a_prefix_is_a_prefix() {
|
||||
assert_eq!(whole[..part.len()], part[..], "prefix of {n} records");
|
||||
}
|
||||
}
|
||||
|
||||
/// A core memory file that exists but cannot be read is an error, not silently absent: the owner
|
||||
/// would otherwise get a session without the memory they curated, and no sign of it.
|
||||
#[test]
|
||||
fn an_unreadable_core_memory_file_is_an_error() {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
if running_as_root() {
|
||||
return; // root can read anything; the check is meaningless there
|
||||
}
|
||||
let home = Home::new();
|
||||
let cfg = home.config(Path::new("/tmp/unused.sock"));
|
||||
home.write("memory/core.md", "secret memory\n");
|
||||
let core = home.dir.join("memory/core.md");
|
||||
std::fs::set_permissions(&core, std::fs::Permissions::from_mode(0o000)).unwrap();
|
||||
let result = Baseline::assemble(&cfg, &Registry::m2b());
|
||||
std::fs::set_permissions(&core, std::fs::Permissions::from_mode(0o644)).unwrap();
|
||||
let e = result.expect_err("an unreadable core.md must not be ignored");
|
||||
assert!(e.to_string().contains("core.md"), "{e}");
|
||||
}
|
||||
|
||||
fn running_as_root() -> bool {
|
||||
std::fs::read_to_string("/proc/self/status")
|
||||
.map(|s| s.lines().any(|l| l.starts_with("Uid:\t0\t")))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ mod support;
|
||||
|
||||
use loopd::channel::{Context, serve};
|
||||
use loopd::llama::Client;
|
||||
use loopd::memory::NoMemory;
|
||||
use loopd::tools::Registry;
|
||||
use proto::{
|
||||
Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, TurnEvent, read_frame,
|
||||
@@ -38,6 +39,7 @@ fn start(replies: Vec<proto::ToolResponse>) -> Loopd {
|
||||
Client::new(cfg),
|
||||
Box::new(ScriptedPort::new(replies)),
|
||||
Registry::m2b(),
|
||||
Box::new(NoMemory),
|
||||
));
|
||||
thread::spawn(move || serve(listener, ctx));
|
||||
Loopd {
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
//! `loopd` and memory: the memory block in a new session's baseline and not in a running one's,
|
||||
//! no session without it, `memoryd` told after a finished turn, and the two core memory tools
|
||||
//! (M5a spec, section 9). Do not edit.
|
||||
|
||||
mod support;
|
||||
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use loopd::channel::{Context, serve};
|
||||
use loopd::llama::Client;
|
||||
use loopd::memory::{MemorySource, UNAVAILABLE};
|
||||
use loopd::tools::Registry;
|
||||
use proto::{
|
||||
Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId, Turn, read_frame, write_frame,
|
||||
};
|
||||
use support::{FakeServer, Home, Reply, ScriptedPort};
|
||||
|
||||
const CHAT: &str = "/v1/chat/completions";
|
||||
|
||||
/// A `memoryd` stand-in: the block it serves, and the sessions it was told about.
|
||||
struct Recorder {
|
||||
block: Mutex<Result<String, String>>,
|
||||
updated: Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
/// Lets the test keep its `Arc<Recorder>` while the context owns a `Box<dyn MemorySource>`.
|
||||
struct Shared(Arc<Recorder>);
|
||||
|
||||
impl MemorySource for Shared {
|
||||
fn block(&self) -> Result<String, String> {
|
||||
self.0.block.lock().unwrap().clone()
|
||||
}
|
||||
fn updated(&self, session: &SessionId) -> Result<(), String> {
|
||||
self.0
|
||||
.updated
|
||||
.lock()
|
||||
.unwrap()
|
||||
.push(session.as_str().to_string());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct Loopd {
|
||||
_home: Home,
|
||||
server: FakeServer,
|
||||
socket: PathBuf,
|
||||
memory: Arc<Recorder>,
|
||||
}
|
||||
|
||||
fn start(block: Result<&str, &str>) -> Loopd {
|
||||
let home = Home::new();
|
||||
let server = FakeServer::start();
|
||||
let cfg = home.config(&server.socket);
|
||||
let socket = cfg.channel_socket();
|
||||
let listener = UnixListener::bind(&socket).unwrap();
|
||||
let memory = Arc::new(Recorder {
|
||||
block: Mutex::new(block.map(str::to_string).map_err(str::to_string)),
|
||||
updated: Mutex::new(Vec::new()),
|
||||
});
|
||||
let ctx = Arc::new(Context::new(
|
||||
cfg.clone(),
|
||||
Client::new(cfg),
|
||||
Box::new(ScriptedPort::new(vec![])),
|
||||
Registry::m2b(),
|
||||
Box::new(Shared(Arc::clone(&memory))),
|
||||
));
|
||||
std::thread::spawn(move || serve(listener, ctx));
|
||||
Loopd {
|
||||
_home: home,
|
||||
server,
|
||||
socket,
|
||||
memory,
|
||||
}
|
||||
}
|
||||
|
||||
/// One turn; the last frame.
|
||||
fn turn(l: &Loopd, session: &str, resume: bool) -> Message {
|
||||
let mut stream = UnixStream::connect(&l.socket).unwrap();
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(10)))
|
||||
.unwrap();
|
||||
let msg = Message::Turn(Turn {
|
||||
session: SessionId::new(session).unwrap(),
|
||||
content: "hello".to_string(),
|
||||
resume,
|
||||
});
|
||||
let env = Envelope {
|
||||
v: PROTOCOL_VERSION,
|
||||
id: 7,
|
||||
r#final: true,
|
||||
msg,
|
||||
};
|
||||
write_frame(&mut stream, &env).unwrap();
|
||||
loop {
|
||||
let frame = read_frame(&mut stream).unwrap();
|
||||
if frame.r#final {
|
||||
return frame.msg;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn system_prompt(l: &Loopd, n: usize) -> String {
|
||||
let requests = l.server.requests_to(CHAT);
|
||||
requests[n].json()["messages"][0]["content"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_new_session_starts_with_the_block_and_keeps_it() {
|
||||
let l = start(Ok("# soul.md\nYou are kind."));
|
||||
l.server
|
||||
.route(CHAT, vec![Reply::fixture("plain"), Reply::fixture("plain")]);
|
||||
assert!(matches!(turn(&l, "a", false), Message::TurnDone(_)));
|
||||
assert_eq!(
|
||||
system_prompt(&l, 0),
|
||||
"You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind."
|
||||
);
|
||||
*l.memory.block.lock().unwrap() = Ok("# soul.md\nYou changed.".to_string());
|
||||
assert!(matches!(turn(&l, "a", true), Message::TurnDone(_)));
|
||||
assert_eq!(
|
||||
system_prompt(&l, 1),
|
||||
"You are Boxmaker, a test agent.\n\n# soul.md\nYou are kind.",
|
||||
"a running session keeps the bytes it started with"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_session_starts_without_the_block() {
|
||||
let l = start(Err(
|
||||
"memoryd at /x: refused; see docs/runbook.md#memory-unavailable",
|
||||
));
|
||||
match turn(&l, "b", false) {
|
||||
Message::Error(e) => {
|
||||
assert_eq!(e.code, ErrorCode::Internal);
|
||||
assert!(e.detail.ends_with(UNAVAILABLE), "{}", e.detail);
|
||||
}
|
||||
other => panic!("started without memory: {other:?}"),
|
||||
}
|
||||
assert!(l.server.requests_to(CHAT).is_empty(), "no request was made");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn memoryd_is_told_after_a_finished_turn_only() {
|
||||
let l = start(Ok(""));
|
||||
l.server.route(
|
||||
CHAT,
|
||||
vec![Reply::fixture("plain"), Reply::fixture("bad_request")],
|
||||
);
|
||||
turn(&l, "c", false);
|
||||
assert!(
|
||||
matches!(turn(&l, "d", false), Message::Error(_)),
|
||||
"a turn that ran and failed"
|
||||
);
|
||||
assert!(matches!(
|
||||
turn(&l, "nobody-made-this", true),
|
||||
Message::Error(_)
|
||||
));
|
||||
assert_eq!(*l.memory.updated.lock().unwrap(), ["c"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recall_and_memory_are_core_tools_of_the_m5a_registry() {
|
||||
let names: Vec<String> = Registry::m5a()
|
||||
.core_schemas()
|
||||
.into_iter()
|
||||
.map(|s| s.name)
|
||||
.collect();
|
||||
assert_eq!(
|
||||
names,
|
||||
["clock", "recall", "memory", "find_tool", "call_tool"]
|
||||
);
|
||||
let m3a: Vec<String> = Registry::m3a()
|
||||
.core_schemas()
|
||||
.into_iter()
|
||||
.map(|s| s.name)
|
||||
.collect();
|
||||
assert_eq!(m3a, ["clock", "find_tool", "call_tool"], "m3a is unchanged");
|
||||
assert!(Registry::m5a().get("read_file").is_some());
|
||||
}
|
||||
@@ -4,13 +4,13 @@
|
||||
|
||||
mod support;
|
||||
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::path::Path;
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::time::Duration;
|
||||
|
||||
use loopd::baseline::Baseline;
|
||||
use loopd::broker_port::{not_configured_line, unavailable_line};
|
||||
use loopd::memory::{MemoryPort, MemorySource};
|
||||
use loopd::session::{Session, SessionError};
|
||||
use loopd::tools::Registry;
|
||||
use proto::SessionId;
|
||||
@@ -64,31 +64,37 @@ fn only_the_torn_error_points_at_the_damaged_log_entry() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unreadable_core_memory_names_its_entry_and_a_missing_system_prompt_does_not() {
|
||||
fn memoryd_unreachable_or_refusing_names_the_memory_entry() {
|
||||
let home = Home::new();
|
||||
let cfg = home.config(Path::new("/tmp/unused.sock"));
|
||||
home.write("memory/core.md", "memory\n");
|
||||
let core = home.dir.join("memory/core.md");
|
||||
if !running_as_root() {
|
||||
std::fs::set_permissions(&core, std::fs::Permissions::from_mode(0o000)).unwrap();
|
||||
let result = Baseline::assemble(&cfg, &Registry::m3a());
|
||||
std::fs::set_permissions(&core, std::fs::Permissions::from_mode(0o644)).unwrap();
|
||||
let message = result.expect_err("unreadable core.md").to_string();
|
||||
assert!(message.contains("core.md"), "{message}");
|
||||
ends_with_pointer(&message, "see docs/runbook.md#core-memory-unreadable");
|
||||
}
|
||||
// The system prompt is a different file with a different remedy: no pointer to this entry.
|
||||
std::fs::remove_file(home.dir.join("system.md")).unwrap();
|
||||
let message = Baseline::assemble(&cfg, &Registry::m3a())
|
||||
.expect_err("no system.md")
|
||||
.to_string();
|
||||
assert!(!message.contains("core-memory-unreadable"), "{message}");
|
||||
}
|
||||
|
||||
fn running_as_root() -> bool {
|
||||
std::fs::read_to_string("/proc/self/status")
|
||||
.map(|s| s.lines().any(|l| l.starts_with("Uid:\t0\t")))
|
||||
.unwrap_or(false)
|
||||
let port = MemoryPort::new(
|
||||
home.dir.join("run/nobody/memory.sock"),
|
||||
Duration::from_secs(1),
|
||||
);
|
||||
let message = port.block().expect_err("nobody is listening");
|
||||
ends_with_pointer(&message, "see docs/runbook.md#memory-unavailable");
|
||||
// A refusal from memoryd itself (a file over its limit) ends with the same entry.
|
||||
let socket = home.dir.join("run/loop-memory/memory.sock");
|
||||
std::fs::create_dir_all(socket.parent().unwrap()).unwrap();
|
||||
let listener = std::os::unix::net::UnixListener::bind(&socket).unwrap();
|
||||
std::thread::spawn(move || {
|
||||
let (mut stream, _) = listener.accept().unwrap();
|
||||
let env = proto::read_frame(&mut stream).unwrap();
|
||||
let reply = proto::Envelope {
|
||||
v: 1,
|
||||
id: env.id,
|
||||
r#final: true,
|
||||
msg: proto::Message::Error(proto::WireError {
|
||||
code: proto::ErrorCode::Internal,
|
||||
detail: "memory/user.md: 1,376 characters; its limit is 1,375".to_string(),
|
||||
}),
|
||||
};
|
||||
proto::write_frame(&mut stream, &reply).unwrap();
|
||||
});
|
||||
let message = MemoryPort::new(socket, Duration::from_secs(5))
|
||||
.block()
|
||||
.expect_err("memoryd refused");
|
||||
assert!(message.contains("user.md: 1,376 characters"), "{message}");
|
||||
ends_with_pointer(&message, "see docs/runbook.md#memory-unavailable");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user