use crate::error::{BundleError, Result};
use crate::event::{FileChange, RawEventRow, SessionRow};
use crate::redact::Detectors;
use crate::store::Store;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::io::Read;
pub const FORMAT_VERSION: u32 = 1;
const ZSTD_LEVEL: i32 = 3;
const MAX_BUNDLE_BYTES: u64 = 2 * 1024 * 1024 * 1024;
#[derive(Debug, Clone)]
pub struct Bundle {
pub hh_version: String,
pub session: serde_json::Value,
pub events: Vec<serde_json::Value>,
pub blobs: BTreeMap<String, Vec<u8>>,
}
pub fn export(
store: &Store,
session_id: &str,
hh_version: &str,
detectors: Option<&Detectors>,
) -> Result<Vec<u8>> {
let session_row = store.get_session(session_id)?;
let mut raw_events: Vec<RawEventRow> = Vec::new();
store.for_each_event_raw(session_id, |row| {
raw_events.push(row);
Ok(())
})?;
let id_to_seq: HashMap<i64, u64> = raw_events
.iter()
.enumerate()
.map(|(i, r)| (r.id, u64::try_from(i).unwrap_or(u64::MAX)))
.collect();
let mut referenced: BTreeSet<String> = BTreeSet::new();
for r in &raw_events {
if let Some(h) = &r.blob_hash {
referenced.insert(h.clone());
}
if let Some(fc) = &r.file_change {
if let Some(h) = &fc.before_hash {
referenced.insert(h.clone());
}
if let Some(h) = &fc.after_hash {
referenced.insert(h.clone());
}
}
}
let mut blobs: BTreeMap<String, Vec<u8>> = BTreeMap::new();
let mut remap: HashMap<String, String> = HashMap::new();
for hash in &referenced {
let content = store.blobs().get(hash)?;
match detectors.and_then(|d| d.redact_bytes(&content)) {
Some(redacted) => {
let new_hash = crate::blob::BlobStore::hash(&redacted.bytes);
if &new_hash != hash {
remap.insert(hash.clone(), new_hash.clone());
}
blobs.insert(new_hash, redacted.bytes);
}
None => {
blobs.insert(hash.clone(), content);
}
}
}
let mut events_json: Vec<serde_json::Value> = Vec::with_capacity(raw_events.len());
for (seq, r) in raw_events.iter().enumerate() {
let correlates_seq = r.correlates.and_then(|cid| id_to_seq.get(&cid).copied());
let file_change_json = r.file_change.as_ref().map(file_change_to_json);
let mut ev = serde_json::json!({
"seq": u64::try_from(seq).unwrap_or(u64::MAX),
"ts_ms": r.ts_ms,
"kind": r.kind.to_string(),
"step": r.step,
"correlates_seq": correlates_seq,
"summary": r.summary,
"body": r.body_json,
"blob_hash": r.blob_hash,
"blob_size": r.blob_size,
"file_change": file_change_json,
});
if let Some(d) = detectors {
let _ = d.redact_json(&mut ev);
}
if !remap.is_empty() {
remap_hashes(&mut ev, &remap);
}
events_json.push(ev);
}
let mut session_json = session_to_bundle_json(&session_row);
if let Some(d) = detectors {
let _ = d.redact_json(&mut session_json);
}
let mut ndjson = String::new();
for ev in &events_json {
let line = serde_json::to_string(ev).map_err(|e| BundleError::Events(e.to_string()))?;
ndjson.push_str(&line);
ndjson.push('\n');
}
let events_bytes = ndjson.into_bytes();
let events_blake3 = blake3::hash(&events_bytes).to_hex().to_string();
let blob_list: Vec<serde_json::Value> = blobs
.iter()
.map(|(hash, content)| serde_json::json!({ "hash": hash, "size": content.len() }))
.collect();
let manifest = serde_json::json!({
"format_version": FORMAT_VERSION,
"hh_version": hh_version,
"session": session_json,
"integrity": {
"events_blake3": events_blake3,
"event_count": events_json.len(),
"blobs": blob_list,
},
});
let manifest_bytes =
serde_json::to_vec(&manifest).map_err(|e| BundleError::Manifest(e.to_string()))?;
let tar_bytes = build_tar(&manifest_bytes, &events_bytes, &blobs)?;
let compressed = zstd::stream::encode_all(&tar_bytes[..], ZSTD_LEVEL)
.map_err(|e| BundleError::Zstd(e.to_string()))?;
Ok(compressed)
}
pub fn parse(bytes: &[u8]) -> Result<Bundle> {
let tar_bytes = bounded_decompress(bytes)?;
let (manifest_bytes, events_bytes, blobs) = extract_tar_entries(&tar_bytes)?;
let manifest_bytes =
manifest_bytes.ok_or_else(|| BundleError::Manifest("missing manifest.json".into()))?;
let manifest: serde_json::Value = serde_json::from_slice(&manifest_bytes)
.map_err(|e| BundleError::Manifest(e.to_string()))?;
let (hh_version, session, expected_digest) = parse_manifest(&manifest)?;
let events_bytes =
events_bytes.ok_or_else(|| BundleError::Events("missing events.ndjson".into()))?;
let actual_digest = blake3::hash(&events_bytes).to_hex().to_string();
if expected_digest != actual_digest {
return Err(BundleError::IntegrityMismatch.into());
}
let events = parse_events(&events_bytes, &blobs)?;
Ok(Bundle {
hh_version,
session,
events,
blobs,
})
}
#[allow(clippy::type_complexity)] fn extract_tar_entries(
tar_bytes: &[u8],
) -> Result<(Option<Vec<u8>>, Option<Vec<u8>>, BTreeMap<String, Vec<u8>>)> {
let mut archive = tar::Archive::new(tar_bytes);
let mut manifest_bytes: Option<Vec<u8>> = None;
let mut events_bytes: Option<Vec<u8>> = None;
let mut blobs: BTreeMap<String, Vec<u8>> = BTreeMap::new();
let entries = archive
.entries()
.map_err(|e| BundleError::Tar(e.to_string()))?;
for entry in entries {
let mut entry = entry.map_err(|e| BundleError::Tar(e.to_string()))?;
if entry.header().entry_type().ne(&tar::EntryType::Regular) {
return Err(BundleError::Tar(
"bundle contains a non-regular-file entry (symlink/hardlink/directory) — refusing"
.into(),
)
.into());
}
let path = entry.path().map_err(|e| BundleError::Tar(e.to_string()))?;
let path_str = path.to_string_lossy().into_owned();
let mut data = Vec::new();
entry
.read_to_end(&mut data)
.map_err(|e| BundleError::Tar(e.to_string()))?;
match classify_entry(&path_str) {
EntryKind::Manifest => manifest_bytes = Some(data),
EntryKind::Events => events_bytes = Some(data),
EntryKind::Blob(hash) => {
let actual = blake3::hash(&data).to_hex().to_string();
if actual != hash {
return Err(BundleError::HashMismatch {
expected: hash,
actual,
}
.into());
}
blobs.insert(hash, data);
}
EntryKind::Unknown => {
return Err(
BundleError::Tar(format!("unexpected entry in bundle: {path_str}")).into(),
);
}
}
}
Ok((manifest_bytes, events_bytes, blobs))
}
fn parse_manifest(manifest: &serde_json::Value) -> Result<(String, serde_json::Value, String)> {
let format_version = manifest
.get("format_version")
.and_then(serde_json::Value::as_u64)
.ok_or_else(|| BundleError::Manifest("missing or non-numeric format_version".into()))?;
let format_version = u32::try_from(format_version)
.map_err(|_| BundleError::Manifest("format_version out of range".into()))?;
if format_version > FORMAT_VERSION {
return Err(BundleError::UnsupportedVersion {
found: format_version,
max: FORMAT_VERSION,
}
.into());
}
let hh_version = manifest
.get("hh_version")
.and_then(serde_json::Value::as_str)
.unwrap_or("unknown")
.to_string();
let session = manifest
.get("session")
.cloned()
.ok_or_else(|| BundleError::Manifest("missing session block".into()))?;
let expected_digest = manifest
.get("integrity")
.and_then(|i| i.get("events_blake3"))
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.to_string();
Ok((hh_version, session, expected_digest))
}
fn parse_events(
events_bytes: &[u8],
blobs: &BTreeMap<String, Vec<u8>>,
) -> Result<Vec<serde_json::Value>> {
let mut events = Vec::new();
for line in events_bytes.split(|&b| b == b'\n') {
if line.is_empty() {
continue;
}
let v: serde_json::Value =
serde_json::from_slice(line).map_err(|e| BundleError::Events(e.to_string()))?;
events.push(v);
}
for ev in &events {
if let Some(h) = ev.get("blob_hash").and_then(serde_json::Value::as_str) {
if !blobs.contains_key(h) {
return Err(BundleError::MissingBlob(h.to_string()).into());
}
}
if let Some(fc) = ev.get("file_change").filter(|v| !v.is_null()) {
for key in ["before_hash", "after_hash"] {
if let Some(h) = fc.get(key).and_then(serde_json::Value::as_str) {
if !blobs.contains_key(h) {
return Err(BundleError::MissingBlob(h.to_string()).into());
}
}
}
}
}
Ok(events)
}
fn bounded_decompress(bytes: &[u8]) -> Result<Vec<u8>> {
let decoder =
zstd::stream::Decoder::new(bytes).map_err(|e| BundleError::Zstd(e.to_string()))?;
let mut limited = decoder.take(MAX_BUNDLE_BYTES + 1);
let mut out = Vec::new();
limited
.read_to_end(&mut out)
.map_err(|e| BundleError::Zstd(e.to_string()))?;
if out.len() as u64 > MAX_BUNDLE_BYTES {
return Err(BundleError::Tar(format!(
"decompressed bundle exceeds the {MAX_BUNDLE_BYTES}-byte safety limit"
))
.into());
}
Ok(out)
}
enum EntryKind {
Manifest,
Events,
Blob(String),
Unknown,
}
fn classify_entry(path: &str) -> EntryKind {
if path == "manifest.json" {
return EntryKind::Manifest;
}
if path == "events.ndjson" {
return EntryKind::Events;
}
if let Some(rest) = path.strip_prefix("blobs/") {
if let Some((prefix, hash)) = rest.split_once('/') {
if prefix.len() == 2
&& hash.len() == 64
&& hash.starts_with(prefix)
&& hash.bytes().all(|b| b.is_ascii_hexdigit())
{
return EntryKind::Blob(hash.to_string());
}
}
}
EntryKind::Unknown
}
fn append_entry(builder: &mut tar::Builder<Vec<u8>>, path: &str, data: &[u8]) -> Result<()> {
let mut header = tar::Header::new_gnu();
header
.set_path(path)
.map_err(|e| BundleError::Tar(e.to_string()))?;
header.set_size(data.len() as u64);
header.set_mtime(0);
header.set_uid(0);
header.set_gid(0);
header.set_mode(0o644);
header.set_cksum();
builder
.append_data(&mut header, path, data)
.map_err(|e| BundleError::Tar(e.to_string()))?;
Ok(())
}
fn build_tar(manifest: &[u8], events: &[u8], blobs: &BTreeMap<String, Vec<u8>>) -> Result<Vec<u8>> {
let mut builder = tar::Builder::new(Vec::new());
append_entry(&mut builder, "manifest.json", manifest)?;
append_entry(&mut builder, "events.ndjson", events)?;
for (hash, content) in blobs {
let path = format!("blobs/{}/{}", &hash[..2], hash);
append_entry(&mut builder, &path, content)?;
}
builder
.into_inner()
.map_err(|e| BundleError::Tar(e.to_string()).into())
}
fn session_to_bundle_json(row: &SessionRow) -> serde_json::Value {
let duration_ms = row.ended_at.map(|end| (end - row.started_at).max(0));
serde_json::json!({
"schema": crate::JSON_SCHEMA_VERSION,
"id": row.id,
"short_id": row.short_id,
"status": row.status.to_string(),
"agent_kind": row.agent_kind.to_string(),
"adapter_status": row.adapter_status.to_string(),
"started_at": row.started_at,
"ended_at": row.ended_at,
"exit_code": row.exit_code,
"duration_ms": duration_ms,
"steps": row.step_count,
"files_changed": row.files_changed,
"command": row.command,
"cwd": row.cwd.to_string_lossy(),
"imported_from": row.imported_from,
})
}
fn file_change_to_json(fc: &FileChange) -> serde_json::Value {
serde_json::json!({
"path": fc.path,
"change_kind": fc.change_kind.to_string(),
"before_hash": fc.before_hash,
"after_hash": fc.after_hash,
"is_binary": fc.is_binary,
})
}
fn remap_hashes(value: &mut serde_json::Value, remap: &HashMap<String, String>) {
match value {
serde_json::Value::String(s) => {
if let Some(new_hash) = remap.get(s.as_str()) {
*s = new_hash.clone();
}
}
serde_json::Value::Array(items) => {
for v in items {
remap_hashes(v, remap);
}
}
serde_json::Value::Object(map) => {
for v in map.values_mut() {
remap_hashes(v, remap);
}
}
_ => {}
}
}
#[cfg(feature = "fuzzing")]
pub mod fuzzing {
pub fn fuzz_parse(bytes: &[u8]) {
let _ = super::parse(bytes);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::RedactionConfig;
use crate::event::{
AdapterStatus, AgentKind, ChangeKind, Event, EventKind, NewSession, SessionStatus,
};
use std::path::PathBuf;
use tempfile::TempDir;
fn new_session() -> NewSession {
NewSession {
id: crate::event::now_v7(),
started_at: 1_700_000_000_000,
agent_kind: AgentKind::Generic,
adapter_status: AdapterStatus::None,
command: vec!["agent".into()],
cwd: PathBuf::from("/tmp/proj"),
hostname: Some("devbox".into()),
hh_version: "test".into(),
model: None,
git_branch: None,
git_sha: None,
git_dirty: None,
}
}
fn fixture() -> (TempDir, Store, String) {
let tmp = TempDir::new().unwrap();
let store = Store::open(&tmp.path().join("hh.db"), &tmp.path().join("blobs")).unwrap();
let created = store.create_session(&new_session()).unwrap();
let sid = created.id.clone();
let writer = store.event_writer().unwrap();
writer
.append_event(Event {
session_id: sid.clone(),
ts_ms: 0,
kind: EventKind::UserMessage,
step: None,
summary: "hello".into(),
body_json: Some(serde_json::json!({"text": "hello there"})),
blob_hash: None,
blob_size: None,
correlates: None,
})
.unwrap();
let call = writer
.append_event(Event {
session_id: sid.clone(),
ts_ms: 1_000,
kind: EventKind::ToolCall,
step: None,
summary: "tool_call: Bash".into(),
body_json: Some(serde_json::json!({"name": "Bash", "input": {"command": "ls"}})),
blob_hash: None,
blob_size: None,
correlates: None,
})
.unwrap();
writer
.append_event(Event {
session_id: sid.clone(),
ts_ms: 1_500,
kind: EventKind::ToolResult,
step: None,
summary: "tool_result: ok".into(),
body_json: Some(serde_json::json!({"content": "file.txt", "is_error": false})),
blob_hash: None,
blob_size: None,
correlates: Some(call),
})
.unwrap();
let after = store.blobs().put(b"created content\n").unwrap();
writer
.append_file_change(
Event {
session_id: sid.clone(),
ts_ms: 2_000,
kind: EventKind::FileChange,
step: None,
summary: "created.txt created".into(),
body_json: Some(serde_json::json!({"path": "created.txt"})),
blob_hash: Some(after.hash.clone()),
blob_size: Some(after.size),
correlates: None,
},
FileChange {
event_id: 0,
path: "created.txt".into(),
change_kind: ChangeKind::Created,
before_hash: None,
after_hash: Some(after.hash.clone()),
is_binary: false,
},
)
.unwrap();
writer.finish().unwrap();
store.assign_steps(&sid).unwrap();
store
.finalize_session(&sid, 1_700_000_003_000, Some(0), SessionStatus::Ok)
.unwrap();
(tmp, store, sid)
}
#[test]
fn export_then_parse_round_trips_the_shape() {
let (_tmp, store, sid) = fixture();
let bytes = export(&store, &sid, "test-version", None).unwrap();
let bundle = parse(&bytes).unwrap();
assert_eq!(bundle.hh_version, "test-version");
assert_eq!(
bundle.session["short_id"],
store.get_session(&sid).unwrap().short_id
);
assert_eq!(bundle.events.len(), 4);
assert_eq!(bundle.events[2]["kind"], "tool_result");
assert_eq!(bundle.events[2]["correlates_seq"], 1);
assert_eq!(bundle.events[1]["correlates_seq"], serde_json::Value::Null);
let after_hash = bundle.events[3]["file_change"]["after_hash"]
.as_str()
.unwrap();
assert_eq!(bundle.blobs[after_hash], b"created content\n");
}
#[test]
fn export_is_deterministic() {
let (_tmp, store, sid) = fixture();
let a = export(&store, &sid, "v", None).unwrap();
let b = export(&store, &sid, "v", None).unwrap();
assert_eq!(
a, b,
"identical session must produce byte-identical bundles"
);
}
#[test]
fn redaction_scrubs_session_events_and_blob_and_remaps_the_hash() {
let (_tmp, store, sid) = fixture();
let writer = store.event_writer().unwrap();
writer
.append_event(Event {
session_id: sid.clone(),
ts_ms: 3_000,
kind: EventKind::AgentMessage,
step: None,
summary: "leaking a key".into(),
body_json: Some(serde_json::json!({"text": "key AKIAIOSFODNN7EXAMPLE end"})),
blob_hash: None,
blob_size: None,
correlates: None,
})
.unwrap();
writer.finish().unwrap();
store.assign_steps(&sid).unwrap();
let secret_blob = store.blobs().put(b"secret AKIAIOSFODNN7EXAMPLE\n").unwrap();
writer_append_file_change_with_secret(&store, &sid, &secret_blob.hash);
let detectors = Detectors::new(&RedactionConfig::default()).unwrap();
let bytes = export(&store, &sid, "v", Some(&detectors)).unwrap();
let bundle = parse(&bytes).unwrap();
let joined = serde_json::to_string(&bundle.events).unwrap();
assert!(
!joined.contains("AKIAIOSFODNN7EXAMPLE"),
"secret must not appear in events: {joined}"
);
for content in bundle.blobs.values() {
let text = String::from_utf8_lossy(content);
assert!(
!text.contains("AKIAIOSFODNN7EXAMPLE"),
"secret must not appear in any blob: {text}"
);
}
assert!(
!bundle.blobs.contains_key(&secret_blob.hash),
"the original (unredacted) blob hash must not survive export"
);
}
fn writer_append_file_change_with_secret(store: &Store, sid: &str, hash: &str) {
let writer = store.event_writer().unwrap();
writer
.append_file_change(
Event {
session_id: sid.to_string(),
ts_ms: 4_000,
kind: EventKind::FileChange,
step: None,
summary: "secret.txt created".into(),
body_json: Some(serde_json::json!({"path": "secret.txt"})),
blob_hash: Some(hash.to_string()),
blob_size: Some(27),
correlates: None,
},
FileChange {
event_id: 0,
path: "secret.txt".into(),
change_kind: ChangeKind::Created,
before_hash: None,
after_hash: Some(hash.to_string()),
is_binary: false,
},
)
.unwrap();
writer.finish().unwrap();
store.assign_steps(sid).unwrap();
}
#[test]
fn parse_rejects_newer_format_version() {
let manifest = serde_json::json!({
"format_version": FORMAT_VERSION + 1,
"hh_version": "future",
"session": {},
"integrity": {"events_blake3": blake3::hash(b"").to_hex().to_string(), "event_count": 0, "blobs": []},
});
let manifest_bytes = serde_json::to_vec(&manifest).unwrap();
let tar_bytes = build_tar(&manifest_bytes, b"", &BTreeMap::new()).unwrap();
let compressed = zstd::stream::encode_all(&tar_bytes[..], ZSTD_LEVEL).unwrap();
let err = parse(&compressed).unwrap_err();
assert!(matches!(
err,
crate::Error::Bundle(BundleError::UnsupportedVersion { .. })
));
}
#[test]
fn parse_rejects_a_tampered_blob() {
let (_tmp, store, sid) = fixture();
let bytes = export(&store, &sid, "v", None).unwrap();
let tar_bytes = bounded_decompress(&bytes).unwrap();
let mut archive = tar::Archive::new(&tar_bytes[..]);
let mut builder = tar::Builder::new(Vec::new());
for entry in archive.entries().unwrap() {
let mut entry = entry.unwrap();
let path = entry.path().unwrap().to_string_lossy().into_owned();
let mut data = Vec::new();
entry.read_to_end(&mut data).unwrap();
if path.starts_with("blobs/") {
data.push(0xFF);
}
append_entry(&mut builder, &path, &data).unwrap();
}
let tampered_tar = builder.into_inner().unwrap();
let compressed = zstd::stream::encode_all(&tampered_tar[..], ZSTD_LEVEL).unwrap();
let err = parse(&compressed).unwrap_err();
assert!(matches!(
err,
crate::Error::Bundle(BundleError::HashMismatch { .. })
));
}
#[test]
fn parse_rejects_garbage_bytes_without_panicking() {
for input in [&b""[..], &b"not a bundle"[..], &[0u8; 4096][..]] {
let err = parse(input);
assert!(err.is_err(), "garbage input must be rejected, not accepted");
}
}
#[test]
fn parse_rejects_an_unexpected_tar_entry() {
let mut builder = tar::Builder::new(Vec::new());
let manifest = serde_json::json!({
"format_version": 1,
"hh_version": "v",
"session": {},
"integrity": {"events_blake3": blake3::hash(b"").to_hex().to_string(), "event_count": 0, "blobs": []},
});
append_entry(
&mut builder,
"manifest.json",
&serde_json::to_vec(&manifest).unwrap(),
)
.unwrap();
append_entry(&mut builder, "events.ndjson", b"").unwrap();
append_entry(&mut builder, "readme.txt", b"not on the allow-list").unwrap();
let tar_bytes = builder.into_inner().unwrap();
let compressed = zstd::stream::encode_all(&tar_bytes[..], ZSTD_LEVEL).unwrap();
let err = parse(&compressed).unwrap_err();
assert!(matches!(err, crate::Error::Bundle(BundleError::Tar(_))));
}
#[test]
fn store_import_mints_a_new_id_and_records_provenance() {
let (_tmp, store, sid) = fixture();
let original = store.get_session(&sid).unwrap();
let bytes = export(&store, &sid, "v", None).unwrap();
let bundle = parse(&bytes).unwrap();
let target_tmp = TempDir::new().unwrap();
let target = Store::open(
&target_tmp.path().join("hh.db"),
&target_tmp.path().join("blobs"),
)
.unwrap();
let created = target.import(&bundle).unwrap();
assert_ne!(created.id, original.id, "import must mint a fresh id");
let imported = target.get_session(&created.id).unwrap();
assert_eq!(
imported.imported_from.as_deref(),
Some(original.id.as_str())
);
assert_eq!(imported.command, original.command);
assert_eq!(imported.status, original.status);
let mut events = Vec::new();
target
.for_each_event_raw(&created.id, |r| {
events.push(r);
Ok(())
})
.unwrap();
assert_eq!(events.len(), 4, "every bundled event must be re-inserted");
let call = events
.iter()
.find(|e| e.kind == EventKind::ToolCall)
.unwrap();
let result = events
.iter()
.find(|e| e.kind == EventKind::ToolResult)
.unwrap();
assert_eq!(result.correlates, Some(call.id));
let fc_event = events
.iter()
.find(|e| e.kind == EventKind::FileChange)
.unwrap();
let fc = fc_event.file_change.as_ref().unwrap();
let after_hash = fc.after_hash.as_ref().unwrap();
assert_eq!(
target.blobs().get(after_hash).unwrap(),
b"created content\n"
);
}
#[test]
fn store_import_into_the_same_store_as_the_source_dedupes_blobs() {
let (_tmp, store, sid) = fixture();
let bytes = export(&store, &sid, "v", None).unwrap();
let bundle = parse(&bytes).unwrap();
let created = store.import(&bundle).unwrap();
assert_ne!(created.id, sid);
let original_still_ok = store.get_session(&sid).unwrap();
assert_eq!(original_still_ok.imported_from, None);
}
}