GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Extract realtime and review routes

Browse source
This commit is contained in:
Meta Vens 2026-05-17 21:15:28 +01:00
commit c86de6fa00
6 changed files with 480 additions and 466 deletions
Download patch file Download diff file Expand all files Collapse all files

1
notes/fresh-workspace-handoff.md
View file

@ -45,6 +45,7 @@ remembers doctrine; the branch remembers the blade.
- Codex account auth route handling remains in vendored Codex but now lives in `codex_message_processor/auth_routes.rs`: API-key login, ChatGPT browser/device-code/external-token login, logout, auth status, account, rate limits, and add-credits nudge handling. This preserves Codex ownership of subscription auth while cutting the root processor to about 8,094 lines.
- One-off command execution and command stream write/resize/terminate route handling now live in `codex_message_processor/command_routes.rs`; `CommandExecManager` remains the lifecycle owner, while the root processor is about 7,800 lines.
- Git diff-to-origin, fuzzy file search/session routes, feedback upload, and Windows sandbox setup now live in `codex_message_processor/utility_routes.rs`; the root processor is about 7,397 lines.
- Realtime conversation routes now live in `codex_message_processor/realtime_routes.rs`, and review-start orchestration now lives in `codex_message_processor/review_routes.rs`; the root processor is about 6,938 lines.
- The May 2026 foundation migration in `notes/epiphany-ideal-architecture-rebuild-plan.md` is complete enough to defend. Read `notes/epiphany-architectural-teardown.md` before touching Epiphany control-plane code, but do not treat it as an open-ended excuse to keep shaving the altar. The rebuild moved scene projection, freshness derivation, pressure policy, targeted context shards, bounded graph traversal, job/progress view derivation, planning view derivation, reorientation resume/regather verdict policy, pure CRRC recommendation policy, fixed-lane coordinator decision policy, role-board projection policy, role/reorient result interpretation, role self-persistence review policy, and role/reorient acceptance bundle policy into `epiphany-core`; typed acceptance receipts replaced summary-string identity for live accept paths, runtime links are dual-written on launch, and result read-back now prefers runtime links.
- Read `notes/epiphany-ideal-architecture-rebuild-plan.md` immediately after the teardown. It defines the smallest coherent replacement architecture: durable `EpiphanyState`, runtime-spine-owned `RuntimeState`, separate role memory/heartbeat state, `epiphany-core` as policy owner, app-server as adapter, Aquarium as reflector, typed acceptance receipts, and derived view lenses instead of a protocol verb zoo.
- Do not continue Aquarium UI, bridge, Face, or dogfood expansion until the teardown has a source-grounded cleanup slice plan. Epiphany is the foundation; patches on patches are not a purification rite, they are how the altar becomes load-bearing garbage.

BIN
state/ledgers.msgpack
View file

Binary file not shown.

11
state/map.yaml
View file

File diff suppressed because one or more lines are too long

463
vendor/codex/codex-rs/app-server/src/codex_message_processor.rs vendored
View file

