Parse Claude and Cursor session records separately (#35623)

## Why

Cursor session messages can place `<cursor_commands>` and `<timestamp>` context
before the actual `<user_query>`. Treating both external session formats with one
parser left that context in the imported message and generated title.

## What changed

- Route detected sessions through format-specific Claude and Cursor parsers.
- Strip recognized Cursor context wrappers when extracting a trailing
  `<user_query>`, while preserving messages with unknown leading context.
- Keep shared content-block, tool-call, and timestamp parsing in a common module.

## Testing

Add parser coverage for wrapped Cursor queries, unknown context, working-directory
precedence, Claude metadata, content hashing, and shared tool annotations.

GitOrigin-RevId: b49328d46ab4937e3c599588a3d3081e8ab798cd
This commit is contained in:
charlesgong-openai
2026-07-27 14:31:50 +00:00
committed by copyberry
parent fea2d8b18f
commit bd2de422aa
13 changed files with 878 additions and 566 deletions
@@ -279,7 +279,7 @@ async fn external_agent_config_secondary_source_imports_session_and_plugin_end_t
"message": {
"content": [{
"type": "text",
"text": "<user_query>first request</user_query>"
"text": "<cursor_commands>\n/verify\n</cursor_commands>\n<timestamp>2026-07-26T18:00:00Z</timestamp>\n<user_query>first request</user_query>"
}]
}
})
@@ -2,6 +2,7 @@ use super::common::SessionFileCandidate;
use super::common::detect_recent_sessions;
use crate::model::ExternalAgentSessionImportLimits;
use crate::sessions::ExternalAgentSessionMigration;
use crate::sessions::SessionRecordFormat;
use std::fs;
use std::io;
use std::path::Path;
@@ -50,6 +51,7 @@ pub(crate) fn detect_recent_cla_sessions_with_limits(
candidates.push(SessionFileCandidate {
path,
fallback_cwd: None,
record_format: SessionRecordFormat::Cla,
});
}
}
@@ -1,9 +1,11 @@
use crate::model::ExternalAgentSessionImportLimits;
use crate::sessions::ExternalAgentSessionMigration;
use crate::sessions::SessionRecordFormat;
use crate::sessions::ledger::load_import_ledger;
use crate::sessions::ledger::save_import_ledger;
use crate::sessions::now_unix_seconds;
use crate::sessions::records::summarize_session_with_cwd;
use crate::sessions::records_cla;
use crate::sessions::records_cur;
use std::cmp::Reverse;
use std::collections::BinaryHeap;
use std::fs;
@@ -14,6 +16,7 @@ use std::path::PathBuf;
pub(super) struct SessionFileCandidate {
pub path: PathBuf,
pub fallback_cwd: Option<PathBuf>,
pub record_format: SessionRecordFormat,
}
pub(super) fn detect_recent_sessions(
@@ -57,6 +60,7 @@ pub(super) fn detect_recent_sessions(
Reverse(modified_at_nanos),
candidate.path,
candidate.fallback_cwd,
candidate.record_format,
));
if recent.len() > limits.max_sessions {
recent.pop();
@@ -66,7 +70,7 @@ pub(super) fn detect_recent_sessions(
drop(source_states);
let mut migrations = Vec::new();
let mut ledger_changed = false;
for (modified_at, path, fallback_cwd) in recent.into_sorted_vec() {
for (modified_at, path, fallback_cwd, record_format) in recent.into_sorted_vec() {
match ledger.refresh_current_source(&path, modified_at.0) {
Ok(false) => {}
Ok(true) => {
@@ -75,7 +79,13 @@ pub(super) fn detect_recent_sessions(
}
Err(_) => continue,
}
let Ok(Some(summary)) = summarize_session_with_cwd(&path, fallback_cwd.as_deref()) else {
let summary = match record_format {
SessionRecordFormat::Cla => records_cla::summarize_session(&path),
SessionRecordFormat::Cur => {
records_cur::summarize_session(&path, fallback_cwd.as_deref())
}
};
let Ok(Some(summary)) = summary else {
continue;
};
if require_existing_cwd && !summary.migration.cwd.is_dir() {
@@ -2,6 +2,7 @@ use super::common::SessionFileCandidate;
use super::common::detect_recent_sessions;
use crate::model::ExternalAgentSessionImportLimits;
use crate::sessions::ExternalAgentSessionMigration;
use crate::sessions::SessionRecordFormat;
use std::fs;
use std::io;
use std::path::Path;
@@ -42,6 +43,7 @@ pub(crate) fn detect_recent_cur_sessions_with_limits(
candidates.push(SessionFileCandidate {
path,
fallback_cwd: fallback_cwd.clone(),
record_format: SessionRecordFormat::Cur,
});
}
}
@@ -1,7 +1,9 @@
use super::ConversationMessage;
use super::ImportedExternalAgentSession;
use super::MessageRole;
use super::records::read_session_import_with_cwd;
use super::SessionRecordFormat;
use super::records_cla;
use super::records_cur;
use super::summarize_for_label;
use super::title::IMPORTED_SESSION_FALLBACK_TITLE;
use super::title::SessionTitleCandidates;
@@ -26,17 +28,23 @@ const EXTERNAL_SESSION_IMPORTED_MARKER: &str = "<EXTERNAL SESSION IMPORTED>";
#[cfg(test)]
fn load_session_for_import(path: &Path) -> io::Result<Option<ImportedExternalAgentSession>> {
Ok(
load_session_for_import_with_content_sha256(path, /*fallback_cwd*/ None)?
.map(|(session, _content_sha256, _attributed_mcp_server_ids)| session),
)
Ok(load_session_for_import_with_content_sha256(
path,
SessionRecordFormat::Cla,
/*fallback_cwd*/ None,
)?
.map(|(session, _content_sha256, _attributed_mcp_server_ids)| session))
}
pub(crate) fn load_session_for_import_with_content_sha256(
path: &Path,
record_format: SessionRecordFormat,
fallback_cwd: Option<&Path>,
) -> io::Result<Option<(ImportedExternalAgentSession, String, BTreeSet<String>)>> {
let parsed = read_session_import_with_cwd(path, fallback_cwd)?;
let parsed = match record_format {
SessionRecordFormat::Cla => records_cla::read_session_import(path)?,
SessionRecordFormat::Cur => records_cur::read_session_import(path, fallback_cwd)?,
};
let Some(cwd) = parsed.cwd else {
return Ok(None);
};
@@ -2,7 +2,9 @@
mod export;
pub(crate) mod ledger;
pub(crate) mod records;
pub(crate) mod records_cla;
mod records_common;
pub(crate) mod records_cur;
mod title;
use codex_protocol::protocol::RolloutItem;
@@ -21,11 +23,30 @@ pub use ledger::ImportedConnectorCandidate;
pub use ledger::has_current_session_been_imported;
pub use ledger::read_imported_connector_candidates;
pub use ledger::record_completed_session_imports;
pub use records::SessionSummary;
pub use records::summarize_session;
pub use records_cla::summarize_session;
const SESSION_TITLE_MAX_LEN: usize = 120;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum SessionRecordFormat {
Cla,
Cur,
}
pub struct SessionSummary {
pub latest_timestamp: i64,
pub migration: ExternalAgentSessionMigration,
}
struct ParsedSessionImport {
cwd: Option<PathBuf>,
custom_title: Option<String>,
ai_title: Option<String>,
messages: Vec<ConversationMessage>,
content_sha256: String,
attributed_mcp_server_ids: BTreeSet<String>,
}
pub(crate) fn normalized_connector_display_name(name: Option<&str>) -> Option<String> {
name.map(str::trim)
.filter(|name| !name.is_empty())
@@ -93,12 +114,12 @@ fn load_importable_session(
metadata_mode: SessionMetadataMode,
) -> io::Result<Option<PendingSessionImport>> {
let source_path = std::fs::canonicalize(path)?;
let fallback_cwd = match metadata_mode {
SessionMetadataMode::Embedded => None,
SessionMetadataMode::MigrationFallback => Some(fallback_cwd),
let (record_format, fallback_cwd) = match metadata_mode {
SessionMetadataMode::Embedded => (SessionRecordFormat::Cla, None),
SessionMetadataMode::MigrationFallback => (SessionRecordFormat::Cur, Some(fallback_cwd)),
};
let Some((imported_session, source_content_sha256, attributed_mcp_server_ids)) =
load_session_for_import_with_content_sha256(&source_path, fallback_cwd)?
load_session_for_import_with_content_sha256(&source_path, record_format, fallback_cwd)?
else {
return Ok(None);
};
@@ -1,549 +0,0 @@
use super::ConversationMessage;
use super::ExternalAgentSessionMigration;
use super::MessageRole;
use super::title::IMPORTED_SESSION_FALLBACK_TITLE;
use super::title::SessionTitleCandidates;
use super::title::fallback_title_from_user_message;
use super::truncate;
use serde_json::Value as JsonValue;
use sha2::Digest;
use sha2::Sha256;
use std::collections::BTreeSet;
use std::fs::File;
use std::io;
use std::io::BufRead;
use std::io::BufReader;
use std::path::Path;
use std::path::PathBuf;
const NOTE_MAX_LEN: usize = 2_000;
const TOOL_RESULT_MAX_LEN: usize = 4_000;
const EXTERNAL_AGENT_TOOL_CALL_TAG: &str = "external_agent_tool_call";
const EXTERNAL_AGENT_TOOL_RESULT_TAG: &str = "external_agent_tool_result";
pub struct SessionSummary {
pub latest_timestamp: i64,
pub migration: ExternalAgentSessionMigration,
}
pub(super) struct ParsedSessionImport {
pub cwd: Option<PathBuf>,
pub custom_title: Option<String>,
pub ai_title: Option<String>,
pub messages: Vec<ConversationMessage>,
pub content_sha256: String,
pub attributed_mcp_server_ids: BTreeSet<String>,
}
pub fn summarize_session(path: &Path) -> io::Result<Option<SessionSummary>> {
summarize_session_with_cwd(path, /*fallback_cwd*/ None)
}
pub(crate) fn summarize_session_with_cwd(
path: &Path,
fallback_cwd: Option<&Path>,
) -> io::Result<Option<SessionSummary>> {
let file = File::open(path)?;
let fallback_timestamp = fallback_cwd.and_then(|_| file_modified_at_seconds(&file));
let reader = BufReader::new(file);
let mut cwd = None;
let mut custom_title = None;
let mut ai_title = None;
let mut fallback_title = None;
let mut saw_user_message = false;
let mut latest_timestamp = None;
let mut saw_message = false;
for line in reader.lines() {
let line = line?;
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if cwd.is_none() {
cwd = record
.get("cwd")
.and_then(JsonValue::as_str)
.map(PathBuf::from);
}
if let Some(title) = custom_title_from_record(&record) {
custom_title = Some(title.to_string());
}
if let Some(title) = ai_title_from_record(&record) {
ai_title = Some(title.to_string());
}
let Some(message) = conversation_message_from_owned_record(&mut record, fallback_timestamp)
else {
continue;
};
saw_message = true;
if message.role == MessageRole::User {
saw_user_message = true;
if fallback_title.is_none() {
fallback_title = fallback_title_from_user_message(&message.text);
}
}
if let Some(timestamp) = message.timestamp {
latest_timestamp =
Some(latest_timestamp.map_or(timestamp, |current: i64| current.max(timestamp)));
}
}
let Some(cwd) = cwd.or_else(|| fallback_cwd.map(Path::to_path_buf)) else {
return Ok(None);
};
if !saw_message {
return Ok(None);
}
let Some(latest_timestamp) = latest_timestamp else {
return Ok(None);
};
Ok(Some(SessionSummary {
latest_timestamp,
migration: ExternalAgentSessionMigration {
path: path.to_path_buf(),
cwd,
title: SessionTitleCandidates {
custom_title,
ai_title,
fallback_title: fallback_title.or_else(|| {
saw_user_message.then(|| IMPORTED_SESSION_FALLBACK_TITLE.to_string())
}),
}
.select(),
},
}))
}
pub(super) fn read_session_import_with_cwd(
path: &Path,
fallback_cwd: Option<&Path>,
) -> io::Result<ParsedSessionImport> {
let file = File::open(path)?;
let fallback_timestamp = fallback_cwd.and_then(|_| file_modified_at_seconds(&file));
let mut reader = BufReader::new(file);
let mut cwd = None;
let mut custom_title = None;
let mut ai_title = None;
let mut messages = Vec::new();
let mut attributed_mcp_server_ids = BTreeSet::new();
let mut line = String::new();
let mut hasher = Sha256::new();
loop {
line.clear();
if reader.read_line(&mut line)? == 0 {
break;
}
hasher.update(line.as_bytes());
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if let Some(server_id) = record
.get("attributionMcpServer")
.and_then(JsonValue::as_str)
.map(str::trim)
.filter(|server_id| !server_id.is_empty())
{
attributed_mcp_server_ids.insert(server_id.to_string());
}
if cwd.is_none() {
cwd = record
.get("cwd")
.and_then(JsonValue::as_str)
.map(PathBuf::from);
}
if let Some(title) = custom_title_from_record(&record) {
custom_title = Some(title.to_string());
}
if let Some(title) = ai_title_from_record(&record) {
ai_title = Some(title.to_string());
}
if let Some(message) =
conversation_message_from_owned_record(&mut record, fallback_timestamp)
{
messages.push(message);
}
}
Ok(ParsedSessionImport {
cwd: cwd.or_else(|| fallback_cwd.map(Path::to_path_buf)),
custom_title,
ai_title,
messages,
content_sha256: format!("{:x}", hasher.finalize()),
attributed_mcp_server_ids,
})
}
fn custom_title_from_record(record: &JsonValue) -> Option<&str> {
title_from_record(record, "custom-title", "customTitle")
}
fn ai_title_from_record(record: &JsonValue) -> Option<&str> {
title_from_record(record, "ai-title", "aiTitle")
}
fn title_from_record<'a>(record: &'a JsonValue, record_type: &str, field: &str) -> Option<&'a str> {
(record.get("type").and_then(JsonValue::as_str) == Some(record_type))
.then(|| record.get(field).and_then(JsonValue::as_str))
.flatten()
.map(str::trim)
.filter(|title| !title.is_empty())
}
fn conversation_message_from_owned_record(
record: &mut JsonValue,
fallback_timestamp: Option<i64>,
) -> Option<ConversationMessage> {
let record_type = record
.get("type")
.and_then(JsonValue::as_str)
.filter(|record_type| matches!(*record_type, "assistant" | "user"))
.or_else(|| record.get("role").and_then(JsonValue::as_str))?;
if !matches!(record_type, "assistant" | "user") {
return None;
}
if record.get("isMeta").and_then(JsonValue::as_bool) == Some(true)
|| record.get("isSidechain").and_then(JsonValue::as_bool) == Some(true)
{
return None;
}
let is_assistant = record_type == "assistant";
let timestamp = record
.get("timestamp")
.and_then(JsonValue::as_str)
.and_then(parse_timestamp)
.or_else(|| {
record
.get("timestamp_ms")
.and_then(JsonValue::as_i64)
.map(|value| value / 1_000)
})
.or(fallback_timestamp);
let content = record.get_mut("message")?.get_mut("content")?.take();
let extracted = match content {
JsonValue::String(text) => {
if text.trim().is_empty() {
return None;
}
ExtractedMessage {
text,
only_tool_result: false,
}
}
content => extract_message_text(&content)?,
};
let role = if is_assistant || extracted.only_tool_result {
MessageRole::Assistant
} else {
MessageRole::User
};
let text = if role == MessageRole::User {
unwrap_user_query(extracted.text)
} else {
extracted.text
};
Some(ConversationMessage {
role,
text,
timestamp,
})
}
fn unwrap_user_query(text: String) -> String {
let trimmed = text.trim();
let Some(inner) = trimmed
.strip_prefix("<user_query>")
.and_then(|inner| inner.strip_suffix("</user_query>"))
.map(str::trim)
.filter(|inner| !inner.is_empty())
else {
return text;
};
inner.to_string()
}
fn file_modified_at_seconds(file: &File) -> Option<i64> {
file.metadata()
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|duration| i64::try_from(duration.as_secs()).ok())
}
struct ExtractedMessage {
text: String,
only_tool_result: bool,
}
fn extract_message_text(content: &JsonValue) -> Option<ExtractedMessage> {
let blocks = content_blocks(content);
let mut parts = Vec::new();
let mut only_tool_result = !blocks.is_empty();
for block in &blocks {
let block_type = block.get("type").and_then(JsonValue::as_str);
match block_type {
Some("text") => {
if let Some(text) = block.get("text").and_then(JsonValue::as_str)
&& !text.is_empty()
{
parts.push(text.to_string());
only_tool_result = false;
}
}
Some("tool_use") => {
parts.push(tool_call_note(block));
only_tool_result = false;
}
Some("tool_result") => {
parts.push(tool_result_note(block));
}
Some("thinking") => {}
Some(other) => {
parts.push(format!("[external unsupported block: {other}]"));
only_tool_result = false;
}
None => {}
}
}
let text = parts
.into_iter()
.filter(|part| !part.trim().is_empty())
.collect::<Vec<_>>()
.join("\n\n");
if text.is_empty() {
None
} else {
Some(ExtractedMessage {
text,
only_tool_result,
})
}
}
fn content_blocks(content: &JsonValue) -> Vec<JsonValue> {
if let Some(text) = content.as_str() {
return vec![serde_json::json!({
"type": "text",
"text": text,
})];
}
content
.as_array()
.map(|items| {
items
.iter()
.filter(|item| item.is_object())
.cloned()
.collect()
})
.unwrap_or_default()
}
fn tool_call_note(block: &JsonValue) -> String {
let name = block
.get("name")
.and_then(JsonValue::as_str)
.unwrap_or("unknown");
let mut lines = vec![format!("[{EXTERNAL_AGENT_TOOL_CALL_TAG}: {name}]")];
if let Some(input) = block.get("input").and_then(JsonValue::as_object) {
if let Some(description) = input.get("description").and_then(JsonValue::as_str) {
lines.push(format!("description: {description}"));
}
if let Some(command) = input.get("command").and_then(JsonValue::as_str) {
lines.push(format!("command: {command}"));
}
if let Some(file) = input
.get("file_path")
.or_else(|| input.get("file"))
.and_then(JsonValue::as_str)
{
lines.push(format!("file: {file}"));
}
if lines.len() == 1 {
lines.push(format!(
"input: {}",
truncate(&JsonValue::Object(input.clone()).to_string(), NOTE_MAX_LEN)
));
}
} else if let Some(input) = block.get("input") {
lines.push(format!(
"input: {}",
truncate(&input.to_string(), NOTE_MAX_LEN)
));
}
lines.push(format!("[/{EXTERNAL_AGENT_TOOL_CALL_TAG}]"));
lines.join("\n")
}
fn tool_result_note(block: &JsonValue) -> String {
let label = if block.get("is_error").and_then(JsonValue::as_bool) == Some(true) {
format!("[{EXTERNAL_AGENT_TOOL_RESULT_TAG}: error]")
} else {
format!("[{EXTERNAL_AGENT_TOOL_RESULT_TAG}]")
};
let text = tool_result_text(block.get("content"));
if text.is_empty() {
format!("{label}\n[/{EXTERNAL_AGENT_TOOL_RESULT_TAG}]")
} else {
format!(
"{label}\n{}\n[/{EXTERNAL_AGENT_TOOL_RESULT_TAG}]",
truncate(&text, TOOL_RESULT_MAX_LEN)
)
}
}
fn tool_result_text(content: Option<&JsonValue>) -> String {
match content {
Some(JsonValue::String(text)) => text.clone(),
Some(JsonValue::Array(items)) => items
.iter()
.filter_map(|item| item.get("text").and_then(JsonValue::as_str))
.filter(|text| !text.is_empty())
.collect::<Vec<_>>()
.join("\n"),
_ => String::new(),
}
}
fn parse_timestamp(timestamp: &str) -> Option<i64> {
chrono::DateTime::parse_from_rfc3339(timestamp)
.ok()
.map(|value| value.timestamp())
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn reads_session_import_in_one_pass() {
let root = TempDir::new().expect("tempdir");
let path = root.path().join("session.jsonl");
let contents = [
serde_json::json!({
"type": "user",
"cwd": root.path(),
"timestamp": "2026-06-03T12:00:00Z",
"message": { "content": "<user_query>\nfirst request\n</user_query>" },
})
.to_string(),
"not json".to_string(),
serde_json::json!({
"type": "ai-title",
"aiTitle": "generated title",
})
.to_string(),
serde_json::json!({
"type": "custom-title",
"customTitle": "custom title",
})
.to_string(),
]
.join("\n");
std::fs::write(&path, &contents).expect("session");
let parsed =
read_session_import_with_cwd(&path, /*fallback_cwd*/ None).expect("parse session");
assert_eq!(parsed.cwd.as_deref(), Some(root.path()));
assert_eq!(parsed.custom_title.as_deref(), Some("custom title"));
assert_eq!(parsed.ai_title.as_deref(), Some("generated title"));
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].text, "first request");
assert_eq!(
parsed.content_sha256,
format!("{:x}", Sha256::digest(contents))
);
}
#[test]
fn embedded_cwd_overrides_migration_fallback() {
let root = TempDir::new().expect("tempdir");
let embedded_cwd = root.path().join("embedded");
let fallback_cwd = root.path().join("fallback");
let path = root.path().join("session.jsonl");
std::fs::write(
&path,
serde_json::json!({
"cwd": embedded_cwd,
"role": "user",
"message": {"content": "first request"},
})
.to_string(),
)
.expect("session");
let parsed =
read_session_import_with_cwd(&path, Some(&fallback_cwd)).expect("parse session");
let summary = summarize_session_with_cwd(&path, Some(&fallback_cwd))
.expect("summarize session")
.expect("session summary");
assert_eq!(parsed.cwd.as_deref(), Some(embedded_cwd.as_path()));
assert_eq!(summary.migration.cwd, embedded_cwd);
}
#[test]
fn converts_tool_use_blocks_to_bounded_external_agent_tags() {
let block = serde_json::json!({
"type": "tool_use",
"name": "Bash",
"input": {
"description": "Check repo status",
"command": "git status --short"
}
});
assert_eq!(
tool_call_note(&block),
"[external_agent_tool_call: Bash]\n\
description: Check repo status\n\
command: git status --short\n\
[/external_agent_tool_call]"
);
}
#[test]
fn converts_tool_result_blocks_to_bounded_external_agent_tags() {
let block = serde_json::json!({
"type": "tool_result",
"content": "codex-rs/external-agent-migration/src/sessions/records.rs"
});
assert_eq!(
tool_result_note(&block),
"[external_agent_tool_result]\n\
codex-rs/external-agent-migration/src/sessions/records.rs\n\
[/external_agent_tool_result]"
);
}
#[test]
fn converts_error_tool_result_blocks_to_bounded_external_agent_tags() {
let block = serde_json::json!({
"type": "tool_result",
"is_error": true,
"content": "command failed"
});
assert_eq!(
tool_result_note(&block),
"[external_agent_tool_result: error]\n\
command failed\n\
[/external_agent_tool_result]"
);
}
}
@@ -0,0 +1,237 @@
use super::ConversationMessage;
use super::ExternalAgentSessionMigration;
use super::MessageRole;
use super::ParsedSessionImport;
use super::SessionSummary;
use super::records_common::ExtractedMessage;
use super::records_common::extract_message_text;
use super::records_common::parse_timestamp;
use super::title::IMPORTED_SESSION_FALLBACK_TITLE;
use super::title::SessionTitleCandidates;
use super::title::fallback_title_from_user_message;
use serde_json::Value as JsonValue;
use sha2::Digest;
use sha2::Sha256;
use std::collections::BTreeSet;
use std::fs::File;
use std::io;
use std::io::BufRead;
use std::io::BufReader;
use std::path::Path;
use std::path::PathBuf;
pub fn summarize_session(path: &Path) -> io::Result<Option<SessionSummary>> {
let file = File::open(path)?;
let reader = BufReader::new(file);
let mut cwd = None;
let mut custom_title = None;
let mut ai_title = None;
let mut fallback_title = None;
let mut saw_user_message = false;
let mut latest_timestamp = None;
let mut saw_message = false;
for line in reader.lines() {
let line = line?;
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if cwd.is_none() {
cwd = record
.get("cwd")
.and_then(JsonValue::as_str)
.map(PathBuf::from);
}
if let Some(title) = custom_title_from_record(&record) {
custom_title = Some(title.to_string());
}
if let Some(title) = ai_title_from_record(&record) {
ai_title = Some(title.to_string());
}
let Some(message) = conversation_message_from_owned_record(&mut record) else {
continue;
};
saw_message = true;
if message.role == MessageRole::User {
saw_user_message = true;
if fallback_title.is_none() {
fallback_title = fallback_title_from_user_message(&message.text);
}
}
if let Some(timestamp) = message.timestamp {
latest_timestamp =
Some(latest_timestamp.map_or(timestamp, |current: i64| current.max(timestamp)));
}
}
let Some(cwd) = cwd else {
return Ok(None);
};
if !saw_message {
return Ok(None);
}
let Some(latest_timestamp) = latest_timestamp else {
return Ok(None);
};
Ok(Some(SessionSummary {
latest_timestamp,
migration: ExternalAgentSessionMigration {
path: path.to_path_buf(),
cwd,
title: SessionTitleCandidates {
custom_title,
ai_title,
fallback_title: fallback_title.or_else(|| {
saw_user_message.then(|| IMPORTED_SESSION_FALLBACK_TITLE.to_string())
}),
}
.select(),
},
}))
}
pub(super) fn read_session_import(path: &Path) -> io::Result<ParsedSessionImport> {
let file = File::open(path)?;
let mut reader = BufReader::new(file);
let mut cwd = None;
let mut custom_title = None;
let mut ai_title = None;
let mut messages = Vec::new();
let mut attributed_mcp_server_ids = BTreeSet::new();
let mut line = String::new();
let mut hasher = Sha256::new();
loop {
line.clear();
if reader.read_line(&mut line)? == 0 {
break;
}
hasher.update(line.as_bytes());
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if let Some(server_id) = record
.get("attributionMcpServer")
.and_then(JsonValue::as_str)
.map(str::trim)
.filter(|server_id| !server_id.is_empty())
{
attributed_mcp_server_ids.insert(server_id.to_string());
}
if cwd.is_none() {
cwd = record
.get("cwd")
.and_then(JsonValue::as_str)
.map(PathBuf::from);
}
if let Some(title) = custom_title_from_record(&record) {
custom_title = Some(title.to_string());
}
if let Some(title) = ai_title_from_record(&record) {
ai_title = Some(title.to_string());
}
if let Some(message) = conversation_message_from_owned_record(&mut record) {
messages.push(message);
}
}
Ok(ParsedSessionImport {
cwd,
custom_title,
ai_title,
messages,
content_sha256: format!("{:x}", hasher.finalize()),
attributed_mcp_server_ids,
})
}
fn custom_title_from_record(record: &JsonValue) -> Option<&str> {
title_from_record(record, "custom-title", "customTitle")
}
fn ai_title_from_record(record: &JsonValue) -> Option<&str> {
title_from_record(record, "ai-title", "aiTitle")
}
fn title_from_record<'a>(record: &'a JsonValue, record_type: &str, field: &str) -> Option<&'a str> {
(record.get("type").and_then(JsonValue::as_str) == Some(record_type))
.then(|| record.get(field).and_then(JsonValue::as_str))
.flatten()
.map(str::trim)
.filter(|title| !title.is_empty())
}
fn conversation_message_from_owned_record(record: &mut JsonValue) -> Option<ConversationMessage> {
let record_type = record
.get("type")
.and_then(JsonValue::as_str)
.filter(|record_type| matches!(*record_type, "assistant" | "user"))?;
if record.get("isMeta").and_then(JsonValue::as_bool) == Some(true)
|| record.get("isSidechain").and_then(JsonValue::as_bool) == Some(true)
{
return None;
}
let is_assistant = record_type == "assistant";
let timestamp = record
.get("timestamp")
.and_then(JsonValue::as_str)
.and_then(parse_timestamp)
.or_else(|| {
record
.get("timestamp_ms")
.and_then(JsonValue::as_i64)
.map(|value| value / 1_000)
});
let content = record.get_mut("message")?.get_mut("content")?.take();
let extracted = match content {
JsonValue::String(text) => {
if text.trim().is_empty() {
return None;
}
ExtractedMessage {
text,
only_tool_result: false,
}
}
content => extract_message_text(&content)?,
};
let role = if is_assistant || extracted.only_tool_result {
MessageRole::Assistant
} else {
MessageRole::User
};
let text = if role == MessageRole::User {
unwrap_user_query(extracted.text)
} else {
extracted.text
};
Some(ConversationMessage {
role,
text,
timestamp,
})
}
fn unwrap_user_query(text: String) -> String {
let trimmed = text.trim();
let Some(inner) = trimmed
.strip_prefix("<user_query>")
.and_then(|inner| inner.strip_suffix("</user_query>"))
.map(str::trim)
.filter(|inner| !inner.is_empty())
else {
return text;
};
inner.to_string()
}
#[cfg(test)]
#[path = "records_cla_tests.rs"]
mod tests;
@@ -0,0 +1,43 @@
use super::*;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
#[test]
fn reads_session_import_in_one_pass() {
let root = TempDir::new().expect("tempdir");
let path = root.path().join("session.jsonl");
let contents = [
serde_json::json!({
"type": "user",
"cwd": root.path(),
"timestamp": "2026-06-03T12:00:00Z",
"message": { "content": "<user_query>\nfirst request\n</user_query>" },
})
.to_string(),
"not json".to_string(),
serde_json::json!({
"type": "ai-title",
"aiTitle": "generated title",
})
.to_string(),
serde_json::json!({
"type": "custom-title",
"customTitle": "custom title",
})
.to_string(),
]
.join("\n");
std::fs::write(&path, &contents).expect("session");
let parsed = read_session_import(&path).expect("parse session");
assert_eq!(parsed.cwd.as_deref(), Some(root.path()));
assert_eq!(parsed.custom_title.as_deref(), Some("custom title"));
assert_eq!(parsed.ai_title.as_deref(), Some("generated title"));
assert_eq!(parsed.messages.len(), 1);
assert_eq!(parsed.messages[0].text, "first request");
assert_eq!(
parsed.content_sha256,
format!("{:x}", Sha256::digest(contents))
);
}
@@ -0,0 +1,154 @@
use super::truncate;
use serde_json::Value as JsonValue;
const NOTE_MAX_LEN: usize = 2_000;
const TOOL_RESULT_MAX_LEN: usize = 4_000;
const EXTERNAL_AGENT_TOOL_CALL_TAG: &str = "external_agent_tool_call";
const EXTERNAL_AGENT_TOOL_RESULT_TAG: &str = "external_agent_tool_result";
pub(super) struct ExtractedMessage {
pub text: String,
pub only_tool_result: bool,
}
pub(super) fn extract_message_text(content: &JsonValue) -> Option<ExtractedMessage> {
let blocks = content_blocks(content);
let mut parts = Vec::new();
let mut only_tool_result = !blocks.is_empty();
for block in &blocks {
let block_type = block.get("type").and_then(JsonValue::as_str);
match block_type {
Some("text") => {
if let Some(text) = block.get("text").and_then(JsonValue::as_str)
&& !text.is_empty()
{
parts.push(text.to_string());
only_tool_result = false;
}
}
Some("tool_use") => {
parts.push(tool_call_note(block));
only_tool_result = false;
}
Some("tool_result") => {
parts.push(tool_result_note(block));
}
Some("thinking") => {}
Some(other) => {
parts.push(format!("[external unsupported block: {other}]"));
only_tool_result = false;
}
None => {}
}
}
let text = parts
.into_iter()
.filter(|part| !part.trim().is_empty())
.collect::<Vec<_>>()
.join("\n\n");
if text.is_empty() {
None
} else {
Some(ExtractedMessage {
text,
only_tool_result,
})
}
}
pub(super) fn parse_timestamp(timestamp: &str) -> Option<i64> {
chrono::DateTime::parse_from_rfc3339(timestamp)
.ok()
.map(|value| value.timestamp())
}
fn content_blocks(content: &JsonValue) -> Vec<JsonValue> {
if let Some(text) = content.as_str() {
return vec![serde_json::json!({
"type": "text",
"text": text,
})];
}
content
.as_array()
.map(|items| {
items
.iter()
.filter(|item| item.is_object())
.cloned()
.collect()
})
.unwrap_or_default()
}
fn tool_call_note(block: &JsonValue) -> String {
let name = block
.get("name")
.and_then(JsonValue::as_str)
.unwrap_or("unknown");
let mut lines = vec![format!("[{EXTERNAL_AGENT_TOOL_CALL_TAG}: {name}]")];
if let Some(input) = block.get("input").and_then(JsonValue::as_object) {
if let Some(description) = input.get("description").and_then(JsonValue::as_str) {
lines.push(format!("description: {description}"));
}
if let Some(command) = input.get("command").and_then(JsonValue::as_str) {
lines.push(format!("command: {command}"));
}
if let Some(file) = input
.get("file_path")
.or_else(|| input.get("file"))
.and_then(JsonValue::as_str)
{
lines.push(format!("file: {file}"));
}
if lines.len() == 1 {
lines.push(format!(
"input: {}",
truncate(&JsonValue::Object(input.clone()).to_string(), NOTE_MAX_LEN)
));
}
} else if let Some(input) = block.get("input") {
lines.push(format!(
"input: {}",
truncate(&input.to_string(), NOTE_MAX_LEN)
));
}
lines.push(format!("[/{EXTERNAL_AGENT_TOOL_CALL_TAG}]"));
lines.join("\n")
}
fn tool_result_note(block: &JsonValue) -> String {
let label = if block.get("is_error").and_then(JsonValue::as_bool) == Some(true) {
format!("[{EXTERNAL_AGENT_TOOL_RESULT_TAG}: error]")
} else {
format!("[{EXTERNAL_AGENT_TOOL_RESULT_TAG}]")
};
let text = tool_result_text(block.get("content"));
if text.is_empty() {
format!("{label}\n[/{EXTERNAL_AGENT_TOOL_RESULT_TAG}]")
} else {
format!(
"{label}\n{}\n[/{EXTERNAL_AGENT_TOOL_RESULT_TAG}]",
truncate(&text, TOOL_RESULT_MAX_LEN)
)
}
}
fn tool_result_text(content: Option<&JsonValue>) -> String {
match content {
Some(JsonValue::String(text)) => text.clone(),
Some(JsonValue::Array(items)) => items
.iter()
.filter_map(|item| item.get("text").and_then(JsonValue::as_str))
.filter(|text| !text.is_empty())
.collect::<Vec<_>>()
.join("\n"),
_ => String::new(),
}
}
#[cfg(test)]
#[path = "records_common_tests.rs"]
mod tests;
@@ -0,0 +1,53 @@
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn converts_tool_use_blocks_to_bounded_external_agent_tags() {
let block = serde_json::json!({
"type": "tool_use",
"name": "Bash",
"input": {
"description": "Check repo status",
"command": "git status --short"
}
});
assert_eq!(
tool_call_note(&block),
"[external_agent_tool_call: Bash]\n\
description: Check repo status\n\
command: git status --short\n\
[/external_agent_tool_call]"
);
}
#[test]
fn converts_tool_result_blocks_to_bounded_external_agent_tags() {
let block = serde_json::json!({
"type": "tool_result",
"content": "codex-rs/external-agent-migration/src/sessions/records_common.rs"
});
assert_eq!(
tool_result_note(&block),
"[external_agent_tool_result]\n\
codex-rs/external-agent-migration/src/sessions/records_common.rs\n\
[/external_agent_tool_result]"
);
}
#[test]
fn converts_error_tool_result_blocks_to_bounded_external_agent_tags() {
let block = serde_json::json!({
"type": "tool_result",
"is_error": true,
"content": "command failed"
});
assert_eq!(
tool_result_note(&block),
"[external_agent_tool_result: error]\n\
command failed\n\
[/external_agent_tool_result]"
);
}
@@ -0,0 +1,255 @@
use super::ConversationMessage;
use super::ExternalAgentSessionMigration;
use super::MessageRole;
use super::ParsedSessionImport;
use super::SessionSummary;
use super::records_common::ExtractedMessage;
use super::records_common::extract_message_text;
use super::records_common::parse_timestamp;
use super::title::IMPORTED_SESSION_FALLBACK_TITLE;
use super::title::SessionTitleCandidates;
use super::title::fallback_title_from_user_message;
use serde_json::Value as JsonValue;
use sha2::Digest;
use sha2::Sha256;
use std::collections::BTreeSet;
use std::fs::File;
use std::io;
use std::io::BufRead;
use std::io::BufReader;
use std::path::Path;
use std::path::PathBuf;
pub(crate) fn summarize_session(
path: &Path,
fallback_cwd: Option<&Path>,
) -> io::Result<Option<SessionSummary>> {
let file = File::open(path)?;
let fallback_timestamp = fallback_cwd.and_then(|_| file_modified_at_seconds(&file));
let reader = BufReader::new(file);
let mut cwd = None;
let mut fallback_title = None;
let mut saw_user_message = false;
let mut latest_timestamp = None;
let mut saw_message = false;
for line in reader.lines() {
let line = line?;
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if cwd.is_none() {
cwd = record
.get("cwd")
.and_then(JsonValue::as_str)
.map(PathBuf::from);
}
let Some(message) = conversation_message_from_owned_record(&mut record, fallback_timestamp)
else {
continue;
};
saw_message = true;
if message.role == MessageRole::User {
saw_user_message = true;
if fallback_title.is_none() {
fallback_title = fallback_title_from_user_message(&message.text);
}
}
if let Some(timestamp) = message.timestamp {
latest_timestamp =
Some(latest_timestamp.map_or(timestamp, |current: i64| current.max(timestamp)));
}
}
let Some(cwd) = cwd.or_else(|| fallback_cwd.map(Path::to_path_buf)) else {
return Ok(None);
};
if !saw_message {
return Ok(None);
}
let Some(latest_timestamp) = latest_timestamp else {
return Ok(None);
};
Ok(Some(SessionSummary {
latest_timestamp,
migration: ExternalAgentSessionMigration {
path: path.to_path_buf(),
cwd,
title: SessionTitleCandidates {
custom_title: None,
ai_title: None,
fallback_title: fallback_title.or_else(|| {
saw_user_message.then(|| IMPORTED_SESSION_FALLBACK_TITLE.to_string())
}),
}
.select(),
},
}))
}
pub(super) fn read_session_import(
path: &Path,
fallback_cwd: Option<&Path>,
) -> io::Result<ParsedSessionImport> {
let file = File::open(path)?;
let fallback_timestamp = fallback_cwd.and_then(|_| file_modified_at_seconds(&file));
let mut reader = BufReader::new(file);
let mut cwd = None;
let mut messages = Vec::new();
let mut line = String::new();
let mut hasher = Sha256::new();
loop {
line.clear();
if reader.read_line(&mut line)? == 0 {
break;
}
hasher.update(line.as_bytes());
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let Ok(mut record) = serde_json::from_str::<JsonValue>(trimmed) else {
continue;
};
if cwd.is_none() {
cwd = record
.get("cwd")
.and_then(JsonValue::as_str)
.map(PathBuf::from);
}
if let Some(message) =
conversation_message_from_owned_record(&mut record, fallback_timestamp)
{
messages.push(message);
}
}
Ok(ParsedSessionImport {
cwd: cwd.or_else(|| fallback_cwd.map(Path::to_path_buf)),
custom_title: None,
ai_title: None,
messages,
content_sha256: format!("{:x}", hasher.finalize()),
attributed_mcp_server_ids: BTreeSet::new(),
})
}
fn conversation_message_from_owned_record(
record: &mut JsonValue,
fallback_timestamp: Option<i64>,
) -> Option<ConversationMessage> {
let record_type = record.get("role").and_then(JsonValue::as_str)?;
if !matches!(record_type, "assistant" | "user") {
return None;
}
if record.get("isMeta").and_then(JsonValue::as_bool) == Some(true)
|| record.get("isSidechain").and_then(JsonValue::as_bool) == Some(true)
{
return None;
}
let is_assistant = record_type == "assistant";
let timestamp = record
.get("timestamp")
.and_then(JsonValue::as_str)
.and_then(parse_timestamp)
.or_else(|| {
record
.get("timestamp_ms")
.and_then(JsonValue::as_i64)
.map(|value| value / 1_000)
})
.or(fallback_timestamp);
let content = record.get_mut("message")?.get_mut("content")?.take();
let extracted = match content {
JsonValue::String(text) => {
if text.trim().is_empty() {
return None;
}
ExtractedMessage {
text,
only_tool_result: false,
}
}
content => extract_message_text(&content)?,
};
let role = if is_assistant || extracted.only_tool_result {
MessageRole::Assistant
} else {
MessageRole::User
};
let text = if role == MessageRole::User {
unwrap_user_query(extracted.text)
} else {
extracted.text
};
Some(ConversationMessage {
role,
text,
timestamp,
})
}
fn unwrap_user_query(text: String) -> String {
let trimmed = text.trim();
const USER_QUERY_OPEN: &str = "<user_query>";
const USER_QUERY_CLOSE: &str = "</user_query>";
const LEADING_CONTEXT_WRAPPERS: [(&str, &str); 2] = [
("<timestamp>", "</timestamp>"),
("<cursor_commands>", "</cursor_commands>"),
];
let Some(query_start) = trimmed.find(USER_QUERY_OPEN) else {
return text;
};
let query_content_start = query_start + USER_QUERY_OPEN.len();
let Some(query_end_offset) = trimmed[query_content_start..].find(USER_QUERY_CLOSE) else {
return text;
};
let query_end = query_content_start + query_end_offset;
if !trimmed[query_end + USER_QUERY_CLOSE.len()..]
.trim()
.is_empty()
{
return text;
}
let mut leading_context = trimmed[..query_start].trim();
while !leading_context.is_empty() {
let Some((opening, closing)) = LEADING_CONTEXT_WRAPPERS
.iter()
.find(|(opening, _)| leading_context.starts_with(*opening))
else {
return text;
};
let Some(wrapper_end_offset) = leading_context[opening.len()..].find(*closing) else {
return text;
};
let wrapper_end = opening.len() + wrapper_end_offset + closing.len();
leading_context = leading_context[wrapper_end..].trim();
}
let inner = trimmed[query_content_start..query_end].trim();
if inner.is_empty() {
text
} else {
inner.to_string()
}
}
fn file_modified_at_seconds(file: &File) -> Option<i64> {
file.metadata()
.ok()?
.modified()
.ok()?
.duration_since(std::time::UNIX_EPOCH)
.ok()
.and_then(|duration| i64::try_from(duration.as_secs()).ok())
}
#[cfg(test)]
#[path = "records_cur_tests.rs"]
mod tests;
@@ -0,0 +1,76 @@
use super::*;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
#[test]
fn unwraps_user_query_after_context_wrappers() {
let root = TempDir::new().expect("tempdir");
let path = root.path().join("session.jsonl");
let contents = serde_json::json!({
"role": "user",
"message": {
"content": [{
"type": "text",
"text": concat!(
"<cursor_commands>\n",
"--- Cursor Command: verify ---\n",
"Run the verification checklist.\n",
"</cursor_commands>\n",
"<timestamp>Saturday, Jul 25, 2026, 10:27 AM (UTC-7)</timestamp>\n",
"<user_query>\n",
"Verify the project without modifying files.\n",
"</user_query>",
),
}],
},
})
.to_string();
std::fs::write(&path, contents).expect("session");
let parsed = read_session_import(&path, Some(root.path())).expect("parse session");
let summary = summarize_session(&path, Some(root.path()))
.expect("summarize session")
.expect("session summary");
assert_eq!(
parsed.messages[0].text,
"Verify the project without modifying files."
);
assert_eq!(
summary.migration.title.as_deref(),
Some("Verify the project without modifying files.")
);
}
#[test]
fn preserves_user_query_tags_after_unrecognized_context() {
let text = "<user_note>keep this</user_note>\n<user_query>query</user_query>".to_string();
assert_eq!(unwrap_user_query(text.clone()), text);
}
#[test]
fn embedded_cwd_overrides_migration_fallback() {
let root = TempDir::new().expect("tempdir");
let embedded_cwd = root.path().join("embedded");
let fallback_cwd = root.path().join("fallback");
let path = root.path().join("session.jsonl");
std::fs::write(
&path,
serde_json::json!({
"cwd": embedded_cwd,
"role": "user",
"message": {"content": "first request"},
})
.to_string(),
)
.expect("session");
let parsed = read_session_import(&path, Some(&fallback_cwd)).expect("parse session");
let summary = summarize_session(&path, Some(&fallback_cwd))
.expect("summarize session")
.expect("session summary");
assert_eq!(parsed.cwd.as_deref(), Some(embedded_cwd.as_path()));
assert_eq!(summary.migration.cwd, embedded_cwd);
}