From 1f03e8a449d987bc8239550e4c5fdc95182e9942 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 14:12:34 +0100 Subject: [PATCH] Move Epiphany state host seam to app-server --- epiphany-codex-bridge/src/mutation_service.rs | 20 ++++- epiphany-codex-bridge/src/state.rs | 76 +++++-------------- .../app-server/src/codex_message_processor.rs | 5 +- .../epiphany_automation.rs | 2 +- .../epiphany_thread_host.rs | 53 +++++++++++++ 5 files changed, 97 insertions(+), 59 deletions(-) create mode 100644 vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_thread_host.rs diff --git a/epiphany-codex-bridge/src/mutation_service.rs b/epiphany-codex-bridge/src/mutation_service.rs index c8844cbe..bd3cdc34 100644 --- a/epiphany-codex-bridge/src/mutation_service.rs +++ b/epiphany-codex-bridge/src/mutation_service.rs @@ -56,7 +56,8 @@ use crate::reorient::map_epiphany_freshness; use crate::reorient::map_epiphany_reorient; use crate::runtime_results::load_completed_epiphany_reorient_finding; use crate::runtime_results::load_completed_epiphany_role_finding; -use crate::state::client_visible_live_thread_epiphany_state; +use crate::state::client_visible_epiphany_state_for_paths; +use crate::state::thread_state_mirror_id_from_rollout_path; use uuid::Uuid; #[derive(Debug, Clone)] @@ -655,3 +656,20 @@ fn validate_expected_revision( } Ok(()) } + +async fn client_visible_live_thread_epiphany_state( + thread: &CodexThread, + fallback: EpiphanyThreadState, +) -> EpiphanyThreadState { + let config = thread.config_snapshot().await; + let codex_home = thread.codex_home().await; + let rollout_path = thread.rollout_path(); + let mirror_thread_id = thread_state_mirror_id_from_rollout_path(rollout_path.as_deref()); + client_visible_epiphany_state_for_paths( + fallback, + config.cwd.as_path(), + codex_home, + Some(mirror_thread_id.as_str()), + ) + .await +} diff --git a/epiphany-codex-bridge/src/state.rs b/epiphany-codex-bridge/src/state.rs index 421290a9..ef8305a6 100644 --- a/epiphany-codex-bridge/src/state.rs +++ b/epiphany-codex-bridge/src/state.rs @@ -2,60 +2,29 @@ use std::fs; use std::path::Path; use std::path::PathBuf; -use codex_core::CodexThread; -use codex_core::RolloutRecorder; -use codex_core::latest_epiphany_state_from_rollout_items; use codex_protocol::protocol::EpiphanyThreadState; -use codex_protocol::protocol::InitialHistory; use epiphany_core::write_thread_state; use tracing::warn; use crate::retrieve::epiphany_retrieval_state_for_paths; -pub async fn load_epiphany_state_from_rollout_path( - rollout_path: &Path, -) -> std::result::Result, String> { - let items = match RolloutRecorder::get_rollout_history(rollout_path) - .await - .map_err(|err| { - format!( - "failed to load rollout `{}` for Epiphany state: {err}", - rollout_path.display() - ) - })? { - InitialHistory::New | InitialHistory::Cleared => Vec::new(), - InitialHistory::Forked(items) => items, - InitialHistory::Resumed(resumed) => resumed.history, - }; - Ok(latest_epiphany_state_from_rollout_items(&items)) -} - -pub async fn live_thread_epiphany_state(thread: &CodexThread) -> Option { - let mut epiphany_state = thread.epiphany_state().await; - if let Some(state) = epiphany_state.as_mut() - && state.retrieval.is_none() - { - let config = thread.config_snapshot().await; - let codex_home = thread.codex_home().await; - state.retrieval = - Some(epiphany_retrieval_state_for_paths(config.cwd.to_path_buf(), codex_home).await); - } - if let Some(state) = epiphany_state.as_ref() { - let config = thread.config_snapshot().await; - let thread_id = thread_state_mirror_id(thread); - if let Err(err) = mirror_thread_state_to_workspace(config.cwd.as_path(), &thread_id, state) - { - warn!("failed to mirror Epiphany thread state into native store: {err}"); - } - } - epiphany_state -} - -pub async fn client_visible_live_thread_epiphany_state( - thread: &CodexThread, - fallback: EpiphanyThreadState, +pub async fn client_visible_epiphany_state_for_paths( + mut state: EpiphanyThreadState, + workspace_root: &Path, + codex_home: PathBuf, + mirror_thread_id: Option<&str>, ) -> EpiphanyThreadState { - live_thread_epiphany_state(thread).await.unwrap_or(fallback) + if state.retrieval.is_none() { + state.retrieval = Some( + epiphany_retrieval_state_for_paths(workspace_root.to_path_buf(), codex_home).await, + ); + } + if let Some(thread_id) = mirror_thread_id + && let Err(err) = mirror_thread_state_to_workspace(workspace_root, thread_id, &state) + { + warn!("failed to mirror Epiphany thread state into native store: {err}"); + } + state } pub fn thread_state_store_path(workspace_root: &Path) -> PathBuf { @@ -75,13 +44,10 @@ pub fn mirror_thread_state_to_workspace( Ok(store_path) } -fn thread_state_mirror_id(thread: &CodexThread) -> String { - thread - .rollout_path() - .and_then(|path| { - path.file_stem() - .and_then(|stem| stem.to_str()) - .map(str::to_string) - }) +pub fn thread_state_mirror_id_from_rollout_path(rollout_path: Option<&Path>) -> String { + rollout_path + .and_then(|path| path.file_stem()) + .and_then(|stem| stem.to_str()) + .map(str::to_string) .unwrap_or_else(|| "live-thread".to_string()) } diff --git a/vendor/codex/codex-rs/app-server/src/codex_message_processor.rs b/vendor/codex/codex-rs/app-server/src/codex_message_processor.rs index 16f8e1e3..36161301 100644 --- a/vendor/codex/codex-rs/app-server/src/codex_message_processor.rs +++ b/vendor/codex/codex-rs/app-server/src/codex_message_processor.rs @@ -4,6 +4,7 @@ mod command_routes; mod epiphany_automation; mod epiphany_mutation_routes; mod epiphany_read_routes; +mod epiphany_thread_host; mod listener_lifecycle; mod mcp_routes; mod realtime_routes; @@ -25,6 +26,8 @@ mod turn_routes; use self::auth_routes::ActiveLogin; pub(crate) use self::epiphany_automation::maybe_run_epiphany_coordinator_automation_for_turn_boundary; pub(crate) use self::epiphany_automation::maybe_run_epiphany_pre_compaction_checkpoint_intervention_for_token_count; +use self::epiphany_thread_host::live_thread_epiphany_state; +use self::epiphany_thread_host::load_epiphany_state_from_rollout_path; pub(super) use self::listener_lifecycle::{EnsureConversationListenerResult, ListenerTaskContext}; pub(crate) use self::running_thread_resume_routes::handle_pending_thread_resume_request; pub(crate) use self::thread_config::{config_load_error, validate_dynamic_tools}; @@ -340,8 +343,6 @@ use codex_thread_store::UpdateThreadMetadataParams as StoreUpdateThreadMetadataP use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP; use epiphany_codex_bridge::invalidation::EpiphanyInvalidationManager; -use epiphany_codex_bridge::state::live_thread_epiphany_state; -use epiphany_codex_bridge::state::load_epiphany_state_from_rollout_path; use std::collections::HashMap; use std::collections::HashSet; use std::io::Error as IoError; diff --git a/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_automation.rs b/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_automation.rs index de3f0ee1..0cd1765b 100644 --- a/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_automation.rs +++ b/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_automation.rs @@ -20,10 +20,10 @@ use epiphany_codex_bridge::pressure::map_epiphany_pressure; use epiphany_codex_bridge::pressure::render_epiphany_pre_compaction_checkpoint_intervention; use epiphany_codex_bridge::pressure::should_run_epiphany_pre_compaction_checkpoint_intervention; use epiphany_codex_bridge::retrieve::epiphany_retrieval_state_for_paths; -use epiphany_codex_bridge::state::client_visible_live_thread_epiphany_state; use tokio::sync::Mutex; use tracing::warn; +use super::epiphany_thread_host::client_visible_live_thread_epiphany_state; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; use crate::thread_state::ThreadState; diff --git a/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_thread_host.rs b/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_thread_host.rs new file mode 100644 index 00000000..d1e780ba --- /dev/null +++ b/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_thread_host.rs @@ -0,0 +1,53 @@ +use std::path::Path; + +use codex_core::CodexThread; +use codex_core::RolloutRecorder; +use codex_core::latest_epiphany_state_from_rollout_items; +use codex_protocol::protocol::EpiphanyThreadState; +use codex_protocol::protocol::InitialHistory; +use epiphany_codex_bridge::state::client_visible_epiphany_state_for_paths; +use epiphany_codex_bridge::state::thread_state_mirror_id_from_rollout_path; + +pub(super) async fn load_epiphany_state_from_rollout_path( + rollout_path: &Path, +) -> std::result::Result, String> { + let items = match RolloutRecorder::get_rollout_history(rollout_path) + .await + .map_err(|err| { + format!( + "failed to load rollout `{}` for Epiphany state: {err}", + rollout_path.display() + ) + })? { + InitialHistory::New | InitialHistory::Cleared => Vec::new(), + InitialHistory::Forked(items) => items, + InitialHistory::Resumed(resumed) => resumed.history, + }; + Ok(latest_epiphany_state_from_rollout_items(&items)) +} + +pub(super) async fn live_thread_epiphany_state( + thread: &CodexThread, +) -> Option { + let state = thread.epiphany_state().await?; + let config = thread.config_snapshot().await; + let codex_home = thread.codex_home().await; + let rollout_path = thread.rollout_path(); + let mirror_thread_id = thread_state_mirror_id_from_rollout_path(rollout_path.as_deref()); + Some( + client_visible_epiphany_state_for_paths( + state, + config.cwd.as_path(), + codex_home, + Some(mirror_thread_id.as_str()), + ) + .await, + ) +} + +pub(super) async fn client_visible_live_thread_epiphany_state( + thread: &CodexThread, + fallback: EpiphanyThreadState, +) -> EpiphanyThreadState { + live_thread_epiphany_state(thread).await.unwrap_or(fallback) +}