Move Epiphany state host seam to app-server
This commit is contained in:
parent
c04c877d15
commit
1f03e8a449
5 changed files with 96 additions and 58 deletions
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Option<EpiphanyThreadState>, 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<EpiphanyThreadState> {
|
||||
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())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
53
vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_thread_host.rs
vendored
Normal file
53
vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_thread_host.rs
vendored
Normal file
|
|
@ -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<Option<EpiphanyThreadState>, 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<EpiphanyThreadState> {
|
||||
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)
|
||||
}
|
||||
Reference in a new issue