feat: complete M0 phase - read-only spine (8/51 tasks)

M0.1 - Cargo workspace + crate skeletons (4 tests)
  ✅ 6-crate workspace with enforced dependency direction
  ✅ GitHub Actions CI pipeline

M0.2 - Domain types and sha256 identity (6 tests)
  ✅ Level, Role, Record, Chunk, MemoryNode types
  ✅ Content-hash identity (sha256) ensuring rebuild idempotence
  ✅ Newtypes (ProjectId, QueryId, RunId) without Default

M0.3 - RecordSource trait + ChunkPolicy (6 tests)
  ✅ RecordSource streaming trait
  ✅ Chunk policy with token budgets and record boundaries
  ✅ Chunking stream that respects budgets without splitting records

M0.4 - Tokenizer-backed chunk sizing (3 tests + 1 ignored)
  ✅ Vendored Qwen2 tokenizer with hash verification
  ✅ QwenTokenCounter for accurate token counting
  ✅ mem tokens CLI subcommand

M0.5 - pi session adapter (5 tests)
  ✅ PiSessionSource implementing RecordSource
  ✅ Project key extraction from cwd field
  ✅ Content flattening for various shapes
  ✅ Shared flatten_content helper module

M0.6 - Claude transcript adapter (4 tests)
  ✅ ClaudeTranscriptSource implementing RecordSource
  ✅ Identical content flattening as pi source
  ✅ Cross-source project key agreement

M0.7 - ingest --dry-run (2 tests)
  ✅ mem ingest --project --dry-run command
  ✅ Zero network calls guarantee

M0.8 - M0 composition gate (5 tests)
  ✅ Both sources compose through chunker identically
  ✅ Sources are swappable via RecordSource trait
  ✅ All role types properly emitted
  ✅ Chunk boundaries respected, t values contiguous

Summary:
- 35 integration tests (34 passing, 1 ignored)
- Zero clippy warnings with -D warnings
- All phases compose and verify correctly
- Read-only spine foundation proves extensibility

crates/mem-ingest/src/claude_transcript.rs

@@ -0,0 +1,178 @@
+use anyhow::anyhow;
+use futures::stream::{Stream, StreamExt};
+use mem_chunk::RecordSource;
+use mem_core::{Provenance, Record, Role};
+use serde_json::Value;
+use std::path::{Path, PathBuf};
+use tokio::fs::File;
+use tokio::io::{AsyncBufReadExt, BufReader};
+use crate::pi_session::flatten_content;
+
+/// A source that reads Claude transcript JSONL files.
+pub struct ClaudeTranscriptSource {
+    file_path: PathBuf,
+}
+
+impl ClaudeTranscriptSource {
+    /// Create a new Claude transcript source from a file path.
+    pub fn new(file_path: PathBuf) -> Self {
+        ClaudeTranscriptSource { file_path }
+    }
+
+    /// Extract the project key from the transcript file's cwd field.
+    pub async fn read_project_key(&self) -> anyhow::Result<String> {
+        let file = File::open(&self.file_path).await?;
+        let reader = BufReader::new(file);
+        let mut lines = reader.lines();
+
+        while let Some(line_result) = lines.next_line().await? {
+            let value: Value = match serde_json::from_str(&line_result) {
+                Ok(v) => v,
+                Err(_) => continue,
+            };
+
+            if let Some(cwd) = value.get("cwd").and_then(|v| v.as_str()) {
+                return Ok(cwd.to_string());
+            }
+        }
+
+        Err(anyhow!("No line with 'cwd' field found in transcript"))
+    }
+}
+
+impl RecordSource for ClaudeTranscriptSource {
+    fn records(self) -> Box<dyn Stream<Item = Result<Record, String>> + Unpin> {
+        let file_path = self.file_path.clone();
+        let session_id = file_path
+            .file_stem()
+            .and_then(|s| s.to_str())
+            .unwrap_or("unknown")
+            .to_string();
+
+        // Create a stream that reads the file asynchronously
+        let stream = futures::stream::once(async move {
+            match read_claude_transcript_lines(&file_path, &session_id).await {
+                Ok(records) => records,
+                Err(e) => vec![Err(e.to_string())],
+            }
+        })
+        .flat_map(futures::stream::iter);
+
+        Box::new(stream.boxed())
+    }
+}
+
+/// Read lines from a claude transcript file and convert them to Records.
+pub async fn read_claude_transcript_lines(
+    path: &Path,
+    session_id: &str,
+) -> anyhow::Result<Vec<Result<Record, String>>> {
+    let file = File::open(path).await?;
+    let reader = BufReader::new(file);
+    let mut lines = reader.lines();
+    let mut records = Vec::new();
+    let mut line_num = 0u64;
+
+    while let Some(line_result) = lines.next_line().await? {
+        line_num += 1;
+
+        match serde_json::from_str::<Value>(&line_result) {
+            Ok(value) => {
+                let record_type = value
+                    .get("type")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("unknown");
+
+                match record_type {
+                    "user" => {
+                        if let Ok(Some(record)) = parse_claude_message(&value, Role::User, session_id, line_num) {
+                            records.push(Ok(record));
+                        }
+                    }
+                    "assistant" => {
+                        if let Ok(Some(record)) = parse_claude_message(&value, Role::Assistant, session_id, line_num) {
+                            records.push(Ok(record));
+                        }
+                    }
+                    "system" => {
+                        // Only emit system messages with api_error subtype
+                        if value.get("subtype").and_then(|v| v.as_str()) == Some("api_error") {
+                            if let Ok(Some(record)) = parse_claude_message(&value, Role::System, session_id, line_num) {
+                                records.push(Ok(record));
+                            }
+                        }
+                    }
+                    "attachment" | "queue-operation" | "file-history-snapshot" | "summary" | "mode" | "ai-title" | "last-prompt" => {
+                        // Skip these record types
+                    }
+                    _ => {}
+                }
+            }
+            Err(_) => {
+                // Malformed line - skip with warning
+                tracing::warn!(
+                    "Skipping malformed JSON at line {}: {}",
+                    line_num,
+                    line_result.trim()
+                );
+            }
+        }
+    }
+
+    Ok(records)
+}
+
+fn parse_claude_message(
+    value: &Value,
+    role: Role,
+    session_id: &str,
+    line_num: u64,
+) -> Result<Option<Record>, String> {
+    let message = value.get("message");
+    let timestamp = parse_claude_timestamp(value);
+
+    let text = if let Some(msg) = message {
+        if let Some(content) = msg.get("content") {
+            flatten_content(content)?
+        } else {
+            "".to_string()
+        }
+    } else {
+        "".to_string()
+    };
+
+    if text.is_empty() && role != Role::System {
+        return Ok(None);
+    }
+
+    Ok(Some(Record {
+        role,
+        text,
+        timestamp,
+        provenance: Provenance {
+            source_id: format!("claude:{}", session_id),
+            offset: line_num,
+        },
+    }))
+}
+
+fn parse_claude_timestamp(value: &Value) -> time::OffsetDateTime {
+    value
+        .get("timestamp")
+        .and_then(|v| v.as_str())
+        .and_then(|s| {
+            time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339).ok()
+        })
+        .unwrap_or_else(time::OffsetDateTime::now_utc)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_claude_transcript_source_creation() {
+        let source = ClaudeTranscriptSource::new(PathBuf::from("test.jsonl"));
+        assert_eq!(source.file_path.to_string_lossy(), "test.jsonl");
+    }
+}

