Shared memory and context tools for agentic work.
Code Rooms
use crate as m1nd_mcp;
use m1nd_core::domain::DomainConfig;
use m1nd_core::graph::Graph;
use m1nd_mcp::brain_runtime::{
BrainActorHandle, BrainSessionCell, UnboundBrainCheckpointAuthority, BRAIN_CHECKPOINT_DIRECTORY,
};
use m1nd_mcp::runtime_jobs::RuntimeJobFailure;
use m1nd_mcp::server::{dispatch_tool, McpConfig};
use m1nd_mcp::session::SessionState;
use serde_json::json;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
fn build_state(root: &Path) -> SessionState {
let config = McpConfig {
graph_source: root.join("graph_snapshot.json"),
plasticity_state: root.join("plasticity_state.json"),
runtime_dir: Some(root.to_path_buf()),
..McpConfig::default()
SessionState::initialize(Graph::new(), &config, DomainConfig::code()).expect("init session")
}
fn call(state: &mut SessionState, tool: &str, params: serde_json::Value) -> serde_json::Value {
dispatch_tool(state, tool, ¶ms).expect("tool call")
fn build_actor(root: &Path) -> Arc<BrainActorHandle> {
let session = Arc::new(BrainSessionCell::new(build_state(root)));
BrainActorHandle::start(
"auto-ingest-test".to_string(),
session,
root.join(BRAIN_CHECKPOINT_DIRECTORY),
Arc::new(UnboundBrainCheckpointAuthority),
8,
None,
)
.expect("start actor")
fn actor_call(
actor: &BrainActorHandle,
tool: &str,
params: serde_json::Value,
mutating: bool,
) -> serde_json::Value {
let tool = tool.to_string();
actor
.try_execute(mutating, move |state| {
dispatch_tool(state, &tool, ¶ms)
.map_err(|error| RuntimeJobFailure::new("test_tool_call_failed", error.to_string()))
})
.expect("actor tool call")
/// Retries the queue-depth polls below are allowed before they call a run stuck.
/// Both loops exit on the FIRST observation that meets the bar, so this only caps a
/// stuck run: a healthy run pays one poll. At 50 retries (≈1s of sleep) it was a bet
/// on how fast the machine drains a scan — the same bet that makes the shutdown
/// deadlines rotate red on the loaded two-core runner.
const QUEUE_POLL_RETRIES: usize = 1_500;
fn wait_for_actor_queue(actor: &BrainActorHandle, expected_min: usize) {
let mut last_queue_depth = 0;
for _ in 0..QUEUE_POLL_RETRIES {
let status = actor_call(
actor,
"auto_ingest_status",
json!({ "agent_id": "tester" }),
false,
);
let queue_depth = status
.get("queue_depth")
.and_then(|value| value.as_u64())
.unwrap_or(0) as usize;
last_queue_depth = queue_depth;
if queue_depth >= expected_min {
return;
thread::sleep(Duration::from_millis(20));
panic!(
"timed out waiting for actor auto-ingest queue depth to reach at least {}; last observed queue depth was {} after {} retries with 20ms sleep",
expected_min, last_queue_depth, QUEUE_POLL_RETRIES
fn search_count(state: &mut SessionState, query: &str) -> usize {
call(
state,
"search",
json!({"agent_id":"tester","query":query,"mode":"literal"}),
.get("results")
.and_then(|value| value.as_array())
.map(|value| value.len())
.unwrap_or(0)
fn search_first_node_id(state: &mut SessionState, query: &str) -> String {
.and_then(|value| value.first())
.and_then(|value| value.get("node_id"))
.and_then(|value| value.as_str())
.unwrap_or("")
.to_string()
fn search_file_paths(state: &mut SessionState, query: &str) -> Vec<String> {
.map(|results| {
results
.iter()
.filter_map(|entry| entry.get("file_path").and_then(|value| value.as_str()))
.map(|value| value.to_string())
.collect::<Vec<_>>()
.unwrap_or_default()
fn wait_for_queue(state: &mut SessionState, expected_min: usize) {
let status = call(state, "auto_ingest_status", json!({ "agent_id": "tester" }));
"timed out waiting for auto-ingest queue depth to reach at least {}; last observed queue depth was {} after {} retries with 20ms sleep",
expected_min,
last_queue_depth,
QUEUE_POLL_RETRIES
fn write(path: &Path, content: &str) {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent).unwrap();
fs::write(path, content).unwrap();
fn light_doc(entity: &str, doi: &str) -> String {
format!(
r#"---
Protocol: L1GHT/1
Node: {entity}
State: active
Color: amber
Glyph: *
Completeness: draft
Proof: working
Depends on:
- {doi}
Next:
- validate
---
## Contract
[⍂ entity: {entity}]
[⟁ depends_on: {doi}]
[𝔻 evidence: ready]
"#
fn pubmed_article(title: &str, doi: &str) -> String {
r#"<?xml version="1.0"?>
<PubmedArticleSet>
<PubmedArticle>
<MedlineCitation>
<PMID>12345678</PMID>
<Article>
<ArticleTitle>{title}</ArticleTitle>
<Journal><Title>Nature</Title></Journal>
<AuthorList>
<Author><ForeName>Jane</ForeName><LastName>Doe</LastName></Author>
</AuthorList>
</Article>
</MedlineCitation>
<PubmedData>
<ReferenceList>
<Reference>
<ArticleIdList>
<ArticleId IdType="doi">{doi}</ArticleId>
</ArticleIdList>
</Reference>
</ReferenceList>
</PubmedData>
</PubmedArticle>
</PubmedArticleSet>"#
fn bibtex_entry(title: &str, doi: &str) -> String {
r#"@article{{shared2026,
author = {{Doe, Jane}},
title = {{{title}}},
journal = {{Nature}},
year = {{2026}},
doi = {{{doi}}}
}}
fn crossref_work(title: &str, doi: &str) -> String {
json!({
"DOI": doi,
"title": [title],
"type": "journal-article",
"publisher": "Nature",
"author": [{"given": "Jane", "family": "Doe", "sequence": "first"}],
"reference": []
fn plain_markdown(title: &str, body: &str) -> String {
format!("# {}\n\n{}\n", title, body)
fn html_doc(title: &str, body: &str) -> String {
"<html><body><h1>{}</h1><p>{}</p><p><a href=\"https://example.com/docs\">Docs</a></p></body></html>",
title, body
fn escape_pdf_text(text: &str) -> String {
text.replace('\\', "\\\\")
.replace('(', "\\(")
.replace(')', "\\)")
fn simple_pdf_bytes(text: &str) -> Vec<u8> {
let content = format!(
"BT\n/F1 18 Tf\n72 120 Td\n({}) Tj\nET\n",
escape_pdf_text(text)
let objects = [
"1 0 obj\n<< /Type /Catalog /Pages 2 0 R >>\nendobj\n".to_string(),
"2 0 obj\n<< /Type /Pages /Kids [3 0 R] /Count 1 >>\nendobj\n".to_string(),
"3 0 obj\n<< /Type /Page /Parent 2 0 R /MediaBox [0 0 300 200] /Resources << /Font << /F1 5 0 R >> >> /Contents 4 0 R >>\nendobj\n".to_string(),
"4 0 obj\n<< /Length {} >>\nstream\n{}endstream\nendobj\n",
content.len(),
content
),
"5 0 obj\n<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>\nendobj\n".to_string(),
];
let mut pdf = b"%PDF-1.4\n".to_vec();
let mut offsets = Vec::with_capacity(objects.len());
for object in &objects {
offsets.push(pdf.len());
pdf.extend_from_slice(object.as_bytes());
let xref_start = pdf.len();
pdf.extend_from_slice(
format!("xref\n0 {}\n0000000000 65535 f \n", objects.len() + 1).as_bytes(),
for offset in offsets {
pdf.extend_from_slice(format!("{offset:010} 00000 n \n").as_bytes());
"trailer\n<< /Root 1 0 R /Size {} >>\nstartxref\n{}\n%%EOF\n",
objects.len() + 1,
xref_start
.as_bytes(),
pdf
#[test]
fn auto_ingest_light_file_lifecycle_end_to_end() {
let temp = tempfile::tempdir().unwrap();
let docs_root = temp.path().join("docs");
fs::create_dir_all(&docs_root).unwrap();
let file = docs_root.join("notes.md");
let mut state = build_state(temp.path());
&mut state,
"auto_ingest_start",
json!({"agent_id":"tester","roots":[docs_root.to_string_lossy().to_string()],"formats":["light"],"debounce_ms":0}),
write(&file, &light_doc("AlphaNode", "10.1000/shared"));
wait_for_queue(&mut state, 1);
call(&mut state, "auto_ingest_tick", json!({"agent_id":"tester"}));
assert!(
search_count(&mut state, "AlphaNode") > 0,
"light entity must be searchable after create"
write(&file, &light_doc("OmegaNode", "10.1000/shared"));
assert_eq!(
search_count(&mut state, "AlphaNode"),
0,
"old entity should disappear after update"
search_count(&mut state, "OmegaNode") > 0,
"new entity must be searchable after update"
fs::remove_file(&file).unwrap();
search_count(&mut state, "OmegaNode"),
"entity should disappear after delete"
fn auto_ingest_supports_article_bibtex_and_crossref() {
let docs_root = temp.path().join("research");
let article = docs_root.join("paper.xml");
let bib = docs_root.join("refs.bib");
let crossref = docs_root.join("work.json");
json!({"agent_id":"tester","roots":[docs_root.to_string_lossy().to_string()],"formats":["article","bibtex","crossref"],"debounce_ms":0}),
write(&article, &pubmed_article("Test Article", "10.1000/shared"));
write(&bib, &bibtex_entry("Shared Bibliography", "10.1000/shared"));
write(&crossref, &crossref_work("Shared Work", "10.1000/shared"));
wait_for_queue(&mut state, 3);
for needle in ["12345678", "Shared Bibliography", "Shared Work"] {
search_count(&mut state, needle) > 0,
"expected searchable result for {}",
needle
fn auto_ingest_mixed_domain_query_returns_multiple_formats() {
let docs_root = temp.path().join("mixed");
write(
&docs_root.join("spec.md"),
&light_doc("BridgeStudy", "10.1000/shared"),
&docs_root.join("paper.xml"),
&pubmed_article("Bridge Article", "10.1000/shared"),
&docs_root.join("refs.bib"),
&bibtex_entry("Bridge Bibliography", "10.1000/shared"),
&docs_root.join("work.json"),
&crossref_work("Bridge CrossRef", "10.1000/shared"),
json!({"agent_id":"tester","roots":[docs_root.to_string_lossy().to_string()],"debounce_ms":0}),
let file_paths = search_file_paths(&mut state, "10.1000/shared");
assert!(file_paths.iter().any(|path| path.ends_with(".md")));
assert!(file_paths.iter().any(|path| path.ends_with(".xml")));
file_paths.len() >= 3,
"expected mixed-domain retrieval, got {:?}",
file_paths
fn auto_ingest_lock_watch_observes_ingest_mutations() {
let docs_root = temp.path().join("locks");
let file = docs_root.join("watch.md");
write(&file, &light_doc("WatchedNode", "10.1000/watch"));
let node_id = search_first_node_id(&mut state, "WatchedNode");
assert!(!node_id.is_empty(), "watched node result");
let created = call(
"lock_create",
json!({"agent_id":"tester","scope":"node","root_nodes":[node_id]}),
let lock_id = created
.get("lock_id")
.expect("lock id")
.to_string();
"lock_watch",
json!({"agent_id":"tester","lock_id":lock_id,"strategy":"on_ingest"}),
write(&file, &light_doc("WatchedNodeV2", "10.1000/watch"));
let diff = call(
"lock_diff",
json!({"agent_id":"tester","lock_id":lock_id}),
diff.get("watcher_events_drained")
>= 1
fn auto_ingest_restart_skips_unchanged_files() {
let docs_root = temp.path().join("restart");
let file = docs_root.join("restart.md");
write(&file, &light_doc("RestartStudy", "10.1000/restart"));
call(&mut state, "auto_ingest_stop", json!({"agent_id":"tester"}));
drop(state);
let mut restarted = build_state(temp.path());
let start = call(
&mut restarted,
let skipped = start
.get("bootstrap")
.and_then(|value| value.get("skipped_paths"))
.unwrap_or(0);
assert!(skipped >= 1, "restart should skip unchanged file");
write(&file, &light_doc("RestartStudyV2", "10.1000/restart"));
wait_for_queue(&mut restarted, 1);
let tick = call(
"auto_ingest_tick",
json!({"agent_id":"tester"}),
let ingested = tick
.get("ingested_paths")
assert_eq!(ingested, 1, "only the changed file should be reingested");
fn auto_ingest_hot_file_storm_converges_to_last_write() {
let docs_root = temp.path().join("storm");
let file = docs_root.join("storm.md");
for i in 0..50 {
write(&file, &light_doc(&format!("Storm{}", i), "10.1000/storm"));
search_count(&mut state, "Storm49") > 0,
"final write should win"
fn auto_ingest_burst_stress_handles_many_documents() {
let docs_root = temp.path().join("burst");
for round in 0..5 {
for index in 0..200 {
&docs_root.join(format!("doc-{}.md", index)),
&light_doc(&format!("Burst{}-{}", round, index), "10.1000/burst"),
thread::sleep(Duration::from_millis(40));
search_count(&mut state, "Burst4-199") > 0,
"burst stress should leave final graph queryable"
fn universal_ingest_actor_checkpoint_writes_canonical_artifacts_and_resolves_document() {
let docs_root = temp.path().join("universal");
&file,
&plain_markdown(
"Universal Notes",
"This document mentions TokenValidator and 10.1000/test.",
let actor = build_actor(temp.path());
let ingest = actor_call(
&actor,
"ingest",
json!({"agent_id":"tester","path":file.to_string_lossy().to_string(),"adapter":"universal","mode":"merge"}),
true,
ingest.get("adapter").and_then(|v| v.as_str()),
Some("universal")
let resolved = actor_call(
"document_resolve",
json!({"agent_id":"tester","path":"notes.md"}),
let canonical_md = resolved
.get("canonical_markdown_path")
.unwrap();
let canonical_json = resolved
.get("canonical_json_path")
let claims_json = resolved
.get("claims_path")
assert!(Path::new(canonical_md).exists());
assert!(Path::new(canonical_json).exists());
assert!(Path::new(claims_json).exists());
let search = actor_call(
json!({"agent_id":"tester","query":"TokenValidator","mode":"literal"}),
assert!(search["results"]
.as_array()
.is_some_and(|results| !results.is_empty()));
actor.stop().expect("stop actor");
fn auto_ingest_universal_handles_markdown_and_html() {
let docs_root = temp.path().join("universal-watch");
let md = docs_root.join("notes.md");
let html = docs_root.join("page.html");
actor_call(
json!({"agent_id":"tester","roots":[docs_root.to_string_lossy().to_string()],"formats":["universal"],"debounce_ms":0}),
&md,
&plain_markdown("Universal Watch", "TokenValidator appears here."),
write(&html, &html_doc("HTML Watch", "DesignSystem appears here."));
wait_for_actor_queue(&actor, 2);
let tick = actor_call(
ingested, 2,
"one tick must ingest both watched files; tick output: {tick}"
// Stronger than the count: the two ingested paths must BE the two watched
// files (canonicalized on both sides — the queue stores canonical paths).
let mut ingested_canonical = tick["ingested_paths"]
.unwrap()
.map(|value| Path::new(value.as_str().unwrap()).canonicalize().unwrap())
.collect::<Vec<_>>();
ingested_canonical.sort();
let mut expected = vec![md.canonicalize().unwrap(), html.canonicalize().unwrap()];
expected.sort();
assert_eq!(ingested_canonical, expected);
for query in ["Universal Watch", "HTML Watch"] {
let md_resolved = actor_call(
let html_resolved = actor_call(
json!({"agent_id":"tester","path":"page.html"}),
assert!(Path::new(md_resolved["canonical_markdown_path"].as_str().unwrap()).exists());
assert!(Path::new(html_resolved["canonical_markdown_path"].as_str().unwrap()).exists());
"auto_ingest_stop",
fn universal_document_bindings_and_drift_surface_work() {
let docs_root = temp.path().join("semantic");
let file = docs_root.join("spec.md");
"# API\n\n`TokenValidator` must validate requests.\n\nSee `src/token_validator.rs`.\n",
{
let mut graph = state.graph.write();
graph
.add_node(
"file::src/token_validator.rs",
"TokenValidator",
m1nd_core::types::NodeType::File,
&["code"],
1.0,
0.1,
graph.finalize().unwrap();
state.rebuild_engines().unwrap();
let bindings = call(
"document_bindings",
json!({"agent_id":"tester","path":"spec.md","top_k":5}),
let bindings_len = bindings["bindings"]
.map(|v| v.len())
assert!(bindings_len > 0);
let drift = call(
"document_drift",
json!({"agent_id":"tester","path":"spec.md"}),
assert!(drift.get("summary").is_some());
fn auto_ingest_status_exposes_semantic_counts() {
let docs_root = temp.path().join("status");
&docs_root.join("notes.md"),
"# Overview\n\n`TokenValidator` must validate requests.\n10.1000/test\n",
let status = call(
assert!(status["semantic_document_count"].as_u64().unwrap_or(0) >= 1);
assert!(status["semantic_section_count"].as_u64().unwrap_or(0) >= 1);
assert!(status["semantic_entity_count"].as_u64().unwrap_or(0) >= 1);
assert!(status["semantic_claim_count"].as_u64().unwrap_or(0) >= 1);
assert_eq!(status["drift_document_count"].as_u64().unwrap_or(0), 1);
status["provider_route_counts"]["universal:internal"]
.as_u64()
.unwrap_or(0),
1
status["provider_fallback_counts"]["universal:internal"]
fn auto_ingest_status_reflects_drift_after_explicit_refresh() {
let docs_root = temp.path().join("status-drift");
state.bump_graph_generation();
let initial = call(
assert_eq!(initial["drift_document_count"].as_u64().unwrap_or(0), 0);
let node = graph.resolve_id("file::src/token_validator.rs").unwrap();
graph.nodes.last_modified[node.as_usize()] = 9999999999.0;
drift["summary"]["code_change_unreflected"]
let refreshed = call(
assert_eq!(refreshed["drift_document_count"].as_u64().unwrap_or(0), 1);
refreshed["semantic_document_count"].as_u64().unwrap_or(0) >= 1,
"semantic counts should remain populated after drift refresh"
fn provider_gated_docling_docx_flow_skips_without_provider_python() {
let Some(provider_python) = std::env::var_os("M1ND_PROVIDER_PYTHON") else {
eprintln!("SKIP: M1ND_PROVIDER_PYTHON not configured");
let provider_python = PathBuf::from(provider_python);
if !provider_python.exists() {
eprintln!("SKIP: configured provider python missing");
let Ok(docling_probe) = std::process::Command::new(&provider_python)
.arg("-c")
.arg("import docling")
.output()
else {
eprintln!("SKIP: failed to spawn configured provider python");
if !docling_probe.status.success() {
eprintln!("SKIP: docling not available in provider env");
let docs_root = temp.path().join("provider");
let docx = docs_root.join("provider.docx");
let Ok(output) = std::process::Command::new(&provider_python)
.arg(format!(
"from docx import Document; d=Document(); d.add_heading('Provider Docx', level=1); d.add_paragraph('DoclingKnowledge appears here.'); d.save(r'{}')",
docx.display()
))
if !output.status.success() {
eprintln!("SKIP: python-docx not available in provider env");
let resolved = call(
json!({"agent_id":"tester","path":"provider.docx"}),
assert_eq!(resolved["producer"].as_str(), Some("universal:docling"));
let source_copy = resolved["original_source_path"].as_str().unwrap();
assert_eq!(fs::read(source_copy).unwrap(), fs::read(&docx).unwrap());
fn provider_gated_trafilatura_html_flow_skips_without_provider_python() {
let Ok(trafilatura_probe) = std::process::Command::new(&provider_python)
.arg("import trafilatura")
if !trafilatura_probe.status.success() {
eprintln!("SKIP: trafilatura not available in provider env");
let docs_root = temp.path().join("provider-html");
let page = docs_root.join("provider.html");
&page,
&html_doc("Provider Html", "SemanticBridge appears here."),
json!({"agent_id":"tester","path":page.to_string_lossy().to_string(),"adapter":"universal","mode":"merge"}),
json!({"agent_id":"tester","path":"provider.html"}),
assert_eq!(resolved["producer"].as_str(), Some("universal:trafilatura"));
assert!(search_count(&mut state, "SemanticBridge") > 0);
fn provider_gated_grobid_pdf_flow_skips_without_provider_env() {
let Ok(requests_probe) = std::process::Command::new(&provider_python)
.arg("import requests")
if !requests_probe.status.success() {
eprintln!("SKIP: requests not available in provider env");
let Some(grobid_url) = std::env::var_os("M1ND_GROBID_URL") else {
eprintln!("SKIP: M1ND_GROBID_URL not configured");
let grobid_url = grobid_url.to_string_lossy().to_string();
if grobid_url.trim().is_empty() {
eprintln!("SKIP: M1ND_GROBID_URL is empty");
let Ok(grobid_probe) = std::process::Command::new(&provider_python)
"import requests, sys; url = {}.rstrip('/') + '/api/isalive'; r = requests.get(url, timeout=10); sys.exit(0 if r.ok else 1)",
serde_json::to_string(&grobid_url).unwrap()
eprintln!("SKIP: failed to probe grobid service");
if !grobid_probe.status.success() {
eprintln!("SKIP: grobid service not reachable");
let docs_root = temp.path().join("provider-pdf");
let pdf = docs_root.join("provider.pdf");
fs::write(&pdf, simple_pdf_bytes("GraphBridge GROBID proof document")).unwrap();
json!({"agent_id":"tester","path":pdf.to_string_lossy().to_string(),"adapter":"universal","mode":"merge"}),
json!({"agent_id":"tester","path":"provider.pdf"}),
assert_eq!(resolved["producer"].as_str(), Some("universal:grobid"));
