use std::path::{Path, PathBuf};
use serde_json::Value;
use crate::event::{Envelope, Source};
use crate::ledger::RunPaths;
pub const MEMBER_SETTLED: &str = "member-settled";
pub const REPORT_PATH: &str = "report_path";
pub const MAX_REPORT_BYTES: u64 = 32 * 1024 * 1024;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Evidence {
pub node: Option<String>,
pub member: Option<String>,
pub named: PathBuf,
pub kept: PathBuf,
}
pub fn retain(paths: &RunPaths, event: &Envelope) {
if event.source != Source::Agentgraph || event.kind.0 != MEMBER_SETTLED {
return;
}
let Some(named) = event
.payload
.get(REPORT_PATH)
.and_then(Value::as_str)
.filter(|path| !path.is_empty())
else {
return;
};
let refuse = |why: &str| {
eprintln!("onepipeline: not retaining the report at '{named}': {why}");
};
if Path::new(named).file_name().and_then(|name| name.to_str())
!= Some(oneagentgraph::member::REPORT_FILE)
{
return refuse(&format!(
"a report the producing library wrote is named {}",
oneagentgraph::member::REPORT_FILE
));
}
if std::fs::symlink_metadata(named).is_ok_and(|about| about.file_type().is_symlink()) {
return refuse("it is a symlink, and a report is a file the producer wrote");
}
let source = match open_no_follow(Path::new(named)) {
Ok(source) => source,
Err(error) => return refuse(&format!("it cannot be opened as a plain file: {error}")),
};
match source.metadata() {
Err(error) => return refuse(&format!("it cannot be read: {error}")),
Ok(about) if !about.is_file() => return refuse("it is not a file"),
Ok(about) if about.len() > MAX_REPORT_BYTES => {
return refuse(&format!("it is larger than {MAX_REPORT_BYTES} bytes"))
}
Ok(_) => {}
}
let reports = paths.reports_dir();
if std::fs::symlink_metadata(&reports).is_ok_and(|about| about.file_type().is_symlink()) {
return refuse(&format!(
"{} is a symlink, and this run's own storage is a directory it owns",
reports.display()
));
}
if let Err(error) = std::fs::create_dir_all(&reports) {
return refuse(&format!("{} cannot be created: {error}", reports.display()));
}
let kept = paths.report_for(&event.stream, event.seq);
let written = match create_new_no_follow(&kept) {
Ok(destination) => destination,
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
if !std::fs::symlink_metadata(&kept).is_ok_and(|about| about.is_file()) {
refuse(&format!(
"{} already exists and is not a plain file, so nothing was written \
through it",
kept.display()
));
}
return;
}
Err(error) => {
return refuse(&format!(
"it cannot be copied to {}: {error}",
kept.display()
))
}
};
use std::io::Read;
let copied = std::io::copy(&mut source.take(MAX_REPORT_BYTES), &mut { written });
if let Err(error) = copied {
refuse(&format!(
"it cannot be copied to {}: {error}",
kept.display()
));
}
}
fn open_no_follow(path: &Path) -> std::io::Result<std::fs::File> {
let mut options = std::fs::OpenOptions::new();
options.read(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.custom_flags(libc::O_NOFOLLOW);
}
#[cfg(not(unix))]
if std::fs::symlink_metadata(path)?.file_type().is_symlink() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"the path is a symlink",
));
}
options.open(path)
}
fn create_new_no_follow(path: &Path) -> std::io::Result<std::fs::File> {
let mut options = std::fs::OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.custom_flags(libc::O_NOFOLLOW);
}
options.open(path)
}
pub fn evidence(paths: &RunPaths, events: &[Envelope]) -> Vec<Evidence> {
events
.iter()
.filter(|event| event.source == Source::Agentgraph && event.kind.0 == MEMBER_SETTLED)
.filter_map(|event| {
let named = event
.payload
.get(REPORT_PATH)
.and_then(Value::as_str)
.filter(|path| !path.is_empty())?;
Some(Evidence {
node: event.labels.node.clone(),
member: event
.labels
.extra
.get("member")
.and_then(Value::as_str)
.map(str::to_string),
named: PathBuf::from(named),
kept: paths.report_for(&event.stream, event.seq),
})
})
.collect()
}
pub fn read(kept: &Path) -> Option<Value> {
let refuse = |why: &str| {
eprintln!(
"onepipeline: not reading the retained report at {}: {why}",
kept.display()
);
};
let file = match open_no_follow(kept) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return None,
Err(error) => {
refuse(&format!("it is not a plain file this run wrote: {error}"));
return None;
}
};
match file.metadata() {
Err(error) => {
refuse(&format!("it cannot be read: {error}"));
return None;
}
Ok(about) if !about.is_file() => {
refuse("it is not a plain file this run wrote");
return None;
}
Ok(about) if about.len() > MAX_REPORT_BYTES => {
refuse(&format!("it is larger than {MAX_REPORT_BYTES} bytes"));
return None;
}
Ok(_) => {}
}
use std::io::Read;
let mut text = String::new();
file.take(MAX_REPORT_BYTES).read_to_string(&mut text).ok()?;
serde_json::from_str(&text).ok()
}
pub fn turns(document: &Value) -> Vec<Turn> {
if let Some(messages) = document
.get("transcript")
.and_then(|transcript| transcript.get("messages"))
.and_then(Value::as_array)
{
return messages.iter().map(Turn::of).collect();
}
results(document).filter_map(Turn::of_result).collect()
}
fn results(document: &Value) -> impl Iterator<Item = &Value> {
document
.get("results")
.and_then(Value::as_array)
.into_iter()
.flatten()
}
pub fn drafted_body(document: &Value) -> Option<String> {
results(document)
.filter(|result| result.get("schema_valid").and_then(Value::as_bool) == Some(true))
.find_map(|result| {
let body = result.get("structured")?.get("body")?.as_str()?.trim();
(!body.is_empty()).then(|| body.to_owned())
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Turn {
pub role: String,
pub text: String,
pub tools: Vec<Tool>,
}
impl Turn {
fn of(message: &Value) -> Self {
Self {
role: string(message, "role"),
text: string(message, "content"),
tools: Self::tools_of(message),
}
}
fn of_result(result: &Value) -> Option<Self> {
let turn = Self {
role: string(result, "harness"),
text: string(result, "text"),
tools: Self::tools_of(result),
};
let said_something = !turn.text.is_empty() || !turn.tools.is_empty();
(!turn.role.is_empty() && said_something).then_some(turn)
}
fn tools_of(value: &Value) -> Vec<Tool> {
value
.get("events")
.and_then(Value::as_array)
.map(|events| events.iter().map(Tool::of).collect())
.unwrap_or_default()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Tool {
pub kind: String,
pub name: String,
pub detail: String,
}
impl Tool {
fn of(event: &Value) -> Self {
Self {
kind: string(event, "kind"),
name: string(event, "name"),
detail: match event.get("input") {
None | Some(Value::Null) => String::new(),
Some(Value::String(text)) => text.clone(),
Some(input) => input.to_string(),
},
}
}
}
fn string(value: &Value, key: &str) -> String {
match value.get(key) {
Some(Value::String(text)) => text.clone(),
Some(Value::Null) | None => String::new(),
Some(other) => other.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::event::{EventKind, Labels, ENVELOPE_VERSION};
use serde_json::json;
fn scratch(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!(
"onepipeline-report-{name}-{}-{:?}",
crate::sys::pid(),
std::thread::current().id()
));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("a scratch root");
dir
}
fn produced(root: &Path, body: &str) -> PathBuf {
let dir = root.join("producer");
std::fs::create_dir_all(&dir).expect("a producer scratch");
let path = dir.join(oneagentgraph::member::REPORT_FILE);
std::fs::write(&path, body).expect("a stored report");
path
}
fn settled(node: Option<&str>, path: Option<&str>) -> Envelope {
let mut labels = Labels {
node: node.map(str::to_string),
..Labels::default()
};
labels.extra.insert("member".into(), "worker".into());
Envelope {
v: ENVELOPE_VERSION,
ts: "2026-08-08T00:00:00.000Z".into(),
stream: "oneagentgraph-1".into(),
seq: 4,
source: Source::Agentgraph,
kind: EventKind(MEMBER_SETTLED.into()),
labels,
payload: crate::journal::payload(&[(REPORT_PATH, json!(path))]),
artifacts: Vec::new(),
}
}
#[test]
fn ingest_keeps_the_run_its_own_copy_and_the_reader_opens_that() {
let root = scratch("kept");
let paths = RunPaths::under(&root, "demo");
paths.create().expect("the run directory");
let produced = produced(&root, r#"{"transcript":{"messages":[]}}"#);
let event = settled(Some("build"), Some(&produced.display().to_string()));
retain(&paths, &event);
let retained = evidence(&paths, &[event]);
assert_eq!(retained.len(), 1);
assert_eq!(retained[0].node.as_deref(), Some("build"));
assert_eq!(retained[0].member.as_deref(), Some("worker"));
assert_eq!(retained[0].named, produced);
assert_eq!(retained[0].kept, paths.report_for("oneagentgraph-1", 4));
assert!(
retained[0].kept.starts_with(paths.reports_dir()),
"the copy is not in the run's own storage: {:?}",
retained[0].kept
);
assert!(read(&retained[0].kept).is_some());
std::fs::remove_file(&produced).expect("the producer's copy is removed");
assert!(read(&retained[0].kept).is_some());
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_settlement_that_stored_none_is_not_listed_with_an_invented_path() {
let paths = RunPaths::under(Path::new("/nowhere"), "demo");
assert!(evidence(&paths, &[settled(Some("build"), None)]).is_empty());
assert!(evidence(&paths, &[settled(Some("build"), Some(""))]).is_empty());
}
#[test]
fn ingest_refuses_anything_that_is_not_the_producers_own_plain_file() {
let root = scratch("refused");
let paths = RunPaths::under(&root, "demo");
paths.create().expect("the run directory");
let secret = root.join("secret.json");
std::fs::write(&secret, r#"{"transcript":{"messages":[]}}"#).expect("a secret");
let planted = root.join("planted");
std::fs::create_dir_all(&planted).expect("a planted directory");
let link = planted.join(oneagentgraph::member::REPORT_FILE);
#[cfg(unix)]
std::os::unix::fs::symlink(&secret, &link).expect("a symlink");
#[cfg(windows)]
std::os::windows::fs::symlink_file(&secret, &link).expect("a symlink");
for named in [
link.display().to_string(),
secret.display().to_string(),
root.join("gone")
.join(oneagentgraph::member::REPORT_FILE)
.display()
.to_string(),
planted.display().to_string(),
] {
let event = settled(Some("build"), Some(&named));
retain(&paths, &event);
let kept = &evidence(&paths, &[event])[0].kept;
assert!(
read(kept).is_none(),
"'{named}' was copied into the run's storage"
);
}
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn ingest_refuses_a_report_past_its_bound() {
let root = scratch("oversize");
let paths = RunPaths::under(&root, "demo");
paths.create().expect("the run directory");
let produced = produced(&root, "x");
let file = std::fs::OpenOptions::new()
.write(true)
.open(&produced)
.expect("the stored report");
file.set_len(MAX_REPORT_BYTES + 1).expect("a large report");
drop(file);
let event = settled(Some("build"), Some(&produced.display().to_string()));
retain(&paths, &event);
assert!(read(&evidence(&paths, &[event])[0].kept).is_none());
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_pipeline_event_of_the_same_shape_is_not_a_members_report() {
let root = scratch("ours");
let paths = RunPaths::under(&root, "demo");
paths.create().expect("the run directory");
let produced = produced(&root, "{}");
let mut ours = settled(Some("build"), Some(&produced.display().to_string()));
ours.source = Source::Pipeline;
retain(&paths, &ours);
assert!(evidence(&paths, &[ours]).is_empty());
assert!(
!paths.reports_dir().exists(),
"this crate's own event was ingested as a sibling's report"
);
std::fs::remove_dir_all(&root).ok();
}
#[test]
fn a_transcripts_turns_carry_their_text_and_their_tools() {
let document = json!({
"transcript": {"messages": [
{"role": "user", "content": "## What\nship it"},
{"role": "assistant", "content": "Ran the gate.", "events": [
{"kind": "tool_call", "name": "bash",
"input": {"command": "just check"}, "index": 0},
{"kind": "tool_result", "output": "ok", "index": 1},
]},
]},
});
let turns = turns(&document);
assert_eq!(turns.len(), 2);
assert_eq!(turns[0].role, "user");
assert!(turns[0].tools.is_empty());
assert_eq!(turns[1].text, "Ran the gate.");
assert_eq!(turns[1].tools[0].name, "bash");
assert!(turns[1].tools[0].detail.contains("just check"));
assert_eq!(turns[1].tools[1].kind, "tool_result");
assert!(turns[1].tools[1].name.is_empty());
}
#[test]
fn a_single_sided_members_report_reads_as_the_turns_its_chain_took() {
let document = json!({
"schema_version": "0.6",
"results": [
{"harness": "codex", "status": "skipped", "text": null},
{"harness": "claude-code", "status": "ok", "text": "Ran the gate.",
"events": [
{"kind": "tool_call", "name": "bash",
"input": {"command": "just check"}, "index": 0},
]},
],
});
let turns = turns(&document);
assert_eq!(turns.len(), 1, "{turns:?}");
assert_eq!(turns[0].role, "claude-code");
assert_eq!(turns[0].text, "Ran the gate.");
assert_eq!(turns[0].tools[0].name, "bash");
assert!(turns[0].tools[0].detail.contains("just check"));
}
#[test]
fn a_report_carrying_no_transcript_has_no_turns_rather_than_a_refusal() {
assert!(turns(&json!({"usage": {"input_tokens": 1}})).is_empty());
assert!(turns(&json!({"transcript": {}})).is_empty());
assert!(turns(&Value::Null).is_empty());
assert!(turns(&json!({"results": [{"harness": "codex", "status": "skipped"}]})).is_empty());
assert!(turns(&json!({"results": [{"text": "done"}]})).is_empty());
}
#[test]
fn a_drafted_body_is_taken_only_from_an_answer_the_schema_accepted() {
let result = |valid: Value, structured: Value| json!({"results": [{"schema_valid": valid, "structured": structured}]});
assert_eq!(
drafted_body(&result(json!(true), json!({"body": "## What\nit landed"}))).as_deref(),
Some("## What\nit landed")
);
assert!(drafted_body(&result(json!(false), json!({"body": "half a "}))).is_none());
assert!(drafted_body(&result(Value::Null, Value::Null)).is_none());
assert!(drafted_body(&result(json!(true), json!({"body": " "}))).is_none());
assert!(drafted_body(&result(json!(true), json!({"title": "feat: x"}))).is_none());
assert!(drafted_body(&json!({"transcript": {"messages": []}})).is_none());
}
#[test]
fn a_report_that_is_not_there_to_read_is_absent() {
assert!(read(Path::new("/nowhere/onepipeline/report.json")).is_none());
}
}