diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 51667c601..308e9b24a 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1853,6 +1853,7 @@ dependencies = [ "codex-core-plugins", "codex-device-key", "codex-exec-server", + "codex-external-agent-sessions", "codex-features", "codex-feedback", "codex-file-search", @@ -2682,6 +2683,20 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "codex-external-agent-sessions" +version = "0.0.0" +dependencies = [ + "chrono", + "codex-app-server-protocol", + "codex-protocol", + "codex-utils-output-truncation", + "serde", + "serde_json", + "sha2", + "tempfile", +] + [[package]] name = "codex-features" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index d39651fed..a0434dd6f 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -40,6 +40,7 @@ members = [ "exec-server", "execpolicy", "execpolicy-legacy", + "external-agent-sessions", "keyring-store", "file-search", "linux-sandbox", @@ -146,6 +147,7 @@ codex-exec = { path = "exec" } codex-file-system = { path = "file-system" } codex-exec-server = { path = "exec-server" } codex-execpolicy = { path = "execpolicy" } +codex-external-agent-sessions = { path = "external-agent-sessions" } codex-experimental-api-macros = { path = "codex-experimental-api-macros" } codex-features = { path = "features" } codex-feedback = { path = "feedback" } diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 31cfd542b..e011a930e 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -849,7 +849,8 @@ "CONFIG", "SKILLS", "PLUGINS", - "MCP_SERVER_CONFIG" + "MCP_SERVER_CONFIG", + "SESSIONS" ], "type": "string" }, @@ -1779,15 +1780,20 @@ "MigrationDetails": { "properties": { "plugins": { + "default": [], "items": { "$ref": "#/definitions/PluginsMigration" }, "type": "array" + }, + "sessions": { + "default": [], + "items": { + "$ref": "#/definitions/SessionMigration" + }, + "type": "array" } }, - "required": [ - "plugins" - ], "type": "object" }, "ModeKind": { @@ -3022,6 +3028,27 @@ ], "type": "string" }, + "SessionMigration": { + "properties": { + "cwd": { + "type": "string" + }, + "path": { + "type": "string" + }, + "title": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "cwd", + "path" + ], + "type": "object" + }, "Settings": { "description": "Settings for a collaboration mode.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 1c0ace7fd..db834cb57 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -8326,7 +8326,8 @@ "CONFIG", "SKILLS", "PLUGINS", - "MCP_SERVER_CONFIG" + "MCP_SERVER_CONFIG", + "SESSIONS" ], "type": "string" }, @@ -10772,15 +10773,20 @@ "MigrationDetails": { "properties": { "plugins": { + "default": [], "items": { "$ref": "#/definitions/v2/PluginsMigration" }, "type": "array" + }, + "sessions": { + "default": [], + "items": { + "$ref": "#/definitions/v2/SessionMigration" + }, + "type": "array" } }, - "required": [ - "plugins" - ], "type": "object" }, "ModeKind": { @@ -13545,6 +13551,27 @@ ], "type": "string" }, + "SessionMigration": { + "properties": { + "cwd": { + "type": "string" + }, + "path": { + "type": "string" + }, + "title": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "cwd", + "path" + ], + "type": "object" + }, "SessionSource": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index cf721d82a..826d0da65 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -4845,7 +4845,8 @@ "CONFIG", "SKILLS", "PLUGINS", - "MCP_SERVER_CONFIG" + "MCP_SERVER_CONFIG", + "SESSIONS" ], "type": "string" }, @@ -7446,15 +7447,20 @@ "MigrationDetails": { "properties": { "plugins": { + "default": [], "items": { "$ref": "#/definitions/PluginsMigration" }, "type": "array" + }, + "sessions": { + "default": [], + "items": { + "$ref": "#/definitions/SessionMigration" + }, + "type": "array" } }, - "required": [ - "plugins" - ], "type": "object" }, "ModeKind": { @@ -11431,6 +11437,27 @@ ], "type": "string" }, + "SessionMigration": { + "properties": { + "cwd": { + "type": "string" + }, + "path": { + "type": "string" + }, + "title": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "cwd", + "path" + ], + "type": "object" + }, "SessionSource": { "oneOf": [ { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigDetectResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigDetectResponse.json index ad8f0f9bd..67f5a43a9 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigDetectResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigDetectResponse.json @@ -39,22 +39,28 @@ "CONFIG", "SKILLS", "PLUGINS", - "MCP_SERVER_CONFIG" + "MCP_SERVER_CONFIG", + "SESSIONS" ], "type": "string" }, "MigrationDetails": { "properties": { "plugins": { + "default": [], "items": { "$ref": "#/definitions/PluginsMigration" }, "type": "array" + }, + "sessions": { + "default": [], + "items": { + "$ref": "#/definitions/SessionMigration" + }, + "type": "array" } }, - "required": [ - "plugins" - ], "type": "object" }, "PluginsMigration": { @@ -74,6 +80,27 @@ "pluginNames" ], "type": "object" + }, + "SessionMigration": { + "properties": { + "cwd": { + "type": "string" + }, + "path": { + "type": "string" + }, + "title": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "cwd", + "path" + ], + "type": "object" } }, "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigImportParams.json b/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigImportParams.json index 4b7ac826c..2f6baf14e 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigImportParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ExternalAgentConfigImportParams.json @@ -39,22 +39,28 @@ "CONFIG", "SKILLS", "PLUGINS", - "MCP_SERVER_CONFIG" + "MCP_SERVER_CONFIG", + "SESSIONS" ], "type": "string" }, "MigrationDetails": { "properties": { "plugins": { + "default": [], "items": { "$ref": "#/definitions/PluginsMigration" }, "type": "array" + }, + "sessions": { + "default": [], + "items": { + "$ref": "#/definitions/SessionMigration" + }, + "type": "array" } }, - "required": [ - "plugins" - ], "type": "object" }, "PluginsMigration": { @@ -74,6 +80,27 @@ "pluginNames" ], "type": "object" + }, + "SessionMigration": { + "properties": { + "cwd": { + "type": "string" + }, + "path": { + "type": "string" + }, + "title": { + "type": [ + "string", + "null" + ] + } + }, + "required": [ + "cwd", + "path" + ], + "type": "object" } }, "properties": { diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ExternalAgentConfigMigrationItemType.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ExternalAgentConfigMigrationItemType.ts index dedc124f0..2a272b280 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ExternalAgentConfigMigrationItemType.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ExternalAgentConfigMigrationItemType.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type ExternalAgentConfigMigrationItemType = "AGENTS_MD" | "CONFIG" | "SKILLS" | "PLUGINS" | "MCP_SERVER_CONFIG"; +export type ExternalAgentConfigMigrationItemType = "AGENTS_MD" | "CONFIG" | "SKILLS" | "PLUGINS" | "MCP_SERVER_CONFIG" | "SESSIONS"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/MigrationDetails.ts b/codex-rs/app-server-protocol/schema/typescript/v2/MigrationDetails.ts index 9305335d9..243ffe86e 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/MigrationDetails.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/MigrationDetails.ts @@ -2,5 +2,6 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { PluginsMigration } from "./PluginsMigration"; +import type { SessionMigration } from "./SessionMigration"; -export type MigrationDetails = { plugins: Array, }; +export type MigrationDetails = { plugins: Array, sessions: Array, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/SessionMigration.ts b/codex-rs/app-server-protocol/schema/typescript/v2/SessionMigration.ts new file mode 100644 index 000000000..526af4dd9 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/SessionMigration.ts @@ -0,0 +1,5 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type SessionMigration = { path: string, cwd: string, title: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 0e43b5a4b..3c0178c29 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -294,6 +294,7 @@ export type { SandboxWorkspaceWrite } from "./SandboxWorkspaceWrite"; export type { SendAddCreditsNudgeEmailParams } from "./SendAddCreditsNudgeEmailParams"; export type { SendAddCreditsNudgeEmailResponse } from "./SendAddCreditsNudgeEmailResponse"; export type { ServerRequestResolvedNotification } from "./ServerRequestResolvedNotification"; +export type { SessionMigration } from "./SessionMigration"; export type { SessionSource } from "./SessionSource"; export type { SkillDependencies } from "./SkillDependencies"; export type { SkillErrorInfo } from "./SkillErrorInfo"; diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 64f16d162..2a9b41392 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1091,6 +1091,9 @@ pub enum ExternalAgentConfigMigrationItemType { #[serde(rename = "MCP_SERVER_CONFIG")] #[ts(rename = "MCP_SERVER_CONFIG")] McpServerConfig, + #[serde(rename = "SESSIONS")] + #[ts(rename = "SESSIONS")] + Sessions, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] @@ -1105,11 +1108,23 @@ pub struct PluginsMigration { pub plugin_names: Vec, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct SessionMigration { + pub path: PathBuf, + pub cwd: PathBuf, + pub title: Option, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct MigrationDetails { + #[serde(default)] pub plugins: Vec, + #[serde(default)] + pub sessions: Vec, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] @@ -7841,11 +7856,50 @@ mod tests { marketplace_name: "team-marketplace".to_string(), plugin_names: vec!["asana".to_string()], }], + sessions: Vec::new(), }), } ); } + #[test] + fn external_agent_config_import_params_accept_legacy_plugin_details() { + let params: ExternalAgentConfigImportParams = serde_json::from_value(json!({ + "migrationItems": [{ + "itemType": "PLUGINS", + "description": "Install supported plugins from Claude settings", + "cwd": absolute_path_string("repo"), + "details": { + "plugins": [ + { + "marketplaceName": "team-marketplace", + "pluginNames": ["asana"] + } + ] + } + }] + })) + .expect("legacy plugin import params should deserialize"); + + assert_eq!( + params, + ExternalAgentConfigImportParams { + migration_items: vec![ExternalAgentConfigMigrationItem { + item_type: ExternalAgentConfigMigrationItemType::Plugins, + description: "Install supported plugins from Claude settings".to_string(), + cwd: Some(PathBuf::from(absolute_path_string("repo"))), + details: Some(MigrationDetails { + plugins: vec![PluginsMigration { + marketplace_name: "team-marketplace".to_string(), + plugin_names: vec!["asana".to_string()], + }], + sessions: Vec::new(), + }), + }], + } + ); + } + #[test] fn command_execution_request_approval_rejects_relative_additional_permission_paths() { let err = serde_json::from_value::(json!({ diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index d1721c99d..dbfbfdc5a 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -38,6 +38,7 @@ codex-core = { workspace = true } codex-core-plugins = { workspace = true } codex-device-key = { workspace = true } codex-exec-server = { workspace = true } +codex-external-agent-sessions = { workspace = true } codex-features = { workspace = true } codex-git-utils = { workspace = true } codex-otel = { workspace = true } diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 13f5bb7eb..1ab9c0c78 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -289,6 +289,7 @@ use codex_core_plugins::remote::RemotePluginServiceConfig; use codex_core_plugins::remote::RemotePluginSummary as RemoteCatalogPluginSummary; use codex_exec_server::EnvironmentManager; use codex_exec_server::LOCAL_FS; +use codex_external_agent_sessions::ImportedExternalAgentSession; use codex_features::FEATURES; use codex_features::Feature; use codex_features::Stage; @@ -2448,6 +2449,64 @@ impl CodexMessageProcessor { .spawn(thread_start_task.instrument(request_context.span())); } + pub(crate) async fn import_external_agent_session( + &self, + session: ImportedExternalAgentSession, + ) -> Result { + let ImportedExternalAgentSession { + cwd, + title, + rollout_items, + } = session; + let typesafe_overrides = self.build_thread_config_overrides( + /*model*/ None, + /*model_provider*/ None, + /*service_tier*/ None, + Some(cwd.to_string_lossy().into_owned()), + /*approval_policy*/ None, + /*approvals_reviewer*/ None, + /*sandbox*/ None, + /*permission_profile*/ None, + /*base_instructions*/ None, + /*developer_instructions*/ None, + /*personality*/ None, + ); + let config = self + .config_manager + .load_with_overrides(/*request_overrides*/ None, typesafe_overrides) + .await + .map_err(|err| { + internal_error(format!("failed to load imported session config: {err}")) + })?; + let environments = self + .thread_manager + .default_environment_selections(&config.cwd); + let imported_thread = self + .thread_manager + .start_thread_with_options(StartThreadOptions { + config, + initial_history: InitialHistory::Forked(rollout_items), + session_source: None, + dynamic_tools: Vec::new(), + persist_extended_history: true, + metrics_service_name: None, + parent_trace: None, + environments, + }) + .await + .map_err(|err| internal_error(format!("failed to import session: {err}")))?; + if let Some(title) = title + && let Some(name) = codex_core::util::normalize_thread_name(&title) + { + imported_thread + .thread + .submit(Op::SetThreadName { name }) + .await + .map_err(|err| internal_error(format!("failed to name imported session: {err}")))?; + } + Ok(imported_thread.thread_id) + } + pub(crate) async fn drain_background_tasks(&self) { self.background_tasks.close(); if tokio::time::timeout(Duration::from_secs(10), self.background_tasks.wait()) diff --git a/codex-rs/app-server/src/config/external_agent_config.rs b/codex-rs/app-server/src/config/external_agent_config.rs index 9ecf06bcf..85582e7a2 100644 --- a/codex-rs/app-server/src/config/external_agent_config.rs +++ b/codex-rs/app-server/src/config/external_agent_config.rs @@ -9,6 +9,8 @@ use codex_core_plugins::marketplace::find_marketplace_manifest_path; use codex_core_plugins::marketplace_add::MarketplaceAddRequest; use codex_core_plugins::marketplace_add::add_marketplace; use codex_core_plugins::marketplace_add::is_local_marketplace_source; +use codex_external_agent_sessions::ExternalAgentSessionMigration; +use codex_external_agent_sessions::detect_recent_sessions; use codex_protocol::protocol::Product; use serde_json::Value as JsonValue; use std::collections::BTreeMap; @@ -41,6 +43,7 @@ pub(crate) enum ExternalAgentConfigMigrationItemType { AgentsMd, Plugins, McpServerConfig, + Sessions, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -52,6 +55,7 @@ pub(crate) struct PluginsMigration { #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct MigrationDetails { pub plugins: Vec, + pub sessions: Vec, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -119,6 +123,10 @@ impl ExternalAgentConfigService { Ok(items) } + pub(crate) fn detect_recent_sessions(&self) -> io::Result> { + detect_recent_sessions(&self.external_agent_home, &self.codex_home) + } + pub(crate) async fn import( &self, migration_items: Vec, @@ -175,6 +183,7 @@ impl ExternalAgentConfigService { ); } ExternalAgentConfigMigrationItemType::McpServerConfig => {} + ExternalAgentConfigMigrationItemType::Sessions => {} } } @@ -337,6 +346,29 @@ impl ExternalAgentConfigService { } } + if repo_root.is_none() { + let sessions = detect_recent_sessions(&self.external_agent_home, &self.codex_home)?; + if !sessions.is_empty() { + items.push(ExternalAgentConfigMigrationItem { + item_type: ExternalAgentConfigMigrationItemType::Sessions, + description: format!( + "Migrate recent sessions from {}", + self.external_agent_home.join("projects").display() + ), + cwd: None, + details: Some(MigrationDetails { + plugins: Vec::new(), + sessions, + }), + }); + emit_migration_metric( + EXTERNAL_AGENT_CONFIG_DETECT_METRIC, + ExternalAgentConfigMigrationItemType::Sessions, + /*skills_count*/ None, + ); + } + } + Ok(()) } @@ -413,9 +445,11 @@ impl ExternalAgentConfigService { let local_details = (!local_plugins.is_empty()).then_some(MigrationDetails { plugins: local_plugins, + sessions: Vec::new(), }); let remote_details = (!remote_plugins.is_empty()).then_some(MigrationDetails { plugins: remote_plugins, + sessions: Vec::new(), }); Ok((local_details, remote_details)) @@ -426,7 +460,7 @@ impl ExternalAgentConfigService { cwd: Option<&Path>, details: Option, ) -> io::Result { - let Some(MigrationDetails { plugins }) = details else { + let Some(MigrationDetails { plugins, .. }) = details else { return Err(invalid_data_error( "plugins migration item is missing details".to_string(), )); @@ -694,7 +728,10 @@ fn extract_plugin_migration_details( return None; } - Some(MigrationDetails { plugins }) + Some(MigrationDetails { + plugins, + sessions: Vec::new(), + }) } fn collect_enabled_plugins(settings: &JsonValue) -> Vec { @@ -1156,6 +1193,7 @@ fn migration_metric_tags( ExternalAgentConfigMigrationItemType::AgentsMd => "agents_md", ExternalAgentConfigMigrationItemType::Plugins => "plugins", ExternalAgentConfigMigrationItemType::McpServerConfig => "mcp_server_config", + ExternalAgentConfigMigrationItemType::Sessions => "sessions", }; let mut tags = vec![("migration_type", migration_type.to_string())]; if item_type == ExternalAgentConfigMigrationItemType::Skills { diff --git a/codex-rs/app-server/src/config/external_agent_config_tests.rs b/codex-rs/app-server/src/config/external_agent_config_tests.rs index b900498c2..da0b7a2a9 100644 --- a/codex-rs/app-server/src/config/external_agent_config_tests.rs +++ b/codex-rs/app-server/src/config/external_agent_config_tests.rs @@ -23,6 +23,7 @@ fn github_plugin_details() -> MigrationDetails { marketplace_name: "acme-tools".to_string(), plugin_names: vec!["formatter".to_string()], }], + sessions: Vec::new(), } } @@ -86,6 +87,58 @@ async fn detect_home_lists_config_skills_and_agents_md() { assert_eq!(items, expected); } +#[tokio::test] +async fn detect_home_lists_recent_sessions() { + let (root, external_agent_home, codex_home) = fixture_paths(); + let project_root = root.path().join("repo"); + let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let session_path = external_agent_home + .join("projects") + .join("repo") + .join("session.jsonl"); + fs::create_dir_all(&project_root).expect("create project root"); + fs::create_dir_all(session_path.parent().expect("session parent")).expect("create sessions"); + fs::write( + &session_path, + serde_json::json!({ + "type": "user", + "cwd": &project_root, + "timestamp": &recent_timestamp, + "message": { "content": "first request" }, + }) + .to_string(), + ) + .expect("write session"); + + let items = service_for_paths(external_agent_home.clone(), codex_home) + .detect(ExternalAgentConfigDetectOptions { + include_home: true, + cwds: None, + }) + .await + .expect("detect"); + + assert_eq!( + items, + vec![ExternalAgentConfigMigrationItem { + item_type: ExternalAgentConfigMigrationItemType::Sessions, + description: format!( + "Migrate recent sessions from {}", + external_agent_home.join("projects").display() + ), + cwd: None, + details: Some(MigrationDetails { + plugins: Vec::new(), + sessions: vec![ExternalAgentSessionMigration { + path: session_path, + cwd: project_root, + title: Some("first request".to_string()), + }], + }), + }] + ); +} + #[tokio::test] async fn detect_repo_lists_agents_md_for_each_cwd() { let root = TempDir::new().expect("create tempdir"); @@ -352,6 +405,7 @@ async fn import_local_plugins_returns_completed_status() { marketplace_name: "my-plugins".to_string(), plugin_names: vec!["cloudflare".to_string()], }], + sessions: Vec::new(), }), }]) .await @@ -392,6 +446,7 @@ async fn import_git_plugins_returns_pending_async_status() { marketplace_name: "acme-tools".to_string(), plugin_names: vec!["formatter".to_string()], }], + sessions: Vec::new(), }), }]) .await @@ -406,6 +461,7 @@ async fn import_git_plugins_returns_pending_async_status() { marketplace_name: "acme-tools".to_string(), plugin_names: vec!["formatter".to_string()], }], + sessions: Vec::new(), }, }] ); @@ -659,6 +715,7 @@ async fn detect_home_lists_enabled_plugins_from_settings() { marketplace_name: "acme-tools".to_string(), plugin_names: vec!["deployer".to_string(), "formatter".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -719,6 +776,7 @@ enabled = true marketplace_name: "acme-tools".to_string(), plugin_names: vec!["deployer".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -877,6 +935,7 @@ enabled = true marketplace_name: "acme-tools".to_string(), plugin_names: vec!["formatter".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -1057,6 +1116,7 @@ source = "owner/debug-marketplace" marketplace_name: "debug".to_string(), plugin_names: vec!["available".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -1090,6 +1150,7 @@ async fn import_plugins_requires_source_marketplace_details() { marketplace_name: "other-tools".to_string(), plugin_names: github_plugin_details().plugins[0].plugin_names.clone(), }], + sessions: Vec::new(), }), ) .await @@ -1197,6 +1258,7 @@ async fn import_plugins_supports_external_agent_plugin_marketplace_layout() { marketplace_name: "my-plugins".to_string(), plugin_names: vec!["cloudflare".to_string()], }], + sessions: Vec::new(), }), ) .await @@ -1284,6 +1346,7 @@ async fn detect_home_supports_relative_external_agent_plugin_marketplace_path() marketplace_name: "my-plugins".to_string(), plugin_names: vec!["cloudflare".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -1327,6 +1390,7 @@ async fn detect_home_infers_claude_official_marketplace_when_missing_from_settin marketplace_name: "claude-plugins-official".to_string(), plugin_names: vec!["sample".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -1386,6 +1450,7 @@ async fn import_plugins_supports_relative_external_agent_plugin_marketplace_path marketplace_name: "my-plugins".to_string(), plugin_names: vec!["cloudflare".to_string()], }], + sessions: Vec::new(), }), ) .await @@ -1429,6 +1494,7 @@ async fn import_plugins_infers_claude_official_marketplace_when_missing_from_set marketplace_name: "claude-plugins-official".to_string(), plugin_names: vec!["sample".to_string()], }], + sessions: Vec::new(), }), ) .await @@ -1518,6 +1584,7 @@ async fn detect_repo_supports_project_relative_external_agent_plugin_marketplace marketplace_name: "my-plugins".to_string(), plugin_names: vec!["cloudflare".to_string()], }], + sessions: Vec::new(), }), }] ); @@ -1582,6 +1649,7 @@ async fn import_plugins_supports_project_relative_external_agent_plugin_marketpl marketplace_name: "my-plugins".to_string(), plugin_names: vec!["cloudflare".to_string()], }], + sessions: Vec::new(), }), ) .await diff --git a/codex-rs/app-server/src/external_agent_config_api.rs b/codex-rs/app-server/src/external_agent_config_api.rs index 34ad572ca..30d4cc8f4 100644 --- a/codex-rs/app-server/src/external_agent_config_api.rs +++ b/codex-rs/app-server/src/external_agent_config_api.rs @@ -4,6 +4,7 @@ use crate::config::external_agent_config::ExternalAgentConfigMigrationItemType a use crate::config::external_agent_config::ExternalAgentConfigService; use crate::config::external_agent_config::PendingPluginImport; use crate::error_code::internal_error; +use crate::error_code::invalid_params; use codex_app_server_protocol::ExternalAgentConfigDetectParams; use codex_app_server_protocol::ExternalAgentConfigDetectResponse; use codex_app_server_protocol::ExternalAgentConfigImportParams; @@ -12,17 +13,25 @@ use codex_app_server_protocol::ExternalAgentConfigMigrationItemType; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::MigrationDetails; use codex_app_server_protocol::PluginsMigration; +use codex_external_agent_sessions::ExternalAgentSessionMigration as CoreSessionMigration; +use codex_external_agent_sessions::PendingSessionImport; +use codex_external_agent_sessions::PrepareSessionImportsError; +use codex_external_agent_sessions::prepare_pending_session_imports; +use codex_external_agent_sessions::record_imported_session; +use codex_protocol::ThreadId; use std::path::PathBuf; #[derive(Clone)] pub(crate) struct ExternalAgentConfigApi { + codex_home: PathBuf, migration_service: ExternalAgentConfigService, } impl ExternalAgentConfigApi { pub(crate) fn new(codex_home: PathBuf) -> Self { Self { - migration_service: ExternalAgentConfigService::new(codex_home), + migration_service: ExternalAgentConfigService::new(codex_home.clone()), + codex_home, } } @@ -59,6 +68,9 @@ impl ExternalAgentConfigApi { CoreMigrationItemType::McpServerConfig => { ExternalAgentConfigMigrationItemType::McpServerConfig } + CoreMigrationItemType::Sessions => { + ExternalAgentConfigMigrationItemType::Sessions + } }, description: migration_item.description, cwd: migration_item.cwd, @@ -71,12 +83,79 @@ impl ExternalAgentConfigApi { plugin_names: plugin.plugin_names, }) .collect(), + sessions: details + .sessions + .into_iter() + .map(|session| codex_app_server_protocol::SessionMigration { + path: session.path, + cwd: session.cwd, + title: session.title, + }) + .collect(), }), }) .collect(), }) } + pub(crate) fn detect_recent_sessions( + &self, + ) -> Result, JSONRPCErrorError> { + self.migration_service + .detect_recent_sessions() + .map_err(|err| internal_error(err.to_string())) + } + + pub(crate) fn prepare_pending_session_imports( + &self, + params: &ExternalAgentConfigImportParams, + ) -> Result, JSONRPCErrorError> { + let sessions = params + .migration_items + .iter() + .filter(|item| { + matches!( + item.item_type, + ExternalAgentConfigMigrationItemType::Sessions + ) + }) + .filter_map(|item| item.details.as_ref()) + .flat_map(|details| details.sessions.clone()) + .map(|session| CoreSessionMigration { + path: session.path, + cwd: session.cwd, + title: session.title, + }) + .collect::>(); + let detected_sessions = if sessions.is_empty() { + Vec::new() + } else { + self.detect_recent_sessions()? + }; + prepare_pending_session_imports(&self.codex_home, sessions, detected_sessions).map_err( + |err| match err { + PrepareSessionImportsError::SessionNotDetected(_) => { + invalid_params(err.to_string()) + } + }, + ) + } + + pub(crate) fn record_imported_session( + &self, + source_path: &std::path::Path, + imported_thread_id: ThreadId, + ) { + if let Err(err) = record_imported_session(&self.codex_home, source_path, imported_thread_id) + { + tracing::warn!( + error = %err, + path = %source_path.display(), + "external agent session import ledger update failed" + ); + } + } + pub(crate) async fn import( &self, params: ExternalAgentConfigImportParams, @@ -103,6 +182,9 @@ impl ExternalAgentConfigApi { ExternalAgentConfigMigrationItemType::McpServerConfig => { CoreMigrationItemType::McpServerConfig } + ExternalAgentConfigMigrationItemType::Sessions => { + CoreMigrationItemType::Sessions + } }, description: migration_item.description, cwd: migration_item.cwd, @@ -118,6 +200,15 @@ impl ExternalAgentConfigApi { } }) .collect(), + sessions: details + .sessions + .into_iter() + .map(|session| CoreSessionMigration { + path: session.path, + cwd: session.cwd, + title: session.title, + }) + .collect(), } }), }) diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 071f69eb6..5cb1c1da3 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1138,11 +1138,21 @@ impl MessageProcessor { ExternalAgentConfigMigrationItemType::Plugins ) }); - + let pending_session_imports = self + .external_agent_config_api + .prepare_pending_session_imports(¶ms)?; let pending_plugin_imports = self.external_agent_config_api.import(params).await?; if has_plugin_imports { self.handle_config_mutation().await; } + for pending_session_import in pending_session_imports { + let imported_thread_id = self + .codex_message_processor + .import_external_agent_session(pending_session_import.session) + .await?; + self.external_agent_config_api + .record_imported_session(&pending_session_import.source_path, imported_thread_id); + } self.outgoing .send_response(request_id, ExternalAgentConfigImportResponse {}) .await; diff --git a/codex-rs/app-server/tests/suite/v2/external_agent_config.rs b/codex-rs/app-server/tests/suite/v2/external_agent_config.rs index 6b1715dc7..1f84b3733 100644 --- a/codex-rs/app-server/tests/suite/v2/external_agent_config.rs +++ b/codex-rs/app-server/tests/suite/v2/external_agent_config.rs @@ -2,13 +2,29 @@ use std::time::Duration; use anyhow::Result; use app_test_support::McpProcess; +use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::to_response; +use app_test_support::write_mock_responses_config_toml; +use codex_app_server::INVALID_PARAMS_ERROR_CODE; +use codex_app_server_protocol::ExternalAgentConfigDetectResponse; use codex_app_server_protocol::ExternalAgentConfigImportResponse; +use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PluginListParams; use codex_app_server_protocol::PluginListResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadListParams; +use codex_app_server_protocol::ThreadListResponse; +use codex_app_server_protocol::ThreadReadParams; +use codex_app_server_protocol::ThreadReadResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::UserInput; +use core_test_support::responses; use pretty_assertions::assert_eq; +use std::collections::BTreeMap; use tempfile::TempDir; use tokio::time::timeout; @@ -183,3 +199,522 @@ async fn external_agent_config_import_sends_completion_notification_after_pendin Ok(()) } + +#[tokio::test] +async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("follow-up answer").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let project_root = codex_home.path().join("repo"); + let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let session_dir = codex_home.path().join(".claude/projects/repo"); + let session_path = session_dir.join("session.jsonl"); + std::fs::create_dir_all(&project_root)?; + std::fs::create_dir_all(&session_dir)?; + std::fs::write( + &session_path, + [ + serde_json::json!({ + "type": "user", + "cwd": &project_root, + "timestamp": &recent_timestamp, + "message": { "content": "first request" }, + }) + .to_string(), + serde_json::json!({ + "type": "assistant", + "cwd": &project_root, + "timestamp": &recent_timestamp, + "message": { "content": "first answer" }, + }) + .to_string(), + serde_json::json!({ + "type": "custom-title", + "customTitle": "source session title", + }) + .to_string(), + ] + .join("\n"), + )?; + + let home_dir = codex_home.path().display().to_string(); + let mut mcp = + McpProcess::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_raw_request( + "externalAgentConfig/detect", + Some(serde_json::json!({ + "includeHome": true, + })), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let detected: ExternalAgentConfigDetectResponse = to_response(response)?; + assert_eq!(detected.items.len(), 1); + + let request_id = mcp + .send_raw_request( + "externalAgentConfig/import", + Some(serde_json::json!({ "migrationItems": detected.items })), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ExternalAgentConfigImportResponse = to_response(response)?; + assert_eq!(response, ExternalAgentConfigImportResponse {}); + + let request_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: None, + sort_key: None, + sort_direction: None, + model_providers: None, + source_kinds: None, + archived: None, + cwd: None, + use_state_db_only: false, + search_term: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ThreadListResponse = to_response(response)?; + let thread = response + .data + .first() + .expect("expected imported thread") + .clone(); + assert_eq!(thread.preview, "first request"); + assert_eq!(thread.name.as_deref(), Some("source session title")); + + let request_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id.clone(), + include_turns: true, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ThreadReadResponse = to_response(response)?; + assert_eq!(response.thread.turns.len(), 1); + assert_eq!(response.thread.turns[0].items.len(), 2); + + let request_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + ..Default::default() + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let _: ThreadResumeResponse = to_response(response)?; + + let request_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "follow up".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let request_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id, + include_turns: true, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ThreadReadResponse = to_response(response)?; + assert_eq!(response.thread.turns.len(), 2); + match &response.thread.turns[1].items[1] { + ThreadItem::AgentMessage { text, .. } => assert_eq!(text, "follow-up answer"), + other => panic!("expected agent message item, got {other:?}"), + } + + Ok(()) +} + +#[tokio::test] +async fn external_agent_config_import_skips_already_imported_session_versions() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("unused").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let project_root = codex_home.path().join("repo"); + let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let session_dir = codex_home.path().join(".claude/projects/repo"); + let session_path = session_dir.join("session.jsonl"); + std::fs::create_dir_all(&project_root)?; + std::fs::create_dir_all(&session_dir)?; + std::fs::write( + &session_path, + serde_json::json!({ + "type": "user", + "cwd": &project_root, + "timestamp": &recent_timestamp, + "message": { "content": "first request" }, + }) + .to_string(), + )?; + + let home_dir = codex_home.path().display().to_string(); + let mut mcp = + McpProcess::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_raw_request( + "externalAgentConfig/detect", + Some(serde_json::json!({ "includeHome": true })), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let detected: ExternalAgentConfigDetectResponse = to_response(response)?; + + for _ in 0..2 { + let request_id = mcp + .send_raw_request( + "externalAgentConfig/import", + Some(serde_json::json!({ "migrationItems": detected.items.clone() })), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let _: ExternalAgentConfigImportResponse = to_response(response)?; + } + + let request_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: None, + sort_key: None, + sort_direction: None, + model_providers: None, + source_kinds: None, + archived: None, + cwd: None, + use_state_db_only: false, + search_term: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ThreadListResponse = to_response(response)?; + assert_eq!(response.data.len(), 1); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn external_agent_config_import_rejects_undetected_session_paths() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("unused").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let project_root = codex_home.path().join("repo"); + let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let session_dir = codex_home.path().join(".claude/projects/repo"); + let detected_session_path = session_dir.join("detected.jsonl"); + let undetected_session_path = codex_home.path().join("outside.jsonl"); + std::fs::create_dir_all(&project_root)?; + std::fs::create_dir_all(&session_dir)?; + for path in [&detected_session_path, &undetected_session_path] { + std::fs::write( + path, + format!( + r#"{{"type":"user","cwd":"{}","timestamp":"{}","message":{{"content":"first request"}}}}"#, + project_root.display(), + recent_timestamp + ), + )?; + } + + let home_dir = codex_home.path().display().to_string(); + let mut mcp = + McpProcess::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_raw_request( + "externalAgentConfig/import", + Some(serde_json::json!({ + "migrationItems": [{ + "itemType": "SESSIONS", + "description": "Migrate recent sessions", + "cwd": null, + "details": { + "sessions": [{ + "path": undetected_session_path, + "cwd": project_root, + "title": "first request" + }] + } + }] + })), + ) + .await?; + let err: JSONRPCError = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(err.error.code, INVALID_PARAMS_ERROR_CODE); + assert!( + err.error + .message + .contains("external agent session was not detected for import") + ); + + let request_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: None, + sort_key: None, + sort_direction: None, + model_providers: None, + source_kinds: None, + archived: None, + cwd: None, + use_state_db_only: false, + search_term: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ThreadListResponse = to_response(response)?; + assert_eq!(response.data, Vec::new()); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn external_agent_config_import_compacts_huge_session_before_first_follow_up() -> Result<()> { + let server = responses::start_mock_server().await; + let response_log = responses::mount_sse_sequence( + &server, + vec![ + responses::sse(vec![ + responses::ev_assistant_message("m1", "LOCAL_SUMMARY"), + responses::ev_completed_with_tokens("r1", /*total_tokens*/ 120), + ]), + responses::sse(vec![ + responses::ev_assistant_message("m2", "follow-up answer"), + responses::ev_completed_with_tokens("r2", /*total_tokens*/ 80), + ]), + ], + ) + .await; + + let codex_home = TempDir::new()?; + write_mock_responses_config_toml( + codex_home.path(), + &server.uri(), + &BTreeMap::default(), + /*auto_compact_limit*/ 200, + /*requires_openai_auth*/ None, + "mock_provider", + "Summarize the conversation.", + )?; + + let project_root = codex_home.path().join("repo"); + let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let session_dir = codex_home.path().join(".claude/projects/repo"); + let session_path = session_dir.join("session.jsonl"); + std::fs::create_dir_all(&project_root)?; + std::fs::create_dir_all(&session_dir)?; + let huge_user = "u".repeat(20_000); + let huge_assistant = "a".repeat(20_000); + std::fs::write( + &session_path, + [ + serde_json::json!({ + "type": "user", + "cwd": &project_root, + "timestamp": &recent_timestamp, + "message": { "content": &huge_user }, + }) + .to_string(), + serde_json::json!({ + "type": "assistant", + "cwd": &project_root, + "timestamp": &recent_timestamp, + "message": { "content": &huge_assistant }, + }) + .to_string(), + ] + .join("\n"), + )?; + + let home_dir = codex_home.path().display().to_string(); + let mut mcp = + McpProcess::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_raw_request( + "externalAgentConfig/detect", + Some(serde_json::json!({ + "includeHome": true, + })), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let detected: ExternalAgentConfigDetectResponse = to_response(response)?; + assert_eq!(detected.items.len(), 1); + + let request_id = mcp + .send_raw_request( + "externalAgentConfig/import", + Some(serde_json::json!({ "migrationItems": detected.items })), + ) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let _: ExternalAgentConfigImportResponse = to_response(response)?; + + let request_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: None, + sort_key: None, + sort_direction: None, + model_providers: None, + source_kinds: None, + archived: None, + cwd: None, + use_state_db_only: false, + search_term: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ThreadListResponse = to_response(response)?; + let thread = response + .data + .first() + .expect("expected imported thread") + .clone(); + + let request_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + ..Default::default() + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let _: ThreadResumeResponse = to_response(response)?; + + let request_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "follow up".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let requests = response_log.requests(); + assert_eq!(requests.len(), 2); + let first = requests[0].body_json().to_string(); + let second = requests[1].body_json().to_string(); + assert!(first.contains("Summarize the conversation.")); + assert!(!first.contains("follow up")); + assert!(second.contains("follow up")); + assert!(second.contains("LOCAL_SUMMARY")); + Ok(()) +} + +fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { + std::fs::write( + codex_home.join("config.toml"), + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/external-agent-sessions/BUILD.bazel b/codex-rs/external-agent-sessions/BUILD.bazel new file mode 100644 index 000000000..5a0998339 --- /dev/null +++ b/codex-rs/external-agent-sessions/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "external-agent-sessions", + crate_name = "codex_external_agent_sessions", +) diff --git a/codex-rs/external-agent-sessions/Cargo.toml b/codex-rs/external-agent-sessions/Cargo.toml new file mode 100644 index 000000000..c3639f8e8 --- /dev/null +++ b/codex-rs/external-agent-sessions/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "codex-external-agent-sessions" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lib] +doctest = false +name = "codex_external_agent_sessions" +path = "src/lib.rs" + +[lints] +workspace = true + +[dependencies] +chrono = { workspace = true } +codex-protocol = { workspace = true } +codex-utils-output-truncation = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +sha2 = { workspace = true } + +[dev-dependencies] +codex-app-server-protocol = { workspace = true } +tempfile = { workspace = true } diff --git a/codex-rs/external-agent-sessions/src/detect.rs b/codex-rs/external-agent-sessions/src/detect.rs new file mode 100644 index 000000000..4b7ea9f05 --- /dev/null +++ b/codex-rs/external-agent-sessions/src/detect.rs @@ -0,0 +1,276 @@ +use crate::ExternalAgentSessionMigration; +use crate::ledger::load_import_ledger; +use crate::now_unix_seconds; +use crate::summarize_session; +use std::fs; +use std::io; +use std::path::Path; +use std::time::Duration; + +const SESSION_IMPORT_MAX_COUNT: usize = 50; +const SESSION_IMPORT_MAX_AGE: Duration = Duration::from_secs(30 * 24 * 60 * 60); + +#[derive(Debug)] +struct SessionCandidate { + latest_timestamp: i64, + migration: ExternalAgentSessionMigration, +} + +pub fn detect_recent_sessions( + external_agent_home: &Path, + codex_home: &Path, +) -> io::Result> { + let projects_root = external_agent_home.join("projects"); + if !projects_root.is_dir() { + return Ok(Vec::new()); + } + + let now = now_unix_seconds(); + let ledger = load_import_ledger(codex_home)?; + let mut candidates = Vec::new(); + for project_entry in fs::read_dir(projects_root)? { + let Ok(project_entry) = project_entry else { + continue; + }; + let project_path = project_entry.path(); + if !project_path.is_dir() { + continue; + } + let Ok(entries) = fs::read_dir(project_path) else { + continue; + }; + for entry in entries { + let Ok(entry) = entry else { + continue; + }; + let path = entry.path(); + if path.extension().and_then(|value| value.to_str()) != Some("jsonl") { + continue; + } + let Ok(Some(summary)) = summarize_session(&path) else { + continue; + }; + let Ok(has_been_imported) = ledger.contains_current_source(&path) else { + continue; + }; + if has_been_imported { + continue; + } + if !is_recent_enough(now, summary.latest_timestamp) { + continue; + } + let migration = summary.migration; + if !migration.cwd.is_dir() { + continue; + } + candidates.push(SessionCandidate { + latest_timestamp: summary.latest_timestamp, + migration, + }); + } + } + + candidates.sort_by(|left, right| { + right + .latest_timestamp + .cmp(&left.latest_timestamp) + .then_with(|| left.migration.path.cmp(&right.migration.path)) + }); + candidates.truncate(SESSION_IMPORT_MAX_COUNT); + Ok(candidates + .into_iter() + .map(|candidate| candidate.migration) + .collect()) +} + +fn is_recent_enough(now: i64, latest_timestamp: i64) -> bool { + latest_timestamp >= now.saturating_sub(SESSION_IMPORT_MAX_AGE.as_secs() as i64) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ledger::record_imported_session; + use codex_protocol::ThreadId; + use serde_json::Value as JsonValue; + use std::path::Path; + use tempfile::TempDir; + + #[test] + fn detects_recent_sessions_with_existing_roots() { + let root = TempDir::new().expect("tempdir"); + let external_agent_home = root.path().join(".external"); + let project_root = root.path().join("repo"); + let session_path = write_session( + &external_agent_home, + &project_root, + "session.jsonl", + &[ + record("user", "hello there", project_root.as_path()), + record("assistant", "ack", project_root.as_path()), + ], + ); + + let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); + + assert_eq!( + sessions, + vec![ExternalAgentSessionMigration { + path: session_path, + cwd: project_root, + title: Some("hello there".to_string()), + }] + ); + } + + #[test] + fn prefers_latest_custom_title_over_first_user_message() { + let root = TempDir::new().expect("tempdir"); + let external_agent_home = root.path().join(".external"); + let project_root = root.path().join("repo"); + let session_path = write_session( + &external_agent_home, + &project_root, + "session.jsonl", + &[ + record("user", "hello there", project_root.as_path()), + custom_title_record("first title"), + custom_title_record("final title"), + ], + ); + + let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); + + assert_eq!( + sessions, + vec![ExternalAgentSessionMigration { + path: session_path, + cwd: project_root, + title: Some("final title".to_string()), + }] + ); + } + + #[test] + fn ignores_old_sessions() { + let root = TempDir::new().expect("tempdir"); + let external_agent_home = root.path().join(".external"); + let project_root = root.path().join("repo"); + write_session( + &external_agent_home, + &project_root, + "session.jsonl", + &[record_at( + "user", + "hello", + &project_root, + "2020-01-01T00:00:00Z", + )], + ); + + assert!( + detect_recent_sessions(&external_agent_home, root.path()) + .expect("detect") + .is_empty() + ); + } + + #[test] + fn skips_already_imported_current_session_versions() { + let root = TempDir::new().expect("tempdir"); + let external_agent_home = root.path().join(".external"); + let project_root = root.path().join("repo"); + let session_path = write_session( + &external_agent_home, + &project_root, + "session.jsonl", + &[record("user", "hello there", project_root.as_path())], + ); + + record_imported_session(root.path(), &session_path, ThreadId::new()) + .expect("record import"); + + assert!( + detect_recent_sessions(&external_agent_home, root.path()) + .expect("detect") + .is_empty() + ); + } + + #[test] + fn redetects_sessions_when_source_contents_change_after_import() { + let root = TempDir::new().expect("tempdir"); + let external_agent_home = root.path().join(".external"); + let project_root = root.path().join("repo"); + let session_path = write_session( + &external_agent_home, + &project_root, + "session.jsonl", + &[record("user", "hello there", project_root.as_path())], + ); + record_imported_session(root.path(), &session_path, ThreadId::new()) + .expect("record import"); + + std::fs::write( + &session_path, + jsonl(&[ + record("user", "hello there", project_root.as_path()), + record("assistant", "new reply", project_root.as_path()), + ]), + ) + .expect("update session"); + + let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); + assert_eq!( + sessions, + vec![ExternalAgentSessionMigration { + path: session_path, + cwd: project_root, + title: Some("hello there".to_string()), + }] + ); + } + + fn write_session( + external_agent_home: &Path, + project_root: &Path, + file_name: &str, + records: &[JsonValue], + ) -> std::path::PathBuf { + let projects_dir = external_agent_home.join("projects").join("repo"); + std::fs::create_dir_all(project_root).expect("project root"); + std::fs::create_dir_all(&projects_dir).expect("projects dir"); + let session_path = projects_dir.join(file_name); + std::fs::write(&session_path, jsonl(records)).expect("session"); + session_path + } + + fn record(role: &str, text: &str, cwd: &Path) -> JsonValue { + let timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + record_at(role, text, cwd, ×tamp) + } + + fn record_at(role: &str, text: &str, cwd: &Path, timestamp: &str) -> JsonValue { + serde_json::json!({ + "type": role, + "cwd": cwd, + "timestamp": timestamp, + "message": { "content": text } + }) + } + + fn custom_title_record(title: &str) -> JsonValue { + serde_json::json!({ + "type": "custom-title", + "customTitle": title, + }) + } + + fn jsonl(records: &[JsonValue]) -> String { + records + .iter() + .map(JsonValue::to_string) + .collect::>() + .join("\n") + } +} diff --git a/codex-rs/external-agent-sessions/src/export.rs b/codex-rs/external-agent-sessions/src/export.rs new file mode 100644 index 000000000..b3b793c59 --- /dev/null +++ b/codex-rs/external-agent-sessions/src/export.rs @@ -0,0 +1,297 @@ +use crate::ConversationMessage; +use crate::ImportedExternalAgentSession; +use crate::MessageRole; +use crate::records::conversation_messages; +use crate::records::custom_title_from_records; +use crate::records::project_root_from_records; +use crate::records::read_records; +use crate::summarize_for_label; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::AgentMessageEvent; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::TokenCountEvent; +use codex_protocol::protocol::TokenUsage; +use codex_protocol::protocol::TokenUsageInfo; +use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnStartedEvent; +use codex_protocol::protocol::UserMessageEvent; +use codex_utils_output_truncation::approx_tokens_from_byte_count_i64; +use std::io; +use std::path::Path; + +pub fn load_session_for_import(path: &Path) -> io::Result> { + let records = read_records(path)?; + let Some(cwd) = project_root_from_records(&records) else { + return Ok(None); + }; + let messages = conversation_messages(&records); + let rollout_items = rollout_items_from_messages(&messages); + if rollout_items.is_empty() { + return Ok(None); + } + let title = custom_title_from_records(&records).or_else(|| { + messages + .iter() + .find(|message| message.role == MessageRole::User) + .map(|message| summarize_for_label(&message.text)) + }); + Ok(Some(ImportedExternalAgentSession { + cwd, + title, + rollout_items, + })) +} + +fn rollout_items_from_messages(messages: &[ConversationMessage]) -> Vec { + let mut items = Vec::new(); + let mut response_items = Vec::new(); + let mut current_turn: Option<(String, Option)> = None; + let mut user_turn_count = 0usize; + + for message in messages { + match message.role { + MessageRole::User => { + if let Some((turn_id, last_agent_message)) = current_turn.take() { + items.push(turn_complete_item( + turn_id, + last_agent_message, + /*completed_at*/ None, + )); + } + user_turn_count += 1; + let turn_id = format!("external-import-turn-{user_turn_count}"); + items.push(RolloutItem::EventMsg(EventMsg::TurnStarted( + TurnStartedEvent { + turn_id: turn_id.clone(), + started_at: message.timestamp, + model_context_window: None, + collaboration_mode_kind: Default::default(), + }, + ))); + let response_item = response_item(message); + response_items.push(response_item.clone()); + items.push(RolloutItem::ResponseItem(response_item)); + items.push(RolloutItem::EventMsg(EventMsg::UserMessage( + UserMessageEvent { + message: message.text.clone(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + }, + ))); + current_turn = Some((turn_id, None)); + } + MessageRole::Assistant => { + let Some((_, last_agent_message)) = current_turn.as_mut() else { + continue; + }; + let response_item = response_item(message); + response_items.push(response_item.clone()); + items.push(RolloutItem::ResponseItem(response_item)); + items.push(RolloutItem::EventMsg(EventMsg::AgentMessage( + AgentMessageEvent { + message: message.text.clone(), + phase: None, + memory_citation: None, + }, + ))); + *last_agent_message = Some(message.text.clone()); + } + } + } + + if let Some((turn_id, last_agent_message)) = current_turn { + items.push(token_count_item(&response_items)); + let completed_at = messages.last().and_then(|message| message.timestamp); + items.push(turn_complete_item( + turn_id, + last_agent_message, + completed_at, + )); + } + + items +} + +fn response_item(message: &ConversationMessage) -> ResponseItem { + let content = match message.role { + MessageRole::Assistant => ContentItem::OutputText { + text: message.text.clone(), + }, + MessageRole::User => ContentItem::InputText { + text: message.text.clone(), + }, + }; + ResponseItem::Message { + id: None, + role: match message.role { + MessageRole::Assistant => "assistant".to_string(), + MessageRole::User => "user".to_string(), + }, + content: vec![content], + phase: None, + } +} + +fn token_count_item(response_items: &[ResponseItem]) -> RolloutItem { + let last_model_generated = response_items.iter().rposition( + |item| matches!(item, ResponseItem::Message { role, .. } if role == "assistant"), + ); + let last_model_visible_tokens = last_model_generated + .map(|index| estimate_response_items_token_count(&response_items[..=index])) + .unwrap_or_default(); + let usage = TokenUsage { + total_tokens: last_model_visible_tokens, + ..TokenUsage::default() + }; + RolloutItem::EventMsg(EventMsg::TokenCount(TokenCountEvent { + info: Some(TokenUsageInfo { + total_token_usage: usage.clone(), + last_token_usage: usage, + model_context_window: None, + }), + rate_limits: None, + })) +} + +fn estimate_response_items_token_count(response_items: &[ResponseItem]) -> i64 { + response_items + .iter() + .map(|item| { + serde_json::to_string(item) + .map(|serialized| i64::try_from(serialized.len()).unwrap_or(i64::MAX)) + .map(approx_tokens_from_byte_count_i64) + .unwrap_or_default() + }) + .fold(0i64, i64::saturating_add) +} + +fn turn_complete_item( + turn_id: String, + last_agent_message: Option, + completed_at: Option, +) -> RolloutItem { + RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { + turn_id, + last_agent_message, + completed_at, + duration_ms: None, + time_to_first_token_ms: None, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_app_server_protocol::build_turns_from_rollout_items; + use serde_json::Value as JsonValue; + use std::path::Path; + use tempfile::TempDir; + + #[test] + fn builds_visible_turns_for_imported_history() { + let root = TempDir::new().expect("tempdir"); + let project_root = root.path().join("repo"); + std::fs::create_dir_all(&project_root).expect("project root"); + let path = root.path().join("session.jsonl"); + std::fs::write( + &path, + jsonl(&[ + record("user", "first request", &project_root), + record("assistant", "first answer", &project_root), + record("user", "second request", &project_root), + ]), + ) + .expect("session"); + + let imported = load_session_for_import(&path) + .expect("load") + .expect("session"); + let turns = build_turns_from_rollout_items(&imported.rollout_items); + + assert_eq!(turns.len(), 2); + assert_eq!(turns[0].items.len(), 2); + assert_eq!(turns[1].items.len(), 1); + } + + #[test] + fn loads_custom_title_for_imported_session() { + let root = TempDir::new().expect("tempdir"); + let project_root = root.path().join("repo"); + std::fs::create_dir_all(&project_root).expect("project root"); + let path = root.path().join("session.jsonl"); + std::fs::write( + &path, + jsonl(&[ + record("user", "first request", &project_root), + custom_title_record("named by source app"), + ]), + ) + .expect("session"); + + let imported = load_session_for_import(&path) + .expect("load") + .expect("session"); + + assert_eq!(imported.title.as_deref(), Some("named by source app")); + } + + #[test] + fn emits_token_usage_for_imported_history() { + let root = TempDir::new().expect("tempdir"); + let project_root = root.path().join("repo"); + std::fs::create_dir_all(&project_root).expect("project root"); + let path = root.path().join("session.jsonl"); + std::fs::write( + &path, + jsonl(&[ + record("user", "first request", &project_root), + record("assistant", "first answer", &project_root), + record("user", "second request", &project_root), + ]), + ) + .expect("session"); + + let imported = load_session_for_import(&path) + .expect("load") + .expect("session"); + let token_count = imported + .rollout_items + .iter() + .find_map(|item| match item { + RolloutItem::EventMsg(EventMsg::TokenCount(event)) => event.info.clone(), + _ => None, + }) + .expect("token count event"); + + assert!(token_count.last_token_usage.total_tokens > 0); + assert_eq!(token_count.total_token_usage, token_count.last_token_usage); + } + + fn record(role: &str, text: &str, cwd: &Path) -> JsonValue { + let timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + serde_json::json!({ + "type": role, + "cwd": cwd, + "timestamp": timestamp, + "message": { "content": text } + }) + } + + fn custom_title_record(title: &str) -> JsonValue { + serde_json::json!({ + "type": "custom-title", + "customTitle": title, + }) + } + + fn jsonl(records: &[JsonValue]) -> String { + records + .iter() + .map(JsonValue::to_string) + .collect::>() + .join("\n") + } +} diff --git a/codex-rs/external-agent-sessions/src/ledger.rs b/codex-rs/external-agent-sessions/src/ledger.rs new file mode 100644 index 000000000..84679581c --- /dev/null +++ b/codex-rs/external-agent-sessions/src/ledger.rs @@ -0,0 +1,108 @@ +use crate::now_unix_seconds; +use codex_protocol::ThreadId; +use serde::Deserialize; +use serde::Serialize; +use sha2::Digest; +use sha2::Sha256; +use std::fs; +use std::io; +use std::path::Path; +use std::path::PathBuf; + +const SESSION_IMPORT_LEDGER_FILE: &str = "external_agent_session_imports.json"; + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub(super) struct ImportedExternalAgentSessionLedger { + records: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct ImportedExternalAgentSessionRecord { + source_path: PathBuf, + content_sha256: String, + imported_thread_id: ThreadId, + imported_at: i64, +} + +pub fn has_current_session_been_imported( + codex_home: &Path, + source_path: &Path, +) -> io::Result { + load_import_ledger(codex_home)?.contains_current_source(source_path) +} + +pub fn record_imported_session( + codex_home: &Path, + source_path: &Path, + imported_thread_id: ThreadId, +) -> io::Result<()> { + let mut ledger = load_import_ledger(codex_home)?; + let source_path = canonical_source_path(source_path)?; + let content_sha256 = session_content_sha256(&source_path)?; + if ledger + .records + .iter() + .any(|record| record.source_path == source_path && record.content_sha256 == content_sha256) + { + return Ok(()); + } + ledger.records.push(ImportedExternalAgentSessionRecord { + source_path, + content_sha256, + imported_thread_id, + imported_at: now_unix_seconds(), + }); + save_import_ledger(codex_home, &ledger) +} + +impl ImportedExternalAgentSessionLedger { + pub(super) fn contains_current_source(&self, source_path: &Path) -> io::Result { + let source_path = canonical_source_path(source_path)?; + let content_sha256 = session_content_sha256(&source_path)?; + Ok(self.records.iter().any(|record| { + record.source_path == source_path && record.content_sha256 == content_sha256 + })) + } +} + +pub(super) fn load_import_ledger( + codex_home: &Path, +) -> io::Result { + let path = import_ledger_path(codex_home); + let raw = match fs::read_to_string(path) { + Ok(raw) => raw, + Err(err) if err.kind() == io::ErrorKind::NotFound => { + return Ok(ImportedExternalAgentSessionLedger::default()); + } + Err(err) => return Err(err), + }; + serde_json::from_str(&raw).map_err(|err| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid external agent session import ledger: {err}"), + ) + }) +} + +fn save_import_ledger( + codex_home: &Path, + ledger: &ImportedExternalAgentSessionLedger, +) -> io::Result<()> { + fs::create_dir_all(codex_home)?; + let path = import_ledger_path(codex_home); + let raw = serde_json::to_vec_pretty(ledger).map_err(io::Error::other)?; + fs::write(path, raw) +} + +fn import_ledger_path(codex_home: &Path) -> PathBuf { + codex_home.join(SESSION_IMPORT_LEDGER_FILE) +} + +fn canonical_source_path(path: &Path) -> io::Result { + fs::canonicalize(path) +} + +fn session_content_sha256(path: &Path) -> io::Result { + let contents = fs::read(path)?; + Ok(format!("{:x}", Sha256::digest(contents))) +} diff --git a/codex-rs/external-agent-sessions/src/lib.rs b/codex-rs/external-agent-sessions/src/lib.rs new file mode 100644 index 000000000..49281893d --- /dev/null +++ b/codex-rs/external-agent-sessions/src/lib.rs @@ -0,0 +1,195 @@ +//! Parsing and export helpers for external-agent session histories. + +mod detect; +mod export; +mod ledger; +mod records; + +use codex_protocol::protocol::RolloutItem; +use std::collections::HashSet; +use std::io; +use std::path::Path; +use std::path::PathBuf; + +pub use detect::detect_recent_sessions; +pub use export::load_session_for_import; +pub use ledger::has_current_session_been_imported; +pub use ledger::record_imported_session; +pub use records::SessionSummary; +pub use records::summarize_session; + +const SESSION_TITLE_MAX_LEN: usize = 120; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ExternalAgentSessionMigration { + pub path: PathBuf, + pub cwd: PathBuf, + pub title: Option, +} + +#[derive(Debug, Clone)] +pub struct ImportedExternalAgentSession { + pub cwd: PathBuf, + pub title: Option, + pub rollout_items: Vec, +} + +#[derive(Debug, Clone)] +pub struct PendingSessionImport { + pub source_path: PathBuf, + pub session: ImportedExternalAgentSession, +} + +#[derive(Debug)] +pub enum PrepareSessionImportsError { + SessionNotDetected(PathBuf), +} + +impl std::fmt::Display for PrepareSessionImportsError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + PrepareSessionImportsError::SessionNotDetected(path) => { + write!( + formatter, + "external agent session was not detected for import: {}", + path.display() + ) + } + } + } +} + +impl std::error::Error for PrepareSessionImportsError {} + +pub fn prepare_pending_session_imports( + codex_home: &Path, + requested_sessions: Vec, + detected_sessions: Vec, +) -> Result, PrepareSessionImportsError> { + let detected_session_paths = detected_sessions + .into_iter() + .map(|session| session.path) + .collect::>(); + let mut pending_session_imports = Vec::new(); + for session in requested_sessions { + let has_been_imported = match has_current_session_been_imported(codex_home, &session.path) { + Ok(has_been_imported) => has_been_imported, + Err(_) => continue, + }; + if !detected_session_paths.contains(&session.path) && !has_been_imported { + return Err(PrepareSessionImportsError::SessionNotDetected(session.path)); + } + if has_been_imported { + continue; + } + let imported_session = match load_importable_session(&session.path) { + Ok(Some(imported_session)) => imported_session, + Ok(None) | Err(_) => continue, + }; + pending_session_imports.push(PendingSessionImport { + source_path: session.path, + session: imported_session, + }); + } + Ok(pending_session_imports) +} + +fn load_importable_session(path: &Path) -> io::Result> { + let Some(imported_session) = load_session_for_import(path)? else { + return Ok(None); + }; + Ok(imported_session.cwd.is_dir().then_some(imported_session)) +} + +#[derive(Debug, Clone)] +struct ConversationMessage { + role: MessageRole, + text: String, + timestamp: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum MessageRole { + Assistant, + User, +} + +fn summarize_for_label(text: &str) -> String { + let first_line = text.lines().next().unwrap_or_default().trim(); + truncate(first_line, SESSION_TITLE_MAX_LEN) +} + +fn truncate(text: &str, max_len: usize) -> String { + if text.chars().count() <= max_len { + return text.to_string(); + } + let prefix = text + .chars() + .take(max_len.saturating_sub(3)) + .collect::(); + format!("{prefix}...") +} + +fn now_unix_seconds() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_secs() as i64) + .unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_protocol::ThreadId; + use tempfile::TempDir; + + #[test] + fn rejects_session_that_was_not_detected() { + let root = TempDir::new().expect("tempdir"); + let codex_home = root.path().join("codex-home"); + let source_path = root.path().join("session.jsonl"); + std::fs::write(&source_path, "{}\n").expect("session"); + + let err = prepare_pending_session_imports( + &codex_home, + vec![session_migration(&source_path)], + Vec::new(), + ) + .expect_err("undetected session should be rejected"); + + match err { + PrepareSessionImportsError::SessionNotDetected(path) => { + assert_eq!(path, source_path); + } + } + } + + #[test] + fn skips_session_that_was_already_imported() { + let root = TempDir::new().expect("tempdir"); + let codex_home = root.path().join("codex-home"); + let source_path = root.path().join("session.jsonl"); + std::fs::write(&source_path, "{}\n").expect("session"); + record_imported_session(&codex_home, &source_path, ThreadId::new()).expect("record import"); + + let pending = prepare_pending_session_imports( + &codex_home, + vec![session_migration(&source_path)], + Vec::new(), + ) + .expect("already imported session should be skipped"); + + assert!(pending.is_empty()); + } + + fn session_migration(path: &Path) -> ExternalAgentSessionMigration { + ExternalAgentSessionMigration { + path: path.to_path_buf(), + cwd: path + .parent() + .expect("source path should have parent") + .to_path_buf(), + title: None, + } + } +} diff --git a/codex-rs/external-agent-sessions/src/records.rs b/codex-rs/external-agent-sessions/src/records.rs new file mode 100644 index 000000000..93eb07feb --- /dev/null +++ b/codex-rs/external-agent-sessions/src/records.rs @@ -0,0 +1,296 @@ +use crate::ConversationMessage; +use crate::ExternalAgentSessionMigration; +use crate::MessageRole; +use crate::summarize_for_label; +use crate::truncate; +use serde_json::Value as JsonValue; +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; + +pub struct SessionSummary { + pub latest_timestamp: i64, + pub migration: ExternalAgentSessionMigration, +} + +pub fn summarize_session(path: &Path) -> io::Result> { + let file = File::open(path)?; + let reader = BufReader::new(file); + let mut cwd = None; + let mut custom_title = None; + let mut title = None; + 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(record) = serde_json::from_str::(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()); + } + let Some(message) = conversation_message_from_record(&record) else { + continue; + }; + saw_message = true; + if title.is_none() && message.role == MessageRole::User { + title = Some(summarize_for_label(&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: custom_title.or(title), + }, + })) +} + +pub(super) fn custom_title_from_records(records: &[JsonValue]) -> Option { + records + .iter() + .filter_map(custom_title_from_record) + .next_back() + .map(ToOwned::to_owned) +} + +pub(super) fn read_records(path: &Path) -> io::Result> { + let file = File::open(path)?; + let reader = BufReader::new(file); + let mut records = Vec::new(); + for line in reader.lines() { + let line = line?; + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let Ok(value) = serde_json::from_str::(trimmed) else { + continue; + }; + if value.is_object() { + records.push(value); + } + } + Ok(records) +} + +pub(super) fn project_root_from_records(records: &[JsonValue]) -> Option { + records + .iter() + .find_map(|record| record.get("cwd").and_then(JsonValue::as_str)) + .map(PathBuf::from) +} + +pub(super) fn conversation_messages(records: &[JsonValue]) -> Vec { + records + .iter() + .filter_map(conversation_message_from_record) + .collect() +} + +fn custom_title_from_record(record: &JsonValue) -> Option<&str> { + (record.get("type").and_then(JsonValue::as_str) == Some("custom-title")) + .then(|| record.get("customTitle").and_then(JsonValue::as_str)) + .flatten() + .map(str::trim) + .filter(|title| !title.is_empty()) +} + +fn conversation_message_from_record(record: &JsonValue) -> Option { + let record_type = record.get("type")?.as_str()?; + if record_type != "assistant" && record_type != "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 extracted = extract_message_text(record.get("message")?.get("content")?)?; + let role = if record_type == "assistant" || extracted.only_tool_result { + MessageRole::Assistant + } else { + MessageRole::User + }; + let timestamp = record + .get("timestamp") + .and_then(JsonValue::as_str) + .and_then(parse_timestamp); + Some(ConversationMessage { + role, + text: extracted.text, + timestamp, + }) +} + +struct ExtractedMessage { + text: String, + only_tool_result: bool, +} + +fn extract_message_text(content: &JsonValue) -> Option { + 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::>() + .join("\n\n"); + if text.is_empty() { + None + } else { + Some(ExtractedMessage { + text, + only_tool_result, + }) + } +} + +fn content_blocks(content: &JsonValue) -> Vec { + 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 tool call: {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.join("\n") +} + +fn tool_result_note(block: &JsonValue) -> String { + let label = if block.get("is_error").and_then(JsonValue::as_bool) == Some(true) { + "[external tool result: error]" + } else { + "[external tool result]" + }; + let text = tool_result_text(block.get("content")); + if text.is_empty() { + label.to_string() + } else { + format!("{label}\n{}", 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::>() + .join("\n"), + _ => String::new(), + } +} + +fn parse_timestamp(timestamp: &str) -> Option { + chrono::DateTime::parse_from_rfc3339(timestamp) + .ok() + .map(|value| value.timestamp()) +} diff --git a/codex-rs/tui/src/external_agent_config_migration.rs b/codex-rs/tui/src/external_agent_config_migration.rs index 0e709f945..3046168a8 100644 --- a/codex-rs/tui/src/external_agent_config_migration.rs +++ b/codex-rs/tui/src/external_agent_config_migration.rs @@ -799,6 +799,7 @@ mod tests { plugin_names: vec!["warehouse".to_string()], }, ], + sessions: Vec::new(), } }