use super::*;
#[cfg(unix)]
pub(super) fn checked_source_length(length: u64) -> anyhow::Result<u64> {
if length > SESSION_EXPORT_MAX_SOURCE_BYTES {
anyhow::bail!(
"session export source-byte limit exceeded: {length} bytes (maximum {SESSION_EXPORT_MAX_SOURCE_BYTES})"
)
}
Ok(length)
}
#[cfg(unix)]
fn hash_file(file: &File, length: u64, cancellation: &AgentCancellation) -> anyhow::Result<String> {
let mut source = file.try_clone()?;
source.seek(SeekFrom::Start(0))?;
let mut source = source.take(length);
let mut hasher = Sha256::new();
let mut copied = 0_u64;
let mut buffer = [0_u8; 64 * 1024];
loop {
cancellation.check()?;
let read = source.read(&mut buffer)?;
if read == 0 {
break;
}
hasher.update(&buffer[..read]);
copied = copied
.checked_add(read as u64)
.ok_or_else(|| anyhow::anyhow!("session export hash length overflowed"))?;
}
cancellation.check()?;
if copied != length || file.metadata()?.len() != length {
anyhow::bail!("session export source changed while hashing")
}
Ok(lower_hex(hasher.finalize()))
}
#[cfg(unix)]
pub(super) fn add_member(
manifest: &mut ExportManifest,
archive_name: String,
file: File,
cancellation: &AgentCancellation,
) -> anyhow::Result<()> {
if manifest.members.len() >= SESSION_EXPORT_MAX_MEMBERS {
anyhow::bail!(
"session export member limit exceeded: maximum {SESSION_EXPORT_MAX_MEMBERS} members"
)
}
if manifest
.members
.iter()
.any(|member| member.archive_name == archive_name)
{
anyhow::bail!("session export contains duplicate archive member")
}
let length = checked_source_length(file.metadata()?.len())?;
let sha256 = hash_file(&file, length, cancellation)?;
let source_bytes = manifest
.source_bytes
.checked_add(length)
.ok_or_else(|| anyhow::anyhow!("session export source-byte count overflowed"))?;
if source_bytes > SESSION_EXPORT_MAX_SOURCE_BYTES {
anyhow::bail!(
"session export source-byte limit exceeded: maximum {SESSION_EXPORT_MAX_SOURCE_BYTES} bytes"
)
}
manifest.source_bytes = source_bytes;
manifest.members.push(SourceMember {
archive_name,
file,
length,
sha256,
});
Ok(())
}
#[cfg(unix)]
pub(super) fn add_optional_member(
manifest: &mut ExportManifest,
root: &Path,
source_name: &str,
archive_name: String,
fatal: bool,
cancellation: &AgentCancellation,
) -> anyhow::Result<()> {
match open_existing_named(root, source_name) {
Ok(Some(file)) => {
if let Err(error) = add_member(manifest, archive_name, file, cancellation) {
if is_export_limit_error(&error) || fatal {
return Err(error);
}
manifest.warnings.push(format!(
"skipped unreadable session export member {source_name}"
));
}
}
Ok(None) => {}
Err(_) if fatal => {
anyhow::bail!("session export member {source_name} is unsafe or unreadable")
}
Err(_) => manifest.warnings.push(format!(
"skipped unsafe or unreadable session export member {source_name}"
)),
}
Ok(())
}
#[cfg(unix)]
fn is_export_limit_error(error: &anyhow::Error) -> bool {
let message = error.to_string();
message.contains("session export member limit")
|| message.contains("session export source-byte")
}
#[cfg(unix)]
fn optional_child_failure(
manifest: &mut ExportManifest,
child_id: &str,
cancellation: &AgentCancellation,
error: anyhow::Error,
) -> anyhow::Result<Option<BTreeSet<String>>> {
if cancellation.is_canceled() || is_export_limit_error(&error) {
return Err(error);
}
manifest.warnings.push(format!(
"skipped child session {child_id}: unsafe, locked, or unreadable"
));
Ok(None)
}
#[cfg(unix)]
pub(super) fn collect_child(
manifest: &mut ExportManifest,
child_root: &Path,
sessions_root: &Path,
child_id: &str,
cancellation: &AgentCancellation,
) -> anyhow::Result<Option<BTreeSet<String>>> {
if validate_session_root(child_root).is_err() {
manifest.warnings.push(format!(
"child session {child_id} root is missing or unsafe"
));
return Ok(None);
}
let child_path = child_root.join(format!("{child_id}.jsonl"));
match fs::symlink_metadata(&child_path) {
Ok(metadata) if metadata.file_type().is_file() => {}
Ok(_) => {
manifest
.warnings
.push(format!("child session {child_id} is unsafe"));
return Ok(None);
}
Err(error) if error.kind() == io::ErrorKind::NotFound => {
manifest
.warnings
.push(format!("child session {child_id} is missing"));
return Ok(None);
}
Err(_) => {
manifest
.warnings
.push(format!("child session {child_id} is unreadable"));
return Ok(None);
}
}
let child = match (|| -> anyhow::Result<Option<(File, u64, BTreeSet<String>)>> {
manifest.lock_session(&child_path, cancellation)?;
let Some(child) = open_existing_named(child_root, &format!("{child_id}.jsonl"))? else {
return Ok(None);
};
let child_length = checked_source_length(child.metadata()?.len())?;
let child_children = discover_children(
&child,
child_length,
sessions_root,
child_id,
manifest,
cancellation,
)?;
Ok(Some((child, child_length, child_children)))
})() {
Ok(Some(value)) => value,
Ok(None) => {
manifest
.warnings
.push(format!("child session {child_id} is missing"));
return Ok(None);
}
Err(error) => return optional_child_failure(manifest, child_id, cancellation, error),
};
let (child, _child_length, child_children) = child;
if let Err(error) = add_member(
manifest,
format!("subagents/{child_id}.jsonl"),
child,
cancellation,
) {
return optional_child_failure(manifest, child_id, cancellation, error);
}
add_optional_member(
manifest,
child_root,
&format!("{child_id}.metadata.json"),
format!("subagents/{child_id}.metadata.json"),
false,
cancellation,
)?;
let history_children = collect_history(
manifest,
child_root,
child_id,
"subagents/.history",
false,
cancellation,
)?;
Ok(Some(
child_children.into_iter().chain(history_children).collect(),
))
}
#[cfg(unix)]
pub(super) fn collect_history(
manifest: &mut ExportManifest,
root: &Path,
session_id: &str,
archive_prefix: &str,
fatal: bool,
cancellation: &AgentCancellation,
) -> anyhow::Result<BTreeSet<String>> {
let mut discovered = BTreeSet::new();
let history_parent = root.join(".history");
let history_parent_exists = match fs::symlink_metadata(&history_parent) {
Ok(_) => true,
Err(error) if error.kind() == io::ErrorKind::NotFound => false,
Err(_) => {
return history_failure(
fatal,
&mut manifest.warnings,
format!("session {session_id} history is unreadable"),
);
}
};
if !history_parent_exists {
return Ok(discovered);
}
if validate_session_root(&history_parent).is_err() {
return history_failure(
fatal,
&mut manifest.warnings,
format!("session {session_id} history is unsafe"),
);
}
let history_root = history_parent.join(session_id);
match fs::symlink_metadata(&history_root) {
Ok(_) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(discovered),
Err(_) => {
return history_failure(
fatal,
&mut manifest.warnings,
format!("session {session_id} history is unreadable"),
);
}
}
let mut entries = Vec::new();
let read_dir = match fs::read_dir(&history_root) {
Ok(entries) => entries,
Err(_error) => {
return history_failure(
fatal,
&mut manifest.warnings,
format!("session {session_id} history is unreadable"),
);
}
};
for entry in read_dir {
if entries.len() >= SESSION_EXPORT_MAX_MEMBERS {
anyhow::bail!(
"session export member limit exceeded: maximum {SESSION_EXPORT_MAX_MEMBERS} members"
);
}
entries.push(entry?);
}
entries.sort_by_key(|entry| entry.file_name());
cancellation.check()?;
for entry in entries {
let Some(name) = entry.file_name().to_str().map(str::to_string) else {
continue;
};
let Some(generation) = name
.strip_suffix(".jsonl")
.and_then(|value| value.parse::<u64>().ok())
else {
continue;
};
let file = match open_existing_named(&history_root, &name) {
Ok(Some(file)) => file,
Ok(None) => continue,
Err(_error) if fatal => {
anyhow::bail!("session export history generation {generation} is unsafe")
}
Err(_error) => {
manifest.warnings.push(format!(
"skipped unsafe session {session_id} history generation {generation}"
));
continue;
}
};
let length = checked_source_length(file.metadata()?.len())?;
cancellation.check()?;
let children = discover_children(
&file,
length,
if archive_prefix == "subagents/.history" {
root.parent().unwrap_or(root)
} else {
root
},
session_id,
manifest,
cancellation,
)?;
discovered.extend(children);
let archive_name = format!("{archive_prefix}/{session_id}/{name}");
if let Err(error) = add_member(manifest, archive_name, file, cancellation) {
if is_export_limit_error(&error) || fatal {
return Err(error);
}
manifest.warnings.push(format!(
"skipped unreadable session {session_id} history generation {generation}"
));
}
}
Ok(discovered)
}
#[cfg(unix)]
fn history_failure(
fatal: bool,
warnings: &mut WarningCollector,
message: String,
) -> anyhow::Result<BTreeSet<String>> {
if fatal {
anyhow::bail!(message)
}
warnings.push(message);
Ok(BTreeSet::new())
}
#[cfg(unix)]
pub(super) fn discover_children(
file: &File,
length: u64,
sessions_root: &Path,
parent_session_id: &str,
manifest: &mut ExportManifest,
cancellation: &AgentCancellation,
) -> anyhow::Result<BTreeSet<String>> {
let mut scan_file = file.try_clone()?;
scan_file.seek(SeekFrom::Start(0))?;
let mut reader = BufReader::new(scan_file.take(length));
let mut children = BTreeSet::new();
let mut total_read = 0u64;
fn read_bounded_line<R: BufRead>(reader: &mut R) -> io::Result<(Option<Vec<u8>>, usize)> {
let mut line = Vec::new();
let mut total = 0usize;
let mut oversized = false;
loop {
let buffer = reader.fill_buf()?;
if buffer.is_empty() {
return Ok((
if total == 0 || oversized {
None
} else {
Some(line)
},
total,
));
}
let newline = buffer.iter().position(|byte| *byte == b'\n');
let consumed = newline.map_or(buffer.len(), |index| index + 1);
if !oversized {
if line.len().saturating_add(consumed) <= SESSION_EXPORT_MAX_DISCOVERY_LINE_BYTES {
line.extend_from_slice(&buffer[..consumed]);
} else {
oversized = true;
line.clear();
}
}
reader.consume(consumed);
total = total.saturating_add(consumed);
if newline.is_some() {
return Ok((if oversized { None } else { Some(line) }, total));
}
}
}
loop {
cancellation.check()?;
let (line, bytes_read) = read_bounded_line(&mut reader)?;
if bytes_read == 0 {
break;
}
total_read = total_read.saturating_add(bytes_read as u64);
if let Some(line) = line
&& let Ok(value) = serde_json::from_slice::<Value>(&line)
{
extract_child_references(
&value,
sessions_root,
parent_session_id,
&mut children,
&mut manifest.relationships,
&mut manifest.warnings,
&mut manifest.graph_nodes,
);
}
}
if total_read != length {
anyhow::bail!("session JSONL changed while preparing export")
}
Ok(children)
}
#[cfg(unix)]
fn extract_child_references(
event: &Value,
sessions_root: &Path,
parent_session_id: &str,
children: &mut BTreeSet<String>,
relationships: &mut BTreeSet<ExportRelationship>,
warnings: &mut WarningCollector,
graph_nodes: &mut usize,
) {
if !matches!(
event.get("event_type").and_then(Value::as_str),
Some("tool_result" | "code_mode_tool_result")
) {
return;
}
let Some(result) = event.get("payload").and_then(|p| p.get("result")) else {
return;
};
if result.get("tool_name").and_then(Value::as_str) != Some("subagents") {
return;
}
let Some(content) = result.get("content").and_then(Value::as_str) else {
return;
};
let Ok(output) = serde_json::from_str::<SubagentsOutput>(content) else {
warnings.push("skipped malformed child session reference");
return;
};
for result in output.results {
let (Some(id), Some(path)) = (result.session_id.as_deref(), result.session_path.as_deref())
else {
warnings.push("skipped child session reference without session_id or session_path");
continue;
};
let Ok(session_id) = validate_session_id(id.to_string()) else {
warnings.push("skipped child session reference with invalid session_id");
continue;
};
let expected = sessions_root
.join("subagents")
.join(format!("{session_id}.jsonl"));
if path != expected {
warnings.push("skipped child session reference with unexpected session_path");
continue;
}
*graph_nodes = graph_nodes.saturating_add(1);
if *graph_nodes > SESSION_EXPORT_MAX_MEMBERS {
continue;
}
relationships.insert(ExportRelationship {
relationship_type: "child_session",
parent_session_id: parent_session_id.to_string(),
child_session_id: session_id.clone(),
});
children.insert(session_id);
}
}