From 98ee32d21e6bf7269c22c3b615a09ded76ab72dc Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 13 May 2026 06:55:37 +0100 Subject: [PATCH] Move Epiphany launch routes into bridge service --- epiphany-codex-bridge/src/mutation_service.rs | 112 +++++++++++++++- notes/fresh-workspace-handoff.md | 11 +- state/ledgers.msgpack | Bin 84770 -> 85185 bytes state/map.yaml | 4 +- .../epiphany_mutation_routes.rs | 122 ++++++------------ 5 files changed, 162 insertions(+), 87 deletions(-) diff --git a/epiphany-codex-bridge/src/mutation_service.rs b/epiphany-codex-bridge/src/mutation_service.rs index 82195501..429fe05b 100644 --- a/epiphany-codex-bridge/src/mutation_service.rs +++ b/epiphany-codex-bridge/src/mutation_service.rs @@ -1,24 +1,32 @@ use chrono::SecondsFormat; use chrono::Utc; use codex_app_server_protocol::ThreadEpiphanyReorientFinding; +use codex_app_server_protocol::ThreadEpiphanyJob; use codex_app_server_protocol::ThreadEpiphanyRoleFinding; use codex_app_server_protocol::ThreadEpiphanyRoleId; use codex_app_server_protocol::ThreadEpiphanyStateUpdatedField; use codex_app_server_protocol::ThreadEpiphanyUpdatePatch; use codex_core::CodexThread; +use codex_core::EpiphanyJobInterruptRequest; +use codex_core::EpiphanyJobLaunchRequest; use codex_core::EpiphanyPromotionInput; use codex_core::evaluate_promotion; use codex_protocol::error::CodexErr; +use codex_protocol::protocol::EpiphanyJobKind as CoreEpiphanyJobKind; use codex_protocol::protocol::EpiphanyEvidenceRecord; use codex_protocol::protocol::EpiphanyThreadState; +use crate::jobs::map_interrupted_epiphany_job; +use crate::jobs::map_launched_epiphany_job; +use crate::launch::build_epiphany_role_launch_request; +use crate::launch::epiphany_role_label; use crate::mutation::build_reorient_acceptance_update; use crate::mutation::build_role_acceptance_update; +use crate::mutation::epiphany_job_launch_changed_fields; use crate::mutation::epiphany_promote_changed_fields; use crate::mutation::epiphany_update_patch_changed_fields; use crate::mutation::state_update_from_thread_patch; use crate::mutation::thread_epiphany_patch_has_state_replacements; -use crate::launch::epiphany_role_label; 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; @@ -60,6 +68,24 @@ pub struct EpiphanyReorientAcceptApplied { pub finding: ThreadEpiphanyReorientFinding, } +#[derive(Debug, Clone)] +pub struct EpiphanyJobLaunchApplied { + pub revision: u64, + pub changed_fields: Vec, + pub epiphany_state: EpiphanyThreadState, + pub job: ThreadEpiphanyJob, +} + +#[derive(Debug, Clone)] +pub struct EpiphanyJobInterruptApplied { + pub cancel_requested: bool, + pub interrupted_thread_ids: Vec, + pub revision: u64, + pub changed_fields: Vec, + pub epiphany_state: EpiphanyThreadState, + pub job: ThreadEpiphanyJob, +} + pub async fn apply_thread_epiphany_update( thread: &CodexThread, expected_revision: Option, @@ -223,6 +249,90 @@ pub async fn apply_thread_epiphany_reorient_accept( }) } +pub async fn launch_thread_epiphany_role( + thread: &CodexThread, + thread_id: &str, + role_id: ThreadEpiphanyRoleId, + expected_revision: Option, + max_runtime_seconds: Option, +) -> Result { + let state = thread.epiphany_state().await.ok_or_else(|| { + CodexErr::InvalidRequest( + "cannot launch an Epiphany role specialist without authoritative Epiphany state" + .to_string(), + ) + })?; + let launch_request = build_epiphany_role_launch_request( + thread_id, + role_id, + expected_revision, + max_runtime_seconds, + &state, + ) + .map_err(CodexErr::InvalidRequest)?; + launch_thread_epiphany_job( + thread, + launch_request, + CoreEpiphanyJobKind::Specialist, + "missing launched role projection", + ) + .await +} + +pub async fn launch_thread_epiphany_job( + thread: &CodexThread, + launch_request: EpiphanyJobLaunchRequest, + kind: CoreEpiphanyJobKind, + missing_projection_reason: &str, +) -> Result { + let changed_fields = epiphany_job_launch_changed_fields(); + let launched = thread.epiphany_launch_job(launch_request).await?; + let epiphany_state = + client_visible_live_thread_epiphany_state(thread, launched.epiphany_state).await; + 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 { + revision: epiphany_state.revision, + changed_fields, + epiphany_state, + job, + }) +} + +pub async fn interrupt_thread_epiphany_job( + thread: &CodexThread, + expected_revision: Option, + binding_id: &str, + reason: Option, +) -> Result { + let changed_fields = vec![ThreadEpiphanyStateUpdatedField::JobBindings]; + let interrupted = thread + .epiphany_interrupt_job(EpiphanyJobInterruptRequest { + expected_revision, + binding_id: binding_id.to_string(), + reason, + }) + .await?; + let epiphany_state = + client_visible_live_thread_epiphany_state(thread, interrupted.epiphany_state).await; + let job = map_interrupted_epiphany_job(&epiphany_state, binding_id); + + Ok(EpiphanyJobInterruptApplied { + cancel_requested: interrupted.cancel_requested, + interrupted_thread_ids: interrupted.interrupted_thread_ids, + revision: epiphany_state.revision, + changed_fields, + epiphany_state, + job, + }) +} + fn validate_expected_revision( expected_revision: Option, actual_revision: u64, diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index 6c9a37a2..1ca39e2d 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -630,8 +630,15 @@ Role/reorient accept have followed into the same bridge service. checks, typed runtime-spine finding load, acceptance id/timestamp generation, acceptance update construction, state application, and client-visible state projection. The vendored mutation route still parses JSON-RPC params, loads the -thread, shapes responses, and emits notifications. Launch/interrupt route -orchestration remains to cut. +thread, shapes responses, and emits notifications. + +Role launch, generic job launch, and job interrupt now route through the bridge +service too. `launch_thread_epiphany_role`, `launch_thread_epiphany_job`, and +`interrupt_thread_epiphany_job` own launch/interrupt application, +changed-field derivation, live-state projection, and job projection. The +remaining thick mutation-route knot is reorient launch: it still composes +freshness, pressure, watcher state, reorientation decision, launch request, and +response shaping inside the vendored route. Also: MCP itself is allowed to be JSON. The target is not "replace MCP JSON"; the target is an Epiphany-owned boundary that speaks typed Epiphany diff --git a/state/ledgers.msgpack b/state/ledgers.msgpack index b4a2712c566a2a59db3f90faf368e1caa56e4f68..7043c82e0fb841eabd20b9dc2e6eb0340bb0b9de 100644 GIT binary patch delta 253 zcmZ29jrHJ6)(Of?KJ6P-r!g|#Yu>}SeGemJfammPRYpNJQ!7(bE2GI1r|URI`{tLW zrYIEU=cFp+B$nnSXXq%Tr{<*=WhN_RV?>A%Qn!dx6Q6jf2H7_N<2xw$(X-Q&9W_}*f?9!4{h0J1wy!>*7-2CE_oJs|+ zV1GZ|pa5rug2bZYRQ;U%#1#Fa)Z&8tycGSs{F2PHN`=&l&lXcNF)(Of?UN1JPPGe-e+q{Qy`yNKd0MF^GeHaB<4Xq5#q5u state, - None => { - self.send_invalid_request_error( - request_id, - "cannot launch an Epiphany role specialist without authoritative Epiphany state" - .to_string(), - ) - .await; - return; - } - }; - - let launch_request = match build_epiphany_role_launch_request( + let applied = match launch_thread_epiphany_role( + loaded_thread.as_ref(), &thread_id, role_id, expected_revision, max_runtime_seconds, - &state, - ) { - Ok(request) => request, - Err(message) => { - self.send_invalid_request_error(request_id, message).await; - return; - } - }; - let changed_fields = epiphany_job_launch_changed_fields(); - let launched = match loaded_thread.epiphany_launch_job(launch_request).await { - Ok(launched) => launched, + ) + .await + { + Ok(applied) => applied, Err(CodexErr::InvalidRequest(message)) => { self.send_invalid_request_error(request_id, message).await; return; @@ -106,20 +85,8 @@ impl CodexMessageProcessor { return; } }; - - let epiphany_state = client_visible_live_thread_epiphany_state( - loaded_thread.as_ref(), - launched.epiphany_state, - ) - .await; - 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(), - CoreEpiphanyJobKind::Specialist, - "missing launched role projection", - ); + let changed_fields = applied.changed_fields; + let epiphany_state = applied.epiphany_state; self.outgoing .send_response( @@ -127,10 +94,10 @@ impl CodexMessageProcessor { ThreadEpiphanyRoleLaunchResponse { thread_id: thread_uuid.to_string(), role_id, - revision: epiphany_state.revision, + revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job, + job: applied.job, }, ) .await; @@ -757,9 +724,9 @@ impl CodexMessageProcessor { }; let launch_document = map_core_worker_launch_document(launch_document); - let changed_fields = epiphany_job_launch_changed_fields(); - let launched = match thread - .epiphany_launch_job(EpiphanyJobLaunchRequest { + let applied = match launch_thread_epiphany_job( + thread.as_ref(), + EpiphanyJobLaunchRequest { expected_revision, binding_id: binding_id.clone(), kind, @@ -772,10 +739,13 @@ impl CodexMessageProcessor { launch_document, output_contract_id, max_runtime_seconds, - }) - .await + }, + kind, + "missing launched job projection", + ) + .await { - Ok(launched) => launched, + Ok(applied) => applied, Err(CodexErr::InvalidRequest(message)) => { self.send_invalid_request_error(request_id, message).await; return; @@ -789,26 +759,17 @@ impl CodexMessageProcessor { return; } }; - let epiphany_state = - client_visible_live_thread_epiphany_state(thread.as_ref(), launched.epiphany_state) - .await; - 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 launched job projection", - ); + let changed_fields = applied.changed_fields; + let epiphany_state = applied.epiphany_state; self.outgoing .send_response( request_id, ThreadEpiphanyJobLaunchResponse { - revision: epiphany_state.revision, + revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job, + job: applied.job, }, ) .await; @@ -858,16 +819,15 @@ impl CodexMessageProcessor { } }; - let changed_fields = vec![ThreadEpiphanyStateUpdatedField::JobBindings]; - let interrupted = match thread - .epiphany_interrupt_job(EpiphanyJobInterruptRequest { - expected_revision, - binding_id: binding_id.clone(), - reason, - }) - .await + let applied = match interrupt_thread_epiphany_job( + thread.as_ref(), + expected_revision, + &binding_id, + reason, + ) + .await { - Ok(interrupted) => interrupted, + Ok(applied) => applied, Err(CodexErr::InvalidRequest(message)) => { self.send_invalid_request_error(request_id, message).await; return; @@ -881,21 +841,19 @@ impl CodexMessageProcessor { return; } }; - let epiphany_state = - client_visible_live_thread_epiphany_state(thread.as_ref(), interrupted.epiphany_state) - .await; - let job = map_interrupted_epiphany_job(&epiphany_state, &binding_id); + let changed_fields = applied.changed_fields; + let epiphany_state = applied.epiphany_state; self.outgoing .send_response( request_id, ThreadEpiphanyJobInterruptResponse { - cancel_requested: interrupted.cancel_requested, - interrupted_thread_ids: interrupted.interrupted_thread_ids.clone(), - revision: epiphany_state.revision, + cancel_requested: applied.cancel_requested, + interrupted_thread_ids: applied.interrupted_thread_ids.clone(), + revision: applied.revision, changed_fields: changed_fields.clone(), epiphany_state: epiphany_state.clone(), - job, + job: applied.job, }, ) .await;