1use std::path::{Path, PathBuf};
27
28use async_stream::stream;
29use chrono::{DateTime, Utc};
30use serde_json::{Value, json};
31
32use crate::{
33 sessions::IngestEvent,
34 wire::{Message, Part, PartKind, Provenance, ProviderOptions, Session},
35};
36
37use super::{
38 Adapter, AdapterError, AdapterFactory, AdapterYield, AdapterYieldStream, DiscoverFuture, Env,
39 RestoreFidelity, RestoredFile, SkipOracle, SkipReason, compact_json, config_path,
40 empty_options,
41 extract::{bound_value, extract_compact_repr, extract_str},
42 extracted_text, part_id, part_ordinal, raw_record, source_options,
43};
44
45const NAME: &str = "claude-ai-export";
46
47const CONVERSATIONS_ENTRY: &str = "conversations.json";
50
51pub struct ClaudeAiExportFactory;
55
56impl AdapterFactory for ClaudeAiExportFactory {
57 fn name(&self) -> &'static str {
58 NAME
59 }
60
61 fn open(&self, config: Value) -> Result<Box<dyn Adapter>, AdapterError> {
62 Ok(Box::new(ClaudeAiExportAdapter::new(config_path(
63 NAME, config,
64 )?)))
65 }
66
67 fn probe_default(&self, _env: &Env) -> Option<Value> {
68 None
71 }
72
73 fn serialize(
74 &self,
75 session: &crate::sessions::SessionWithMessages,
76 fidelity: RestoreFidelity,
77 ) -> Result<Vec<RestoredFile>, AdapterError> {
78 serialize_session(session, fidelity)
79 }
80}
81
82#[derive(Debug, Clone)]
85pub struct ClaudeAiExportAdapter {
86 path: PathBuf,
87}
88
89impl ClaudeAiExportAdapter {
90 pub fn new(path: impl Into<PathBuf>) -> Self {
91 Self { path: path.into() }
92 }
93}
94
95impl Adapter for ClaudeAiExportAdapter {
96 fn discover(&self) -> DiscoverFuture<'_> {
97 let path = self.path.clone();
98 Box::pin(async move {
99 tokio::task::spawn_blocking(move || {
100 read_conversations(&path).map(|conversations| {
101 conversations
102 .iter()
103 .filter(|conv| {
106 conv.get("uuid").and_then(Value::as_str).is_some()
107 && !messages_of(conv).is_empty()
108 })
109 .count()
110 })
111 })
112 .await
113 .map_err(join_error)?
114 })
115 }
116
117 fn events_with<'a>(&'a self, oracle: &'a dyn SkipOracle) -> AdapterYieldStream<'a> {
118 let path = self.path.clone();
119 Box::pin(stream! {
120 let parsed = tokio::task::spawn_blocking(move || read_conversations(&path)).await;
124 let conversations = match parsed {
125 Ok(Ok(conversations)) => conversations,
126 Ok(Err(error)) => { yield Err(error); return; }
127 Err(join) => { yield Err(join_error(join)); return; }
128 };
129
130 for mut conv in conversations {
131 bound_value(&mut conv);
132 let Some(session_id) = conv.get("uuid").and_then(Value::as_str).map(ToOwned::to_owned)
133 else {
134 yield Err(AdapterError::schema(
135 NAME,
136 CONVERSATIONS_ENTRY,
137 "conversation missing `uuid`",
138 ));
139 continue;
140 };
141 if messages_of(&conv).is_empty() {
142 yield Ok(AdapterYield::Skipped {
144 session_id: Some(session_id),
145 project: conv
146 .get("account")
147 .and_then(|account| account.get("uuid"))
148 .and_then(Value::as_str)
149 .map(ToOwned::to_owned),
150 reason: SkipReason::Empty,
151 });
152 continue;
153 }
154 let source_last_ts = messages_of(&conv)
159 .last()
160 .and_then(|message| rfc3339(message, "created_at"))
161 .map(|timestamp| timestamp.timestamp_micros());
162 if crate::adapter::is_session_fresh(oracle, &session_id, source_last_ts) {
163 yield Ok(AdapterYield::Skipped {
164 session_id: Some(session_id),
165 project: None,
166 reason: SkipReason::Fresh,
167 });
168 continue;
169 }
170
171 let session = match build_session(&conv, &session_id) {
172 Ok(session) => session,
173 Err(error) => { yield Err(error); continue; }
174 };
175 let created_at = session.created_at;
176 yield Ok(AdapterYield::Event(IngestEvent::Session(session)));
177
178 for (index, message) in messages_of(&conv).iter().enumerate() {
179 for event in message_events(&session_id, message, index, created_at) {
180 yield Ok(AdapterYield::Event(event));
181 }
182 }
183 }
184 })
185 }
186}
187
188fn join_error(join: tokio::task::JoinError) -> AdapterError {
190 AdapterError::io(
191 NAME,
192 "blocking read task",
193 std::io::Error::other(join.to_string()),
194 )
195}
196
197fn messages_of(conv: &Value) -> &[Value] {
198 conv.get("chat_messages")
199 .and_then(Value::as_array)
200 .map(Vec::as_slice)
201 .unwrap_or(&[])
202}
203
204fn read_conversations(path: &Path) -> Result<Vec<Value>, AdapterError> {
207 use std::io::Read;
208 let io = |location: String, source| AdapterError::io(NAME, location, source);
209
210 let bytes = if path.is_dir() {
211 let file = path.join(CONVERSATIONS_ENTRY);
212 std::fs::read(&file).map_err(|error| io(file.display().to_string(), error))?
213 } else if path.extension().and_then(|ext| ext.to_str()) == Some("zip") {
214 let file =
215 std::fs::File::open(path).map_err(|error| io(path.display().to_string(), error))?;
216 let mut archive = zip::ZipArchive::new(file).map_err(|error| {
217 AdapterError::schema(
218 NAME,
219 path.display().to_string(),
220 format!("bad zip: {error}"),
221 )
222 })?;
223 let mut entry = archive.by_name(CONVERSATIONS_ENTRY).map_err(|error| {
224 AdapterError::schema(
225 NAME,
226 path.display().to_string(),
227 format!("export zip has no `{CONVERSATIONS_ENTRY}`: {error}"),
228 )
229 })?;
230 let hint = entry.size().min(64 * 1024 * 1024) as usize;
234 let mut buf = Vec::with_capacity(hint);
235 entry
236 .read_to_end(&mut buf)
237 .map_err(|error| io(path.display().to_string(), error))?;
238 buf
239 } else {
240 std::fs::read(path).map_err(|error| io(path.display().to_string(), error))?
241 };
242
243 let value: Value = serde_json::from_slice(&bytes)
244 .map_err(|error| AdapterError::parse(NAME, path.display().to_string(), 1, error))?;
245 match value {
246 Value::Array(conversations) => Ok(conversations),
247 _ => Err(AdapterError::schema(
248 NAME,
249 path.display().to_string(),
250 format!("`{CONVERSATIONS_ENTRY}` is not a JSON array"),
251 )),
252 }
253}
254
255fn build_session(conv: &Value, session_id: &str) -> Result<Session, AdapterError> {
256 let created_at = rfc3339(conv, "created_at").ok_or_else(|| {
257 AdapterError::schema(
258 NAME,
259 session_id.to_owned(),
260 "conversation missing/invalid `created_at`",
261 )
262 })?;
263 let project = conv
266 .get("account")
267 .and_then(|account| extract_str(account, "uuid"))
268 .filter(|uuid| !uuid.trim().is_empty())
271 .ok_or_else(|| {
272 AdapterError::schema(
273 NAME,
274 session_id.to_owned(),
275 "conversation missing/empty `account.uuid` for the project",
276 )
277 })?;
278
279 let mut options = source_options(NAME, conv);
280 if let Some(source) = options.get_mut("source").and_then(Value::as_object_mut) {
281 for key in ["name", "summary", "updated_at"] {
282 if let Some(value) = conv.get(key) {
283 source.insert(key.to_owned(), value.clone());
284 }
285 }
286 }
287
288 Ok(Session {
289 id: session_id.to_owned(),
290 parent_session_id: None,
291 parent_message_id: None,
292 source_agent: NAME.to_owned(),
293 created_at,
294 project,
295 options,
296 })
297}
298
299fn message_events(
300 session_id: &str,
301 message: &Value,
302 index: usize,
303 default_ts: DateTime<Utc>,
304) -> Vec<IngestEvent> {
305 let message_id = message
309 .get("uuid")
310 .and_then(Value::as_str)
311 .map_or_else(|| format!("{session_id}:{index}"), ToOwned::to_owned);
312 let timestamp = rfc3339(message, "created_at").unwrap_or(default_ts);
313 let blocks = message
314 .get("content")
315 .and_then(Value::as_array)
316 .map(Vec::as_slice)
317 .unwrap_or(&[]);
318 let sender = message.get("sender").and_then(Value::as_str);
319 let all_tool_results = !blocks.is_empty() && blocks.iter().all(is_tool_result);
320
321 let parts: Vec<Part> = blocks
322 .iter()
323 .enumerate()
324 .map(|(ordinal, block)| content_part(session_id, &message_id, ordinal, block))
325 .collect();
326
327 let options = message_options(message);
328 let header = match (sender, all_tool_results) {
329 (Some("human"), true) => Message::Tool {
331 id: message_id.clone(),
332 session_id: session_id.to_owned(),
333 timestamp,
334 options,
335 },
336 (Some("assistant"), _) => Message::Assistant {
337 id: message_id.clone(),
338 session_id: session_id.to_owned(),
339 timestamp,
340 options,
341 },
342 _ => Message::User {
345 id: message_id.clone(),
346 session_id: session_id.to_owned(),
347 timestamp,
348 options,
349 },
350 };
351
352 let mut events = Vec::with_capacity(parts.len() + 1);
353 events.push(IngestEvent::Message(header));
354 events.extend(parts.into_iter().map(IngestEvent::Part));
355 events
356}
357
358fn content_part(session_id: &str, message_id: &str, ordinal: usize, block: &Value) -> Part {
359 let (provenance, kind) = match block.get("type").and_then(Value::as_str) {
360 Some("text") => (
361 Provenance::Conversational,
362 PartKind::Text {
363 text: extract_str(block, "text"),
364 },
365 ),
366 Some("thinking") => (
367 Provenance::Conversational,
368 PartKind::Reasoning {
369 text: extract_str(block, "thinking"),
370 },
371 ),
372 Some("tool_use") => (
373 Provenance::Conversational,
374 PartKind::ToolCall {
375 call_id: extract_str(block, "id"),
376 name: extract_str(block, "name"),
377 params: block.get("input").cloned().unwrap_or(Value::Null),
378 provider_executed: true,
379 },
380 ),
381 Some("tool_result") => (
382 Provenance::Injected,
383 PartKind::ToolResult {
384 call_id: None,
389 name: extract_str(block, "name"),
390 is_failure: block
391 .get("is_error")
392 .and_then(Value::as_bool)
393 .unwrap_or(false),
394 result: block.get("content").cloned().unwrap_or(Value::Null),
395 },
396 ),
397 _ => (
400 Provenance::Conversational,
401 PartKind::Text {
402 text: Some(extract_compact_repr(block)),
403 },
404 ),
405 };
406 Part {
407 session_id: session_id.to_owned(),
408 id: part_id(message_id, ordinal),
409 message_id: message_id.to_owned(),
410 ordinal: part_ordinal(ordinal),
411 provenance,
412 options: empty_options(),
413 kind,
414 }
415}
416
417fn message_options(message: &Value) -> ProviderOptions {
418 let mut options = source_options(NAME, message);
419 if let Some(source) = options.get_mut("source").and_then(Value::as_object_mut) {
420 for key in ["sender", "updated_at", "attachments", "files"] {
421 if let Some(value) = message.get(key) {
422 source.insert(key.to_owned(), value.clone());
423 }
424 }
425 }
426 options
427}
428
429fn is_tool_result(block: &Value) -> bool {
430 block.get("type").and_then(Value::as_str) == Some("tool_result")
431}
432
433fn rfc3339(value: &Value, key: &str) -> Option<DateTime<Utc>> {
434 value
435 .get(key)
436 .and_then(Value::as_str)
437 .and_then(|text| DateTime::parse_from_rfc3339(text).ok())
438 .map(|dt| dt.with_timezone(&Utc))
439}
440
441fn serialize_session(
442 session: &crate::sessions::SessionWithMessages,
443 fidelity: RestoreFidelity,
444) -> Result<Vec<RestoredFile>, AdapterError> {
445 let conversation = match fidelity {
450 RestoreFidelity::Native => raw_record(&session.session.options),
451 RestoreFidelity::Foreign => None,
452 };
453 let actual_fidelity = if conversation.is_some() {
454 RestoreFidelity::Native
455 } else {
456 RestoreFidelity::Foreign
457 };
458 let conversation = conversation.unwrap_or_else(|| foreign_conversation(session));
459
460 Ok(vec![RestoredFile::new(
461 PathBuf::from(CONVERSATIONS_ENTRY),
462 serde_json::to_vec(&Value::Array(vec![conversation])).map_err(|error| {
463 AdapterError::schema(
464 NAME,
465 &session.session.id,
466 format!("json encode failed: {error}"),
467 )
468 })?,
469 actual_fidelity,
470 )])
471}
472
473fn foreign_conversation(session: &crate::sessions::SessionWithMessages) -> Value {
476 let chat_messages: Vec<Value> = session
477 .messages
478 .iter()
479 .map(|message| {
480 let sender = match message.message {
481 Message::Assistant { .. } => "assistant",
482 _ => "human",
483 };
484 json!({
485 "uuid": message.message.id(),
486 "sender": sender,
487 "created_at": message.message.timestamp().to_rfc3339(),
488 "content": message.parts.iter().map(content_block).collect::<Vec<_>>(),
489 })
490 })
491 .collect();
492 json!({
493 "uuid": session.session.id,
494 "account": { "uuid": &*session.session.project },
495 "created_at": session.session.created_at.to_rfc3339(),
496 "chat_messages": chat_messages,
497 })
498}
499
500fn content_block(part: &Part) -> Value {
501 match &part.kind {
502 PartKind::Text { text } => json!({ "type": "text", "text": extracted_text(text) }),
503 PartKind::Reasoning { text } => {
504 json!({ "type": "thinking", "thinking": extracted_text(text) })
505 }
506 PartKind::ToolCall {
507 call_id,
508 name,
509 params,
510 ..
511 } => json!({
512 "type": "tool_use",
513 "id": extracted_text(call_id),
514 "name": extracted_text(name),
515 "input": params,
516 }),
517 PartKind::ToolResult {
518 name,
519 is_failure,
520 result,
521 ..
522 } => json!({
523 "type": "tool_result",
524 "name": extracted_text(name),
525 "is_error": is_failure,
526 "content": result,
527 }),
528 other => json!({
529 "type": "text",
530 "text": compact_json(&serde_json::to_value(other).unwrap_or(Value::Null)),
531 }),
532 }
533}
534
535#[cfg(test)]
536mod tests {
537 #![allow(clippy::expect_used, clippy::unwrap_used)]
542
543 use super::*;
544 use crate::{handlers::ingest_adapter, sessions::Store};
545 use tempfile::TempDir;
546
547 const FIXTURE_DIR: &str = concat!(
550 env!("CARGO_MANIFEST_DIR"),
551 "/tests/fixtures/adapter/claude_ai_export"
552 );
553 const ACCOUNT: &str = "ffffffff-ffff-ffff-ffff-ffffffffffff";
554 const TOOL_CONV: &str = "33333333-3333-3333-3333-333333333333";
555 const EMPTY_NAME_CONV: &str = "44444444-4444-4444-4444-444444444444";
556 const ZERO_MESSAGE_CONV: &str = "55555555-5555-5555-5555-555555555555";
557
558 #[test]
559 fn probe_default_returns_none_no_autodiscovery() {
560 assert!(
562 ClaudeAiExportFactory
563 .probe_default(&Env::with_home("/tmp"))
564 .is_none()
565 );
566 }
567
568 #[tokio::test(flavor = "multi_thread")]
569 async fn ingests_export_directory_into_canonical_shape() -> anyhow::Result<()> {
570 let temp = TempDir::new()?;
571 let store = Store::open_local(temp.path()).await?;
572 let summary = ingest_adapter(
573 &store,
574 &ClaudeAiExportAdapter::new(FIXTURE_DIR),
575 &crate::adapter::NoopOracle,
576 |_| {},
577 )
578 .await?;
579 assert_eq!(summary.dropped_sessions, 0);
580
581 let ids = store.session_ids().await?;
582 assert_eq!(ids.len(), 4, "the 0-message conversation is skipped");
584 assert!(
585 !ids.iter().any(|id| id == ZERO_MESSAGE_CONV),
586 "0-message conversation must not become a session",
587 );
588 assert!(
589 ids.iter().any(|id| id == EMPTY_NAME_CONV),
590 "an empty-name conversation still ingests (title can't gate it)",
591 );
592
593 for id in &ids {
594 let session = store.get_session(id).await?.expect("round-trips");
595 assert_eq!(session.session.source_agent, NAME);
596 assert_eq!(
597 &*session.session.project, ACCOUNT,
598 "spec.md#model-project-non-empty: project = account.uuid",
599 );
600 }
601
602 let tool = store
603 .get_session(TOOL_CONV)
604 .await?
605 .expect("tool conversation");
606 let mut saw_call = false;
607 let mut saw_reasoning = false;
608 let mut tool_result = None;
609 let mut saw_tool_message = false;
610 for stored in &tool.messages {
611 if matches!(stored.message, Message::Tool { .. }) {
612 saw_tool_message = true;
613 }
614 for part in &stored.parts {
615 match &part.kind {
616 PartKind::ToolCall { name, .. }
617 if name.as_deref().map(String::as_str) == Some("web_search") =>
618 {
619 saw_call = true;
620 }
621 PartKind::Reasoning { .. } => saw_reasoning = true,
622 PartKind::ToolResult { call_id, name, .. } => {
623 tool_result = Some((call_id.as_deref().cloned(), name.as_deref().cloned()));
624 }
625 _ => {}
626 }
627 }
628 }
629 let thinking = store
631 .get_session("22222222-2222-2222-2222-222222222222")
632 .await?
633 .expect("thinking conversation");
634 for stored in &thinking.messages {
635 for part in &stored.parts {
636 if matches!(part.kind, PartKind::Reasoning { .. }) {
637 saw_reasoning = true;
638 }
639 }
640 }
641 assert!(saw_call, "tool_use -> ToolCall named web_search");
642 assert!(saw_reasoning, "thinking -> Reasoning");
643 assert!(
644 saw_tool_message,
645 "a human turn of pure tool_result is a Tool message",
646 );
647 let (call_id, name) = tool_result.expect("tool conversation has a ToolResult");
648 assert_eq!(
649 name.as_deref(),
650 Some("web_search"),
651 "tool_result name comes straight off the block",
652 );
653 assert_eq!(
654 call_id, None,
655 "the export carries no tool_use_id on tool_result, so call_id is honestly None",
656 );
657 Ok(())
658 }
659
660 #[test]
664 fn uuid_less_message_ingests_under_synthetic_id() {
665 let ts = DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
666 .unwrap()
667 .with_timezone(&Utc);
668 let message = json!({
669 "sender": "human",
670 "content": [{ "type": "text", "text": "no uuid here" }],
671 });
672 let events = message_events("conv-xyz", &message, 3, ts);
673 assert_eq!(
674 events.len(),
675 2,
676 "uuid-less message still emits its Message + Part, not dropped",
677 );
678 match &events[0] {
679 IngestEvent::Message(message) => {
680 assert_eq!(message.id(), "conv-xyz:3", "deterministic synthetic id");
681 }
682 _ => panic!("first event must be the Message"),
683 }
684 }
685
686 #[tokio::test(flavor = "multi_thread")]
687 async fn ingests_export_zip() -> anyhow::Result<()> {
688 use std::io::Write;
689 let temp = TempDir::new()?;
690 let zip_path = temp.path().join("data-2026-01-15-00-00-00-batch-0000.zip");
691 let conversations = std::fs::read(format!("{FIXTURE_DIR}/conversations.json"))?;
692 {
693 let file = std::fs::File::create(&zip_path)?;
694 let mut zip = zip::ZipWriter::new(file);
695 zip.start_file(
696 "conversations.json",
697 zip::write::SimpleFileOptions::default(),
698 )?;
699 zip.write_all(&conversations)?;
700 zip.finish()?;
701 }
702
703 let store = Store::open_local(temp.path().join("store")).await?;
704 ingest_adapter(
705 &store,
706 &ClaudeAiExportAdapter::new(&zip_path),
707 &crate::adapter::NoopOracle,
708 |_| {},
709 )
710 .await?;
711 assert_eq!(
712 store.session_ids().await?.len(),
713 4,
714 "the same four sessions ingest from the zip",
715 );
716 Ok(())
717 }
718
719 #[tokio::test(flavor = "multi_thread")]
720 async fn native_restore_round_trips_one_conversation() -> anyhow::Result<()> {
721 let temp = TempDir::new()?;
722 let store = Store::open_local(temp.path().join("store")).await?;
723 ingest_adapter(
724 &store,
725 &ClaudeAiExportAdapter::new(FIXTURE_DIR),
726 &crate::adapter::NoopOracle,
727 |_| {},
728 )
729 .await?;
730 let session = store
731 .get_session(TOOL_CONV)
732 .await?
733 .expect("tool conversation");
734
735 let files = ClaudeAiExportFactory.serialize(&session, RestoreFidelity::Native)?;
736 assert_eq!(files.len(), 1);
737 assert_eq!(
738 files[0].relative_path,
739 std::path::Path::new(CONVERSATIONS_ENTRY)
740 );
741 let value: Value = serde_json::from_slice(&files[0].bytes)?;
742 let array = value.as_array().expect("conversations.json is an array");
743 assert_eq!(
744 array.len(),
745 1,
746 "per-session restore is a one-conversation export"
747 );
748 assert_eq!(
749 array[0].get("uuid").and_then(Value::as_str),
750 Some(TOOL_CONV),
751 );
752
753 let restore_dir = temp.path().join("restore");
754 std::fs::create_dir_all(&restore_dir)?;
755 std::fs::write(restore_dir.join(CONVERSATIONS_ENTRY), &files[0].bytes)?;
756 let restore_store = Store::open_local(temp.path().join("restore-store")).await?;
757 ingest_adapter(
758 &restore_store,
759 &ClaudeAiExportAdapter::new(&restore_dir),
760 &crate::adapter::NoopOracle,
761 |_| {},
762 )
763 .await?;
764 let restored = restore_store
765 .get_session(TOOL_CONV)
766 .await?
767 .expect("restored");
768 assert_eq!(
769 restored.messages.len(),
770 session.messages.len(),
771 "native restore replays every message",
772 );
773 Ok(())
774 }
775}