From e4052d557f6961b023ee2cbf2e0be8243ff01647 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 16:42:58 +0100 Subject: [PATCH] Move job mutation surfaces to typed state --- epiphany-codex-bridge/src/cultnet.rs | 3 + epiphany-codex-bridge/src/jobs.rs | 42 +++++++------ epiphany-codex-bridge/src/mutation_service.rs | 24 ++++--- epiphany-core/src/surfaces/jobs.rs | 18 +++++- .../epiphany_mutation_routes.rs | 63 +++++++++++++++++-- 5 files changed, 116 insertions(+), 34 deletions(-) diff --git a/epiphany-codex-bridge/src/cultnet.rs b/epiphany-codex-bridge/src/cultnet.rs index fe94435c..24ac2497 100644 --- a/epiphany-codex-bridge/src/cultnet.rs +++ b/epiphany-codex-bridge/src/cultnet.rs @@ -3,7 +3,10 @@ use epiphany_core::EpiphanyInvalidationInput; use epiphany_core::EpiphanyRetrievalFreshness; pub use epiphany_core::EpiphanyGraphFreshnessStatus; pub use epiphany_core::EpiphanyInvalidationStatus; +pub use epiphany_core::EpiphanyJobStatus; +pub use epiphany_core::EpiphanyJobView; pub use epiphany_core::EpiphanyRetrievalFreshnessStatus; +pub use epiphany_state_model::EpiphanyJobKind; use serde::Deserialize; use serde::Serialize; diff --git a/epiphany-codex-bridge/src/jobs.rs b/epiphany-codex-bridge/src/jobs.rs index 2d82af40..7a1c6de0 100644 --- a/epiphany-codex-bridge/src/jobs.rs +++ b/epiphany-codex-bridge/src/jobs.rs @@ -9,14 +9,21 @@ use epiphany_state_model::EpiphanyJobKind as CoreEpiphanyJobKind; use epiphany_state_model::EpiphanyRetrievalState; use epiphany_state_model::EpiphanyThreadState; -pub fn map_epiphany_jobs( +pub fn derive_epiphany_jobs( state: Option<&EpiphanyThreadState>, retrieval_override: Option<&EpiphanyRetrievalState>, -) -> Vec { +) -> Vec { derive_jobs(EpiphanyJobsInput { state, retrieval_override, }) +} + +pub fn map_epiphany_jobs( + state: Option<&EpiphanyThreadState>, + retrieval_override: Option<&EpiphanyRetrievalState>, +) -> Vec { + derive_epiphany_jobs(state, retrieval_override) .into_iter() .map(map_core_epiphany_job_view) .collect() @@ -55,23 +62,21 @@ pub fn map_core_epiphany_job_kind(kind: CoreEpiphanyJobKind) -> ThreadEpiphanyJo pub fn map_launched_epiphany_job( state: &EpiphanyThreadState, binding_id: &str, - launcher_job_id: &str, backend_job_id: &str, fallback_kind: CoreEpiphanyJobKind, fallback_scope: &str, -) -> ThreadEpiphanyJob { - map_epiphany_jobs(Some(state), None) +) -> EpiphanyJobView { + derive_epiphany_jobs(Some(state), None) .into_iter() .find(|job| job.id == binding_id) - .unwrap_or_else(|| ThreadEpiphanyJob { + .unwrap_or_else(|| EpiphanyJobView { id: binding_id.to_string(), - kind: map_core_epiphany_job_kind(fallback_kind), + kind: fallback_kind, scope: fallback_scope.to_string(), owner_role: "epiphany-harness".to_string(), - launcher_job_id: Some(launcher_job_id.to_string()), authority_scope: None, - backend_job_id: Some(backend_job_id.to_string()), - status: ThreadEpiphanyJobStatus::Pending, + runtime_job_id: Some(backend_job_id.to_string()), + status: CoreEpiphanyJobStatus::Pending, items_processed: None, items_total: None, progress_note: None, @@ -86,14 +91,14 @@ pub fn map_launched_epiphany_job( pub fn map_interrupted_epiphany_job( state: &EpiphanyThreadState, binding_id: &str, -) -> ThreadEpiphanyJob { - map_epiphany_jobs(Some(state), None) +) -> EpiphanyJobView { + derive_epiphany_jobs(Some(state), None) .into_iter() .find(|job| job.id == binding_id) .unwrap_or_else(|| { epiphany_blocked_state_job( binding_id, - ThreadEpiphanyJobKind::Specialist, + CoreEpiphanyJobKind::Specialist, "role-scoped specialist work", "Interrupted job binding was not reflected in Epiphany state.", ) @@ -116,19 +121,18 @@ fn map_core_epiphany_job_status(status: CoreEpiphanyJobStatus) -> ThreadEpiphany pub fn epiphany_blocked_state_job( id: &str, - kind: ThreadEpiphanyJobKind, + kind: CoreEpiphanyJobKind, scope: &str, blocking_reason: &str, -) -> ThreadEpiphanyJob { - ThreadEpiphanyJob { +) -> EpiphanyJobView { + EpiphanyJobView { id: id.to_string(), kind, scope: scope.to_string(), owner_role: "epiphany-harness".to_string(), - launcher_job_id: None, authority_scope: None, - backend_job_id: None, - status: ThreadEpiphanyJobStatus::Blocked, + runtime_job_id: None, + status: CoreEpiphanyJobStatus::Blocked, items_processed: None, items_total: None, progress_note: None, diff --git a/epiphany-codex-bridge/src/mutation_service.rs b/epiphany-codex-bridge/src/mutation_service.rs index dd85ab98..98b42eba 100644 --- a/epiphany-codex-bridge/src/mutation_service.rs +++ b/epiphany-codex-bridge/src/mutation_service.rs @@ -3,8 +3,6 @@ use chrono::Utc; use std::path::Path; use std::path::PathBuf; -use codex_app_server_protocol::ThreadEpiphanyJob; -use codex_app_server_protocol::ThreadEpiphanyJobKind; use codex_app_server_protocol::ThreadEpiphanyReorientDecision; use codex_app_server_protocol::ThreadEpiphanyReorientFinding; use codex_app_server_protocol::ThreadEpiphanyReorientSource; @@ -17,6 +15,7 @@ use epiphany_core::EpiphanyJobInterruptRequest; use epiphany_core::EpiphanyJobInterruptResult; use epiphany_core::EpiphanyJobLaunchRequest; use epiphany_core::EpiphanyJobLaunchResult; +use epiphany_core::EpiphanyJobView; use epiphany_core::EpiphanyPromotionInput; use epiphany_core::EpiphanyStateUpdate; use epiphany_core::EpiphanyTokenUsageSnapshot; @@ -37,7 +36,7 @@ use epiphany_state_model::EpiphanyThreadState; use crate::error::EpiphanyBridgeError; use crate::error::Result as BridgeResult; use crate::jobs::epiphany_blocked_state_job; -use crate::jobs::map_epiphany_jobs; +use crate::jobs::derive_epiphany_jobs; use crate::jobs::map_interrupted_epiphany_job; use crate::jobs::map_launched_epiphany_job; use crate::launch::EPIPHANY_REORIENT_LAUNCH_BINDING_ID; @@ -112,10 +111,12 @@ pub struct EpiphanyReorientAcceptApplied { #[derive(Debug, Clone)] pub struct EpiphanyJobLaunchApplied { + pub launcher_job_id: String, + pub backend_job_id: String, pub revision: u64, pub changed_fields: Vec, pub epiphany_state: EpiphanyThreadState, - pub job: ThreadEpiphanyJob, + pub job: EpiphanyJobView, } #[derive(Debug, Clone)] @@ -125,11 +126,13 @@ pub struct EpiphanyJobInterruptApplied { pub revision: u64, pub changed_fields: Vec, pub epiphany_state: EpiphanyThreadState, - pub job: ThreadEpiphanyJob, + pub job: EpiphanyJobView, } #[derive(Debug, Clone)] pub struct EpiphanyReorientLaunchApplied { + pub launcher_job_id: String, + pub backend_job_id: String, pub source: ThreadEpiphanyReorientSource, pub state_status: ThreadEpiphanyReorientStateStatus, pub state_revision: Option, @@ -137,7 +140,7 @@ pub struct EpiphanyReorientLaunchApplied { pub revision: u64, pub changed_fields: Vec, pub epiphany_state: EpiphanyThreadState, - pub job: ThreadEpiphanyJob, + pub job: EpiphanyJobView, } pub async fn apply_epiphany_state_update_to_thread( @@ -522,12 +525,13 @@ pub async fn launch_thread_epiphany_job( let job = map_launched_epiphany_job( &epiphany_state, launched.binding_id.as_str(), - launched.launcher_job_id.as_str(), launched.backend_job_id.as_str(), kind, missing_projection_reason, ); Ok(EpiphanyJobLaunchApplied { + launcher_job_id: launched.launcher_job_id, + backend_job_id: launched.backend_job_id, revision: epiphany_state.revision, changed_fields, epiphany_state, @@ -608,19 +612,21 @@ pub async fn launch_thread_epiphany_reorient( let epiphany_state = thread .client_visible_epiphany_state(launched.epiphany_state) .await; - let job = map_epiphany_jobs(Some(&epiphany_state), None) + let job = derive_epiphany_jobs(Some(&epiphany_state), None) .into_iter() .find(|job| job.id == EPIPHANY_REORIENT_LAUNCH_BINDING_ID) .unwrap_or_else(|| { epiphany_blocked_state_job( EPIPHANY_REORIENT_LAUNCH_BINDING_ID, - ThreadEpiphanyJobKind::Specialist, + CoreEpiphanyJobKind::Specialist, "reorient-guided checkpoint regather", "Launched reorientation worker was not reflected in Epiphany state.", ) }); Ok(EpiphanyReorientLaunchApplied { + launcher_job_id: launched.launcher_job_id, + backend_job_id: launched.backend_job_id, source: ThreadEpiphanyReorientSource::Live, state_status, state_revision, diff --git a/epiphany-core/src/surfaces/jobs.rs b/epiphany-core/src/surfaces/jobs.rs index ea0828af..228a54e0 100644 --- a/epiphany-core/src/surfaces/jobs.rs +++ b/epiphany-core/src/surfaces/jobs.rs @@ -4,8 +4,11 @@ use epiphany_state_model::EpiphanyRetrievalState; use epiphany_state_model::EpiphanyRetrievalStatus; use epiphany_state_model::EpiphanyRuntimeLink; use epiphany_state_model::EpiphanyThreadState; +use serde::Deserialize; +use serde::Serialize; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] pub enum EpiphanyJobStatus { Idle, Needed, @@ -18,22 +21,33 @@ pub enum EpiphanyJobStatus { Unavailable, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] pub struct EpiphanyJobView { pub id: String, pub kind: EpiphanyJobKind, pub scope: String, pub owner_role: String, + #[serde(default, skip_serializing_if = "Option::is_none")] pub authority_scope: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub runtime_job_id: Option, pub status: EpiphanyJobStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] pub items_processed: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub items_total: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub progress_note: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub last_checkpoint_at_unix_seconds: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub blocking_reason: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub active_thread_ids: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub linked_subgoal_ids: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] pub linked_graph_node_ids: Vec, } diff --git a/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_mutation_routes.rs b/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_mutation_routes.rs index 9fc73a37..a5c56cc3 100644 --- a/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_mutation_routes.rs +++ b/vendor/codex/codex-rs/app-server/src/codex_message_processor/epiphany_mutation_routes.rs @@ -3,6 +3,9 @@ use codex_core::CodexThread; use codex_protocol::ThreadId; use codex_protocol::protocol::EpiphanyRetrievalState; use codex_protocol::protocol::EpiphanyThreadState; +use epiphany_codex_bridge::cultnet::EpiphanyJobKind; +use epiphany_codex_bridge::cultnet::EpiphanyJobStatus; +use epiphany_codex_bridge::cultnet::EpiphanyJobView; use epiphany_codex_bridge::error::EpiphanyBridgeError; use epiphany_codex_bridge::invalidation::epiphany_freshness_watcher_snapshot; use epiphany_codex_bridge::launch::EPIPHANY_REORIENT_LAUNCH_BINDING_ID; @@ -134,7 +137,11 @@ impl CodexMessageProcessor { revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job: applied.job, + job: thread_epiphany_job_from_surface( + applied.job, + Some(applied.launcher_job_id), + Some(applied.backend_job_id), + ), }, ) .await; @@ -304,7 +311,11 @@ impl CodexMessageProcessor { revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job: applied.job, + job: thread_epiphany_job_from_surface( + applied.job, + Some(applied.launcher_job_id), + Some(applied.backend_job_id), + ), }, ) .await; @@ -622,7 +633,11 @@ impl CodexMessageProcessor { revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job: applied.job, + job: thread_epiphany_job_from_surface( + applied.job, + Some(applied.launcher_job_id), + Some(applied.backend_job_id), + ), }, ) .await; @@ -687,7 +702,7 @@ impl CodexMessageProcessor { revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job: applied.job, + job: thread_epiphany_job_from_surface(applied.job, None, None), }, ) .await; @@ -701,6 +716,46 @@ impl CodexMessageProcessor { } } +fn thread_epiphany_job_from_surface( + job: EpiphanyJobView, + launcher_job_id: Option, + backend_job_id_override: Option, +) -> ThreadEpiphanyJob { + ThreadEpiphanyJob { + id: job.id, + kind: match job.kind { + EpiphanyJobKind::Indexing => ThreadEpiphanyJobKind::Indexing, + EpiphanyJobKind::Remap => ThreadEpiphanyJobKind::Remap, + EpiphanyJobKind::Verification => ThreadEpiphanyJobKind::Verification, + EpiphanyJobKind::Specialist => ThreadEpiphanyJobKind::Specialist, + }, + scope: job.scope, + owner_role: job.owner_role, + launcher_job_id, + authority_scope: job.authority_scope, + backend_job_id: backend_job_id_override.or(job.runtime_job_id), + status: match job.status { + EpiphanyJobStatus::Idle => ThreadEpiphanyJobStatus::Idle, + EpiphanyJobStatus::Needed => ThreadEpiphanyJobStatus::Needed, + EpiphanyJobStatus::Pending => ThreadEpiphanyJobStatus::Pending, + EpiphanyJobStatus::Running => ThreadEpiphanyJobStatus::Running, + EpiphanyJobStatus::Completed => ThreadEpiphanyJobStatus::Completed, + EpiphanyJobStatus::Failed => ThreadEpiphanyJobStatus::Failed, + EpiphanyJobStatus::Cancelled => ThreadEpiphanyJobStatus::Cancelled, + EpiphanyJobStatus::Blocked => ThreadEpiphanyJobStatus::Blocked, + EpiphanyJobStatus::Unavailable => ThreadEpiphanyJobStatus::Unavailable, + }, + items_processed: job.items_processed, + items_total: job.items_total, + progress_note: job.progress_note, + last_checkpoint_at_unix_seconds: job.last_checkpoint_at_unix_seconds, + blocking_reason: job.blocking_reason, + active_thread_ids: job.active_thread_ids, + linked_subgoal_ids: job.linked_subgoal_ids, + linked_graph_node_ids: job.linked_graph_node_ids, + } +} + async fn thread_epiphany_retrieval_state(thread: &CodexThread) -> EpiphanyRetrievalState { let config = thread.config_snapshot().await; let codex_home = thread.codex_home().await;