@ -4,6 +4,8 @@ mod epiphany_automation;
mod epiphany_mutation_routes;
mod epiphany_read_routes;
mod mcp_routes;
mod realtime_routes;
mod review_routes;
use self::auth_routes::ActiveLogin;
pub(crate) use self::epiphany_automation::maybe_run_epiphany_coordinator_automation_for_turn_boundary;
@ -5027,467 +5029,6 @@ impl CodexMessageProcessor {
}
}
async fn prepare_realtime_conversation_thread(
&self,
request_id: ConnectionRequestId,
thread_id: &str,
) -> Option<(ThreadId, Arc<CodexThread>)> {
let (thread_id, thread) = match self.load_thread(thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return None;
}
};
match self
.ensure_conversation_listener(
thread_id,
request_id.connection_id,
/*raw_events_enabled*/ false,
ApiVersion::V2,
)
.await
{
Ok(EnsureConversationListenerResult::Attached) => {}
Ok(EnsureConversationListenerResult::ConnectionClosed) => {
return None;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return None;
}
}
if !thread.enabled(Feature::RealtimeConversation) {
self.send_invalid_request_error(
request_id,
format!("thread {thread_id} does not support realtime conversation"),
)
.await;
return None;
}
Some((thread_id, thread))
}
async fn thread_realtime_start(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeStartParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::RealtimeConversationStart(ConversationStartParams {
output_modality: params.output_modality,
prompt: params.prompt,
session_id: params.session_id,
transport: params.transport.map(|transport| match transport {
ThreadRealtimeStartTransport::Websocket => {
ConversationStartTransport::Websocket
}
ThreadRealtimeStartTransport::Webrtc { sdp } => {
ConversationStartTransport::Webrtc { sdp }
}
}),
voice: params.voice,
}),
)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeStartResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to start realtime conversation: {err}"),
)
.await;
}
}
}
async fn thread_realtime_append_audio(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeAppendAudioParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::RealtimeConversationAudio(ConversationAudioParams {
frame: params.audio.into(),
}),
)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeAppendAudioResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to append realtime conversation audio: {err}"),
)
.await;
}
}
}
async fn thread_realtime_append_text(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeAppendTextParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::RealtimeConversationText(ConversationTextParams { text: params.text }),
)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeAppendTextResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to append realtime conversation text: {err}"),
)
.await;
}
}
}
async fn thread_realtime_stop(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeStopParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(&request_id, thread.as_ref(), Op::RealtimeConversationClose)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeStopResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to stop realtime conversation: {err}"),
)
.await;
}
}
}
async fn thread_realtime_list_voices(
&self,
request_id: ConnectionRequestId,
_params: ThreadRealtimeListVoicesParams,
) {
self.outgoing
.send_response(
request_id,
ThreadRealtimeListVoicesResponse {
voices: RealtimeVoicesList::builtin(),
},
)
.await;
}
fn build_review_turn(turn_id: String, display_text: &str) -> Turn {
let items = if display_text.is_empty() {
Vec::new()
} else {
vec![ThreadItem::UserMessage {
id: turn_id.clone(),
content: vec![V2UserInput::Text {
text: display_text.to_string(),
// Review prompt display text is synthesized; no UI element ranges to preserve.
text_elements: Vec::new(),
}],
}]
};
Turn {
id: turn_id,
items,
error: None,
status: TurnStatus::InProgress,
started_at: None,
completed_at: None,
duration_ms: None,
}
}
async fn emit_review_started(
&self,
request_id: &ConnectionRequestId,
turn: Turn,
review_thread_id: String,
) {
let response = ReviewStartResponse {
turn,
review_thread_id,
};
self.outgoing
.send_response(request_id.clone(), response)
.await;
}
async fn start_inline_review(
&self,
request_id: &ConnectionRequestId,
parent_thread: Arc<CodexThread>,
review_request: ReviewRequest,
display_text: &str,
parent_thread_id: String,
) -> std::result::Result<(), JSONRPCErrorError> {
let turn_id = self
.submit_core_op(
request_id,
parent_thread.as_ref(),
Op::Review { review_request },
)
.await;
match turn_id {
Ok(turn_id) => {
let turn = Self::build_review_turn(turn_id, display_text);
self.emit_review_started(request_id, turn, parent_thread_id)
.await;
Ok(())
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start review: {err}"),
data: None,
}),
}
}
async fn start_detached_review(
&self,
request_id: &ConnectionRequestId,
parent_thread_id: ThreadId,
parent_thread: Arc<CodexThread>,
review_request: ReviewRequest,
display_text: &str,
) -> std::result::Result<(), JSONRPCErrorError> {
let rollout_path = if let Some(path) = parent_thread.rollout_path() {
path
} else {
find_thread_path_by_id_str(&self.config.codex_home, &parent_thread_id.to_string())
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to locate thread id {parent_thread_id}: {err}"),
data: None,
})?
.ok_or_else(|| JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("no rollout found for thread id {parent_thread_id}"),
data: None,
})?
};
let mut config = self.config.as_ref().clone();
if let Some(review_model) = &config.review_model {
config.model = Some(review_model.clone());
}
let NewThread {
thread_id,
thread: review_thread,
session_configured,
..
} = self
.thread_manager
.fork_thread(
ForkSnapshot::Interrupted,
config,
rollout_path,
/*persist_extended_history*/ false,
self.request_trace_context(request_id).await,
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("error creating detached review thread: {err}"),
data: None,
})?;
Self::log_listener_attach_result(
self.ensure_conversation_listener(
thread_id,
request_id.connection_id,
/*raw_events_enabled*/ false,
ApiVersion::V2,
)
.await,
thread_id,
request_id.connection_id,
"review thread",
);
let fallback_provider = self.config.model_provider_id.as_str();
if let Some(rollout_path) = review_thread.rollout_path() {
match read_summary_from_rollout(rollout_path.as_path(), fallback_provider).await {
Ok(summary) => {
let mut thread = summary_to_thread(summary, &self.config.cwd);
thread.epiphany_state =
live_thread_epiphany_state(review_thread.as_ref()).await;
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.await;
thread.status = resolve_thread_status(
self.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await,
/*has_in_progress_turn*/ false,
);
let notif = ThreadStartedNotification { thread };
self.outgoing
.send_server_notification(ServerNotification::ThreadStarted(notif))
.await;
}
Err(err) => {
tracing::warn!(
"failed to load summary for review thread {}: {}",
session_configured.session_id,
err
);
}
}
} else {
tracing::warn!(
"review thread {} has no rollout path",
session_configured.session_id
);
}
let turn_id = self
.submit_core_op(
request_id,
review_thread.as_ref(),
Op::Review { review_request },
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start detached review turn: {err}"),
data: None,
})?;
let turn = Self::build_review_turn(turn_id, display_text);
let review_thread_id = thread_id.to_string();
self.emit_review_started(request_id, turn, review_thread_id)
.await;
Ok(())
}
async fn review_start(&self, request_id: ConnectionRequestId, params: ReviewStartParams) {
let ReviewStartParams {
thread_id,
target,
delivery,
} = params;
let (parent_thread_id, parent_thread) = match self.load_thread(&thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let (review_request, display_text) = match Self::review_request_from_target(target) {
Ok(value) => value,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
};
let delivery = delivery.unwrap_or(ApiReviewDelivery::Inline).to_core();
match delivery {
CoreReviewDelivery::Inline => {
if let Err(err) = self
.start_inline_review(
&request_id,
parent_thread,
review_request,
display_text.as_str(),
thread_id.clone(),
)
.await
{
self.outgoing.send_error(request_id, err).await;
}
}
CoreReviewDelivery::Detached => {
if let Err(err) = self
.start_detached_review(
&request_id,
parent_thread_id,
parent_thread,
review_request,
display_text.as_str(),
)
.await
{
self.outgoing.send_error(request_id, err).await;
}
}
}
}
async fn turn_interrupt(&self, request_id: ConnectionRequestId, params: TurnInterruptParams) {
let TurnInterruptParams { thread_id, turn_id } = params;
let is_startup_interrupt = turn_id.is_empty();

217
vendor/codex/codex-rs/app-server/src/codex_message_processor/realtime_routes.rs vendored Normal file
View file

@ -0,0 +1,217 @@
use super::*;
impl CodexMessageProcessor {
async fn prepare_realtime_conversation_thread(
&self,
request_id: ConnectionRequestId,
thread_id: &str,
) -> Option<(ThreadId, Arc<CodexThread>)> {
let (thread_id, thread) = match self.load_thread(thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return None;
}
};
match self
.ensure_conversation_listener(
thread_id,
request_id.connection_id,
/*raw_events_enabled*/ false,
ApiVersion::V2,
)
.await
{
Ok(EnsureConversationListenerResult::Attached) => {}
Ok(EnsureConversationListenerResult::ConnectionClosed) => {
return None;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return None;
}
}
if !thread.enabled(Feature::RealtimeConversation) {
self.send_invalid_request_error(
request_id,
format!("thread {thread_id} does not support realtime conversation"),
)
.await;
return None;
}
Some((thread_id, thread))
}
pub(super) async fn thread_realtime_start(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeStartParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::RealtimeConversationStart(ConversationStartParams {
output_modality: params.output_modality,
prompt: params.prompt,
session_id: params.session_id,
transport: params.transport.map(|transport| match transport {
ThreadRealtimeStartTransport::Websocket => {
ConversationStartTransport::Websocket
}
ThreadRealtimeStartTransport::Webrtc { sdp } => {
ConversationStartTransport::Webrtc { sdp }
}
}),
voice: params.voice,
}),
)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeStartResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to start realtime conversation: {err}"),
)
.await;
}
}
}
pub(super) async fn thread_realtime_append_audio(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeAppendAudioParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::RealtimeConversationAudio(ConversationAudioParams {
frame: params.audio.into(),
}),
)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeAppendAudioResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to append realtime conversation audio: {err}"),
)
.await;
}
}
}
pub(super) async fn thread_realtime_append_text(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeAppendTextParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(
&request_id,
thread.as_ref(),
Op::RealtimeConversationText(ConversationTextParams { text: params.text }),
)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeAppendTextResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to append realtime conversation text: {err}"),
)
.await;
}
}
}
pub(super) async fn thread_realtime_stop(
&self,
request_id: ConnectionRequestId,
params: ThreadRealtimeStopParams,
) {
let Some((_, thread)) = self
.prepare_realtime_conversation_thread(request_id.clone(), &params.thread_id)
.await
else {
return;
};
let submit = self
.submit_core_op(&request_id, thread.as_ref(), Op::RealtimeConversationClose)
.await;
match submit {
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadRealtimeStopResponse::default())
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to stop realtime conversation: {err}"),
)
.await;
}
}
}
pub(super) async fn thread_realtime_list_voices(
&self,
request_id: ConnectionRequestId,
_params: ThreadRealtimeListVoicesParams,
) {
self.outgoing
.send_response(
request_id,
ThreadRealtimeListVoicesResponse {
voices: RealtimeVoicesList::builtin(),
},
)
.await;
}
}