crates/mem-ingest/src/lib.rs

@@ -1 +1,5 @@
-pub mod placeholder {}
+pub mod pi_session;
+pub mod claude_transcript;
+
+pub use pi_session::PiSessionSource;
+pub use claude_transcript::ClaudeTranscriptSource;

crates/mem-ingest/src/pi_session.rs

@@ -0,0 +1,251 @@
+use anyhow::anyhow;
+use futures::stream::{Stream, StreamExt};
+use mem_chunk::RecordSource;
+use mem_core::{Provenance, Record, Role};
+use serde_json::Value;
+use std::path::{Path, PathBuf};
+use tokio::fs::File;
+use tokio::io::{AsyncBufReadExt, BufReader};
+
+/// A source that reads pi session JSONL files.
+pub struct PiSessionSource {
+    file_path: PathBuf,
+}
+
+impl PiSessionSource {
+    /// Create a new Pi session source from a file path.
+    pub fn new(file_path: PathBuf) -> Self {
+        PiSessionSource { file_path }
+    }
+
+    /// Extract the project key from the session file's cwd.
+    pub async fn read_project_key(&self) -> anyhow::Result<String> {
+        let file = File::open(&self.file_path).await?;
+        let mut reader = BufReader::new(file);
+        let mut first_line = String::new();
+        reader.read_line(&mut first_line).await?;
+
+        let value: Value = serde_json::from_str(&first_line)?;
+        let session_type = value
+            .get("type")
+            .and_then(|v| v.as_str())
+            .ok_or_else(|| anyhow!("Missing 'type' field in session header"))?;
+
+        if session_type != "session" {
+            return Err(anyhow!(
+                "Expected first line to have type='session', got '{}'",
+                session_type
+            ));
+        }
+
+        let cwd = value
+            .get("cwd")
+            .and_then(|v| v.as_str())
+            .ok_or_else(|| anyhow!("Missing 'cwd' field in session header"))?;
+
+        Ok(cwd.to_string())
+    }
+}
+
+impl RecordSource for PiSessionSource {
+    fn records(self) -> Box<dyn Stream<Item = Result<Record, String>> + Unpin> {
+        let file_path = self.file_path.clone();
+        let session_id = file_path
+            .file_stem()
+            .and_then(|s| s.to_str())
+            .unwrap_or("unknown")
+            .to_string();
+
+        // Create a stream that reads the file asynchronously
+        let stream = futures::stream::once(async move {
+            match read_pi_session_lines(&file_path, &session_id).await {
+                Ok(records) => records,
+                Err(e) => vec![Err(e.to_string())],
+            }
+        })
+        .flat_map(futures::stream::iter);
+
+        Box::new(stream.boxed())
+    }
+}
+
+/// Read lines from a pi session file and convert them to Records.
+pub async fn read_pi_session_lines(
+    path: &Path,
+    session_id: &str,
+) -> anyhow::Result<Vec<Result<Record, String>>> {
+    let file = File::open(path).await?;
+    let reader = BufReader::new(file);
+    let mut lines = reader.lines();
+    let mut records = Vec::new();
+    let mut line_num = 0u64;
+    let mut skip_first = true;
+
+    while let Some(line_result) = lines.next_line().await? {
+        line_num += 1;
+        
+        // Skip the first line (session header)
+        if skip_first {
+            skip_first = false;
+            continue;
+        }
+
+        match serde_json::from_str::<Value>(&line_result) {
+            Ok(value) => {
+                let record_type = value
+                    .get("type")
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("unknown");
+
+                match record_type {
+                    "message" => {
+                        if let Ok(Some(record)) = parse_message(&value, session_id, line_num) {
+                            records.push(Ok(record));
+                        }
+                    }
+                    "compaction" => {
+                        let msg = value
+                            .get("message")
+                            .and_then(|v| v.as_str())
+                            .unwrap_or("Context compacted");
+                        
+                        let timestamp = parse_timestamp(&value);
+                        let record = Record {
+                            role: Role::System,
+                            text: msg.to_string(),
+                            timestamp,
+                            provenance: Provenance {
+                                source_id: format!("pi:{}", session_id),
+                                offset: line_num,
+                            },
+                        };
+                        records.push(Ok(record));
+                    }
+                    "model_change" | "thinking_level_change" => {
+                        // Skip these record types
+                    }
+                    _ => {}
+                }
+            }
+            Err(_) => {
+                // Malformed line - skip with warning
+                tracing::warn!(
+                    "Skipping malformed JSON at line {}: {}",
+                    line_num,
+                    line_result.trim()
+                );
+            }
+        }
+    }
+
+    Ok(records)
+}
+
+fn parse_message(
+    value: &Value,
+    session_id: &str,
+    line_num: u64,
+) -> Result<Option<Record>, String> {
+    let message = value
+        .get("message")
+        .ok_or_else(|| "Missing 'message' field".to_string())?;
+
+    let role_str = message
+        .get("role")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| "Missing 'role' field in message".to_string())?;
+
+    let role = match role_str {
+        "user" => Role::User,
+        "assistant" => Role::Assistant,
+        "toolResult" => Role::ToolResult,
+        _ => return Ok(None),
+    };
+
+    let content = message
+        .get("content")
+        .ok_or_else(|| "Missing 'content'".to_string())?;
+    let text = flatten_content(content)?;
+
+    let timestamp = parse_timestamp(message);
+    let record_id = value
+        .get("id")
+        .and_then(|v| v.as_str())
+        .unwrap_or("unknown");
+
+    Ok(Some(Record {
+        role,
+        text,
+        timestamp,
+        provenance: Provenance {
+            source_id: format!("pi:{}:{}", session_id, record_id),
+            offset: line_num,
+        },
+    }))
+}
+
+/// Shared content flattening logic for all sources.
+/// Handles: string, array of blocks, and structured objects.
+pub fn flatten_content(content: &Value) -> Result<String, String> {
+    if let Some(s) = content.as_str() {
+        // Simple string content
+        return Ok(s.to_string());
+    }
+
+    if let Some(arr) = content.as_array() {
+        // Array of content blocks
+        let mut text_parts = Vec::new();
+        for block in arr {
+            if let Some(block_type) = block.get("type").and_then(|v| v.as_str()) {
+                match block_type {
+                    "text" => {
+                        if let Some(text) = block.get("text").and_then(|v| v.as_str()) {
+                            text_parts.push(text.to_string());
+                        }
+                    }
+                    "tool_use" => {
+                        if let Some(name) = block.get("tool_name").and_then(|v| v.as_str()) {
+                            text_parts.push(format!("[tool_use: {}]", name));
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        }
+        if !text_parts.is_empty() {
+            return Ok(text_parts.join(" "));
+        }
+    }
+
+    if let Some(obj) = content.as_object() {
+        // Structured content (e.g., tool result)
+        if let Some(text) = obj.get("text").and_then(|v| v.as_str()) {
+            return Ok(text.to_string());
+        }
+        // Fallback: serialize the whole object
+        return Ok(content.to_string());
+    }
+
+    Err("Could not flatten content".to_string())
+}
+
+fn parse_timestamp(value: &Value) -> time::OffsetDateTime {
+    value
+        .get("timestamp")
+        .and_then(|v| v.as_str())
+        .and_then(|s| {
+            time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339).ok()
+        })
+        .unwrap_or_else(time::OffsetDateTime::now_utc)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_pi_session_source_creation() {
+        let source = PiSessionSource::new(PathBuf::from("test.jsonl"));
+        assert_eq!(source.file_path.to_string_lossy(), "test.jsonl");
+    }
+}