Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 92 additions & 8 deletions bin/memayu/src/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,14 @@
//! single-user by design: the user id comes from the config (default `default`)
//! and no auth is required for a local instance.

use crate::service::build_service;
use memayu_config::Config;
use memayu_core::{
BatchItemResult, BatchMemory, Memory, MemoryService, MetadataFilter, MAX_PAGE_SIZE,
};
use std::collections::{HashMap, HashSet};

use crate::service::build_service;
use memayu_core::{Memory, MetadataFilter, MAX_PAGE_SIZE};
use std::fs::File;
use std::io::{self, BufRead, BufReader};

/// Default user id for the local single-user CLI.
pub const DEFAULT_USER_ID: &str = "default";
Expand Down Expand Up @@ -53,7 +56,7 @@ fn parse_args(args: impl Iterator<Item = String>) -> Parsed {
if let Some(rest) = a.strip_prefix("--") {
if let Some((k, v)) = rest.split_once('=') {
insert_flag(&mut p, k, v.to_string());
} else if matches!(rest, "limit" | "cursor" | "filter") {
} else if matches!(rest, "limit" | "cursor" | "filter" | "batch") {
let v = it.next().unwrap_or_default();
insert_flag(&mut p, rest, v);
} else {
Expand Down Expand Up @@ -158,13 +161,21 @@ fn reject_unknown(
/// `memayu add "<content>"` — run ADD/UPDATE extraction and store the memory.
pub async fn cmd_add(config: &Config, args: impl Iterator<Item = String>) -> Result<(), String> {
let p = parse_args(args);
reject_unknown(&p, &[], &[])?;
reject_unknown(&p, &["batch"], &[])?;
guard_local(config)?;
let (service, _) = build_service(config).await.map_err(|e| e.to_string())?;

if let Some(path) = p.opts.get("batch") {
if !p.positionals.is_empty() {
return Err("--batch cannot be combined with positional content".to_string());
}
return cmd_add_batch(&service, path).await;
}

let content = p.positionals.join(" ").trim().to_string();
if content.is_empty() {
return Err("usage: memayu add \"<content>\"".to_string());
return Err("usage: memayu add \"<content>\" | --batch <file.jsonl | ->".to_string());
}
guard_local(config)?;
let (service, _) = build_service(config).await.map_err(|e| e.to_string())?;
let mem = service
.add_memory(DEFAULT_USER_ID, &content, &Default::default())
.await
Expand All @@ -173,6 +184,79 @@ pub async fn cmd_add(config: &Config, args: impl Iterator<Item = String>) -> Res
Ok(())
}

/// `memayu add --batch <file.jsonl | ->` — store many memories at once.
/// Each line is a JSON object with a `content` field (and optional `metadata`).
/// A `-` path reads from stdin. Per-item failures don't abort the batch.
async fn cmd_add_batch(service: &MemoryService, path: &str) -> Result<(), String> {
let reader: Box<dyn BufRead> = if path == "-" {
Box::new(BufReader::new(io::stdin()))
} else {
let file = File::open(path).map_err(|e| format!("cannot open {path}: {e}"))?;
Box::new(BufReader::new(file))
};

let mut items = Vec::new();
for (idx, line) in reader.lines().enumerate() {
let line_no = idx + 1;
let line = line
.map_err(|e| format!("read {path} line {line_no}: {e}"))?
.trim()
.to_string();
if line.is_empty() {
continue;
}
let value: serde_json::Value = serde_json::from_str(&line)
.map_err(|e| format!("{path}:{line_no}: invalid JSON: {e}"))?;
let content = value
.get("content")
.and_then(|c| c.as_str())
.unwrap_or("")
.to_string();
let metadata = value
.get("metadata")
.and_then(|m| m.as_object())
.map(|obj| {
obj.iter()
.filter_map(|(k, val)| val.as_str().map(|s| (k.clone(), s.to_string())))
.collect()
})
.unwrap_or_default();
items.push(BatchMemory { content, metadata });
}

if items.is_empty() {
return Err(format!(
"{path}: no memories found (file is empty or all blank)"
));
}

let results = service
.add_memories_batch(DEFAULT_USER_ID, &items)
.await
.map_err(|e| e.to_string())?;

let mut added = 0usize;
let mut failed = 0usize;
for result in &results {
match result {
BatchItemResult::Stored { memory_id } => {
added += 1;
println!("stored: {memory_id}");
}
BatchItemResult::Failed { error } => {
failed += 1;
eprintln!("failed: {error}");
}
}
}
println!("added: {added} memory(ies)");
if failed > 0 {
Err(format!("{failed} item(s) failed"))
} else {
Ok(())
}
}

/// `memayu search "<query>" [--limit N] [--filter key=value] [--json]` —
/// ranked semantic results.
pub async fn cmd_search(config: &Config, args: impl Iterator<Item = String>) -> Result<(), String> {
Expand Down
124 changes: 124 additions & 0 deletions bin/memayu/tests/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,73 @@ fn start_mock_embedder() -> u16 {
port
}

/// Like [`start_mock_embedder`] but derives a distinct embedding from each
/// request's full body. Used by batch tests where every item must get its own
/// vector so the storage layer doesn't treat them as duplicates and merge them.
fn start_distinct_embedder() -> u16 {
use std::hash::{Hash, Hasher};
let listener = TcpListener::bind("127.0.0.1:0").expect("bind distinct embedder");
let port = listener.local_addr().unwrap().port();
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(mut stream) = stream else { break };
// Read the full request: headers plus Content-Length bytes of body.
let mut buf = Vec::new();
let mut tmp = [0u8; 2048];
loop {
match stream.read(&mut tmp) {
Ok(0) => break,
Ok(n) => {
buf.extend_from_slice(&tmp[..n]);
if let Some(header_end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
let header = String::from_utf8_lossy(&buf[..header_end]).into_owned();
let content_length = header
.lines()
.find_map(|line| {
let mut parts = line.splitn(2, ':');
if parts
.next()
.map(|k| k.trim().eq_ignore_ascii_case("content-length"))
== Some(true)
{
parts.next()?.trim().parse::<usize>().ok()
} else {
None
}
})
.unwrap_or(0);
if buf.len() >= header_end + 4 + content_length {
break;
}
}
}
Err(_) => break,
}
}
let mut h = std::collections::hash_map::DefaultHasher::new();
buf.hash(&mut h);
let v = h.finish();
let emb = [
((v & 0xffff) as f32) / 65535.0,
(((v >> 16) & 0xffff) as f32) / 65535.0,
(((v >> 32) & 0xffff) as f32) / 65535.0,
];
let body = format!(
r#"{{"data":[{{"embedding":[{:.6},{:.6},{:.6}]}}]}}"#,
emb[0], emb[1], emb[2]
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
let _ = stream.write_all(resp.as_bytes());
let _ = stream.flush();
}
});
port
}

/// A temp libsql db path unique per test process.
fn temp_db_path() -> std::path::PathBuf {
static COUNTER: AtomicU16 = AtomicU16::new(0);
Expand Down Expand Up @@ -229,3 +296,60 @@ fn list_exposes_paging_contract() {
assert_eq!(obj["memories"].as_array().unwrap().len(), 0);
assert_eq!(obj["total"].as_u64(), Some(0));
}

#[test]
fn add_batch_stores_all_items() {
let port = start_distinct_embedder();
let db = temp_db_path();
let batch = temp_db_path().with_extension("batch.jsonl");
std::fs::write(
&batch,
concat!(
"{\"content\":\"alpha\",\"metadata\":{\"source\":\"batch\"}}\n",
"{\"content\":\"beta\"}\n",
"{\"content\":\"gamma\"}\n",
),
)
.unwrap();

let (code, stdout) = run_memayu(&["add", "--batch", batch.to_str().unwrap()], port, &db);
assert_eq!(code, 0, "batch add should succeed");
assert_eq!(stdout.matches("stored:").count(), 3);
assert!(stdout.contains("added: 3 memory(ies)"), "stdout: {stdout}");

// All three are persisted.
let (code, stdout) = run_memayu(&["list", "--json"], port, &db);
assert_eq!(code, 0);
let obj: serde_json::Value = serde_json::from_str(stdout.trim()).unwrap();
let contents: Vec<String> = obj["memories"]
.as_array()
.unwrap()
.iter()
.map(|m| m["content"].as_str().unwrap().to_string())
.collect();
for c in ["alpha", "beta", "gamma"] {
assert!(
contents.iter().any(|x| x == c),
"missing {c} in {contents:?}"
);
}
}

#[test]
fn add_batch_reports_partial_failures() {
let port = start_distinct_embedder();
let db = temp_db_path();
let batch = temp_db_path().with_extension("batch.jsonl");
std::fs::write(&batch, "{\"content\":\"ok\"}\n{\"content\":\" \"}\n").unwrap();

let (code, stdout, stderr) =
run_memayu_full(&["add", "--batch", batch.to_str().unwrap()], port, &db);
// The valid item is stored, but the blank one fails and we exit non-zero.
assert_ne!(code, 0, "partial failure should exit non-zero");
assert_eq!(stdout.matches("stored:").count(), 1);
assert!(
stderr.contains("failed: memory content is required"),
"stderr: {stderr}"
);
assert!(stderr.contains("1 item(s) failed"), "stderr: {stderr}");
}
1 change: 1 addition & 0 deletions crates/memayu-api/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ impl From<CoreError> for ApiError {
let status = match &e {
CoreError::DimensionMismatch { .. } => 422,
CoreError::InvalidExtraction(_) => 422,
CoreError::InvalidInput(_) => 400,
CoreError::NotFound(_) => 404,
CoreError::InvalidCursor(_) => 400,
CoreError::LimitExceeded { .. } => 400,
Expand Down
33 changes: 33 additions & 0 deletions crates/memayu-api/src/modules/memory/dto.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,39 @@ pub struct AddMemoryRequest {
pub metadata: Metadata,
}

/// One memory within a batch add request.
#[derive(Debug, Deserialize, ToSchema)]
pub struct AddMemoryItem {
pub content: String,
#[serde(default)]
pub metadata: Metadata,
}

/// Batch add request — `{"memories": [{"content": ..., "metadata": ...}, ...]}`.
#[derive(Debug, Deserialize, ToSchema)]
pub struct AddMemoriesBatchRequest {
pub memories: Vec<AddMemoryItem>,
}

/// A single failed item within a batch response.
#[derive(Debug, Serialize, ToSchema)]
pub struct BatchMemoryError {
/// Zero-based index of the failed item in the request's `memories` array.
pub index: usize,
pub error: String,
}

#[derive(Debug, Serialize, ToSchema)]
pub struct AddMemoriesBatchResponse {
pub status: String,
/// Number of memories actually stored (successes).
pub added: usize,
/// Memory ids of the successfully stored items, in request order.
pub memory_ids: Vec<String>,
/// Per-item failures. Empty when every item was stored.
pub errors: Vec<BatchMemoryError>,
}

#[derive(Debug, Serialize, ToSchema)]
pub struct AddMemoryResponse {
pub status: String,
Expand Down
64 changes: 60 additions & 4 deletions crates/memayu-api/src/transport/handlers/memory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,15 @@
/// the core MemoryService.
use crate::error::{ApiError, ApiErrorBody, ApiResult};
use crate::modules::memory::dto::{
AddMemoryRequest, AddMemoryResponse, ListMemoryResponse, ListQuery, ListedMemory,
SearchMemoryRequest, SearchMemoryResponse, SearchResult, UpdateMemoryRequest,
UpdateMemoryResponse,
AddMemoriesBatchRequest, AddMemoriesBatchResponse, AddMemoryRequest, AddMemoryResponse,
BatchMemoryError, ListMemoryResponse, ListQuery, ListedMemory, SearchMemoryRequest,
SearchMemoryResponse, SearchResult, UpdateMemoryRequest, UpdateMemoryResponse,
};
use crate::transport::middleware::{AccountId, ApiState};
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::Json;
use memayu_core::MetadataFilter;
use memayu_core::{BatchItemResult, BatchMemory, MetadataFilter};

/// Add a new memory for a user
#[utoipa::path(
Expand Down Expand Up @@ -49,6 +49,62 @@ pub async fn add_memory(
))
}

/// Add many memories for a user in a single call. A failure on one item does
/// not fail the rest; successes are reported via `added`/`memory_ids` and
/// failures via `errors`.
#[utoipa::path(
post,
path = "/api/memories/batch",
request_body = AddMemoriesBatchRequest,
responses(
(status = 200, description = "Memories added", body = ApiResult<AddMemoriesBatchResponse>),
(status = 400, description = "Bad request", body = ApiErrorBody),
)
)]
pub async fn add_memories_batch(
State(state): State<ApiState>,
_account: AccountId,
Json(req): Json<AddMemoriesBatchRequest>,
) -> Result<Json<ApiResult<AddMemoriesBatchResponse>>, ApiError> {
let user_id = &_account.0;
if req.memories.is_empty() {
return Err(ApiError::bad_request("memories must not be empty"));
}

let items: Vec<BatchMemory> = req
.memories
.into_iter()
.map(|m| BatchMemory {
content: m.content,
metadata: m.metadata,
})
.collect();

let results = state.service.add_memories_batch(user_id, &items).await?;

let mut added = 0usize;
let mut memory_ids = Vec::new();
let mut errors = Vec::new();
for (index, outcome) in results.into_iter().enumerate() {
match outcome {
BatchItemResult::Stored { memory_id } => {
added += 1;
memory_ids.push(memory_id);
}
BatchItemResult::Failed { error } => errors.push(BatchMemoryError { index, error }),
}
}

Ok(Json(ApiResult {
result: AddMemoriesBatchResponse {
status: "success".into(),
added,
memory_ids,
errors,
},
}))
}

/// Search memories by semantic similarity
#[utoipa::path(
post,
Expand Down
Loading
Loading