254
vendor/codex/codex-rs/app-server/src/codex_message_processor/review_routes.rs vendored Normal file
View file

@ -0,0 +1,254 @@
use super::*;
impl CodexMessageProcessor {
fn build_review_turn(turn_id: String, display_text: &str) -> Turn {
let items = if display_text.is_empty() {
Vec::new()
} else {
vec![ThreadItem::UserMessage {
id: turn_id.clone(),
content: vec![V2UserInput::Text {
text: display_text.to_string(),
// Review prompt display text is synthesized; no UI element ranges to preserve.
text_elements: Vec::new(),
}],
}]
};
Turn {
id: turn_id,
items,
error: None,
status: TurnStatus::InProgress,
started_at: None,
completed_at: None,
duration_ms: None,
}
}
async fn emit_review_started(
&self,
request_id: &ConnectionRequestId,
turn: Turn,
review_thread_id: String,
) {
let response = ReviewStartResponse {
turn,
review_thread_id,
};
self.outgoing
.send_response(request_id.clone(), response)
.await;
}
async fn start_inline_review(
&self,
request_id: &ConnectionRequestId,
parent_thread: Arc<CodexThread>,
review_request: ReviewRequest,
display_text: &str,
parent_thread_id: String,
) -> std::result::Result<(), JSONRPCErrorError> {
let turn_id = self
.submit_core_op(
request_id,
parent_thread.as_ref(),
Op::Review { review_request },
)
.await;
match turn_id {
Ok(turn_id) => {
let turn = Self::build_review_turn(turn_id, display_text);
self.emit_review_started(request_id, turn, parent_thread_id)
.await;
Ok(())
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start review: {err}"),
data: None,
}),
}
}
async fn start_detached_review(
&self,
request_id: &ConnectionRequestId,
parent_thread_id: ThreadId,
parent_thread: Arc<CodexThread>,
review_request: ReviewRequest,
display_text: &str,
) -> std::result::Result<(), JSONRPCErrorError> {
let rollout_path = if let Some(path) = parent_thread.rollout_path() {
path
} else {
find_thread_path_by_id_str(&self.config.codex_home, &parent_thread_id.to_string())
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to locate thread id {parent_thread_id}: {err}"),
data: None,
})?
.ok_or_else(|| JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("no rollout found for thread id {parent_thread_id}"),
data: None,
})?
};
let mut config = self.config.as_ref().clone();
if let Some(review_model) = &config.review_model {
config.model = Some(review_model.clone());
}
let NewThread {
thread_id,
thread: review_thread,
session_configured,
..
} = self
.thread_manager
.fork_thread(
ForkSnapshot::Interrupted,
config,
rollout_path,
/*persist_extended_history*/ false,
self.request_trace_context(request_id).await,
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("error creating detached review thread: {err}"),
data: None,
})?;
Self::log_listener_attach_result(
self.ensure_conversation_listener(
thread_id,
request_id.connection_id,
/*raw_events_enabled*/ false,
ApiVersion::V2,
)
.await,
thread_id,
request_id.connection_id,
"review thread",
);
let fallback_provider = self.config.model_provider_id.as_str();
if let Some(rollout_path) = review_thread.rollout_path() {
match read_summary_from_rollout(rollout_path.as_path(), fallback_provider).await {
Ok(summary) => {
let mut thread = summary_to_thread(summary, &self.config.cwd);
thread.epiphany_state =
live_thread_epiphany_state(review_thread.as_ref()).await;
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.await;
thread.status = resolve_thread_status(
self.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await,
/*has_in_progress_turn*/ false,
);
let notif = ThreadStartedNotification { thread };
self.outgoing
.send_server_notification(ServerNotification::ThreadStarted(notif))
.await;
}
Err(err) => {
tracing::warn!(
"failed to load summary for review thread {}: {}",
session_configured.session_id,
err
);
}
}
} else {
tracing::warn!(
"review thread {} has no rollout path",
session_configured.session_id
);
}
let turn_id = self
.submit_core_op(
request_id,
review_thread.as_ref(),
Op::Review { review_request },
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start detached review turn: {err}"),
data: None,
})?;
let turn = Self::build_review_turn(turn_id, display_text);
let review_thread_id = thread_id.to_string();
self.emit_review_started(request_id, turn, review_thread_id)
.await;
Ok(())
}
pub(super) async fn review_start(
&self,
request_id: ConnectionRequestId,
params: ReviewStartParams,
) {
let ReviewStartParams {
thread_id,
target,
delivery,
} = params;
let (parent_thread_id, parent_thread) = match self.load_thread(&thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let (review_request, display_text) = match Self::review_request_from_target(target) {
Ok(value) => value,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
};
let delivery = delivery.unwrap_or(ApiReviewDelivery::Inline).to_core();
match delivery {
CoreReviewDelivery::Inline => {
if let Err(err) = self
.start_inline_review(
&request_id,
parent_thread,
review_request,
display_text.as_str(),
thread_id.clone(),
)
.await
{
self.outgoing.send_error(request_id, err).await;
}
}
CoreReviewDelivery::Detached => {
if let Err(err) = self
.start_detached_review(
&request_id,
parent_thread_id,
parent_thread,
review_request,
display_text.as_str(),
)
.await
{
self.outgoing.send_error(request_id, err).await;
}
}
}
}
}