use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct MemoryRecord {
pub namespace: String,
pub timestamp_ms: u64,
pub content: String,
#[serde(default, deserialize_with = "null_as_no_tags")]
pub tags: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub meta: Option<serde_json::Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub project: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub provenance: Option<MemoryProvenance>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub sensitivity: Option<MemorySensitivity>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub retention: Option<MemoryRetention>,
}
fn null_as_no_tags<'de, D: serde::Deserializer<'de>>(reader: D) -> Result<Vec<String>, D::Error> {
Ok(Option::<Vec<String>>::deserialize(reader)?.unwrap_or_default())
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct MemoryProvenance {
pub source: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session: Option<String>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum MemorySensitivity {
Normal,
Sensitive,
Secret,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum MemoryRetention {
Standard,
MaxAgeDays(u32),
}
#[derive(Debug, Clone, Default)]
pub struct MemoryQuery {
pub content_contains: Option<String>,
pub tag_prefix: Option<String>,
pub since_ms: Option<u64>,
pub until_ms: Option<u64>,
pub limit: Option<usize>,
}
pub const DEFAULT_LIMIT: usize = 50;
pub const MAX_CONTENT_BYTES: usize = 16 * 1024;
#[derive(Debug)]
pub enum MemoryError {
InvalidNamespace(String),
ContentTooLarge {
len: usize,
max: usize,
},
Io(String),
Serde(String),
NotFound(String),
InvalidProjectRoot(String),
WrittenUnconfirmed { id: Option<String>, cause: String },
}
impl MemoryError {
fn unconfirmed(cause: impl std::fmt::Display) -> Self {
MemoryError::WrittenUnconfirmed {
id: None,
cause: cause.to_string(),
}
}
fn with_id(self, id: Option<&str>) -> Self {
match self {
MemoryError::WrittenUnconfirmed { id: None, cause } => {
MemoryError::WrittenUnconfirmed {
id: id.map(str::to_string),
cause,
}
}
other => other,
}
}
}
impl std::fmt::Display for MemoryError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
MemoryError::InvalidNamespace(s) => write!(f, "invalid namespace: {s:?}"),
MemoryError::ContentTooLarge { len, max } => {
write!(f, "content too large: {len} bytes (max {max})")
}
MemoryError::Io(s) => write!(f, "memory io error: {s}"),
MemoryError::Serde(s) => write!(f, "memory serde error: {s}"),
MemoryError::NotFound(id) => {
write!(f, "memory record not found in this project scope: {id}")
}
MemoryError::InvalidProjectRoot(s) => write!(f, "invalid project root: {s}"),
MemoryError::WrittenUnconfirmed { id: None, cause } => write!(
f,
"written, but not confirmed on disk (the sync failed: {cause}); \
it is in the store now, so look before doing it again"
),
MemoryError::WrittenUnconfirmed {
id: Some(id),
cause,
} => write!(
f,
"written, but not confirmed on disk (the sync failed: {cause}); \
{id} is in the store now, so fetch {id} to check before doing it again"
),
}
}
}
impl std::error::Error for MemoryError {}
impl From<std::io::Error> for MemoryError {
fn from(e: std::io::Error) -> Self {
MemoryError::Io(e.to_string())
}
}
impl From<serde_json::Error> for MemoryError {
fn from(e: serde_json::Error) -> Self {
MemoryError::Serde(e.to_string())
}
}
pub fn now_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
fn validate_namespace(ns: &str) -> Result<(), MemoryError> {
if ns.is_empty()
|| ns.len() > 64
|| ns.contains('/')
|| ns.contains('\\')
|| ns.contains("..")
|| ns.chars().any(|c| c.is_whitespace())
{
return Err(MemoryError::InvalidNamespace(ns.to_string()));
}
Ok(())
}
pub fn memory_dir() -> PathBuf {
crate::config::base_dir().join("memory")
}
pub fn memory_dir_in(base: &Path) -> PathBuf {
base.join("memory")
}
fn namespace_path(dir: &Path, ns: &str) -> PathBuf {
dir.join(format!("{ns}.jsonl"))
}
pub fn append(record: &MemoryRecord) -> Result<(), MemoryError> {
append_to(&crate::config::base_dir(), record)
}
pub fn append_to(base: &Path, record: &MemoryRecord) -> Result<(), MemoryError> {
append_record(base, record, true)
}
fn append_record(base: &Path, record: &MemoryRecord, sync_now: bool) -> Result<(), MemoryError> {
validate_namespace(&record.namespace)?;
if record.content.len() > MAX_CONTENT_BYTES {
return Err(MemoryError::ContentTooLarge {
len: record.content.len(),
max: MAX_CONTENT_BYTES,
});
}
let dir = memory_dir_in(base);
let new_dir = !dir.exists();
crate::core::private_fs::ensure_private_dir(&dir).map_err(std::io::Error::from)?;
if new_dir {
if let Some(parent) = dir.parent() {
sync_dir(parent)?;
}
}
let path = namespace_path(&dir, &record.namespace);
let line = serde_json::to_string(record)?;
append_jsonl_line(&path, &line, sync_now).map_err(|e| e.with_id(record.id.as_deref()))
}
fn append_jsonl_line(path: &Path, line: &str, sync_now: bool) -> Result<(), MemoryError> {
let _lock = lock_notes_file(path)?;
let needs_heal = match fs::metadata(path) {
Ok(meta) if meta.len() > 0 => {
use std::io::{Read, Seek, SeekFrom};
let mut f = fs::File::open(path)?;
f.seek(SeekFrom::End(-1))?;
let mut last = [0u8; 1];
f.read_exact(&mut last)?;
last[0] != b'\n'
}
_ => false,
};
let mut f = crate::core::private_fs::open_private_append(path).map_err(std::io::Error::from)?;
write_whole_line(&mut f, line, needs_heal)?;
if sync_now {
let synced = sync_file(&f).and_then(|()| match path.parent() {
Some(dir) => sync_dir(dir),
None => Ok(()),
});
synced.map_err(MemoryError::unconfirmed)?;
}
Ok(())
}
pub(crate) const NOTES_LOCK_WAIT: std::time::Duration =
crate::context_archive::PROCESS_LOCK_WAIT;
const NOTES_LOCK_POLL: std::time::Duration = std::time::Duration::from_millis(5);
pub(crate) fn notes_lock_path(path: &Path) -> PathBuf {
let mut name = path.as_os_str().to_os_string();
name.push(".lock");
PathBuf::from(name)
}
pub(crate) fn lock_notes_file(
path: &Path,
) -> std::io::Result<crate::core::lock_file::ExclusiveLock> {
let lock_path = notes_lock_path(path);
let wait = notes_lock_wait();
crate::core::lock_file::lock_within(wait, NOTES_LOCK_POLL, || {
let file = open_lock_file(&lock_path).map_err(|e| {
std::io::Error::new(
e.kind(),
format!(
"cannot open the notes lock file {}: {e}; nothing was written",
lock_path.display()
),
)
})?;
crate::core::lock_file::try_lock(file)
})?
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::WouldBlock,
format!(
"notes file {} is locked by another writer (waited {} ms); nothing was written",
path.file_name().unwrap_or_default().to_string_lossy(),
wait.as_millis()
),
)
})
}
pub fn open_lock_file(path: &Path) -> std::io::Result<fs::File> {
let mut options = fs::OpenOptions::new();
options.read(true).write(true).create(true).truncate(false);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options
.mode(crate::core::private_fs::FILE_MODE)
.custom_flags(libc::O_NOFOLLOW);
}
options.open(path)
}
#[cfg(test)]
thread_local! {
static NOTES_LOCK_WAIT_OVERRIDE: std::cell::Cell<Option<std::time::Duration>> =
const { std::cell::Cell::new(None) };
}
#[cfg(test)]
pub(crate) fn set_notes_lock_wait(wait: Option<std::time::Duration>) {
NOTES_LOCK_WAIT_OVERRIDE.with(|cell| cell.set(wait));
}
fn notes_lock_wait() -> std::time::Duration {
#[cfg(test)]
if let Some(wait) = NOTES_LOCK_WAIT_OVERRIDE.with(|cell| cell.get()) {
return wait;
}
NOTES_LOCK_WAIT
}
const LINE_WRITE_ATTEMPTS: usize = 3;
fn write_whole_line<W: Write>(
file: &mut W,
line: &str,
mut lead_newline: bool,
) -> std::io::Result<()> {
let mut attempts = 0;
while attempts < LINE_WRITE_ATTEMPTS {
let mut out = Vec::with_capacity(line.len() + 2);
if lead_newline {
out.push(b'\n');
}
out.extend_from_slice(line.as_bytes());
out.push(b'\n');
step("write");
match file.write(&out) {
Ok(n) if n == out.len() => return Ok(()),
Ok(n) => {
attempts += 1;
lead_newline |= n > 0;
}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(e) => return Err(e),
}
}
Err(std::io::Error::new(
std::io::ErrorKind::WriteZero,
"a line could not be appended with one write",
))
}
fn sync_file(file: &fs::File) -> std::io::Result<()> {
step("sync file");
failing("sync file")?;
file.sync_all()
}
fn sync_dir(dir: &Path) -> std::io::Result<()> {
step("sync dir");
failing("sync dir")?;
crate::core::private_fs::sync_dir(dir)
}
#[cfg(test)]
thread_local! {
static IO_STEPS: std::cell::RefCell<Vec<&'static str>> =
const { std::cell::RefCell::new(Vec::new()) };
}
#[cfg(test)]
fn step(name: &'static str) {
IO_STEPS.with(|steps| steps.borrow_mut().push(name));
}
#[cfg(not(test))]
fn step(_name: &'static str) {}
#[cfg(test)]
thread_local! {
static FAILING_STEP: std::cell::Cell<Option<&'static str>> =
const { std::cell::Cell::new(None) };
}
#[cfg(test)]
fn failing(name: &'static str) -> std::io::Result<()> {
if FAILING_STEP.with(|step| step.get()) == Some(name) {
return Err(std::io::Error::other(format!("{name} failed (test)")));
}
Ok(())
}
#[cfg(not(test))]
fn failing(_name: &'static str) -> std::io::Result<()> {
Ok(())
}
pub fn query(namespace: &str, q: &MemoryQuery) -> Result<Vec<MemoryRecord>, MemoryError> {
query_in(&crate::config::base_dir(), namespace, q)
}
pub fn query_in(
base: &Path,
namespace: &str,
q: &MemoryQuery,
) -> Result<Vec<MemoryRecord>, MemoryError> {
validate_namespace(namespace)?;
let path = namespace_path(&memory_dir_in(base), namespace);
if !path.exists() {
return Ok(Vec::new());
}
let f = fs::File::open(&path)?;
let reader = BufReader::new(f);
let needle = q.content_contains.as_ref().map(|s| s.to_lowercase());
let mut out: Vec<MemoryRecord> = Vec::new();
for line in reader.lines() {
let line = match line {
Ok(l) => l,
Err(e) if e.kind() == std::io::ErrorKind::InvalidData => continue,
Err(e) => return Err(e.into()),
};
if line.trim().is_empty() {
continue;
}
let rec: MemoryRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(_) => continue,
};
if let Some(since) = q.since_ms {
if rec.timestamp_ms < since {
continue;
}
}
if let Some(until) = q.until_ms {
if rec.timestamp_ms > until {
continue;
}
}
if let Some(needle) = &needle {
if !rec.content.to_lowercase().contains(needle) {
continue;
}
}
if let Some(prefix) = &q.tag_prefix {
if !rec.tags.iter().any(|t| t.starts_with(prefix)) {
continue;
}
}
out.push(rec);
}
out.sort_by_key(|b| std::cmp::Reverse(b.timestamp_ms));
let limit = q.limit.unwrap_or(DEFAULT_LIMIT);
out.truncate(limit);
Ok(out)
}
pub fn list_namespaces() -> Result<Vec<String>, MemoryError> {
list_namespaces_in(&crate::config::base_dir())
}
pub(crate) fn list_namespaces_in(base: &Path) -> Result<Vec<String>, MemoryError> {
let dir = memory_dir_in(base);
if !dir.exists() {
return Ok(Vec::new());
}
let mut out = Vec::new();
for entry in fs::read_dir(&dir)? {
let entry = entry?;
let path = entry.path();
if path.extension().and_then(|s| s.to_str()) == Some("jsonl") {
if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
out.push(stem.to_string());
}
}
}
out.sort();
Ok(out)
}
pub fn new_record(
namespace: impl Into<String>,
content: impl Into<String>,
tags: Vec<String>,
meta: Option<serde_json::Value>,
) -> MemoryRecord {
MemoryRecord {
namespace: namespace.into(),
timestamp_ms: now_ms(),
content: content.into(),
tags,
meta,
id: None,
project: None,
provenance: None,
sensitivity: None,
retention: None,
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProjectScope {
key: String,
root: PathBuf,
}
impl ProjectScope {
pub fn for_root(root: &Path) -> Result<Self, MemoryError> {
let canonical = root
.canonicalize()
.map_err(|e| MemoryError::InvalidProjectRoot(format!("{}: {e}", root.display())))?;
Ok(Self::for_canonical_root(canonical))
}
fn for_canonical_root(canonical: PathBuf) -> Self {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(canonical.as_os_str().as_encoded_bytes());
let digest = hasher.finalize();
let mut key = String::with_capacity(17);
key.push('p');
for byte in digest.iter().take(8) {
use std::fmt::Write;
let _ = write!(key, "{byte:02x}");
}
Self {
key,
root: canonical,
}
}
fn validate_repository_key(key: &str) -> Result<(), MemoryError> {
if key.len() != 17
|| !key.starts_with('p')
|| !key.as_bytes()[1..]
.iter()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(b))
|| key == "p0000000000000000"
{
return Err(MemoryError::InvalidProjectRoot(
"invalid or reserved repository scope key".into(),
));
}
Ok(())
}
pub fn key(&self) -> &str {
&self.key
}
pub fn discover(start: &Path) -> Result<Self, MemoryError> {
let override_root = std::env::var_os("SYNAPS_PROJECT_ROOT").map(std::path::PathBuf::from);
Self::discover_with_override(start, override_root.as_deref())
}
pub fn discover_with_override(
start: &Path,
override_root: Option<&Path>,
) -> Result<Self, MemoryError> {
if let Some(root) = override_root {
return Self::for_root(root);
}
const MARKERS: [&str; 2] = [".git", ".synaps-project"];
const MAX_ASCENT: usize = 64;
let canonical_start = start
.canonicalize()
.map_err(|e| MemoryError::InvalidProjectRoot(format!("{}: {e}", start.display())))?;
let mut dir = canonical_start.as_path();
for _ in 0..MAX_ASCENT {
if MARKERS
.iter()
.any(|marker| dir.join(marker).symlink_metadata().is_ok())
{
return Self::for_root(dir);
}
match dir.parent() {
Some(parent) => dir = parent,
None => break,
}
}
Self::for_root(&canonical_start)
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn namespace(&self) -> String {
project_namespace(&self.key)
}
}
fn project_namespace(key: &str) -> String {
format!("project-{key}")
}
#[derive(Debug, Clone)]
pub struct NewMemoryRecord {
pub content: String,
pub tags: Vec<String>,
pub provenance: MemoryProvenance,
pub sensitivity: MemorySensitivity,
pub retention: MemoryRetention,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemoryDescriptor {
pub id: String,
pub project: String,
pub timestamp_ms: u64,
pub tags: Vec<String>,
pub snippet: String,
pub truncated: bool,
pub content_bytes: usize,
pub sensitivity: MemorySensitivity,
pub retention: MemoryRetention,
}
pub const MAX_SEARCH_LIMIT: usize = 25;
pub const DEFAULT_SEARCH_LIMIT: usize = 8;
pub const MAX_SNIPPET_BYTES: usize = 400;
pub const DEFAULT_SNIPPET_BYTES: usize = 160;
#[derive(Debug, Clone, Default)]
pub struct ProjectMemoryQuery {
pub content_contains: Option<String>,
pub tag_prefix: Option<String>,
pub since_ms: Option<u64>,
pub until_ms: Option<u64>,
pub limit: Option<usize>,
pub snippet_bytes: Option<usize>,
}
#[derive(Debug, Serialize, Deserialize)]
struct TombstoneLine {
tombstone: String,
timestamp_ms: u64,
}
pub fn store_record_in(
base: &Path,
scope: &ProjectScope,
new: NewMemoryRecord,
) -> Result<MemoryRecord, MemoryError> {
let record = MemoryRecord {
namespace: scope.namespace(),
timestamp_ms: now_ms(),
content: new.content,
tags: new.tags,
meta: None,
id: Some(format!("mem-{}", uuid::Uuid::new_v4().simple())),
project: Some(scope.key().to_string()),
provenance: Some(new.provenance),
sensitivity: Some(new.sensitivity),
retention: Some(new.retention),
};
append_to(base, &record)?;
Ok(record)
}
struct ProjectLines {
records: Vec<MemoryRecord>,
tombstoned: std::collections::HashSet<String>,
}
fn scan_project_file(base: &Path, key: &str) -> Result<ProjectLines, MemoryError> {
let mut lines = ProjectLines {
records: Vec::new(),
tombstoned: std::collections::HashSet::new(),
};
let path = namespace_path(&memory_dir_in(base), &project_namespace(key));
let f = match fs::File::open(path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(lines),
Err(e) => return Err(e.into()),
};
for line in BufReader::new(f).lines() {
let line = match line {
Ok(l) => l,
Err(e) if e.kind() == std::io::ErrorKind::InvalidData => continue,
Err(e) => return Err(e.into()),
};
if line.trim().is_empty() {
continue;
}
if let Ok(tomb) = serde_json::from_str::<TombstoneLine>(&line) {
lines.tombstoned.insert(tomb.tombstone);
continue;
}
let rec: MemoryRecord = match serde_json::from_str(&line) {
Ok(r) => r,
Err(_) => continue,
};
if rec.project.as_deref() != Some(key) || rec.id.is_none() {
continue;
}
lines.records.push(rec);
}
Ok(lines)
}
fn load_live_project_records(
base: &Path,
scope: &ProjectScope,
) -> Result<Vec<MemoryRecord>, MemoryError> {
let ProjectLines {
mut records,
tombstoned,
} = scan_project_file(base, scope.key())?;
records.retain(|r| r.id.as_ref().is_some_and(|id| !tombstoned.contains(id)));
Ok(records)
}
pub fn search_project_in(
base: &Path,
scope: &ProjectScope,
q: &ProjectMemoryQuery,
) -> Result<Vec<MemoryDescriptor>, MemoryError> {
let needle = q.content_contains.as_ref().map(|s| s.to_lowercase());
let snippet_bytes = q
.snippet_bytes
.unwrap_or(DEFAULT_SNIPPET_BYTES)
.min(MAX_SNIPPET_BYTES);
let limit = q
.limit
.unwrap_or(DEFAULT_SEARCH_LIMIT)
.min(MAX_SEARCH_LIMIT);
let mut records = load_live_project_records(base, scope)?;
records.retain(|rec| {
let secret = rec.sensitivity == Some(MemorySensitivity::Secret);
if let Some(since) = q.since_ms {
if rec.timestamp_ms < since {
return false;
}
}
if let Some(until) = q.until_ms {
if rec.timestamp_ms > until {
return false;
}
}
if let Some(needle) = &needle {
if secret || !rec.content.to_lowercase().contains(needle) {
return false;
}
}
if let Some(prefix) = &q.tag_prefix {
if !rec.tags.iter().any(|t| t.starts_with(prefix)) {
return false;
}
}
true
});
records.sort_by_key(|r| std::cmp::Reverse(r.timestamp_ms));
records.truncate(limit);
Ok(records
.into_iter()
.map(|rec| {
let sensitivity = rec.sensitivity.unwrap_or(MemorySensitivity::Normal);
let (snippet, truncated) = if sensitivity == MemorySensitivity::Secret {
(String::new(), false)
} else {
let bounded = crate::text::BoundedText::new(&rec.content, snippet_bytes);
(bounded.text, bounded.truncated)
};
MemoryDescriptor {
id: rec.id.clone().unwrap_or_default(),
project: scope.key().to_string(),
timestamp_ms: rec.timestamp_ms,
tags: rec.tags,
snippet,
truncated,
content_bytes: rec.content.len(),
sensitivity,
retention: rec.retention.unwrap_or(MemoryRetention::Standard),
}
})
.collect())
}
pub fn fetch_exact_in(
base: &Path,
scope: &ProjectScope,
ids: &[&str],
) -> Result<Vec<MemoryRecord>, MemoryError> {
let live = load_live_project_records(base, scope)?;
let mut out = Vec::with_capacity(ids.len());
for wanted in ids {
match live.iter().find(|r| r.id.as_deref() == Some(*wanted)) {
Some(rec) => out.push(rec.clone()),
None => return Err(MemoryError::NotFound((*wanted).to_string())),
}
}
Ok(out)
}
pub fn forget_in(base: &Path, scope: &ProjectScope, id: &str) -> Result<(), MemoryError> {
let live = load_live_project_records(base, scope)?;
if !live.iter().any(|r| r.id.as_deref() == Some(id)) {
return Err(MemoryError::NotFound(id.to_string()));
}
let tombstone = TombstoneLine {
tombstone: id.to_string(),
timestamp_ms: now_ms(),
};
append_tombstone(base, &scope.namespace(), &tombstone, true)
}
fn append_tombstone(
base: &Path,
namespace: &str,
tombstone: &TombstoneLine,
sync_now: bool,
) -> Result<(), MemoryError> {
let dir = memory_dir_in(base);
crate::core::private_fs::ensure_private_dir(&dir).map_err(std::io::Error::from)?;
let path = namespace_path(&dir, namespace);
let line = serde_json::to_string(tombstone)?;
append_jsonl_line(&path, &line, sync_now).map_err(|e| e.with_id(Some(&tombstone.tombstone)))
}
pub(crate) fn namespace_file(base: &Path, namespace: &str) -> PathBuf {
namespace_path(&memory_dir_in(base), namespace)
}
pub fn project_notes_file(base: &Path, key: &str) -> PathBuf {
namespace_file(base, &project_namespace(key))
}
const USER_WIDE_KEY: &str = "p0000000000000000";
const RECORD_FIELDS: [&str; 10] = [
"namespace",
"timestamp_ms",
"content",
"tags",
"meta",
"id",
"project",
"provenance",
"sensitivity",
"retention",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ImportTarget<'a> {
AsRecorded,
Project(&'a str),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ImportOutcome {
Stored,
AlreadyPresent,
KeptForgotten,
Forgotten,
Refused(String),
}
impl ImportOutcome {
pub fn name(&self) -> &'static str {
match self {
Self::Stored => "stored",
Self::AlreadyPresent => "already_present",
Self::KeptForgotten => "kept_forgotten",
Self::Forgotten => "forgotten",
Self::Refused(_) => "refused",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImportedLine {
pub outcome: ImportOutcome,
pub id: Option<String>,
pub project: Option<String>,
pub tombstone: bool,
}
#[derive(Debug)]
#[must_use = "end an import run with `finish()`: nothing it stored is durable before that"]
pub struct ImportRun {
dry_run: bool,
files: std::collections::HashMap<PathBuf, ImportFile>,
unfiled_tombstones: std::collections::HashSet<String>,
read_every_project: bool,
finished: bool,
}
#[derive(Debug)]
struct ImportFile {
key: String,
notes: std::collections::HashMap<String, Vec<MemoryRecord>>,
tombstoned: std::collections::HashSet<String>,
appended: bool,
}
impl ImportRun {
pub fn new(dry_run: bool) -> Self {
Self {
dry_run,
files: std::collections::HashMap::new(),
unfiled_tombstones: std::collections::HashSet::new(),
read_every_project: false,
finished: false,
}
}
pub fn is_dry_run(&self) -> bool {
self.dry_run
}
fn file(&mut self, base: &Path, key: &str) -> Result<&mut ImportFile, MemoryError> {
use std::collections::hash_map::Entry;
match self
.files
.entry(namespace_file(base, &project_namespace(key)))
{
Entry::Occupied(known) => Ok(known.into_mut()),
Entry::Vacant(vacant) => {
let lines = scan_project_file(base, key)?;
let mut notes: std::collections::HashMap<String, Vec<MemoryRecord>> =
std::collections::HashMap::new();
for record in lines.records {
if let Some(id) = record.id.clone() {
notes.entry(id).or_default().push(record);
}
}
Ok(vacant.insert(ImportFile {
key: key.to_string(),
notes,
tombstoned: lines.tombstoned,
appended: false,
}))
}
}
}
fn read_every_project(&mut self, base: &Path) -> Result<(), MemoryError> {
if self.read_every_project {
return Ok(());
}
let entries = match fs::read_dir(memory_dir_in(base)) {
Ok(entries) => Some(entries),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
Err(e) => return Err(e.into()),
};
for entry in entries.into_iter().flatten() {
let name = entry?.file_name();
let key = name
.to_str()
.and_then(|name| name.strip_prefix("project-"))
.and_then(|name| name.strip_suffix(".jsonl"));
if let Some(key) = key.filter(|key| ProjectScope::validate_repository_key(key).is_ok())
{
self.file(base, key)?;
}
}
self.read_every_project = true;
Ok(())
}
fn projects_holding(&self, id: &str) -> Vec<String> {
let mut keys: Vec<String> = self
.files
.values()
.filter(|file| file.notes.contains_key(id) || file.tombstoned.contains(id))
.map(|file| file.key.clone())
.collect();
keys.sort();
keys
}
#[must_use = "an import is not durable unless `finish` returned Ok"]
pub fn finish(mut self) -> Result<Vec<PathBuf>, MemoryError> {
self.finished = true;
self.sync()
.map_err(|(path, e)| MemoryError::unconfirmed(format!("{}: {e}", path.display())))
}
fn sync(&self) -> Result<Vec<PathBuf>, (PathBuf, std::io::Error)> {
let mut synced: Vec<PathBuf> = self
.files
.iter()
.filter(|(_, file)| file.appended)
.map(|(path, _)| path.clone())
.collect();
synced.sort();
let mut first_error: Option<(PathBuf, std::io::Error)> = None;
let mut dirs: std::collections::BTreeSet<&Path> = std::collections::BTreeSet::new();
for path in &synced {
let result = fs::OpenOptions::new()
.append(true)
.open(path)
.and_then(|file| sync_file(&file));
if let Err(e) = result {
first_error.get_or_insert((path.clone(), e));
}
dirs.extend(path.parent());
}
for dir in dirs {
if let Err(e) = sync_dir(dir) {
first_error.get_or_insert((dir.to_path_buf(), e));
}
}
match first_error {
Some(error) => Err(error),
None => Ok(synced),
}
}
}
impl Drop for ImportRun {
fn drop(&mut self) {
if self.finished || !self.files.values().any(|file| file.appended) {
return;
}
if let Err((path, error)) = self.sync() {
tracing::warn!(
"an import run was dropped unfinished and its sync of {} failed: {error}",
path.display()
);
}
}
}
pub fn import_line_in(
base: &Path,
line: &str,
target: ImportTarget<'_>,
run: &mut ImportRun,
) -> Result<ImportedLine, MemoryError> {
if let ImportTarget::Project(key) = target {
ProjectScope::validate_repository_key(key)?;
}
let mut about = ImportedLine {
outcome: ImportOutcome::Refused(String::new()),
id: None,
project: None,
tombstone: false,
};
let outcome = match parse_import_line(line) {
Err(reason) => ImportOutcome::Refused(reason),
Ok(ImportLine::Tombstone(tombstone)) => {
about.tombstone = true;
import_tombstone(base, tombstone, target, run, &mut about)?
}
Ok(ImportLine::Note(note)) => import_note(base, *note, target, run, &mut about)?,
};
about.outcome = outcome;
Ok(about)
}
const UNFILED_TOMBSTONE: &str = "its tombstone in this run could not be filed";
const UNPLACED_TOMBSTONE: &str = "a tombstone line names no project and no project's notes hold \
its id; pass --cwd or --project-key";
fn refused(reason: &str) -> Result<ImportOutcome, MemoryError> {
Ok(ImportOutcome::Refused(reason.to_string()))
}
fn import_note(
base: &Path,
mut note: MemoryRecord,
target: ImportTarget<'_>,
run: &mut ImportRun,
about: &mut ImportedLine,
) -> Result<ImportOutcome, MemoryError> {
let Some(id) = note.id.clone() else {
return refused("missing field id");
};
if !valid_import_id(&id) {
return refused("invalid id");
}
about.id = Some(id.clone());
if run.unfiled_tombstones.contains(&id) {
return refused(UNFILED_TOMBSTONE);
}
let Some(recorded) = note.project.clone() else {
return refused("missing field project");
};
if recorded != USER_WIDE_KEY && ProjectScope::validate_repository_key(&recorded).is_err() {
return refused("invalid project");
}
if note.namespace != project_namespace(&recorded) {
return refused("namespace is not project-<project>");
}
if note.content.len() > MAX_CONTENT_BYTES {
return refused(&format!("content is over {MAX_CONTENT_BYTES} bytes"));
}
let key = match target {
ImportTarget::Project(key) => {
note.project = Some(key.to_string());
note.namespace = project_namespace(key);
key.to_string()
}
ImportTarget::AsRecorded if recorded == USER_WIDE_KEY => {
return refused("user-wide note; pass --cwd to file it under a project");
}
ImportTarget::AsRecorded => recorded,
};
about.project = Some(key.clone());
let dry_run = run.dry_run;
let file = run.file(base, &key)?;
if file.tombstoned.contains(&id) {
return Ok(ImportOutcome::KeptForgotten);
}
if let Some(held) = file.notes.get(&id) {
if held.iter().any(|other| same_note(other, ¬e)) {
return Ok(ImportOutcome::AlreadyPresent);
}
return refused("id already holds a different note");
}
if !dry_run {
append_record(base, ¬e, false)?;
file.appended = true;
}
file.notes.insert(id, vec![note]);
Ok(ImportOutcome::Stored)
}
fn import_tombstone(
base: &Path,
tombstone: TombstoneLine,
target: ImportTarget<'_>,
run: &mut ImportRun,
about: &mut ImportedLine,
) -> Result<ImportOutcome, MemoryError> {
if !valid_import_id(&tombstone.tombstone) {
return refused("invalid id");
}
about.id = Some(tombstone.tombstone.clone());
let keys = match target {
ImportTarget::Project(key) => vec![key.to_string()],
ImportTarget::AsRecorded => {
let mut keys = run.projects_holding(&tombstone.tombstone);
if keys.is_empty() {
run.read_every_project(base)?;
keys = run.projects_holding(&tombstone.tombstone);
}
keys
}
};
if keys.is_empty() {
run.unfiled_tombstones.insert(tombstone.tombstone);
return refused(UNPLACED_TOMBSTONE);
}
about.project = keys.first().cloned();
let dry_run = run.dry_run;
let (mut forgot, mut stored) = (false, false);
for key in &keys {
let file = run.file(base, key)?;
if file.tombstoned.contains(&tombstone.tombstone) {
continue;
}
if !dry_run {
append_tombstone(base, &project_namespace(key), &tombstone, false)?;
file.appended = true;
}
if file.notes.contains_key(&tombstone.tombstone) {
forgot = true;
} else {
stored = true;
}
file.tombstoned.insert(tombstone.tombstone.clone());
}
Ok(if forgot {
ImportOutcome::Forgotten
} else if stored {
ImportOutcome::Stored
} else {
ImportOutcome::AlreadyPresent
})
}
pub(crate) fn valid_import_id(id: &str) -> bool {
(1..=128).contains(&id.len()) && !id.chars().any(|c| c.is_whitespace() || c.is_control())
}
fn same_note(a: &MemoryRecord, b: &MemoryRecord) -> bool {
a.content == b.content
&& a.tags == b.tags
&& a.timestamp_ms == b.timestamp_ms
&& a.provenance == b.provenance
&& a.sensitivity == b.sensitivity
&& a.retention == b.retention
&& a.meta == b.meta
}
enum ImportLine {
Note(Box<MemoryRecord>),
Tombstone(TombstoneLine),
}
fn parse_import_line(line: &str) -> Result<ImportLine, String> {
let object: serde_json::Map<String, serde_json::Value> =
serde_json::from_str(line).map_err(|e| match e.classify() {
serde_json::error::Category::Data => "not a JSON object".to_string(),
_ => format!("not JSON at column {}", e.column()),
})?;
if object.contains_key("tombstone") {
if let Some(extra) = object
.keys()
.find(|name| !["tombstone", "timestamp_ms"].contains(&name.as_str()))
{
return Err(unknown_field(extra));
}
return serde_json::from_str(line)
.map(ImportLine::Tombstone)
.map_err(|e| unread("tombstone line", &e));
}
if let Some(extra) = object
.keys()
.find(|name| !RECORD_FIELDS.contains(&name.as_str()))
{
return Err(unknown_field(extra));
}
let note: MemoryRecord = serde_json::from_str(line).map_err(|e| unread("note", &e))?;
let kept = serde_json::to_value(¬e)
.map(|kept| without_nulls(&kept))
.map_err(|_| "not a note".to_string())?;
for (name, value) in &object {
if !value.is_null() && kept.get(name) != Some(&without_nulls(value)) {
return Err(format!("{name} holds something a note cannot keep"));
}
}
Ok(ImportLine::Note(Box::new(note)))
}
fn unknown_field(name: &str) -> String {
let plain = (1..=32).contains(&name.len())
&& name.bytes().all(|b| b.is_ascii_lowercase() || b == b'_');
if plain {
format!("unknown field {name}")
} else {
"unknown field".to_string()
}
}
fn unread(what: &str, e: &serde_json::Error) -> String {
let message = e.to_string();
for kind in ["missing field", "duplicate field"] {
let name = message
.strip_prefix(kind)
.and_then(|rest| rest.split('`').nth(1));
if let Some(name) = name {
return format!("{kind} {name}");
}
}
format!("not a {what}: wrong value at column {}", e.column())
}
fn without_nulls(value: &serde_json::Value) -> serde_json::Value {
use serde_json::Value;
match value {
Value::Object(members) => Value::Object(
members
.iter()
.filter(|(_, member)| !member.is_null())
.map(|(name, member)| (name.clone(), without_nulls(member)))
.collect(),
),
Value::Array(items) => Value::Array(items.iter().map(without_nulls).collect()),
other => other.clone(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn a_folder_maps_to_the_same_project_key_forever() {
for (path, key) in [
("/fixed/project/root", "p6e2cb4c61e23315c"),
("/home/haseeb", "pa465e6cee4a607a1"),
] {
let scope = ProjectScope::for_canonical_root(PathBuf::from(path));
assert_eq!(scope.key(), key, "{path}");
}
}
fn rec(ns: &str, ts: u64, content: &str, tags: Vec<&str>) -> MemoryRecord {
MemoryRecord {
namespace: ns.to_string(),
timestamp_ms: ts,
content: content.to_string(),
tags: tags.into_iter().map(String::from).collect(),
meta: None,
id: None,
project: None,
provenance: None,
sensitivity: None,
retention: None,
}
}
#[test]
fn append_then_query_returns_record() {
let tmp = TempDir::new().unwrap();
let r = rec("ns", 100, "hello world", vec!["@user"]);
append_to(tmp.path(), &r).unwrap();
let got = query_in(tmp.path(), "ns", &MemoryQuery::default()).unwrap();
assert_eq!(got, vec![r]);
}
#[test]
fn query_filters_by_content_contains() {
let tmp = TempDir::new().unwrap();
append_to(tmp.path(), &rec("ns", 100, "Hello World", vec![])).unwrap();
append_to(tmp.path(), &rec("ns", 200, "goodbye", vec![])).unwrap();
let q = MemoryQuery {
content_contains: Some("hello".to_string()),
..Default::default()
};
let got = query_in(tmp.path(), "ns", &q).unwrap();
assert_eq!(got.len(), 1);
assert_eq!(got[0].content, "Hello World");
}
#[test]
fn query_filters_by_tag_prefix() {
let tmp = TempDir::new().unwrap();
append_to(
tmp.path(),
&rec("ns", 100, "x", vec!["@user", "preference"]),
)
.unwrap();
let got = query_in(
tmp.path(),
"ns",
&MemoryQuery {
tag_prefix: Some("@u".into()),
..Default::default()
},
)
.unwrap();
assert_eq!(got.len(), 1);
let got = query_in(
tmp.path(),
"ns",
&MemoryQuery {
tag_prefix: Some("@x".into()),
..Default::default()
},
)
.unwrap();
assert!(got.is_empty());
}
#[test]
fn query_filters_by_time_range() {
let tmp = TempDir::new().unwrap();
append_to(tmp.path(), &rec("ns", 100, "a", vec![])).unwrap();
append_to(tmp.path(), &rec("ns", 200, "b", vec![])).unwrap();
append_to(tmp.path(), &rec("ns", 300, "c", vec![])).unwrap();
let got = query_in(
tmp.path(),
"ns",
&MemoryQuery {
since_ms: Some(150),
until_ms: Some(250),
..Default::default()
},
)
.unwrap();
assert_eq!(got.len(), 1);
assert_eq!(got[0].timestamp_ms, 200);
}
#[test]
fn query_returns_most_recent_first() {
let tmp = TempDir::new().unwrap();
append_to(tmp.path(), &rec("ns", 100, "a", vec![])).unwrap();
append_to(tmp.path(), &rec("ns", 300, "c", vec![])).unwrap();
append_to(tmp.path(), &rec("ns", 200, "b", vec![])).unwrap();
let got = query_in(tmp.path(), "ns", &MemoryQuery::default()).unwrap();
let ts: Vec<u64> = got.iter().map(|r| r.timestamp_ms).collect();
assert_eq!(ts, vec![300, 200, 100]);
}
#[test]
fn query_respects_limit() {
let tmp = TempDir::new().unwrap();
for i in 1..=5 {
append_to(tmp.path(), &rec("ns", i * 100, "x", vec![])).unwrap();
}
let got = query_in(
tmp.path(),
"ns",
&MemoryQuery {
limit: Some(2),
..Default::default()
},
)
.unwrap();
assert_eq!(got.len(), 2);
assert_eq!(got[0].timestamp_ms, 500);
assert_eq!(got[1].timestamp_ms, 400);
}
#[test]
fn query_skips_malformed_lines() {
let tmp = TempDir::new().unwrap();
let dir = memory_dir_in(tmp.path());
fs::create_dir_all(&dir).unwrap();
let path = namespace_path(&dir, "ns");
let v1 = serde_json::to_string(&rec("ns", 100, "a", vec![])).unwrap();
let v2 = serde_json::to_string(&rec("ns", 200, "b", vec![])).unwrap();
let body = format!("invalid json\n{v1}\nnot json either\n{v2}\n");
fs::write(&path, body).unwrap();
let got = query_in(tmp.path(), "ns", &MemoryQuery::default()).unwrap();
assert_eq!(got.len(), 2);
}
#[test]
fn query_skips_a_line_that_is_not_utf8_and_stops_on_a_lasting_read_error() {
let tmp = TempDir::new().unwrap();
append_to(tmp.path(), &rec("ns", 1, "one", vec![])).unwrap();
let path = memory_dir_in(tmp.path()).join("ns.jsonl");
let mut bytes = fs::read(&path).unwrap();
bytes.extend_from_slice(b"{\"content\":\"caf\xc3\n");
fs::write(&path, bytes).unwrap();
append_to(tmp.path(), &rec("ns", 2, "two", vec![])).unwrap();
let found = query_in(tmp.path(), "ns", &MemoryQuery::default()).unwrap();
let contents: Vec<&str> = found.iter().map(|r| r.content.as_str()).collect();
assert_eq!(contents, ["two", "one"]);
#[cfg(unix)]
{
fs::create_dir(memory_dir_in(tmp.path()).join("dir.jsonl")).unwrap();
let error = query_in(tmp.path(), "dir", &MemoryQuery::default()).unwrap_err();
assert!(matches!(error, MemoryError::Io(_)), "{error}");
}
}
#[test]
fn append_rejects_oversized_content() {
let tmp = TempDir::new().unwrap();
let big = "x".repeat(MAX_CONTENT_BYTES + 1);
let r = rec("ns", 1, &big, vec![]);
match append_to(tmp.path(), &r) {
Err(MemoryError::ContentTooLarge { .. }) => {}
other => panic!("expected ContentTooLarge, got {other:?}"),
}
}
#[test]
fn append_rejects_invalid_namespace() {
let tmp = TempDir::new().unwrap();
let cases = [
"",
"a/b",
"a\\b",
"..",
"a..b",
"has space",
"tab\there",
&"x".repeat(65),
];
for ns in cases {
let r = rec(ns, 1, "x", vec![]);
match append_to(tmp.path(), &r) {
Err(MemoryError::InvalidNamespace(_)) => {}
other => panic!("expected InvalidNamespace for {ns:?}, got {other:?}"),
}
}
}
#[test]
fn list_namespaces_returns_existing_files() {
let tmp = TempDir::new().unwrap();
append_to(tmp.path(), &rec("alpha", 1, "x", vec![])).unwrap();
append_to(tmp.path(), &rec("beta", 1, "x", vec![])).unwrap();
let got = list_namespaces_in(tmp.path()).unwrap();
assert_eq!(got, vec!["alpha".to_string(), "beta".to_string()]);
}
#[test]
fn list_namespaces_on_missing_dir_returns_empty() {
let tmp = TempDir::new().unwrap();
let got = list_namespaces_in(tmp.path()).unwrap();
assert!(got.is_empty());
}
#[test]
fn query_on_missing_namespace_returns_empty() {
let tmp = TempDir::new().unwrap();
let got = query_in(tmp.path(), "nope", &MemoryQuery::default()).unwrap();
assert!(got.is_empty());
}
fn project_scope(tmp: &TempDir) -> ProjectScope {
let root = tmp.path().join("project");
fs::create_dir_all(&root).unwrap();
ProjectScope::for_root(&root).unwrap()
}
fn note(content: &str) -> NewMemoryRecord {
NewMemoryRecord {
content: content.to_string(),
tags: vec!["t".into()],
provenance: MemoryProvenance {
source: "model".into(),
session: None,
},
sensitivity: MemorySensitivity::Normal,
retention: MemoryRetention::Standard,
}
}
fn take_steps() -> Vec<&'static str> {
IO_STEPS.with(|steps| std::mem::take(&mut *steps.borrow_mut()))
}
#[test]
fn a_save_and_a_forget_are_synced_before_they_return() {
let tmp = TempDir::new().unwrap();
let scope = project_scope(&tmp);
take_steps();
let first = store_record_in(tmp.path(), &scope, note("first")).unwrap();
assert_eq!(
take_steps(),
["sync dir", "write", "sync file", "sync dir"],
"the first save makes `memory/` and syncs the folder it is in; \
then the line, the file, and the file's directory"
);
store_record_in(tmp.path(), &scope, note("second")).unwrap();
assert_eq!(
take_steps(),
["write", "sync file", "sync dir"],
"every save syncs the directory: that the file has bytes does not \
say its name is on disk"
);
forget_in(tmp.path(), &scope, first.id.as_deref().unwrap()).unwrap();
assert_eq!(
take_steps(),
["write", "sync file", "sync dir"],
"a tombstone line is synced like a note"
);
append_to(tmp.path(), &rec("ns", 1, "x", vec![])).unwrap();
assert_eq!(take_steps(), ["write", "sync file", "sync dir"]);
fs::write(memory_dir_in(tmp.path()).join("empty.jsonl"), "").unwrap();
append_to(tmp.path(), &rec("empty", 1, "x", vec![])).unwrap();
assert_eq!(take_steps(), ["write", "sync file", "sync dir"]);
}
#[test]
fn a_sync_that_fails_after_the_write_is_written_unconfirmed() {
struct Failing;
impl Failing {
fn step(name: &'static str) -> Self {
FAILING_STEP.with(|step| step.set(Some(name)));
Failing
}
}
impl Drop for Failing {
fn drop(&mut self) {
FAILING_STEP.with(|step| step.set(None));
}
}
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let scope = project_scope(&tmp);
let kept = store_record_in(base, &scope, note("kept")).unwrap();
let contents = || -> Vec<String> {
let live = load_live_project_records(base, &scope).unwrap();
live.into_iter().map(|record| record.content).collect()
};
for step in ["sync file", "sync dir"] {
let failing = Failing::step(step);
let error = store_record_in(base, &scope, note(step)).unwrap_err();
drop(failing);
let live = load_live_project_records(base, &scope).unwrap();
let saved = live.iter().find(|r| r.content == step).unwrap();
let saved_id = saved.id.clone().unwrap();
assert!(
matches!(&error, MemoryError::WrittenUnconfirmed { id: Some(id), .. } if *id == saved_id),
"{error:?}"
);
let message = error.to_string();
assert!(message.starts_with("written, but not confirmed on disk"), "{message}");
assert!(message.contains(&format!("{step} failed")), "{message}");
assert!(message.contains(&format!("fetch {saved_id} to check")), "{message}");
}
assert_eq!(contents(), ["kept", "sync file", "sync dir"], "both notes are there");
let id = kept.id.as_deref().unwrap();
let failing = Failing::step("sync file");
let error = forget_in(base, &scope, id).unwrap_err();
drop(failing);
assert!(
matches!(&error, MemoryError::WrittenUnconfirmed { id: Some(got), .. } if got == id),
"{error:?}"
);
assert_eq!(contents(), ["sync file", "sync dir"]);
assert!(matches!(forget_in(base, &scope, id), Err(MemoryError::NotFound(_))));
let failing = Failing::step("sync dir");
let error = append_to(base, &rec("ns", 1, "x", vec![])).unwrap_err();
drop(failing);
assert!(
matches!(error, MemoryError::WrittenUnconfirmed { id: None, .. }),
"{error}"
);
assert_eq!(query_in(base, "ns", &MemoryQuery::default()).unwrap().len(), 1);
let mut run = ImportRun::new(false);
let line = exported("mem-i", scope.key(), 1, "imported");
import(base, &line, ImportTarget::AsRecorded, &mut run);
let failing = Failing::step("sync file");
let error = run.finish().unwrap_err();
drop(failing);
assert!(
matches!(error, MemoryError::WrittenUnconfirmed { id: None, .. }),
"{error}"
);
assert_eq!(contents(), ["sync file", "sync dir", "imported"]);
let fresh = TempDir::new().unwrap();
let failing = Failing::step("sync dir");
let error = append_to(fresh.path(), &rec("ns", 1, "x", vec![])).unwrap_err();
drop(failing);
assert!(matches!(error, MemoryError::Io(_)), "{error}");
assert!(!namespace_file(fresh.path(), "ns").exists());
}
#[test]
fn finish_syncs_every_file_and_names_the_one_that_failed() {
let tmp = TempDir::new().unwrap();
let other = TempDir::new().unwrap();
let base = tmp.path();
let keys = [
project_scope(&tmp).key().to_string(),
project_scope(&other).key().to_string(),
];
let mut run = ImportRun::new(false);
for (n, key) in keys.iter().enumerate() {
let line = exported(&format!("mem-{n}"), key, 1, "imported");
let imported = import(base, &line, ImportTarget::AsRecorded, &mut run);
assert_eq!(imported.outcome, ImportOutcome::Stored);
}
let mut paths: Vec<PathBuf> = keys
.iter()
.map(|key| namespace_file(base, &project_namespace(key)))
.collect();
paths.sort();
take_steps();
FAILING_STEP.with(|step| step.set(Some("sync file")));
let error = run.finish().unwrap_err();
FAILING_STEP.with(|step| step.set(None));
let syncs: Vec<&str> = take_steps()
.into_iter()
.filter(|step| step.starts_with("sync"))
.collect();
assert_eq!(syncs, ["sync file", "sync file", "sync dir"], "all tried");
assert!(
matches!(error, MemoryError::WrittenUnconfirmed { id: None, .. }),
"{error}"
);
let message = error.to_string();
assert!(message.contains(&paths[0].display().to_string()), "{message}");
assert!(!message.contains(&paths[1].display().to_string()), "{message}");
}
#[test]
fn a_save_that_cannot_get_the_lock_fails_and_writes_nothing() {
let tmp = TempDir::new().unwrap();
let scope = project_scope(&tmp);
store_record_in(tmp.path(), &scope, note("first")).unwrap();
let path = namespace_file(tmp.path(), &project_namespace(scope.key()));
let before = fs::read(&path).unwrap();
let held = lock_notes_file(&path).unwrap();
set_notes_lock_wait(Some(std::time::Duration::from_millis(60)));
let started = std::time::Instant::now();
let refused = store_record_in(tmp.path(), &scope, note("second")).unwrap_err();
let other = append_to(tmp.path(), &rec("other", 1, "x", vec![]));
let _other_lock = lock_notes_file(&namespace_file(tmp.path(), "held")).unwrap();
let unborn = append_to(tmp.path(), &rec("held", 1, "x", vec![])).unwrap_err();
set_notes_lock_wait(None);
assert!(started.elapsed() >= std::time::Duration::from_millis(60));
let message = refused.to_string();
assert!(message.contains("is locked by another writer"), "{message}");
assert!(message.contains("nothing was written"), "{message}");
assert_eq!(fs::read(&path).unwrap(), before);
other.expect("another file's lock is free");
assert!(unborn.to_string().contains("held.jsonl is locked"));
assert!(!namespace_file(tmp.path(), "held").exists());
drop(held);
store_record_in(tmp.path(), &scope, note("second")).unwrap();
assert_eq!(fs::read_to_string(&path).unwrap().lines().count(), 2);
}
struct SharedAppendFile {
bytes: Vec<u8>,
accept: std::collections::VecDeque<usize>,
interloper: Option<String>,
calls: Vec<Vec<u8>>,
}
impl SharedAppendFile {
fn taking(accept: &[usize], interloper: Option<String>) -> Self {
Self {
bytes: Vec::new(),
accept: accept.iter().copied().collect(),
interloper,
calls: Vec::new(),
}
}
fn lines(&self) -> Vec<&str> {
std::str::from_utf8(&self.bytes).unwrap().lines().collect()
}
}
impl Write for SharedAppendFile {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let n = self.accept.pop_front().unwrap_or(buf.len()).min(buf.len());
self.calls.push(buf.to_vec());
self.bytes.extend_from_slice(&buf[..n]);
if n < buf.len() {
if let Some(other) = self.interloper.take() {
self.bytes.extend_from_slice(other.as_bytes());
}
}
Ok(n)
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
const LINE: &str = r#"{"namespace":"ns","timestamp_ms":1,"content":"mine","tags":[]}"#;
const OTHER: &str = r#"{"namespace":"ns","timestamp_ms":2,"content":"theirs","tags":[]}"#;
#[test]
fn a_line_is_one_write_call() {
let mut file = SharedAppendFile::taking(&[], None);
write_whole_line(&mut file, LINE, false).unwrap();
assert_eq!(file.calls, [format!("{LINE}\n").into_bytes()]);
let mut file = SharedAppendFile::taking(&[], None);
write_whole_line(&mut file, LINE, true).unwrap();
assert_eq!(file.calls, [format!("\n{LINE}\n").into_bytes()]);
}
#[test]
fn a_short_write_is_abandoned_and_the_whole_line_written_again() {
let mut file = SharedAppendFile::taking(&[20], Some(format!("\n{OTHER}\n")));
write_whole_line(&mut file, LINE, false).unwrap();
assert_eq!(file.lines(), [&LINE[..20], OTHER, "", LINE]);
assert_eq!(
file.calls,
[format!("{LINE}\n").into_bytes(), format!("\n{LINE}\n").into_bytes()],
"each call carries the whole line; none carries a remainder"
);
let mut file = SharedAppendFile::taking(&[20], Some(format!("{OTHER}\n")));
write_whole_line(&mut file, LINE, false).unwrap();
let merged = format!("{}{OTHER}", &LINE[..20]);
assert_eq!(file.lines(), [merged.as_str(), "", LINE]);
for line in file.lines() {
let parsed = serde_json::from_str::<MemoryRecord>(line);
assert_eq!(parsed.is_ok(), line == LINE, "{line:?}");
}
}
#[test]
fn a_line_that_cannot_go_in_one_write_is_an_error_not_a_split_line() {
let mut file = SharedAppendFile::taking(&[5, 5, 5, 5, 5, 5], None);
let err = write_whole_line(&mut file, LINE, false).unwrap_err();
assert_eq!(err.kind(), std::io::ErrorKind::WriteZero);
assert_eq!(file.calls.len(), LINE_WRITE_ATTEMPTS);
assert!(
file.lines().iter().all(|line| line.len() <= 5),
"only fragments, each on a line of its own: {:?}",
file.lines()
);
let mut file = SharedAppendFile::taking(&[0, 0, 0, 0], None);
assert!(write_whole_line(&mut file, LINE, false).is_err());
assert!(file.bytes.is_empty());
}
#[test]
fn concurrent_appends_keep_every_line_whole_and_lose_none() {
const WRITERS: usize = 8;
const ROUNDS: usize = 24;
let tmp = TempDir::new().unwrap();
let scope = project_scope(&tmp);
let start = std::sync::Barrier::new(WRITERS);
let mut forgotten: Vec<String> = Vec::new();
let mut failures: Vec<String> = Vec::new();
std::thread::scope(|threads| {
let handles: Vec<_> = (0..WRITERS)
.map(|writer| {
let (base, scope, start) = (tmp.path(), &scope, &start);
threads.spawn(move || {
let (mut ids, mut errors) = (Vec::new(), Vec::new());
for round in 0..ROUNDS {
let pad = "x".repeat((round % 6) * 3_000);
let content = format!("w{writer}-r{round}-{pad}");
start.wait();
let stored = match store_record_in(base, scope, note(&content)) {
Ok(stored) => stored,
Err(e) => {
errors.push(format!("save w{writer}-r{round}: {e}"));
continue;
}
};
if round % 8 == 7 {
let id = stored.id.unwrap_or_default();
match forget_in(base, scope, &id) {
Ok(()) => ids.push(id),
Err(e) => errors.push(format!("forget w{writer}-r{round}: {e}")),
}
}
}
(ids, errors)
})
})
.collect();
for handle in handles {
let (ids, errors) = handle.join().unwrap();
forgotten.extend(ids);
failures.extend(errors);
}
});
assert!(
failures.is_empty(),
"a save or a forget failed among concurrent writers: {failures:?}"
);
let raw = fs::read_to_string(namespace_file(tmp.path(), &scope.namespace())).unwrap();
assert!(raw.ends_with('\n'));
let mut contents = std::collections::HashSet::new();
let mut tombstones = std::collections::HashSet::new();
for line in raw.lines().filter(|line| !line.is_empty()) {
if let Ok(tomb) = serde_json::from_str::<TombstoneLine>(line) {
assert!(tombstones.insert(tomb.tombstone), "a tombstone twice");
continue;
}
let record: MemoryRecord = serde_json::from_str(line)
.unwrap_or_else(|e| panic!("a line does not parse ({e}): {:.80}", line));
let head: String = record.content.chars().take(12).collect();
assert!(contents.insert(record.content), "a note twice: {head}");
}
assert_eq!(contents.len(), WRITERS * ROUNDS, "no note is lost");
for writer in 0..WRITERS {
for round in 0..ROUNDS {
let pad = "x".repeat((round % 6) * 3_000);
assert!(contents.contains(&format!("w{writer}-r{round}-{pad}")));
}
}
assert_eq!(forgotten.len(), WRITERS * ROUNDS / 8);
assert_eq!(
tombstones,
forgotten.iter().cloned().collect(),
"no tombstone is lost"
);
let live = load_live_project_records(tmp.path(), &scope).unwrap();
assert_eq!(live.len(), WRITERS * ROUNDS - forgotten.len());
}
#[test]
fn a_torn_last_line_is_fenced_off_and_no_good_line_is_touched() {
let tmp = TempDir::new().unwrap();
let scope = project_scope(&tmp);
let kept = store_record_in(tmp.path(), &scope, note("kept")).unwrap();
let path = namespace_file(tmp.path(), &scope.namespace());
let before = fs::read(&path).unwrap();
let mut whole = MemoryRecord {
content: "caf\u{e9}".into(),
..kept.clone()
};
whole.id = Some("mem-torn".into());
let torn_line = serde_json::to_string(&whole).unwrap();
let cut = torn_line.find('\u{e9}').unwrap() + 1;
let fragment = &torn_line.as_bytes()[..cut];
assert!(std::str::from_utf8(fragment).is_err(), "cut inside a character");
let mut torn = before.clone();
torn.extend_from_slice(fragment);
fs::write(&path, &torn).unwrap();
let after = store_record_in(tmp.path(), &scope, note("after the tear")).unwrap();
let bytes = fs::read(&path).unwrap();
assert!(bytes.starts_with(&torn), "nothing before the append changed");
let mut expected = torn.clone();
expected.push(b'\n');
expected.extend_from_slice(serde_json::to_string(&after).unwrap().as_bytes());
expected.push(b'\n');
assert_eq!(bytes, expected, "fragment, newline, the new line, newline");
let live = load_live_project_records(tmp.path(), &scope).unwrap();
assert_eq!(live, [kept.clone(), after.clone()]);
let ids = [kept.id.as_deref().unwrap(), after.id.as_deref().unwrap()];
assert_eq!(fetch_exact_in(tmp.path(), &scope, &ids).unwrap().len(), 2);
assert_eq!(
query_in(tmp.path(), &scope.namespace(), &MemoryQuery::default())
.unwrap()
.len(),
2
);
}
#[test]
fn a_last_line_missing_only_its_newline_is_kept() {
let tmp = TempDir::new().unwrap();
let scope = project_scope(&tmp);
let first = store_record_in(tmp.path(), &scope, note("first")).unwrap();
let path = namespace_file(tmp.path(), &scope.namespace());
let mut bytes = fs::read(&path).unwrap();
assert_eq!(bytes.pop(), Some(b'\n'));
fs::write(&path, bytes).unwrap();
let second = store_record_in(tmp.path(), &scope, note("second")).unwrap();
let live = load_live_project_records(tmp.path(), &scope).unwrap();
assert_eq!(live, [first, second]);
let raw = fs::read_to_string(&path).unwrap();
assert_eq!(raw.lines().count(), 2, "no empty line, nothing merged");
}
fn exported(id: &str, project: &str, timestamp_ms: u64, content: &str) -> String {
serde_json::json!({
"namespace": format!("project-{project}"),
"timestamp_ms": timestamp_ms,
"content": content,
"tags": ["imported", "@user"],
"meta": {"_synaps_repository": {"source_worktree_root": "/elsewhere"}, "blank": null},
"id": id,
"project": project,
"provenance": {"source": "model", "session": "sess-old"},
"sensitivity": "sensitive",
"retention": {"max_age_days": 30},
})
.to_string()
}
fn import(
base: &Path,
line: &str,
target: ImportTarget<'_>,
run: &mut ImportRun,
) -> ImportedLine {
import_line_in(base, line, target, run).unwrap()
}
fn refusal(imported: &ImportedLine) -> &str {
match &imported.outcome {
ImportOutcome::Refused(reason) => reason,
other => panic!("expected a refusal, got {other:?}"),
}
}
#[test]
fn import_notes_t_a10() {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let scope = project_scope(&tmp);
let key = scope.key().to_string();
let here = ImportTarget::Project(&key);
let path = namespace_file(base, &scope.namespace());
let old: Vec<String> = (0..3)
.map(|i| {
let stored = store_record_in(base, &scope, note(&format!("old note {i}")));
stored.unwrap().id.unwrap()
})
.collect();
let lines: Vec<String> = (0..8u64)
.map(|i| exported(&format!("mem-new-{i}"), &key, 1_000 + i, &format!("new note {i}")))
.collect();
let mut run = ImportRun::new(false);
for (i, line) in lines.iter().enumerate() {
assert_eq!(
import(base, line, ImportTarget::AsRecorded, &mut run),
ImportedLine {
outcome: ImportOutcome::Stored,
id: Some(format!("mem-new-{i}")),
project: Some(key.clone()),
tombstone: false,
}
);
}
assert_eq!(run.finish().unwrap(), std::slice::from_ref(&path));
assert_eq!(load_live_project_records(base, &scope).unwrap().len(), 11);
let search = |needle: &str| {
let query = ProjectMemoryQuery {
content_contains: Some(needle.into()),
limit: Some(MAX_SEARCH_LIMIT),
..Default::default()
};
search_project_in(base, &scope, &query).unwrap().len()
};
assert_eq!(
(search("note"), search("old note"), search("new note")),
(11, 3, 8)
);
let mut ids = old.clone();
ids.extend((0..8).map(|i| format!("mem-new-{i}")));
let wanted: Vec<&str> = ids.iter().map(String::as_str).collect();
let fetched = fetch_exact_in(base, &scope, &wanted).unwrap();
assert_eq!(fetched.len(), 11);
for (line, stored) in lines.iter().zip(&fetched[3..]) {
assert_eq!(
serde_json::to_value(stored).unwrap(),
serde_json::from_str::<serde_json::Value>(line).unwrap()
);
}
let before = fs::read(&path).unwrap();
let mut again = ImportRun::new(false);
for line in &lines {
assert_eq!(
import(base, line, ImportTarget::AsRecorded, &mut again).outcome,
ImportOutcome::AlreadyPresent
);
}
let changed = exported("mem-new-3", &key, 1_003, "another text");
assert_eq!(
refusal(&import(base, &changed, here, &mut again)),
"id already holds a different note"
);
assert!(again.finish().unwrap().is_empty());
assert_eq!(fs::read(&path).unwrap(), before);
forget_in(base, &scope, "mem-new-5").unwrap();
let before = fs::read(&path).unwrap();
let mut run = ImportRun::new(false);
assert_eq!(
import(base, &lines[5], here, &mut run).outcome,
ImportOutcome::KeptForgotten
);
let extra = lines[0].replacen('{', r#"{"colour":"red","#, 1);
assert_eq!(
refusal(&import(base, &extra, here, &mut run)),
"unknown field colour"
);
let user_wide = exported("mem-user", USER_WIDE_KEY, 3_000, "a user-wide note");
assert_eq!(
refusal(&import(base, &user_wide, ImportTarget::AsRecorded, &mut run)),
"user-wide note; pass --cwd to file it under a project"
);
let to_user_wide = ImportTarget::Project(USER_WIDE_KEY);
let err = import_line_in(base, &lines[0], to_user_wide, &mut run).unwrap_err();
assert!(matches!(err, MemoryError::InvalidProjectRoot(_)), "{err}");
assert!(run.finish().unwrap().is_empty());
assert_eq!(fs::read(&path).unwrap(), before, "none of these wrote");
assert_eq!(list_namespaces_in(base).unwrap(), [scope.namespace()]);
let mut run = ImportRun::new(false);
let tombstone = r#"{"tombstone":"mem-late","timestamp_ms":77}"#;
let late = exported("mem-late", &key, 2_000, "forgotten before it came");
let first = import(base, tombstone, here, &mut run);
assert_eq!(
(first.outcome, first.tombstone),
(ImportOutcome::Stored, true)
);
assert_eq!(
import(base, &late, here, &mut run).outcome,
ImportOutcome::KeptForgotten
);
run.finish().unwrap();
let mut run = ImportRun::new(false);
assert_eq!(
fs::read_to_string(&path).unwrap().lines().last(),
Some(tombstone),
"the tombstone is stored with the time it came with"
);
assert!(matches!(
fetch_exact_in(base, &scope, &["mem-late"]),
Err(MemoryError::NotFound(_))
));
assert_eq!(
import(base, &late, here, &mut ImportRun::new(false)).outcome,
ImportOutcome::KeptForgotten
);
let elsewhere = "pc4d9f8d15dc3bd8f";
let moved = exported("mem-moved", elsewhere, 4_000, "from another project");
let imported = import(base, &moved, here, &mut run);
assert_eq!(
(imported.outcome, imported.project),
(ImportOutcome::Stored, Some(key.clone()))
);
let stored = fetch_exact_in(base, &scope, &["mem-moved"]).unwrap().remove(0);
let mut expected: MemoryRecord = serde_json::from_str(&moved).unwrap();
expected.project = Some(key.clone());
expected.namespace = scope.namespace();
assert_eq!(stored, expected);
assert!(!namespace_file(base, &project_namespace(elsewhere)).exists());
assert_eq!(
import(base, &user_wide, here, &mut run).outcome,
ImportOutcome::Stored
);
let filed = fetch_exact_in(base, &scope, &["mem-user"]).unwrap().remove(0);
assert_eq!(filed.project.as_deref(), Some(key.as_str()));
assert_eq!(filed.namespace, scope.namespace());
run.finish().unwrap();
}
#[test]
fn a_dry_run_changes_no_byte_and_says_what_the_real_run_then_does() {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let scope = project_scope(&tmp);
let key = scope.key().to_string();
let here = ImportTarget::Project(&key);
let path = namespace_file(base, &scope.namespace());
let mut setup = ImportRun::new(false);
for id in ["mem-live", "mem-gone"] {
import(base, &exported(id, &key, 10, "there before"), here, &mut setup);
}
setup.finish().unwrap();
forget_in(base, &scope, "mem-gone").unwrap();
let before = fs::read(&path).unwrap();
let lines = [
exported("mem-a", &key, 20, "new"),
exported("mem-a", &key, 20, "new"),
exported("mem-a", &key, 20, "not the same"),
exported("mem-live", &key, 10, "there before"),
exported("mem-gone", &key, 10, "there before"),
r#"{"tombstone":"mem-live","timestamp_ms":30}"#.to_string(),
r#"{"tombstone":"mem-live","timestamp_ms":31}"#.to_string(),
exported("mem-live", &key, 10, "there before"),
r#"{"tombstone":"mem-never","timestamp_ms":32}"#.to_string(),
exported("mem-never", &key, 40, "comes after its tombstone"),
"not json".to_string(),
];
let expected = [
"stored",
"already_present",
"refused",
"already_present",
"kept_forgotten",
"forgotten",
"already_present",
"kept_forgotten",
"stored",
"kept_forgotten",
"refused",
];
let run_all = |dry_run: bool| -> Vec<ImportedLine> {
let mut run = ImportRun::new(dry_run);
assert_eq!(run.is_dry_run(), dry_run);
let all: Vec<ImportedLine> = lines
.iter()
.map(|line| import_line_in(base, line, here, &mut run).unwrap())
.collect();
assert_eq!(run.finish().unwrap().is_empty(), dry_run);
all
};
take_steps();
let dry = run_all(true);
assert!(take_steps().is_empty(), "a dry run writes and syncs nothing");
assert_eq!(fs::read(&path).unwrap(), before, "byte for byte");
assert_eq!(list_namespaces_in(base).unwrap(), [scope.namespace()]);
let names: Vec<&str> = dry.iter().map(|line| line.outcome.name()).collect();
assert_eq!(names, expected);
let real = run_all(false);
assert_eq!(real, dry, "the real run does what the dry run said");
let live = load_live_project_records(base, &scope).unwrap();
let ids: Vec<&str> = live.iter().filter_map(|r| r.id.as_deref()).collect();
assert_eq!(ids, ["mem-a"], "and the file agrees");
let empty = TempDir::new().unwrap();
let mut run = ImportRun::new(true);
let imported = import_line_in(empty.path(), &lines[0], here, &mut run).unwrap();
assert_eq!(imported.outcome, ImportOutcome::Stored);
assert!(run.finish().unwrap().is_empty());
assert_eq!(fs::read_dir(empty.path()).unwrap().count(), 0);
}
#[test]
fn a_preview_and_the_import_after_it_are_two_runs() {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let key = "p1111111111111111";
let line = exported("mem-a", key, 1, "x");
for dry_run in [true, true, false] {
let mut run = ImportRun::new(dry_run);
let first = import(base, &line, ImportTarget::AsRecorded, &mut run);
assert_eq!(first.outcome, ImportOutcome::Stored, "dry_run {dry_run}");
let again = import(base, &line, ImportTarget::AsRecorded, &mut run);
assert_eq!(again.outcome, ImportOutcome::AlreadyPresent);
assert_eq!(run.finish().unwrap().len(), usize::from(!dry_run));
}
let text = fs::read_to_string(namespace_file(base, &project_namespace(key))).unwrap();
assert_eq!(text.lines().count(), 1);
}
#[test]
fn a_run_dropped_unfinished_still_syncs_what_it_appended() {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let key = "p1111111111111111";
take_steps();
{
let mut run = ImportRun::new(false);
import(base, &exported("mem-a", key, 1, "x"), ImportTarget::AsRecorded, &mut run);
assert_eq!(take_steps(), ["sync dir", "write"]);
}
assert_eq!(take_steps(), ["sync file", "sync dir"], "synced on drop");
{
let mut run = ImportRun::new(false);
import(base, &exported("mem-a", key, 1, "x"), ImportTarget::AsRecorded, &mut run);
let mut dry = ImportRun::new(true);
import(base, &exported("mem-b", key, 1, "x"), ImportTarget::AsRecorded, &mut dry);
}
assert!(take_steps().is_empty(), "nothing appended, nothing synced");
let mut run = ImportRun::new(false);
import(base, &exported("mem-c", key, 1, "x"), ImportTarget::AsRecorded, &mut run);
run.finish().unwrap();
assert_eq!(take_steps(), ["write", "sync file", "sync dir"]);
}
#[test]
fn an_import_syncs_each_file_once_at_the_end_of_the_run() {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let (a, b) = ("p1111111111111111", "p2222222222222222");
let mut run = ImportRun::new(false);
take_steps();
for (i, project) in [a, a, b, a].into_iter().enumerate() {
let line = exported(&format!("mem-{i}"), project, i as u64, "x");
assert_eq!(
import(base, &line, ImportTarget::AsRecorded, &mut run).outcome,
ImportOutcome::Stored
);
}
let tombstone = r#"{"tombstone":"mem-0","timestamp_ms":9}"#;
assert_eq!(
import(base, tombstone, ImportTarget::Project(a), &mut run).outcome,
ImportOutcome::Forgotten
);
assert_eq!(
take_steps(),
["sync dir", "write", "write", "write", "write", "write"],
"the folder the new `memory/` is in, then five lines and no sync yet"
);
let files = [a, b].map(|key| namespace_file(base, &project_namespace(key)));
assert_eq!(run.finish().unwrap(), files);
assert_eq!(take_steps(), ["sync file", "sync file", "sync dir"]);
}
#[test]
fn a_line_that_could_not_be_kept_whole_is_refused_and_writes_nothing() {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let key = "p1111111111111111";
let good = exported("mem-x", key, 5, "secret words");
let with = |from: &str, to: &str| {
assert!(good.contains(from), "{from}");
good.replacen(from, to, 1)
};
let long_id = format!(r#""id":"{}""#, "i".repeat(129));
let big = format!(r#""content":"{}""#, "c".repeat(MAX_CONTENT_BYTES + 1));
let long_key = format!(r#"{{"{}":1,"content""#, "secret".repeat(6));
let wrong_value = "not a note: wrong value at column ";
let cases: Vec<(String, &str)> = vec![
(with(r#","timestamp_ms":5"#, ""), "missing field timestamp_ms"),
(with(r#""timestamp_ms":5"#, r#""timestamp_ms":"5""#), wrong_value),
(with(r#"{"content""#, r#"{"extra":1,"content""#), "unknown field extra"),
(with(r#"{"content""#, r#"{"secret words":1,"content""#), "unknown field"),
(with(r#"{"content""#, r#"{"Secret":1,"content""#), "unknown field"),
(with(r#"{"content""#, r#"{"secret-token-1234":1,"content""#), "unknown field"),
(with(r#"{"content""#, &long_key), "unknown field"),
(with(r#"{"content""#, r#"{"":1,"content""#), "unknown field"),
(
r#"{"tombstone":"mem-x","timestamp_ms":1,"secret words":2}"#.to_string(),
"unknown field",
),
(
with(r#""source":"model""#, r#""source":"model","host":"x""#),
"provenance holds something a note cannot keep",
),
(
with(r#"{"content""#, r#"{"content":"first","content""#),
"duplicate field content",
),
(with(r#""retention":{"max_age_days":30}"#, r#""retention":"forever""#), wrong_value),
(with(r#""id":"mem-x","#, ""), "missing field id"),
(with(r#""id":"mem-x""#, r#""id":"mem x""#), "invalid id"),
(with(r#""id":"mem-x""#, r#""id":"""#), "invalid id"),
(with(r#""id":"mem-x""#, &long_id), "invalid id"),
(with(r#""project":"p1111111111111111","#, ""), "missing field project"),
(
with(r#""project":"p1111111111111111""#, r#""project":"P1111111111111111""#),
"invalid project",
),
(
with(r#""namespace":"project-p1111111111111111""#, r#""namespace":"notes""#),
"namespace is not project-<project>",
),
(with(r#""content":"secret words""#, &big), "content is over 16384 bytes"),
(r#"{"content":"secret words""#.to_string(), "not JSON at column "),
(r#"["secret words"]"#.to_string(), "not a JSON object"),
(r#""secret words""#.to_string(), "not a JSON object"),
(
r#"{"tombstone":"mem-x","timestamp_ms":1,"content":"secret words"}"#.to_string(),
"unknown field content",
),
(r#"{"tombstone":"mem-x"}"#.to_string(), "missing field timestamp_ms"),
(
r#"{"tombstone":"mem-gone","timestamp_ms":1}"#.to_string(),
"a tombstone line names no project and no project's notes hold its id; \
pass --cwd or --project-key",
),
];
let mut run = ImportRun::new(false);
for (line, expected) in &cases {
let imported = import(base, line, ImportTarget::AsRecorded, &mut run);
let reason = refusal(&imported);
assert_eq!(
reason.trim_end_matches(|c: char| c.is_ascii_digit()),
*expected,
"{line:.70}"
);
let quoted = reason.to_lowercase().contains("secret");
assert!(!quoted, "a reason quotes nothing: {reason}");
}
assert!(!memory_dir_in(base).exists(), "a refused line writes nothing");
assert_eq!(
import(base, &good, ImportTarget::AsRecorded, &mut run).outcome,
ImportOutcome::Stored
);
assert_eq!(run.finish().unwrap().len(), 1);
}
#[test]
fn a_tombstone_as_recorded_finds_its_note_in_a_file_the_run_has_not_opened() {
let (a, b) = ("p1111111111111111", "p2222222222222222");
let as_recorded = ImportTarget::AsRecorded;
let tombstone = |id: &str| format!(r#"{{"tombstone":"{id}","timestamp_ms":9}}"#);
let file = |base: &Path, key: &str| namespace_file(base, &project_namespace(key));
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let mut run = ImportRun::new(false);
for (id, key) in [("mem-1", a), ("mem-5", b)] {
let stored = import(base, &exported(id, key, 1, "x"), as_recorded, &mut run);
assert_eq!(stored.outcome, ImportOutcome::Stored);
}
let _ = run.finish().unwrap();
let memory = memory_dir_in(base);
fs::write(memory.join("project-p0000000000000000.jsonl"), exported("mem-u", a, 1, "x"))
.unwrap();
fs::write(memory.join("session-notes.jsonl"), "not a note\n").unwrap();
fs::write(memory.join(".import.lock"), "").unwrap();
for dry_run in [true, false] {
let before = fs::read_to_string(file(base, a)).unwrap();
let mut run = ImportRun::new(dry_run);
let late = import(base, &tombstone("mem-1"), as_recorded, &mut run);
assert_eq!(
(late.outcome, late.project.as_deref()),
(ImportOutcome::Forgotten, Some(a)),
"dry run {dry_run}"
);
for id in ["mem-nowhere", "mem-u"] {
let nowhere = import(base, &tombstone(id), as_recorded, &mut run);
assert_eq!(refusal(&nowhere), UNPLACED_TOMBSTONE, "{id}");
}
let files = run.finish().unwrap();
let text = fs::read_to_string(file(base, a)).unwrap();
if dry_run {
assert!(files.is_empty());
assert_eq!(text, before);
} else {
assert_eq!(files, [file(base, a)]);
assert_eq!(text.lines().last(), Some(tombstone("mem-1").as_str()));
assert!(!fs::read_to_string(file(base, b)).unwrap().contains("tombstone"));
}
}
let mut run = ImportRun::new(false);
assert_eq!(
import(base, &tombstone("mem-1"), as_recorded, &mut run).outcome,
ImportOutcome::AlreadyPresent
);
assert!(run.finish().unwrap().is_empty());
}
#[test]
fn a_tombstone_as_recorded_is_filed_where_the_run_saw_its_id_or_keeps_its_note_out() {
let (a, b) = ("p1111111111111111", "p2222222222222222");
let as_recorded = ImportTarget::AsRecorded;
let tombstone = |id: &str| format!(r#"{{"tombstone":"{id}","timestamp_ms":9}}"#);
let file = |base: &Path, key: &str| namespace_file(base, &project_namespace(key));
let live = |base: &Path, key: &str| -> Vec<String> {
let scanned = scan_project_file(base, key).unwrap();
let mut ids: Vec<String> = scanned
.records
.iter()
.filter_map(|record| record.id.clone())
.filter(|id| !scanned.tombstoned.contains(id))
.collect();
ids.sort();
ids
};
for dry_run in [true, false] {
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let mut run = ImportRun::new(dry_run);
for (id, key) in [("mem-1", a), ("mem-2", a), ("mem-3", b)] {
import(base, &exported(id, key, 1, "x"), as_recorded, &mut run);
}
let forgotten = import(base, &tombstone("mem-1"), as_recorded, &mut run);
assert_eq!(
forgotten,
ImportedLine {
outcome: ImportOutcome::Forgotten,
id: Some("mem-1".into()),
project: Some(a.into()),
tombstone: true,
}
);
assert_eq!(
import(base, &tombstone("mem-1"), as_recorded, &mut run).outcome,
ImportOutcome::AlreadyPresent
);
assert_eq!(
import(base, &exported("mem-1", a, 1, "x"), as_recorded, &mut run).outcome,
ImportOutcome::KeptForgotten
);
let refused = import(base, &tombstone("mem-9"), as_recorded, &mut run);
assert_eq!(
refusal(&refused),
"a tombstone line names no project and no project's notes hold its id; \
pass --cwd or --project-key"
);
assert_eq!((refused.id.as_deref(), refused.project), (Some("mem-9"), None));
let late = import(base, &exported("mem-9", a, 1, "x"), as_recorded, &mut run);
assert_eq!(refusal(&late), "its tombstone in this run could not be filed");
let late = import(base, &exported("mem-9", a, 1, "x"), ImportTarget::Project(b), &mut run);
assert_eq!(refusal(&late), "its tombstone in this run could not be filed");
assert_eq!(
import(base, &exported("mem-4", b, 1, "x"), as_recorded, &mut run).outcome,
ImportOutcome::Stored
);
let files = run.finish().unwrap();
if dry_run {
assert!(files.is_empty());
assert!(!memory_dir_in(base).exists(), "a dry run writes nothing");
continue;
}
assert_eq!(files, [file(base, a), file(base, b)]);
assert_eq!(live(base, a), ["mem-2"]);
assert_eq!(live(base, b), ["mem-3", "mem-4"]);
let text = fs::read_to_string(file(base, a)).unwrap();
assert_eq!(text.lines().last(), Some(tombstone("mem-1").as_str()));
assert_eq!(text.matches("tombstone").count(), 1);
assert!(!text.contains("mem-9"));
assert!(!fs::read_to_string(file(base, b)).unwrap().contains("mem-9"));
}
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let mut run = ImportRun::new(false);
let first = import(base, &tombstone("mem-1"), as_recorded, &mut run);
let second = import(base, &exported("mem-1", a, 1, "x"), as_recorded, &mut run);
assert_eq!(first.outcome.name(), "refused");
assert_eq!(refusal(&second), "its tombstone in this run could not be filed");
assert!(run.finish().unwrap().is_empty());
assert!(!memory_dir_in(base).exists());
let mut run = ImportRun::new(false);
assert_eq!(
import(base, &exported("mem-1", a, 1, "x"), as_recorded, &mut run).outcome,
ImportOutcome::Stored
);
let again = exported("mem-1", a, 1, "x");
assert_eq!(
import(base, &again, ImportTarget::Project(b), &mut run).outcome,
ImportOutcome::Stored
);
let both = import(base, &tombstone("mem-1"), as_recorded, &mut run);
assert_eq!(
(both.outcome, both.project.as_deref()),
(ImportOutcome::Forgotten, Some(a))
);
run.finish().unwrap();
assert!(live(base, a).is_empty() && live(base, b).is_empty());
let tmp = TempDir::new().unwrap();
let base = tmp.path();
let mut earlier = ImportRun::new(false);
import(base, &exported("mem-1", a, 1, "x"), as_recorded, &mut earlier);
earlier.finish().unwrap();
let mut run = ImportRun::new(false);
assert_eq!(
import(base, &exported("mem-1", a, 1, "x"), as_recorded, &mut run).outcome,
ImportOutcome::AlreadyPresent
);
assert_eq!(
import(base, &tombstone("mem-1"), as_recorded, &mut run).outcome,
ImportOutcome::Forgotten
);
run.finish().unwrap();
assert!(live(base, a).is_empty());
}
#[test]
fn a_null_field_and_absent_tags_are_not_a_loss() {
let tmp = TempDir::new().unwrap();
let scope = project_scope(&tmp);
let key = scope.key();
let line = format!(
r#"{{"namespace":"project-{key}","timestamp_ms":7,"content":"c","id":"mem-n","project":"{key}","tags":null,"meta":null,"provenance":{{"source":"model","session":null}}}}"#
);
let mut run = ImportRun::new(false);
assert_eq!(
import(tmp.path(), &line, ImportTarget::AsRecorded, &mut run).outcome,
ImportOutcome::Stored
);
let stored = fetch_exact_in(tmp.path(), &scope, &["mem-n"]).unwrap().remove(0);
assert_eq!(stored, serde_json::from_str::<MemoryRecord>(&line).unwrap());
assert_eq!(stored.timestamp_ms, 7);
assert!(stored.tags.is_empty(), "`\"tags\": null` is no tags");
let absent = line.replace(r#""tags":null,"#, "").replace("mem-n", "mem-o");
assert_eq!(
import(tmp.path(), &absent, ImportTarget::AsRecorded, &mut run).outcome,
ImportOutcome::Stored
);
let wrong = line.replace(r#""tags":null"#, r#""tags":"one""#).replace("mem-n", "mem-p");
assert!(refusal(&import(tmp.path(), &wrong, ImportTarget::AsRecorded, &mut run))
.starts_with("not a note: wrong value at column "));
run.finish().unwrap();
}
#[cfg(unix)]
mod private_modes {
use super::*;
use crate::core::private_fs::test_support::UmaskGuard;
use serial_test::serial;
use std::os::unix::fs::PermissionsExt;
fn mode_of(path: &std::path::Path) -> u32 {
fs::metadata(path).unwrap().permissions().mode() & 0o777
}
#[test]
#[serial(umask)]
fn append_creates_0600_file_and_0700_dir_under_permissive_umask() {
let _umask = UmaskGuard::set(0);
let tmp = TempDir::new().unwrap();
append_to(tmp.path(), &rec("ns", 1, "x", vec![])).unwrap();
let dir = memory_dir_in(tmp.path());
assert_eq!(mode_of(&dir), 0o700, "memory dir must be 0700");
assert_eq!(
mode_of(&dir.join("ns.jsonl")),
0o600,
"memory file must be 0600"
);
}
#[test]
fn append_refuses_symlink_target() {
let tmp = TempDir::new().unwrap();
let dir = memory_dir_in(tmp.path());
fs::create_dir_all(&dir).unwrap();
let victim = tmp.path().join("victim.txt");
fs::write(&victim, "").unwrap();
std::os::unix::fs::symlink(&victim, dir.join("ns.jsonl")).unwrap();
let res = append_to(tmp.path(), &rec("ns", 1, "secret", vec![]));
assert!(res.is_err(), "append through a symlink must fail");
assert_eq!(
fs::read_to_string(&victim).unwrap(),
"",
"no bytes may be written through the planted symlink"
);
}
#[test]
fn a_lock_file_that_cannot_be_opened_is_named_in_the_error() {
let tmp = TempDir::new().unwrap();
let dir = memory_dir_in(tmp.path());
fs::create_dir_all(&dir).unwrap();
let victim = tmp.path().join("victim.txt");
fs::write(&victim, "").unwrap();
let lock = dir.join("ns.jsonl.lock");
std::os::unix::fs::symlink(&victim, &lock).unwrap();
let error = append_to(tmp.path(), &rec("ns", 1, "x", vec![])).unwrap_err();
assert!(matches!(error, MemoryError::Io(_)), "{error}");
let message = error.to_string();
assert!(
message.contains(&format!("cannot open the notes lock file {}", lock.display())),
"{message}"
);
assert!(!dir.join("ns.jsonl").exists());
assert_eq!(fs::read_to_string(&victim).unwrap(), "");
}
#[test]
fn append_repairs_preexisting_broad_mode() {
let tmp = TempDir::new().unwrap();
let dir = memory_dir_in(tmp.path());
fs::create_dir_all(&dir).unwrap();
let file = dir.join("ns.jsonl");
fs::write(&file, "").unwrap();
fs::set_permissions(&file, fs::Permissions::from_mode(0o666)).unwrap();
append_to(tmp.path(), &rec("ns", 1, "x", vec![])).unwrap();
assert_eq!(
mode_of(&file),
0o600,
"broad pre-existing mode must be repaired"
);
}
}
}