memoryd: the admin socket's requests move to serve/admin.rs

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
This commit is contained in:
2026-09-29 09:54:14 -07:00
co-authored by ornith-1.5-35b-a3b
parent 132e816322
commit 165145edb5
4 changed files with 254 additions and 67 deletions
+5 -67
View File
@@ -8,9 +8,9 @@ use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;
use bxcore::{
DataClass, Empty, Envelope, ErrorCode, MemoryAdminReply, MemoryArgs, MemoryBlockReply,
MemoryCall, MemoryCommand, Message, PROTOCOL_VERSION, RecallArgs, Reindexed, SessionId,
SessionUpdated, Timestamp, ToolResponse, WireError, read_frame, write_frame,
DataClass, Empty, Envelope, ErrorCode, MemoryArgs, MemoryBlockReply, MemoryCall, Message,
PROTOCOL_VERSION, RecallArgs, SessionId, SessionUpdated, Timestamp, ToolResponse, WireError,
read_frame, write_frame,
};
use crate::config::Config;
@@ -18,7 +18,6 @@ use crate::edit::{self, Context};
use crate::files::Store;
use crate::honcho::ledger::{self, LedgerError};
use crate::honcho::mirror::Mirror;
use crate::honcho::report;
use crate::honcho::search;
use crate::honcho::state;
use crate::honcho::turns::TurnSync;
@@ -27,6 +26,8 @@ use crate::index::{self, Index};
use crate::recall::Recalled;
use crate::{journal, recall, review};
mod admin;
/// How long a peer may take to send its request.
pub const READ_WAIT: Duration = Duration::from_secs(10);
@@ -429,67 +430,4 @@ impl Memoryd {
(self.log)(format!("memoryd: notes not indexed: {e}").as_str());
}
}
fn reindex(&self) -> Message {
let mut state = self.lock();
let max_class = self.config.index.max_class;
match index::rebuild(
&self.config.index_path(),
max_class,
&self.config.sessions_dir(),
&self.store.path("notes.md"),
) {
Ok((counts, problems)) => {
for problem in problems {
(self.log)(format!("memoryd: not indexed: {problem}").as_str());
}
match Index::open(&self.config.index_path(), max_class) {
Ok(index) => {
state.index = index;
Message::Reindexed(Reindexed {
logs: counts.logs,
records: counts.records,
notes: counts.notes,
})
}
Err(e) => error(e),
}
}
Err(e) => error(e),
}
}
fn admin(&self, command: MemoryCommand, arg: &str) -> Message {
let state = self.lock();
let now = Timestamp::now();
let result = match command {
MemoryCommand::Log => {
let n = if arg.is_empty() {
10
} else {
match arg.parse::<usize>() {
Ok(n) => n,
Err(_) => return error(format!("log takes a count; {arg:?} is not one")),
}
};
journal::log(&self.store, n)
}
MemoryCommand::Undo => journal::undo(&self.store, arg, now),
MemoryCommand::Pending => review::list(&self.store),
MemoryCommand::Accept => review::accept(&self.store, arg, now),
MemoryCommand::Reject => review::reject(&self.store, arg, now),
MemoryCommand::Sync => Ok(match &self.mirror {
Some(mirror) => mirror.report(),
None => report::NOT_CONFIGURED.to_string(),
}),
};
if result.is_ok() && matches!(command, MemoryCommand::Undo | MemoryCommand::Accept) {
self.mirror(None);
}
self.index_notes(&state);
match result {
Ok(text) => Message::MemoryAdminReply(MemoryAdminReply { text }),
Err(e) => error(e),
}
}
}
+75
View File
@@ -0,0 +1,75 @@
//! The owner's requests on `memoryd`'s admin socket: `reindex`, and the memory commands `log`,
//! `undo`, `pending`, `accept`, `reject` and `sync` (M5a spec, section 10; M5c spec, section 5).
//! Moved out of `serve.rs` as they were, to keep it under the line limit.
use bxcore::{MemoryAdminReply, MemoryCommand, Message, Reindexed, Timestamp};
use super::{Memoryd, error};
use crate::honcho::report;
use crate::index::{self, Index};
use crate::{journal, review};
impl Memoryd {
pub(super) fn reindex(&self) -> Message {
let mut state = self.lock();
let max_class = self.config.index.max_class;
match index::rebuild(
&self.config.index_path(),
max_class,
&self.config.sessions_dir(),
&self.store.path("notes.md"),
) {
Ok((counts, problems)) => {
for problem in problems {
(self.log)(format!("memoryd: not indexed: {problem}").as_str());
}
match Index::open(&self.config.index_path(), max_class) {
Ok(index) => {
state.index = index;
Message::Reindexed(Reindexed {
logs: counts.logs,
records: counts.records,
notes: counts.notes,
})
}
Err(e) => error(e),
}
}
Err(e) => error(e),
}
}
pub(super) fn admin(&self, command: MemoryCommand, arg: &str) -> Message {
let state = self.lock();
let now = Timestamp::now();
let result = match command {
MemoryCommand::Log => {
let n = if arg.is_empty() {
10
} else {
match arg.parse::<usize>() {
Ok(n) => n,
Err(_) => return error(format!("log takes a count; {arg:?} is not one")),
}
};
journal::log(&self.store, n)
}
MemoryCommand::Undo => journal::undo(&self.store, arg, now),
MemoryCommand::Pending => review::list(&self.store),
MemoryCommand::Accept => review::accept(&self.store, arg, now),
MemoryCommand::Reject => review::reject(&self.store, arg, now),
MemoryCommand::Sync => Ok(match &self.mirror {
Some(mirror) => mirror.report(),
None => report::NOT_CONFIGURED.to_string(),
}),
};
if result.is_ok() && matches!(command, MemoryCommand::Undo | MemoryCommand::Accept) {
self.mirror(None);
}
self.index_notes(&state);
match result {
Ok(text) => Message::MemoryAdminReply(MemoryAdminReply { text }),
Err(e) => error(e),
}
}
}
+173
View File
@@ -0,0 +1,173 @@
//! The owner's commands on `memoryd`'s admin socket, end to end (M5a spec, section 10): `log`
//! shows the last ten edits unless told how many; `reject` discards a held write and `accept`
//! applies one; what an accepted write moves to `notes.md` is found by `recall` at once; a
//! reindex counts the notes. They pin `serve/admin.rs`, which M5c-2 task 11 moved out of
//! `serve.rs`. Do not edit.
#[path = "support/tmp.rs"]
mod tmp;
use std::os::unix::net::UnixStream;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use bxcore::{
CallId, DataClass, Empty, Envelope, MemoryAdmin, MemoryCall, MemoryCommand, Message, SessionId,
ToolRequest, ToolResponse, read_frame, write_frame,
};
use memoryd::config::Config;
use memoryd::serve;
use tmp::TempDir;
struct Running {
config: Config,
/// Removed when the test ends; declared last, so it outlives what writes into it (T44).
dir: TempDir,
}
fn start(tag: &str, review: bool) -> Running {
let dir = TempDir::new(tag);
dir.write("etc/soul.md", "You are Boxmaker.\n");
let text = format!(
"[paths]\nhome = \"{}\"\n[memory]\nsoul_starter = \"{}\"\nreview = {review}\n",
dir.path().display(),
dir.path().join("etc/soul.md").display()
);
let config = Config::parse(&text).unwrap();
let log: serve::Log = Arc::new(|_: &str| {});
let running = serve::start(config.clone(), log).unwrap();
std::thread::spawn(move || running.run());
Running { config, dir }
}
fn ask(socket: &Path, msg: Message) -> Message {
let mut stream = UnixStream::connect(socket).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.unwrap();
let env = Envelope {
v: 1,
id: 9,
r#final: true,
msg,
};
write_frame(&mut stream, &env).unwrap();
read_frame(&mut stream).unwrap().msg
}
/// A `memory` or `recall` call from a clean session; the answer's text.
fn call(r: &Running, tool: &str, arguments: &str) -> String {
let msg = Message::MemoryCall(MemoryCall {
request: ToolRequest {
session: SessionId::new("mm-a").unwrap(),
call: CallId(1),
tool: tool.to_string(),
arguments: arguments.to_string(),
},
taint: DataClass::Private,
untrusted: false,
});
match ask(&r.config.broker_socket(), msg) {
Message::ToolResponse(ToolResponse::Result { content, .. }) => content,
other => panic!("not a result: {other:?}"),
}
}
fn admin(r: &Running, command: MemoryCommand, arg: &str) -> String {
let msg = Message::MemoryAdmin(MemoryAdmin {
command,
arg: arg.to_string(),
});
match ask(&r.config.admin_socket(), msg) {
Message::MemoryAdminReply(reply) => reply.text,
other => panic!("{other:?}"),
}
}
fn memory_md(r: &Running) -> String {
std::fs::read_to_string(r.dir.path().join("memory/memory.md")).unwrap_or_default()
}
/// The id of the newest write waiting for review.
fn newest_pending(r: &Running) -> String {
let pending = admin(r, MemoryCommand::Pending, "");
let last = pending.lines().last().unwrap_or_default();
last.split_whitespace()
.next()
.unwrap_or_default()
.to_string()
}
#[test]
fn log_shows_the_last_ten_edits_unless_told_how_many() {
let r = start("admin-log", false);
for n in 1..=12 {
call(
&r,
"memory",
&format!(r#"{{"action":"note","text":"note {n}"}}"#),
);
}
assert_eq!(admin(&r, MemoryCommand::Log, "").lines().count(), 10);
assert_eq!(admin(&r, MemoryCommand::Log, "3").lines().count(), 3);
assert_eq!(admin(&r, MemoryCommand::Log, "20").lines().count(), 12);
}
#[test]
fn reject_discards_a_held_write_and_accept_applies_one() {
let r = start("admin-review", true);
call(
&r,
"memory",
r#"{"action":"add","file":"memory","text":"Rejected fact."}"#,
);
let rejected = newest_pending(&r);
assert!(admin(&r, MemoryCommand::Reject, &rejected).starts_with("rejected: "));
assert_eq!(memory_md(&r), "");
call(
&r,
"memory",
r#"{"action":"add","file":"memory","text":"Accepted fact."}"#,
);
let accepted = newest_pending(&r);
assert!(admin(&r, MemoryCommand::Accept, &accepted).starts_with("accepted: "));
assert_eq!(memory_md(&r), "Accepted fact.\n");
}
/// An accepted `consolidate` moves an entry to `notes.md`; the admin command indexes it, so
/// `recall` finds it at once, as identity moved out of `memory.md`.
#[test]
fn what_an_accepted_write_moves_to_notes_is_recalled_at_once() {
let r = start("admin-notes", true);
call(
&r,
"memory",
r#"{"action":"add","file":"memory","text":"Zebra crossings are painted on Mondays."}"#,
);
admin(&r, MemoryCommand::Accept, &newest_pending(&r));
assert_eq!(
call(&r, "recall", r#"{"query":"zebra"}"#),
"(no matches for \"zebra\")"
);
call(
&r,
"memory",
r#"{"action":"consolidate","file":"memory","entries":[{"old":"Zebra crossings are painted on Mondays.","move":true}]}"#,
);
admin(&r, MemoryCommand::Accept, &newest_pending(&r));
assert_eq!(memory_md(&r), "");
let found = call(&r, "recall", r#"{"query":"zebra"}"#);
assert!(found.starts_with("[1] identity · moved "), "{found}");
}
#[test]
fn a_reindex_counts_the_notes() {
let r = start("admin-reindex", false);
call(&r, "memory", r#"{"action":"note","text":"one note"}"#);
let Message::Reindexed(counts) = ask(&r.config.admin_socket(), Message::Reindex(Empty {}))
else {
panic!("not reindexed")
};
assert_eq!((counts.logs, counts.records, counts.notes), (0, 0, 1));
}