Shared memory and context tools for agentic work.
Code Rooms
use crate::canonical::{
short_hash, short_hash_bytes, source_kind_from_extension, CanonicalDocument, ClaimModality,
ConfidenceLevel, DocumentBlock, DocumentBlockKind, DocumentCitation, DocumentClaimCandidate,
DocumentClaimKind, DocumentCodeCandidate, DocumentEntityCandidate, DocumentEntityKind,
DocumentLink, DocumentMetadata, DocumentSection, DocumentSectionKind, DocumentSpan,
DocumentTable, DocumentTableCell, DocumentTableRow, ProvenanceSpan, SourceKind,
};
use crate::{extension_of, relative_source_path};
use crate::{
BibTexAdapter, CrossRefAdapter, IngestAdapter, IngestStats, JatsArticleAdapter,
L1ghtIngestAdapter, PatentIngestAdapter, RfcAdapter,
use m1nd_core::error::{M1ndError, M1ndResult};
use m1nd_core::graph::{Graph, NodeProvenanceInput};
use m1nd_core::types::{EdgeDirection, FiniteF32, NodeType};
use regex::Regex;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::fs;
use std::io::{self, Read};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};
use walkdir::WalkDir;
#[derive(Clone, Debug, Serialize, Deserialize, Default)]
pub struct ProviderAvailability {
pub magika: bool,
pub docling: bool,
pub markitdown: bool,
pub trafilatura: bool,
pub grobid: bool,
pub marker: bool,
pub mineru: bool,
}
const MAX_UNIVERSAL_DIAGNOSTICS: usize = 32;
const MAX_UNIVERSAL_PROVIDER_CHARS: usize = 80;
const MAX_UNIVERSAL_REASON_CHARS: usize = 240;
const DEFAULT_PROVIDER_TIMEOUT_MS: u64 = 30_000;
const MAX_PROVIDER_TIMEOUT_MS: u64 = 300_000;
const MAX_PROVIDER_STDOUT_BYTES: usize = 8 * 1024 * 1024;
const MAX_PROVIDER_STDERR_BYTES: usize = 64 * 1024;
const PROVIDER_POLL_INTERVAL_MS: u64 = 5;
#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum ProviderFailureKind {
SpawnFailed,
Crashed,
TimedOut,
Corrupt,
Encrypted,
InvalidOutput,
OutputLimitExceeded,
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(
tag = "status",
content = "failure",
rename_all = "SCREAMING_SNAKE_CASE"
)]
pub enum ProviderExtractionOutcome {
Extracted,
Empty,
Failed(ProviderFailureKind),
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ProviderExtractionResult {
Extracted(String),
Failed {
kind: ProviderFailureKind,
detail: String,
},
impl ProviderExtractionResult {
fn outcome(&self) -> ProviderExtractionOutcome {
match self {
Self::Extracted(_) => ProviderExtractionOutcome::Extracted,
Self::Empty => ProviderExtractionOutcome::Empty,
Self::Failed { kind, .. } => ProviderExtractionOutcome::Failed(*kind),
pub enum UniversalIngestStatus {
Ingested,
Degraded,
Unsupported,
Failed,
impl UniversalIngestStatus {
pub fn as_str(self) -> &'static str {
Self::Ingested => "INGESTED",
Self::Degraded => "DEGRADED",
Self::Unsupported => "UNSUPPORTED",
Self::Failed => "FAILED",
Self::Empty => "EMPTY",
pub struct UniversalDocumentOutcome {
pub source_path: String,
pub source_kind: SourceKind,
pub status: UniversalIngestStatus,
pub parsed: bool,
pub provider: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub provider_outcome: Option<ProviderExtractionOutcome>,
#[serde(skip_serializing_if = "Option::is_none")]
pub reason: Option<String>,
impl UniversalDocumentOutcome {
fn new(
source_path: String,
source_kind: SourceKind,
status: UniversalIngestStatus,
parsed: bool,
provider: &str,
provider_outcome: Option<ProviderExtractionOutcome>,
reason: Option<&str>,
) -> Self {
Self {
source_path,
source_kind,
status,
parsed,
provider: bounded_text(provider, MAX_UNIVERSAL_PROVIDER_CHARS),
provider_outcome,
reason: reason.map(|value| bounded_text(value, MAX_UNIVERSAL_REASON_CHARS)),
pub struct UniversalIngestSummary {
pub candidate_count: u64,
pub parsed_count: u64,
pub ingested_count: u64,
pub degraded_count: u64,
pub unsupported_count: u64,
pub failed_count: u64,
pub diagnostics: Vec<UniversalDocumentOutcome>,
pub diagnostics_omitted: u64,
pub struct UniversalIngestBundle {
pub graph: Graph,
pub stats: IngestStats,
pub documents: Vec<CanonicalDocument>,
pub outcomes: Vec<UniversalDocumentOutcome>,
pub providers: ProviderAvailability,
impl UniversalIngestBundle {
pub fn summary(&self) -> UniversalIngestSummary {
summarize_outcomes(self.status, &self.outcomes)
pub fn is_committable(&self) -> bool {
self.stats.files_parsed > 0
&& matches!(
self.status,
UniversalIngestStatus::Ingested | UniversalIngestStatus::Degraded
)
pub struct UniversalIngestAdapter {
namespace: String,
impl UniversalIngestAdapter {
pub fn new(namespace: Option<String>) -> Self {
namespace: namespace.unwrap_or_else(|| "universal".to_string()),
pub fn provider_availability() -> ProviderAvailability {
ProviderAvailability {
magika: python_module_available("magika"),
docling: python_module_available("docling"),
markitdown: python_module_available("markitdown"),
trafilatura: python_module_available("trafilatura"),
grobid: grobid_configured(),
marker: command_available("marker"),
mineru: command_available("mineru"),
pub fn provider_python_command() -> String {
provider_python()
pub fn can_handle_path(path: &Path) -> bool {
let ext = extension_of(path);
matches!(
ext.as_str(),
"md" | "markdown"
| "txt"
| "rst"
| "adoc"
| "html"
| "htm"
| "pdf"
| "docx"
| "pptx"
| "xlsx"
| "xml"
| "nxml"
| "json"
| "bib"
| "bibtex"
pub fn ingest_bundle(&self, root: &Path) -> M1ndResult<UniversalIngestBundle> {
let providers = Self::provider_availability();
self.ingest_bundle_with(root, providers, extract_with_provider)
fn ingest_bundle_with<F>(
&self,
root: &Path,
providers: ProviderAvailability,
extract: F,
) -> M1ndResult<UniversalIngestBundle>
where
F: Fn(&str, &Path) -> ProviderExtractionResult,
{
let start = std::time::Instant::now();
let mut stats = IngestStats::default();
let files = collect_candidate_files(root)?;
stats.files_scanned = files.len() as u64;
let mut documents = Vec::new();
let mut outcomes = Vec::with_capacity(files.len());
for path in files {
match self.canonicalize_path_with(root, &path, &providers, &extract) {
Ok(canonicalized) => {
if let Some(document) = canonicalized.document {
stats.files_parsed += 1;
documents.push(document);
outcomes.push(canonicalized.outcome);
Err(error) => {
outcomes.push(UniversalDocumentOutcome::new(
relative_source_path(root, &path),
source_kind_from_extension(&path),
UniversalIngestStatus::Failed,
false,
"universal:reader",
None,
Some(&error.to_string()),
));
let status = bundle_status(&outcomes, stats.files_parsed);
let graph = graphify_documents(&documents, &self.namespace)?;
stats.nodes_created = graph.num_nodes() as u64;
stats.edges_created = graph.num_edges() as u64;
stats.elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
Ok(UniversalIngestBundle {
graph,
stats,
documents,
outcomes,
providers,
})
fn canonicalize_path_with<F>(
path: &Path,
availability: &ProviderAvailability,
extract: &F,
) -> M1ndResult<CanonicalizedFile>
let rel_path = relative_source_path(root, path);
let source_kind = source_kind_from_extension(path);
let bytes = fs::read(path).map_err(|error| M1ndError::InvalidParams {
tool: "universal_ingest".into(),
detail: format!("failed to read {}: {}", path.display(), error),
})?;
let source_text = String::from_utf8_lossy(&bytes).to_string();
let (mut document, status, provider, provider_outcome, reason) = match source_kind {
SourceKind::Markdown | SourceKind::Text => {
let document = canonicalize_plain_text(
&rel_path,
source_kind.clone(),
"universal:internal",
&source_text,
);
(
Some(document),
UniversalIngestStatus::Ingested,
SourceKind::Html => {
if availability.trafilatura {
let extraction = normalize_provider_extraction(extract("trafilatura", path));
let provider_outcome = Some(extraction.outcome());
match extraction {
ProviderExtractionResult::Extracted(extracted) => (
Some(canonicalize_html_with_fallback(
"universal:trafilatura",
"universal:internal-html",
&extracted,
)),
),
ProviderExtractionResult::Empty => (
UniversalIngestStatus::Degraded,
Some("provider extraction returned no content; internal HTML fallback used".to_string()),
ProviderExtractionResult::Failed { kind, detail } => (
Some(ProviderExtractionOutcome::Failed(kind)),
Some(format!("provider extraction failed: {detail}; internal HTML fallback used")),
} else if availability.docling {
let extraction = normalize_provider_extraction(extract("docling", path));
"universal:docling",
} else {
SourceKind::Pdf | SourceKind::Docx | SourceKind::Pptx | SourceKind::Xlsx => {
let selected_provider =
if matches!(source_kind, SourceKind::Pdf) && availability.grobid {
Some(("grobid", "universal:grobid"))
Some(("docling", "universal:docling"))
} else if availability.markitdown {
Some(("markitdown", "universal:markitdown"))
None
match selected_provider {
Some((provider_key, producer)) => {
let extraction = normalize_provider_extraction(extract(provider_key, path));
Some(canonicalize_binary_placeholder(
producer,
Some("provider extraction returned no content".to_string()),
Some(detail),
None => (
UniversalIngestStatus::Unsupported,
"universal:none",
Some(if matches!(source_kind, SourceKind::Pdf) {
"no configured provider (grobid, docling, or markitdown)".to_string()
"no configured provider (docling or markitdown)".to_string()
}),
SourceKind::Unknown => {
if let Some(native) =
self.wrap_native_document(root, path, &rel_path, &source_text)?
let provider = native.producer.clone();
Some(native),
if provider == "universal:native-wrap" {
"universal:native-wrap"
"universal:internal"
Some("unsupported document type or unrecognized native format".to_string()),
_ => (
Some("unsupported document type".to_string()),
if let Some(document) = document.as_mut() {
document.content_hash = short_hash_bytes(&bytes);
let outcome = UniversalDocumentOutcome::new(
rel_path,
document.is_some(),
provider,
reason.as_deref(),
Ok(CanonicalizedFile { document, outcome })
fn wrap_native_document(
_root: &Path,
rel_path: &str,
source_text: &str,
) -> M1ndResult<Option<CanonicalDocument>> {
let ext = path
.extension()
.and_then(|value| value.to_str())
.map(|value| value.to_ascii_lowercase())
.unwrap_or_default();
let native_kind = if matches!(ext.as_str(), "md" | "markdown")
&& L1ghtIngestAdapter::looks_like_l1ght(source_text)
Some(SourceKind::NativeLight)
} else if matches!(ext.as_str(), "bib" | "bibtex") {
Some(SourceKind::NativeBibtex)
} else if matches!(ext.as_str(), "xml" | "nxml") {
if source_text.contains("<PubmedArticle")
|| source_text.contains("<PubmedArticleSet")
|| source_text.contains("NLM//DTD")
|| (source_text.contains("<article") && source_text.contains("dtd-version"))
Some(SourceKind::NativeArticle)
} else if source_text.contains("<rfc ") || source_text.contains("<rfc>") {
Some(SourceKind::NativeRfc)
} else if source_text.contains("<us-patent-grant")
|| source_text.contains("<us-patent-application")
|| source_text.contains("<ep-patent-document")
Some(SourceKind::NativePatent)
} else if ext == "json"
&& source_text.contains("\"DOI\"")
&& source_text.contains("\"publisher\"")
&& source_text.contains("\"type\"")
Some(SourceKind::NativeCrossref)
let Some(native_kind) = native_kind else {
return Ok(None);
Ok(Some(canonicalize_plain_text(
native_kind,
"universal:native-wrap",
source_text,
)))
impl IngestAdapter for UniversalIngestAdapter {
fn domain(&self) -> &str {
"universal"
fn ingest(&self, root: &Path) -> M1ndResult<(Graph, IngestStats)> {
let bundle = self.ingest_bundle(root)?;
if !bundle.is_committable() {
let summary = bundle.summary();
return Err(M1ndError::InvalidParams {
detail: format!(
"universal ingest is noncommittable: status={}, candidate_count={}, parsed_count={}, unsupported_count={}, failed_count={}",
summary.status.as_str(),
summary.candidate_count,
summary.parsed_count,
summary.unsupported_count,
summary.failed_count
});
Ok((bundle.graph, bundle.stats))
struct CanonicalizedFile {
document: Option<CanonicalDocument>,
outcome: UniversalDocumentOutcome,
fn bundle_status(
outcomes: &[UniversalDocumentOutcome],
parsed_count: u64,
) -> UniversalIngestStatus {
if outcomes.is_empty() {
UniversalIngestStatus::Empty
} else if parsed_count == 0
&& outcomes
.iter()
.any(|outcome| outcome.status == UniversalIngestStatus::Failed)
UniversalIngestStatus::Failed
} else if parsed_count == 0 {
UniversalIngestStatus::Unsupported
} else if outcomes
.all(|outcome| outcome.status == UniversalIngestStatus::Ingested)
UniversalIngestStatus::Ingested
UniversalIngestStatus::Degraded
fn summarize_outcomes(
) -> UniversalIngestSummary {
let parsed_count = outcomes.iter().filter(|outcome| outcome.parsed).count() as u64;
let ingested_count = outcomes
.filter(|outcome| outcome.status == UniversalIngestStatus::Ingested)
.count() as u64;
let degraded_count = outcomes
.filter(|outcome| outcome.status == UniversalIngestStatus::Degraded)
let unsupported_count = outcomes
.filter(|outcome| outcome.status == UniversalIngestStatus::Unsupported)
let failed_count = outcomes
.filter(|outcome| outcome.status == UniversalIngestStatus::Failed)
let diagnostic_total = (degraded_count + unsupported_count + failed_count) as usize;
let diagnostics = outcomes
.filter(|outcome| {
outcome.status,
| UniversalIngestStatus::Unsupported
| UniversalIngestStatus::Failed
.take(MAX_UNIVERSAL_DIAGNOSTICS)
.cloned()
.collect::<Vec<_>>();
UniversalIngestSummary {
candidate_count: outcomes.len() as u64,
parsed_count,
ingested_count,
degraded_count,
unsupported_count,
failed_count,
diagnostics_omitted: diagnostic_total.saturating_sub(diagnostics.len()) as u64,
diagnostics,
fn bounded_text(input: &str, max_chars: usize) -> String {
input.chars().take(max_chars).collect()
fn normalize_provider_extraction(result: ProviderExtractionResult) -> ProviderExtractionResult {
match result {
ProviderExtractionResult::Extracted(text) if text.trim().is_empty() => {
ProviderExtractionResult::Empty
ProviderExtractionResult::Extracted(text) => {
ProviderExtractionResult::Extracted(text.trim().to_string())
other => other,
fn extract_with_provider(provider: &str, path: &Path) -> ProviderExtractionResult {
match provider {
"docling" => docling_extract(path),
"trafilatura" => trafilatura_extract(path),
"markitdown" => markitdown_extract(path),
"grobid" => grobid_extract(path),
_ => ProviderExtractionResult::Failed {
kind: ProviderFailureKind::InvalidOutput,
detail: format!("unknown universal document provider '{provider}'"),
fn python_module_available(module_name: &str) -> bool {
let mut command = Command::new(provider_python());
command.arg("-c").arg(format!(
"import importlib.util; print('1' if importlib.util.find_spec('{module_name}') else '0')"
run_provider_command(&mut command, Duration::from_secs(2), 1024, 1024),
ProviderExtractionResult::Extracted(value) if value == "1"
fn command_available(name: &str) -> bool {
Command::new("sh")
.arg("-c")
.arg(format!("command -v {} >/dev/null 2>&1", name))
.status()
.map(|status| status.success())
.unwrap_or(false)
fn grobid_configured() -> bool {
configured_grobid_endpoint().is_ok()
fn configured_grobid_endpoint() -> Result<url::Url, String> {
let raw = std::env::var("M1ND_GROBID_URL")
.map_err(|_| "M1ND_GROBID_URL is not configured".to_string())?;
let allowed_hosts = std::env::var("M1ND_GROBID_ALLOWED_HOSTS")
.unwrap_or_default()
.split(',')
.map(str::trim)
.filter(|host| !host.is_empty())
.map(str::to_ascii_lowercase)
validate_grobid_endpoint(&raw, &allowed_hosts)
fn validate_grobid_endpoint(raw: &str, allowed_hosts: &[String]) -> Result<url::Url, String> {
let endpoint = url::Url::parse(raw.trim())
.map_err(|_| "GROBID endpoint is not a valid absolute URL".to_string())?;
if endpoint.username() != "" || endpoint.password().is_some() {
return Err(
"GROBID endpoint userinfo is forbidden; configure credentials separately".to_string(),
if endpoint.query().is_some() || endpoint.fragment().is_some() {
return Err("GROBID endpoint query and fragment are forbidden".to_string());
let host = endpoint
.host_str()
.ok_or_else(|| "GROBID endpoint requires a host".to_string())?
.to_ascii_lowercase();
let loopback = match endpoint.host() {
Some(url::Host::Ipv4(address)) => address.is_loopback(),
Some(url::Host::Ipv6(address)) => address.is_loopback(),
Some(url::Host::Domain(domain)) => domain.eq_ignore_ascii_case("localhost"),
None => false,
match endpoint.scheme() {
"http" if loopback => {}
"https" if loopback || allowed_hosts.iter().any(|allowed| allowed == &host) => {}
"http" => return Err("plaintext GROBID is allowed only on loopback".to_string()),
"https" => {
"remote GROBID host is not present in M1ND_GROBID_ALLOWED_HOSTS".to_string(),
_ => return Err("GROBID endpoint must use HTTPS or loopback HTTP".to_string()),
Ok(endpoint)
/// Redacted diagnostic identity: origin only, never credentials, path, query,
/// or fragment. Invalid configured endpoints are reported without echoing input.
pub fn grobid_endpoint_summary() -> Option<String> {
match std::env::var("M1ND_GROBID_URL") {
Err(_) => None,
Ok(_) => Some(
configured_grobid_endpoint()
.map(|endpoint| endpoint.origin().ascii_serialization())
.unwrap_or_else(|_| "configured_but_refused".to_string()),
fn provider_timeout() -> Duration {
let timeout_ms = std::env::var("M1ND_PROVIDER_TIMEOUT_MS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
.filter(|value| *value > 0)
.unwrap_or(DEFAULT_PROVIDER_TIMEOUT_MS)
.min(MAX_PROVIDER_TIMEOUT_MS);
Duration::from_millis(timeout_ms)
#[derive(Debug)]
struct CapturedOutput {
bytes: Vec<u8>,
exceeded_limit: bool,
fn read_bounded(mut reader: impl Read, limit: usize) -> io::Result<CapturedOutput> {
let mut bytes = Vec::with_capacity(limit.min(8 * 1024));
let mut exceeded_limit = false;
let mut buffer = [0_u8; 8 * 1024];
loop {
let count = reader.read(&mut buffer)?;
if count == 0 {
break;
let remaining = limit.saturating_sub(bytes.len());
let retained = remaining.min(count);
bytes.extend_from_slice(&buffer[..retained]);
exceeded_limit |= retained < count;
Ok(CapturedOutput {
bytes,
exceeded_limit,
fn join_capture(
handle: thread::JoinHandle<io::Result<CapturedOutput>>,
stream: &str,
) -> Result<CapturedOutput, ProviderExtractionResult> {
match handle.join() {
Ok(Ok(output)) => Ok(output),
Ok(Err(error)) => Err(ProviderExtractionResult::Failed {
detail: format!("failed to read provider {stream}: {error}"),
Err(_) => Err(ProviderExtractionResult::Failed {
detail: format!("provider {stream} reader panicked"),
fn failure_kind_from_stderr(stderr: &str) -> ProviderFailureKind {
let lower = stderr.to_ascii_lowercase();
if lower.contains("encrypt") || lower.contains("password") {
ProviderFailureKind::Encrypted
} else if lower.contains("corrupt") || lower.contains("malformed") || lower.contains("damaged")
ProviderFailureKind::Corrupt
ProviderFailureKind::Crashed
fn terminate_provider_process(child: &mut std::process::Child) {
#[cfg(unix)]
let process_group = -(child.id() as i32);
// SAFETY: the child is placed in a fresh process group immediately
// before spawn below. The negative pid therefore targets only that
// provider group, including helpers that inherited its output pipes.
unsafe {
libc::kill(process_group, libc::SIGKILL);
let _ = child.kill();
let _ = child.wait();
fn run_provider_command(
command: &mut Command,
timeout: Duration,
stdout_limit: usize,
stderr_limit: usize,
) -> ProviderExtractionResult {
command.stdout(Stdio::piped()).stderr(Stdio::piped());
use std::os::unix::process::CommandExt;
command.process_group(0);
let mut child = match command.spawn() {
Ok(child) => child,
return ProviderExtractionResult::Failed {
kind: ProviderFailureKind::SpawnFailed,
detail: format!("failed to spawn provider: {error}"),
let stdout = child
.stdout
.take()
.expect("provider stdout is piped before spawn");
let stderr = child
.stderr
.expect("provider stderr is piped before spawn");
let stdout_reader = thread::spawn(move || read_bounded(stdout, stdout_limit));
let stderr_reader = thread::spawn(move || read_bounded(stderr, stderr_limit));
let started = Instant::now();
let status = loop {
match child.try_wait() {
Ok(Some(status)) => break status,
Ok(None) if started.elapsed() >= timeout => {
terminate_provider_process(&mut child);
let _ = join_capture(stdout_reader, "stdout");
let _ = join_capture(stderr_reader, "stderr");
kind: ProviderFailureKind::TimedOut,
detail: format!("provider timed out after {} ms", timeout.as_millis()),
Ok(None) => {
let remaining = timeout.saturating_sub(started.elapsed());
thread::sleep(remaining.min(Duration::from_millis(PROVIDER_POLL_INTERVAL_MS)));
kind: ProviderFailureKind::Crashed,
detail: format!("failed while waiting for provider: {error}"),
let stdout = match join_capture(stdout_reader, "stdout") {
Ok(output) => output,
Err(result) => return result,
let stderr = match join_capture(stderr_reader, "stderr") {
if stdout.exceeded_limit || stderr.exceeded_limit {
kind: ProviderFailureKind::OutputLimitExceeded,
"provider output exceeded bounds (stdout={} bytes, stderr={} bytes)",
stdout_limit, stderr_limit
let stderr_text = String::from_utf8_lossy(&stderr.bytes).trim().to_string();
if !status.success() {
let kind = failure_kind_from_stderr(&stderr_text);
let detail = if stderr_text.is_empty() {
format!("provider exited unsuccessfully: {status}")
bounded_text(&stderr_text, MAX_UNIVERSAL_REASON_CHARS)
return ProviderExtractionResult::Failed { kind, detail };
let stdout_text = match String::from_utf8(stdout.bytes) {
Ok(text) => text,
detail: format!("provider stdout was not UTF-8: {error}"),
normalize_provider_extraction(ProviderExtractionResult::Extracted(stdout_text))
fn python_inline(script: &str, arg: &Path) -> ProviderExtractionResult {
command.arg("-c").arg(script).arg(arg);
run_provider_command(
&mut command,
provider_timeout(),
MAX_PROVIDER_STDOUT_BYTES,
MAX_PROVIDER_STDERR_BYTES,
fn provider_python() -> String {
std::env::var("M1ND_PROVIDER_PYTHON").unwrap_or_else(|_| "python3".to_string())
fn docling_extract(path: &Path) -> ProviderExtractionResult {
python_inline(
r#"
import sys
from docling.document_converter import DocumentConverter
converter = DocumentConverter()
result = converter.convert(sys.argv[1])
doc = getattr(result, 'document', None)
if doc is not None and hasattr(doc, 'export_to_markdown'):
text = doc.export_to_markdown()
if text:
print(text)
"#,
path,
fn trafilatura_extract(path: &Path) -> ProviderExtractionResult {
import sys, pathlib, trafilatura
path = pathlib.Path(sys.argv[1])
text = path.read_text(encoding='utf-8', errors='ignore')
extracted = trafilatura.extract(text, output_format='markdown', include_links=True, include_tables=True)
if extracted:
print(extracted)
fn markitdown_extract(path: &Path) -> ProviderExtractionResult {
from markitdown import MarkItDown
md = MarkItDown()
result = md.convert(sys.argv[1])
text = getattr(result, 'text_content', None) or getattr(result, 'markdown', None) or str(result)
fn grobid_extract(path: &Path) -> ProviderExtractionResult {
let url = match configured_grobid_endpoint() {
Ok(url) => url.to_string(),
Err(detail) => {
detail,
let script = format!(
import sys, requests
path = sys.argv[1]
url = {url_repr}.rstrip('/') + '/api/processFulltextDocument'
with open(path, 'rb') as fh:
resp = requests.post(url, files={{'input': fh}}, timeout=30, allow_redirects=False)
if 300 <= resp.status_code < 400:
raise RuntimeError('GROBID redirect refused')
resp.raise_for_status()
text = resp.text.strip()
url_repr = serde_json::to_string(&url).expect("URL string always serializes")
python_inline(&script, path)
fn collect_candidate_files(root: &Path) -> M1ndResult<Vec<PathBuf>> {
let metadata = fs::symlink_metadata(root)?;
if metadata.file_type().is_symlink() {
"refusing symlink ingest root {}; pass its canonical target explicitly",
root.display()
if metadata.is_file() {
return Ok(vec![root.to_path_buf()]);
if !metadata.is_dir() {
"ingest root is neither a regular file nor directory: {}",
let mut files = Vec::new();
for entry in WalkDir::new(root).follow_links(false) {
let entry = entry.map_err(|error| {
M1ndError::IngestError(format!(
"failed to traverse universal ingest root {}: {error}",
))
if entry.file_type().is_file() {
files.push(entry.into_path());
files.sort();
Ok(files)
fn slugify(raw: &str) -> String {
let mut out = String::with_capacity(raw.len());
let mut prev_dash = false;
for ch in raw.chars() {
if ch.is_ascii_alphanumeric() {
out.push(ch.to_ascii_lowercase());
prev_dash = false;
} else if !prev_dash {
out.push('-');
prev_dash = true;
out.trim_matches('-').to_string()
fn content_hash(input: &str) -> String {
short_hash(input)
fn default_title(source_path: &str, plain_text: &str) -> String {
plain_text
.lines()
.find(|line| !line.is_empty())
.map(|line| line.trim_start_matches('#').trim().to_string())
.filter(|line| !line.is_empty())
.unwrap_or_else(|| {
Path::new(source_path)
.file_name()
.unwrap_or(source_path)
.to_string()
fn section_id(source_path: &str, heading: &str, index: usize) -> String {
format!(
"section::{}::{}-{}",
short_hash(source_path),
slugify(heading),
index
fn block_id(source_path: &str, heading: &str, index: usize) -> String {
"block::{}::{}-{}",
fn claim_id(source_path: &str, label: &str, index: usize) -> String {
"claim::{}::{}-{}",
slugify(label),
fn classify_section_kind(heading: &str) -> DocumentSectionKind {
let lower = heading.to_ascii_lowercase();
if lower.contains("api") || lower.contains("contract") || lower.contains("interface") {
DocumentSectionKind::Api
} else if lower.contains("constraint") || lower.contains("invariant") || lower.contains("guard")
DocumentSectionKind::Constraints
} else if lower.contains("test") || lower.contains("verification") {
DocumentSectionKind::Tests
} else if lower.contains("rollout") || lower.contains("migration") {
DocumentSectionKind::Rollout
} else if lower.contains("reference") || lower.contains("bibliography") {
DocumentSectionKind::Reference
} else if lower.contains("appendix") {
DocumentSectionKind::Appendix
} else if lower.contains("overview")
|| lower.contains("introduction")
|| lower.contains("summary")
DocumentSectionKind::Overview
DocumentSectionKind::Unknown
fn classify_entity_kind(label: &str) -> DocumentEntityKind {
if label.contains("m1nd.") {
DocumentEntityKind::ToolId
} else if label.contains('/')
|| label.contains(".rs")
|| label.contains(".py")
|| label.contains(".ts")
|| label.contains(".md")
DocumentEntityKind::FilePath
} else if label.contains("::") || label.contains('.') {
DocumentEntityKind::Symbol
} else if label.to_ascii_lowercase().contains("test") {
DocumentEntityKind::TestName
DocumentEntityKind::NamedTerm
fn classify_claim(line: &str) -> (DocumentClaimKind, ClaimModality, bool) {
let lower = line.to_ascii_lowercase();
let modality = if lower.contains(" must ") || lower.starts_with("must ") {
ClaimModality::Must
} else if lower.contains(" should ") || lower.starts_with("should ") {
ClaimModality::Should
} else if lower.contains(" may ") || lower.starts_with("may ") {
ClaimModality::May
} else if lower.contains(" is ") || lower.starts_with("is ") {
ClaimModality::Is
ClaimModality::Unknown
let negated = lower.contains(" not ") || lower.contains(" no ");
let kind = if lower.contains("warning") || lower.contains("danger") || lower.contains("risk") {
DocumentClaimKind::Warning
} else if lower.contains("test") || lower.contains("assert") || lower.contains("expect") {
DocumentClaimKind::TestExpectation
} else if matches!(modality, ClaimModality::Must | ClaimModality::Should) {
DocumentClaimKind::Requirement
} else if lower.contains("decision") || lower.contains("we choose") || lower.contains("chosen")
DocumentClaimKind::Decision
} else if lower.contains("always") || lower.contains("never") || lower.contains("invariant") {
DocumentClaimKind::Invariant
} else if lower.contains("can ") || lower.contains("supports") || lower.contains("capability") {
DocumentClaimKind::Capability
DocumentClaimKind::Unknown
(kind, modality, negated)
fn canonicalize_plain_text(
source_path: &str,
producer: &str,
raw_text: &str,
) -> CanonicalDocument {
let heading_re = Regex::new(r"^(#{1,6})\s+(.+?)\s*$").unwrap();
let link_re = Regex::new(r"\[([^\]]+)\]\(([^)]+)\)").unwrap();
let doi_re = Regex::new(r"(10\.\d{4,9}/[-._;()/:A-Za-z0-9]+)").unwrap();
let code_ref_re = Regex::new(r"`([^`]+)`").unwrap();
let entity_re = Regex::new(r"\b([A-Z][A-Za-z0-9_.:-]{3,})\b").unwrap();
let mut sections = Vec::new();
let mut links = Vec::new();
let mut citations = Vec::new();
let mut entities = Vec::new();
let mut claims = Vec::new();
let mut code_candidates = Vec::new();
let mut tables = Vec::new();
let mut current_heading = "Document".to_string();
let mut current_level = 1u8;
let mut current_parent_section_id: Option<String> = None;
let mut current_blocks = Vec::new();
let mut section_index = 0usize;
let mut block_index = 0usize;
let mut claim_index = 0usize;
let mut seen_entities = HashSet::new();
let mut seen_links = HashSet::new();
let mut seen_citations = HashSet::new();
let mut seen_code_candidates = HashSet::new();
let mut section_stack: Vec<(u8, String)> = Vec::new();
let mut in_code_block = false;
let mut code_lang: Option<String> = None;
let mut flush_section = |sections: &mut Vec<DocumentSection>,
current_heading: &mut String,
current_level: &mut u8,
current_parent_section_id: &mut Option<String>,
current_blocks: &mut Vec<DocumentBlock>,
section_index: &mut usize| {
if current_blocks.is_empty() {
return;
*section_index += 1;
sections.push(DocumentSection {
section_id: section_id(source_path, current_heading, *section_index),
heading: current_heading.clone(),
level: *current_level,
kind: classify_section_kind(current_heading),
parent_section_id: current_parent_section_id.clone(),
blocks: std::mem::take(current_blocks),
provenance: ProvenanceSpan {
line_start: None,
line_end: None,
excerpt: Some(current_heading.clone()),
for (idx, line) in raw_text.lines().enumerate() {
let line_no = idx as u32 + 1;
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
if let Some(caps) = heading_re.captures(trimmed) {
flush_section(
&mut sections,
&mut current_heading,
&mut current_level,
&mut current_parent_section_id,
&mut current_blocks,
&mut section_index,
current_heading = caps.get(2).unwrap().as_str().trim().to_string();
current_level = caps.get(1).unwrap().as_str().len() as u8;
while section_stack
.last()
.is_some_and(|(level, _)| *level >= current_level)
section_stack.pop();
current_parent_section_id = section_stack.last().map(|(_, id)| id.clone());
section_stack.push((
current_level,
section_id(source_path, ¤t_heading, section_index + 1),
if trimmed.starts_with("```") {
in_code_block = !in_code_block;
code_lang = trimmed
.strip_prefix("```")
.filter(|value| !value.is_empty())
.map(|value| value.to_string());
let is_table_line = trimmed.contains('|')
&& trimmed.matches('|').count() >= 2
&& !trimmed.starts_with("http");
if is_table_line {
let cells = trimmed
.trim_matches('|')
.split('|')
.map(|cell| DocumentTableCell {
text: cell.trim().to_string(),
line_start: Some(line_no),
line_end: Some(line_no),
excerpt: Some(trimmed.to_string()),
let table_id = format!("table::{}::{}", short_hash(source_path), tables.len() + 1);
let headers = if tables.last().is_none() {
cells.iter().map(|cell| cell.text.clone()).collect()
Vec::new()
tables.push(DocumentTable {
table_id,
headers,
rows: vec![DocumentTableRow { cells }],
confidence: ConfidenceLevel::Parsed,
let kind = if in_code_block {
DocumentBlockKind::Code
} else if trimmed.starts_with("- ") || trimmed.starts_with("* ") {
DocumentBlockKind::ListItem
} else if trimmed.starts_with('>') {
DocumentBlockKind::Quote
} else if is_table_line {
DocumentBlockKind::Table
DocumentBlockKind::Paragraph
let mut spans = Vec::new();
block_index += 1;
current_blocks.push(DocumentBlock {
block_id: block_id(source_path, ¤t_heading, block_index),
kind,
text: trimmed.to_string(),
excerpt: Some(trimmed.chars().take(200).collect()),
language: code_lang.clone().filter(|_| in_code_block),
spans: Vec::new(),
for caps in link_re.captures_iter(trimmed) {
let label = caps.get(1).unwrap().as_str().to_string();
let target = caps.get(2).unwrap().as_str().to_string();
let key = format!("{label}|{target}");
if seen_links.insert(key) {
links.push(DocumentLink {
label,
target: target.clone(),
spans.push(DocumentSpan {
text: target,
kind: "link".into(),
for caps in doi_re.captures_iter(trimmed) {
let target = caps.get(1).unwrap().as_str().to_string();
if seen_citations.insert(target.clone()) {
citations.push(DocumentCitation {
label: target.clone(),
citation_kind: "paper".into(),
title: None,
authors: Vec::new(),
venue: None,
year: None,
kind: "doi".into(),
for caps in code_ref_re.captures_iter(trimmed) {
let label = caps.get(1).unwrap().as_str().trim().to_string();
if !label.is_empty() && seen_entities.insert(label.clone()) {
entities.push(DocumentEntityCandidate {
label: label.clone(),
kind: DocumentEntityKind::CodeRef,
aliases: Vec::new(),
if !label.is_empty() && seen_code_candidates.insert(label.clone()) {
code_candidates.push(DocumentCodeCandidate {
candidate_kind: classify_entity_kind(&label),
text: label,
kind: "code_ref".into(),
for caps in entity_re.captures_iter(trimmed) {
if seen_entities.insert(label.clone()) {
kind: classify_entity_kind(caps.get(1).unwrap().as_str().trim()),
confidence: ConfidenceLevel::Inferred,
if trimmed.ends_with('.') && trimmed.len() > 20 {
claim_index += 1;
let (kind, modality, negated) = classify_claim(trimmed);
let subject = code_ref_re
.captures(trimmed)
.and_then(|caps| caps.get(1).map(|m| m.as_str().trim().to_string()))
.or_else(|| {
entity_re
let predicate = match modality {
ClaimModality::Must => Some("must".to_string()),
ClaimModality::Should => Some("should".to_string()),
ClaimModality::May => Some("may".to_string()),
ClaimModality::Is => Some("is".to_string()),
ClaimModality::Unknown => None,
let object = predicate.as_ref().and_then(|verb| {
let needle = format!(" {} ", verb);
trimmed
.to_ascii_lowercase()
.find(&needle)
.map(|idx| {
trimmed[idx + needle.len()..]
.trim_end_matches('.')
.trim()
claims.push(DocumentClaimCandidate {
claim_id: claim_id(source_path, trimmed, claim_index),
label: trimmed.to_string(),
modality,
subject,
predicate,
object,
negated,
if let Some(last) = current_blocks.last_mut() {
last.spans = spans;
let title = default_title(source_path, raw_text);
let mut metadata = DocumentMetadata {
title: Some(title.clone()),
..Default::default()
if let Some(first) = citations.first() {
metadata.doi = Some(first.target.clone());
CanonicalDocument {
doc_id: format!("canon::{}", short_hash(source_path)),
source_path: source_path.to_string(),
source_kind: source_kind.clone(),
detected_type: match source_kind {
SourceKind::Markdown => "markdown".into(),
SourceKind::Text => "text".into(),
SourceKind::Html => "html".into(),
SourceKind::NativeLight => "light".into(),
SourceKind::NativeArticle => "article".into(),
SourceKind::NativeBibtex => "bibtex".into(),
SourceKind::NativeCrossref => "crossref".into(),
SourceKind::NativeRfc => "rfc".into(),
SourceKind::NativePatent => "patent".into(),
other => format!("{:?}", other).to_lowercase(),
producer: producer.to_string(),
content_hash: content_hash(raw_text),
title,
plain_text: raw_text.to_string(),
metadata,
sections,
tables,
links,
citations,
entities,
claims,
code_candidates,
structured_origin: serde_json::json!({ "producer": producer }),
fn strip_html_tags(input: &str) -> String {
let tag_re = Regex::new(r"<[^>]+>").unwrap();
let heading_re = Regex::new(r"(?i)<h([1-6])[^>]*>(.*?)</h[1-6]>").unwrap();
let mut text = input.to_string();
for caps in heading_re.captures_iter(input) {
let level = caps.get(1).unwrap().as_str();
let body = tag_re.replace_all(caps.get(2).unwrap().as_str(), "");
text = text.replace(
caps.get(0).unwrap().as_str(),
&format!(
"\n{} {}\n",
"#".repeat(level.parse::<usize>().unwrap_or(1)),
body.trim()
tag_re.replace_all(&text, " ").to_string()
fn canonicalize_html_with_fallback(
fallback_producer: &str,
let text = strip_html_tags(raw_text);
let producer = if text.trim().is_empty() {
fallback_producer
producer
canonicalize_plain_text(source_path, SourceKind::Html, producer, &text)
fn canonicalize_binary_placeholder(
let fallback = format!(
"# Imported Document\n\nSource: {}\n\nThis document was detected and reserved for optional provider-based canonicalization.\n",
source_path
let text = if source_text.trim().is_empty() {
fallback
source_text.to_string()
canonicalize_plain_text(source_path, source_kind, producer, &text)
pub fn graphify_documents(documents: &[CanonicalDocument], namespace: &str) -> M1ndResult<Graph> {
let mut graph = Graph::with_capacity(documents.len() * 24, documents.len() * 48);
let mut entity_nodes = HashSet::new();
let mut citation_nodes = HashSet::new();
let mut binding_nodes = HashSet::new();
for document in documents {
let doc_id = format!("universal::{}::doc::{}", namespace, document.doc_id);
let doc_tags = [
"universal".to_string(),
format!("universal:type:{}", document.detected_type),
format!("namespace:{}", namespace),
format!("producer:{}", document.producer),
];
let doc_tag_refs: Vec<&str> = doc_tags.iter().map(String::as_str).collect();
let doc_node = graph.add_node(
&doc_id,
&document.title,
NodeType::File,
&doc_tag_refs,
0.0,
0.5,
)?;
graph.set_node_provenance(
doc_node,
NodeProvenanceInput {
source_path: Some(&document.source_path),
excerpt: Some(&document.title),
namespace: Some(namespace),
canonical: true,
for section in &document.sections {
let section_id = format!("universal::{}::{}", namespace, section.section_id);
let section_tags = [
"universal:section".to_string(),
format!("section:kind:{:?}", section.kind).to_lowercase(),
let section_refs: Vec<&str> = section_tags.iter().map(String::as_str).collect();
let section_node = graph.add_node(
§ion_id,
§ion.heading,
NodeType::Module,
§ion_refs,
0.4,
section_node,
line_start: section.provenance.line_start,
line_end: section.provenance.line_end,
excerpt: section.provenance.excerpt.as_deref(),
graph.add_edge(
"contains_section",
FiniteF32::ONE,
EdgeDirection::Forward,
FiniteF32::new(0.8),
if let Some(parent_id) = §ion.parent_section_id {
let graph_parent_id = format!("universal::{}::{}", namespace, parent_id);
if let Some(parent_node) = graph.resolve_id(&graph_parent_id) {
parent_node,
"subsection_of",
FiniteF32::new(0.7),
FiniteF32::new(0.5),
for block in §ion.blocks {
let block_node_id = format!("universal::{}::{}", namespace, block.block_id);
let (node_type, relation, kind_tag) = match block.kind {
DocumentBlockKind::Code => {
(NodeType::Module, "contains_code", "universal:code")
DocumentBlockKind::Table => {
(NodeType::System, "contains_table", "universal:table")
_ => (NodeType::Concept, "contains_block", "universal:block"),
let block_tags = [
kind_tag.to_string(),
format!("confidence:{:?}", block.confidence).to_lowercase(),
let block_refs: Vec<&str> = block_tags.iter().map(String::as_str).collect();
let block_node = graph.add_node(
&block_node_id,
&block.text.chars().take(80).collect::<String>(),
node_type,
&block_refs,
0.3,
block_node,
line_start: block.provenance.line_start,
line_end: block.provenance.line_end,
excerpt: block.provenance.excerpt.as_deref(),
relation,
FiniteF32::new(0.9),
FiniteF32::new(0.6),
for table in &document.tables {
let table_id = format!("universal::{}::{}", namespace, table.table_id);
let table_tags = ["universal", "universal:table"];
let table_node = graph.add_node(
&table_id,
&format!("Table {}", table.table_id),
NodeType::System,
&table_tags,
0.35,
table_node,
line_start: table.provenance.line_start,
line_end: table.provenance.line_end,
excerpt: table.provenance.excerpt.as_deref(),
"contains_table",
FiniteF32::new(0.85),
for entity in &document.entities {
let entity_id = format!(
"universal::{}::entity::{}",
namespace,
slugify(&entity.label)
if entity_nodes.insert(entity_id.clone()) {
let tags = [
format!("entity:kind:{:?}", entity.kind).to_lowercase(),
format!("confidence:{:?}", entity.confidence).to_lowercase(),
let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
let entity_node = graph.add_node(
&entity_id,
&entity.label,
NodeType::Concept,
&tag_refs,
entity_node,
line_start: entity.provenance.line_start,
line_end: entity.provenance.line_end,
excerpt: entity.provenance.excerpt.as_deref(),
if let Some(entity_node) = graph.resolve_id(&entity_id) {
"declares_entity",
for citation in &document.citations {
let citation_id = format!(
"universal::{}::citation::{}",
slugify(&citation.target)
if citation_nodes.insert(citation_id.clone()) {
"universal:citation".to_string(),
format!("target:{}", citation.target),
let citation_node = graph.add_node(
&citation_id,
&citation.label,
NodeType::Reference,
0.25,
citation_node,
line_start: citation.provenance.line_start,
line_end: citation.provenance.line_end,
excerpt: citation.provenance.excerpt.as_deref(),
if let Some(citation_node) = graph.resolve_id(&citation_id) {
"references",
for link in &document.links {
let link_id = format!("universal::{}::link::{}", namespace, slugify(&link.target));
if graph.resolve_id(&link_id).is_none() {
let tags = ["universal".to_string(), "universal:link".to_string()];
let link_node = graph.add_node(
&link_id,
&link.label,
0.2,
link_node,
line_start: link.provenance.line_start,
line_end: link.provenance.line_end,
excerpt: link.provenance.excerpt.as_deref(),
if let Some(link_node) = graph.resolve_id(&link_id) {
"binds_to",
for claim in &document.claims {
let claim_id = format!("universal::{}::{}", namespace, claim.claim_id);
if graph.resolve_id(&claim_id).is_none() {
format!("confidence:{:?}", claim.confidence).to_lowercase(),
"universal:claim".to_string(),
format!("claim:kind:{:?}", claim.kind).to_lowercase(),
format!("claim:modality:{:?}", claim.modality).to_lowercase(),
let claim_node = graph.add_node(
&claim_id,
&claim.label.chars().take(80).collect::<String>(),
claim_node,
line_start: claim.provenance.line_start,
line_end: claim.provenance.line_end,
excerpt: claim.provenance.excerpt.as_deref(),
if let Some(claim_node) = graph.resolve_id(&claim_id) {
"declares_claim",
let excerpt = claim.provenance.excerpt.as_deref().unwrap_or(&claim.label);
if !excerpt.contains(&citation.target) && !excerpt.contains(&citation.label) {
"supports",
FiniteF32::new(0.4),
for candidate in &document.code_candidates {
let binding_id = format!(
"universal::{}::binding::{}",
slugify(&candidate.label)
if binding_nodes.insert(binding_id.clone()) {
"universal:binding".to_string(),
format!("candidate:{:?}", candidate.candidate_kind).to_lowercase(),
let binding_node = graph.add_node(
&binding_id,
&candidate.label,
binding_node,
line_start: candidate.provenance.line_start,
line_end: candidate.provenance.line_end,
excerpt: candidate.provenance.excerpt.as_deref(),
if let Some(binding_node) = graph.resolve_id(&binding_id) {
"mentions_symbol",
FiniteF32::new(0.75),
if graph.num_nodes() > 0 {
graph.finalize()?;
Ok(graph)
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn canonicalizes_markdown_without_l1ght() {
let doc = canonicalize_plain_text(
"docs/example.md",
SourceKind::Markdown,
"test",
"# API\n\n`TokenValidator` must validate requests.\n\nSee [Docs](https://example.com).\n\n`TokenValidator`\n",
assert_eq!(doc.detected_type, "markdown");
assert!(!doc.sections.is_empty());
assert!(matches!(doc.sections[0].kind, DocumentSectionKind::Api));
assert!(!doc.links.is_empty());
assert!(!doc.entities.is_empty());
assert!(!doc.claims.is_empty());
assert!(!doc.code_candidates.is_empty());
assert!(matches!(doc.claims[0].kind, DocumentClaimKind::Requirement));
assert!(matches!(doc.claims[0].modality, ClaimModality::Must));
fn graphify_documents_creates_sections_entities_and_refs() {
"# Overview\n\nHello World.\n\nSee [Docs](https://example.com).\n\n`TokenValidator`\n10.1000/test\n",
let graph = graphify_documents(&[doc], "universal").unwrap();
assert!(graph.num_nodes() >= 6);
assert!(graph.num_edges() >= 5);
fn canonicalize_extracts_simple_tables() {
"docs/table.md",
"# Overview\n\n| Name | Value |\n| A | B |\n",
assert!(!doc.tables.is_empty());
assert_eq!(doc.tables[0].rows.len(), 1);
fn graphify_documents_emits_code_table_and_subsection_edges() {
"docs/semantic.md",
"# API\n\n```rust\nfn validate() {}\n```\n\n## Tables\n\n| Name | Value |\n| A | B |\n",
let mut contains_code = 0;
let mut contains_table = 0;
let mut subsection_of = 0;
for src in 0..graph.num_nodes() as usize {
for edge_idx in graph
.csr
.out_range(m1nd_core::types::NodeId::new(src as u32))
let rel = graph.strings.resolve(graph.csr.relations[edge_idx]);
match rel {
"contains_code" => contains_code += 1,
"contains_table" => contains_table += 1,
"subsection_of" => subsection_of += 1,
_ => {}
assert!(contains_code >= 1);
assert!(contains_table >= 1);
assert!(subsection_of >= 1);
fn claim_support_edges_require_local_citation_evidence() {
"docs/evidence.md",
"# API\n\nTokenValidator must validate requests.\n\n10.1000/alpha\n10.1000/beta\n",
let mut support_edges = 0;
if rel == "supports" {
support_edges += 1;
assert_eq!(support_edges, 0);
fn provider_probe_is_stable() {
let providers = UniversalIngestAdapter::provider_availability();
let _ = serde_json::to_string(&providers).unwrap();
fn content_hash_tracks_original_source_bytes_for_html_documents() {
let temp = std::env::temp_dir().join(format!(
"m1nd-universal-hash-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
std::fs::create_dir_all(&temp).unwrap();
let file = temp.join("page.html");
let raw = "<html><body><h1>Hash Title</h1><p>TokenValidator must validate requests.</p></body></html>";
std::fs::write(&file, raw).unwrap();
let adapter = UniversalIngestAdapter::new(Some("universal".into()));
let bundle = adapter.ingest_bundle(&file).unwrap();
let document = bundle.documents.first().unwrap();
assert_eq!(document.content_hash, short_hash_bytes(raw.as_bytes()));
assert_ne!(document.content_hash, short_hash(&document.plain_text));
let _ = std::fs::remove_dir_all(&temp);
fn honesty_temp_dir(label: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!(
"m1nd-universal-honesty-{}-{}-{}",
std::fs::create_dir_all(&dir).unwrap();
dir
fn empty_candidate_set_is_explicit_empty() {
let dir = honesty_temp_dir("empty");
let adapter = UniversalIngestAdapter::new(Some("test".into()));
let bundle = adapter
.ingest_bundle_with(&dir, ProviderAvailability::default(), |_, _| {
.unwrap();
assert_eq!(bundle.status, UniversalIngestStatus::Empty);
assert_eq!(summary.status, UniversalIngestStatus::Empty);
assert_eq!(summary.candidate_count, 0);
assert_eq!(summary.parsed_count, 0);
assert!(summary.diagnostics.is_empty());
assert_eq!(bundle.graph.num_nodes(), 0);
let _ = std::fs::remove_dir_all(&dir);
fn missing_ingest_root_is_an_error_not_empty() {
let dir = honesty_temp_dir("missing-root");
let missing = dir.join("does-not-exist");
let error =
match adapter.ingest_bundle_with(&missing, ProviderAvailability::default(), |_, _| {
}) {
Err(error) => error,
Ok(_) => panic!("a missing root must never be reported as an empty ingest"),
assert!(matches!(error, M1ndError::Io(_)), "{error}");
fn symlinks_cannot_expand_the_ingest_boundary() {
use std::os::unix::fs::symlink;
let container = honesty_temp_dir("symlink-boundary");
let root = container.join("root");
let outside = container.join("outside");
std::fs::create_dir_all(&root).unwrap();
std::fs::create_dir_all(&outside).unwrap();
let inside_file = root.join("inside.md");
let outside_file = outside.join("secret.md");
std::fs::write(&inside_file, "# Inside\n").unwrap();
std::fs::write(&outside_file, "# Secret\n").unwrap();
symlink(&outside_file, root.join("file-link.md")).unwrap();
symlink(&outside, root.join("directory-link")).unwrap();
let files = collect_candidate_files(&root).unwrap();
assert_eq!(files, vec![inside_file]);
let linked_root = container.join("linked-root");
symlink(&root, &linked_root).unwrap();
let error = collect_candidate_files(&linked_root)
.expect_err("a symlink root must require an explicit canonical target");
assert!(matches!(error, M1ndError::InvalidParams { .. }), "{error}");
let _ = std::fs::remove_dir_all(&container);
fn pdf_and_office_without_providers_are_typed_noncommittable_unsupported() {
let dir = honesty_temp_dir("unsupported-binary");
for extension in ["pdf", "docx", "pptx", "xlsx"] {
std::fs::write(dir.join(format!("sample.{extension}")), b"binary fixture").unwrap();
assert_eq!(summary.status, UniversalIngestStatus::Unsupported);
assert_eq!(summary.candidate_count, 4);
assert_eq!(summary.unsupported_count, 4);
assert_eq!(summary.failed_count, 0);
assert!(!bundle.is_committable());
fn available_provider_returning_empty_is_typed_failed_not_placeholder() {
let dir = honesty_temp_dir("provider-none");
let file = dir.join("sample.docx");
std::fs::write(&file, b"PK fake docx bytes").unwrap();
let providers = ProviderAvailability {
docling: true,
..ProviderAvailability::default()
let extraction_calls = std::cell::Cell::new(0_u64);
.ingest_bundle_with(&file, providers, |provider, _| {
assert_eq!(provider, "docling");
extraction_calls.set(extraction_calls.get() + 1);
assert_eq!(extraction_calls.get(), 1);
assert_eq!(summary.status, UniversalIngestStatus::Failed);
assert_eq!(summary.failed_count, 1);
assert_eq!(summary.diagnostics[0].provider, "universal:docling");
assert_eq!(
summary.diagnostics[0].provider_outcome,
Some(ProviderExtractionOutcome::Empty)
assert!(summary.diagnostics[0]
.reason
.as_deref()
.contains("returned no content"));
fn mixed_batch_returns_degraded_with_parsed_documents_and_diagnostics() {
let dir = honesty_temp_dir("mixed");
std::fs::write(dir.join("good.md"), "# Good\n\nParsed content.\n").unwrap();
std::fs::write(dir.join("unsupported.pdf"), b"%PDF fixture").unwrap();
assert_eq!(bundle.status, UniversalIngestStatus::Degraded);
assert_eq!(bundle.documents.len(), 1);
assert_eq!(summary.candidate_count, 2);
assert_eq!(summary.parsed_count, 1);
assert_eq!(summary.ingested_count, 1);
assert_eq!(summary.unsupported_count, 1);
assert_eq!(summary.diagnostics.len(), 1);
summary.diagnostics[0].status,
fn provider_failure_uses_honest_html_fallback_provenance() {
let dir = honesty_temp_dir("html-fallback");
let file = dir.join("page.html");
std::fs::write(&file, "<h1>Fallback</h1><p>Useful text.</p>").unwrap();
trafilatura: true,
assert_eq!(provider, "trafilatura");
assert_eq!(bundle.documents[0].producer, "universal:internal-html");
assert!(bundle.documents[0].plain_text.contains("Fallback"));
assert_eq!(summary.degraded_count, 1);
assert_eq!(summary.diagnostics[0].provider, "universal:trafilatura");
fn diagnostics_are_bounded_and_report_omissions() {
let dir = honesty_temp_dir("bounded");
std::fs::write(dir.join("good.md"), "# Good\n").unwrap();
for index in 0..40 {
std::fs::write(dir.join(format!("unsupported-{index}.bin")), b"fixture").unwrap();
assert_eq!(summary.status, UniversalIngestStatus::Degraded);
assert_eq!(summary.unsupported_count, 40);
assert_eq!(summary.diagnostics.len(), MAX_UNIVERSAL_DIAGNOSTICS);
assert_eq!(summary.diagnostics_omitted, 8);
fn run_shell_provider(script: &str, timeout: Duration) -> ProviderExtractionResult {
let mut command = Command::new("sh");
command.arg("-c").arg(script);
run_provider_command(&mut command, timeout, 4096, 4096)
fn provider_process_result_distinguishes_positive_empty_crash_timeout_corrupt_and_encrypted() {
run_shell_provider("printf 'canonical text'", Duration::from_secs(1)),
ProviderExtractionResult::Extracted("canonical text".to_string())
run_shell_provider("exit 0", Duration::from_secs(1)),
let crashed = run_shell_provider("printf 'boom' >&2; exit 7", Duration::from_secs(1));
assert!(matches!(
crashed,
ProviderExtractionResult::Failed {
..
let timed_out = run_shell_provider("sleep 5 & wait", Duration::from_millis(30));
timed_out,
assert!(
started.elapsed() < Duration::from_secs(2),
"timed-out provider must be killed instead of waiting for natural exit"
let corrupt = run_shell_provider(
"printf 'corrupt document' >&2; exit 2",
Duration::from_secs(1),
corrupt,
kind: ProviderFailureKind::Corrupt,
let encrypted = run_shell_provider(
"printf 'password protected encrypted file' >&2; exit 2",
encrypted,
kind: ProviderFailureKind::Encrypted,
let mut missing = Command::new("m1nd-provider-command-that-does-not-exist");
run_provider_command(&mut missing, Duration::from_secs(1), 16, 16),
let mut oversized = Command::new("sh");
oversized.arg("-c").arg("printf '0123456789'");
run_provider_command(&mut oversized, Duration::from_secs(1), 4, 16),
let mut invalid_utf8 = Command::new("sh");
invalid_utf8.arg("-c").arg("printf '\\377'");
run_provider_command(&mut invalid_utf8, Duration::from_secs(1), 16, 16),
fn positive_provider_fixture_is_committable_and_persists_typed_outcome() {
let dir = honesty_temp_dir("provider-positive");
let file = dir.join("sample.pdf");
std::fs::write(&file, b"%PDF fixture").unwrap();
ProviderExtractionResult::Extracted("# Extracted\n\nUseful text.".to_string())
assert_eq!(bundle.status, UniversalIngestStatus::Ingested);
assert!(bundle.is_committable());
bundle.outcomes[0].provider_outcome,
Some(ProviderExtractionOutcome::Extracted)
fn provider_failure_fixtures_are_failed_noncommittable_and_never_create_graph_nodes() {
let cases = [
ProviderFailureKind::SpawnFailed,
ProviderFailureKind::Crashed,
ProviderFailureKind::TimedOut,
ProviderFailureKind::Corrupt,
ProviderFailureKind::Encrypted,
ProviderFailureKind::InvalidOutput,
ProviderFailureKind::OutputLimitExceeded,
for kind in cases {
let dir = honesty_temp_dir(&format!("provider-{kind:?}"));
.ingest_bundle_with(&file, providers, |_, _| ProviderExtractionResult::Failed {
detail: format!("fixture {kind:?}"),
assert_eq!(bundle.status, UniversalIngestStatus::Failed);
assert_eq!(bundle.summary().failed_count, 1);
Some(ProviderExtractionOutcome::Failed(kind))
fn per_file_read_error_is_failed_without_discarding_other_documents() {
let dir = honesty_temp_dir("per-file-read-error");
let first = dir.join("a.pdf");
let removed_before_read = dir.join("b.md");
std::fs::write(&first, b"%PDF fixture").unwrap();
std::fs::write(&removed_before_read, b"# removed before read").unwrap();
.ingest_bundle_with(&dir, providers, |_, _| {
std::fs::remove_file(&removed_before_read).unwrap();
ProviderExtractionResult::Extracted("# First\n\nParsed.".to_string())
let failed = bundle
.outcomes
.find(|outcome| outcome.status == UniversalIngestStatus::Failed)
assert_eq!(failed.provider, "universal:reader");
assert!(failed.reason.as_deref().unwrap().contains("failed to read"));
fn mixed_success_and_provider_failure_is_degraded_not_ingested() {
let dir = honesty_temp_dir("mixed-provider-failure");
std::fs::write(dir.join("good.md"), "# Good\n\nParsed.").unwrap();
std::fs::write(dir.join("failed.pdf"), b"%PDF fixture").unwrap();
.ingest_bundle_with(&dir, providers, |_, _| ProviderExtractionResult::Failed {
detail: "corrupt fixture".to_string(),
assert_eq!(summary.degraded_count, 0);
fn grobid_endpoint_policy_allows_loopback_and_explicit_https_only() {
let no_hosts = Vec::<String>::new();
assert!(validate_grobid_endpoint("http://127.0.0.1:8070", &no_hosts).is_ok());
assert!(validate_grobid_endpoint("http://[::1]:8070/base", &no_hosts).is_ok());
assert!(validate_grobid_endpoint("https://localhost:8070", &no_hosts).is_ok());
assert!(validate_grobid_endpoint("http://grobid.example", &no_hosts).is_err());
assert!(validate_grobid_endpoint("https://grobid.example", &no_hosts).is_err());
assert!(validate_grobid_endpoint(
"https://grobid.example/base",
&["grobid.example".to_string()]
.is_ok());
fn grobid_endpoint_policy_refuses_secret_bearing_or_ambiguous_urls() {
let hosts = vec!["grobid.example".to_string()];
for endpoint in [
"https://user:secret@grobid.example",
"https://grobid.example?token=secret",
"https://grobid.example/#secret",
"file:///tmp/grobid",
"javascript:alert(1)",
] {
validate_grobid_endpoint(endpoint, &hosts).is_err(),
"{endpoint}"
