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

Move read helpers into read routes

Browse source
This commit is contained in:
Meta Vens 2026-05-17 21:28:11 +01:00
commit 934be2f7cf
5 changed files with 174 additions and 177 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -48,7 +48,7 @@ remembers doctrine; the branch remembers the blade.
- 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.
- Model/catalog routes now live in `codex_message_processor/catalog_routes.rs`: model list, collaboration-mode list, experimental feature list, and mock experimental method. The root processor is about 6,738 lines.
- Thread administrative mutation routes now live in `codex_message_processor/thread_admin_routes.rs`: archive/unarchive, elicitation counters, set-name, memory mode, memory reset, metadata update, rollback, compact, background terminal cleanup, shell command submission, and guardian approval. The root processor is about 5,750 lines.
- Thread read/list projection routes now live in `codex_message_processor/thread_read_routes.rs`: persisted/live thread listing, loaded-thread listing, `thread/read`, Epiphany state attachment for reads, rollout turn paging, and read-side rollout field hydration. The root processor is about 5,241 lines.
- Thread read/list projection routes now live in `codex_message_processor/thread_read_routes.rs`: persisted/live thread listing, loaded-thread listing, `thread/read`, conversation summary reads, shared list pagination, Epiphany state attachment for reads, rollout turn paging, and read-side rollout field hydration. The root processor is about 3,105 lines after later route extractions.
- Thread resume/fork orchestration now lives in `codex_message_processor/thread_resume_routes.rs`; it owns resume from history/rollout, running-thread resume response replay, persisted resume metadata hydration, fork response construction, and the shared thread-title attachment helper. The root processor is about 4,336 lines.
- Thread start route handling now lives in `codex_message_processor/thread_start_routes.rs`; the route shell delegates to the root lifecycle `thread_start_task`. The root processor is about 4,250 lines.
- Turn control routes now live in `codex_message_processor/turn_routes.rs`: turn start, thread item injection, app-server client info setting, turn steer, and turn interrupt. The root processor is about 3,803 lines.

BIN
state/ledgers.msgpack
View file

Binary file not shown.

10
state/map.yaml
View file

File diff suppressed because one or more lines are too long

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

@ -1534,176 +1534,6 @@ impl CodexMessageProcessor {
}
}
async fn get_thread_summary(
&self,
request_id: ConnectionRequestId,
params: GetConversationSummaryParams,
) {
let fallback_provider = self.config.model_provider_id.as_str();
let read_result = match params {
GetConversationSummaryParams::ThreadId { conversation_id } => self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id: conversation_id,
include_archived: true,
include_history: false,
})
.await
.map_err(|err| conversation_summary_thread_id_read_error(conversation_id, err)),
GetConversationSummaryParams::RolloutPath { rollout_path } => {
let Some(local_thread_store) = self
.thread_store
.as_any()
.downcast_ref::<LocalThreadStore>()
else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message:
"rollout path queries are only supported with the local thread store"
.to_string(),
data: None,
};
return self.outgoing.send_error(request_id, error).await;
};
local_thread_store
.read_thread_by_rollout_path(
rollout_path.clone(),
/*include_archived*/ true,
/*include_history*/ false,
)
.await
.map_err(|err| conversation_summary_rollout_path_read_error(&rollout_path, err))
}
};
match read_result {
Ok(stored_thread) => {
let Some(summary) = summary_from_stored_thread(stored_thread, fallback_provider)
else {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message:
"failed to load conversation summary: thread is missing rollout path"
.to_string(),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
};
let response = GetConversationSummaryResponse { summary };
self.outgoing.send_response(request_id, response).await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn list_threads_common(
&self,
requested_page_size: usize,
cursor: Option<String>,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
filters: ThreadListFilters,
) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
let ThreadListFilters {
model_providers,
source_kinds,
archived,
cwd_filters,
search_term,
use_state_db_only,
} = filters;
let mut cursor_obj = cursor;
let mut last_cursor = cursor_obj.clone();
let mut remaining = requested_page_size;
let mut items = Vec::with_capacity(requested_page_size);
let mut next_cursor: Option<String> = None;
let model_provider_filter = match model_providers {
Some(providers) => {
if providers.is_empty() {
None
} else {
Some(providers)
}
}
None => Some(vec![self.config.model_provider_id.clone()]),
};
let fallback_provider = self.config.model_provider_id.clone();
let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds);
let allowed_sources = allowed_sources_vec.as_slice();
let store_sort_direction = match sort_direction {
SortDirection::Asc => StoreSortDirection::Asc,
SortDirection::Desc => StoreSortDirection::Desc,
};
while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
let page = self
.thread_store
.list_threads(StoreListThreadsParams {
page_size,
cursor: cursor_obj.clone(),
sort_key,
sort_direction: store_sort_direction,
allowed_sources: allowed_sources.to_vec(),
model_providers: model_provider_filter.clone(),
cwd_filters: cwd_filters.clone(),
archived,
search_term: search_term.clone(),
use_state_db_only,
})
.await
.map_err(thread_store_list_error)?;
let mut filtered = Vec::with_capacity(page.items.len());
for it in page.items {
let Some(summary) = summary_from_stored_thread(it, fallback_provider.as_str())
else {
continue;
};
if source_kind_filter
.as_ref()
.is_none_or(|filter| source_kind_matches(&summary.source, filter))
&& cwd_filters.as_ref().is_none_or(|expected_cwds| {
expected_cwds.iter().any(|expected_cwd| {
path_utils::paths_match_after_normalization(&summary.cwd, expected_cwd)
})
})
{
filtered.push(summary);
if filtered.len() >= remaining {
break;
}
}
}
items.extend(filtered);
remaining = requested_page_size.saturating_sub(items.len());
next_cursor = page.next_cursor;
if remaining == 0 {
break;
}
let Some(cursor_val) = next_cursor.clone() else {
break;
};
// Break if our pagination would reuse the same cursor again; this avoids
// an infinite loop when filtering drops everything on the page.
if last_cursor.as_ref() == Some(&cursor_val) {
next_cursor = None;
break;
}
last_cursor = Some(cursor_val.clone());
cursor_obj = Some(cursor_val);
}
Ok((items, next_cursor))
}
async fn send_invalid_request_error(&self, request_id: ConnectionRequestId, message: String) {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,

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

@ -518,4 +518,173 @@ impl CodexMessageProcessor {
}
}
}
pub(super) async fn get_thread_summary(
&self,
request_id: ConnectionRequestId,
params: GetConversationSummaryParams,
) {
let fallback_provider = self.config.model_provider_id.as_str();
let read_result = match params {
GetConversationSummaryParams::ThreadId { conversation_id } => self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id: conversation_id,
include_archived: true,
include_history: false,
})
.await
.map_err(|err| conversation_summary_thread_id_read_error(conversation_id, err)),
GetConversationSummaryParams::RolloutPath { rollout_path } => {
let Some(local_thread_store) = self
.thread_store
.as_any()
.downcast_ref::<LocalThreadStore>()
else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message:
"rollout path queries are only supported with the local thread store"
.to_string(),
data: None,
};
return self.outgoing.send_error(request_id, error).await;
};
local_thread_store
.read_thread_by_rollout_path(
rollout_path.clone(),
/*include_archived*/ true,
/*include_history*/ false,
)
.await
.map_err(|err| conversation_summary_rollout_path_read_error(&rollout_path, err))
}
};
match read_result {
Ok(stored_thread) => {
let Some(summary) = summary_from_stored_thread(stored_thread, fallback_provider)
else {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message:
"failed to load conversation summary: thread is missing rollout path"
.to_string(),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
};
let response = GetConversationSummaryResponse { summary };
self.outgoing.send_response(request_id, response).await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn list_threads_common(
&self,
requested_page_size: usize,
cursor: Option<String>,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
filters: ThreadListFilters,
) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
let ThreadListFilters {
model_providers,
source_kinds,
archived,
cwd_filters,
search_term,
use_state_db_only,
} = filters;
let mut cursor_obj = cursor;
let mut last_cursor = cursor_obj.clone();
let mut remaining = requested_page_size;
let mut items = Vec::with_capacity(requested_page_size);
let mut next_cursor: Option<String> = None;
let model_provider_filter = match model_providers {
Some(providers) => {
if providers.is_empty() {
None
} else {
Some(providers)
}
}
None => Some(vec![self.config.model_provider_id.clone()]),
};
let fallback_provider = self.config.model_provider_id.clone();
let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds);
let allowed_sources = allowed_sources_vec.as_slice();
let store_sort_direction = match sort_direction {
SortDirection::Asc => StoreSortDirection::Asc,
SortDirection::Desc => StoreSortDirection::Desc,
};
while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
let page = self
.thread_store
.list_threads(StoreListThreadsParams {
page_size,
cursor: cursor_obj.clone(),
sort_key,
sort_direction: store_sort_direction,
allowed_sources: allowed_sources.to_vec(),
model_providers: model_provider_filter.clone(),
cwd_filters: cwd_filters.clone(),
archived,
search_term: search_term.clone(),
use_state_db_only,
})
.await
.map_err(thread_store_list_error)?;
let mut filtered = Vec::with_capacity(page.items.len());
for it in page.items {
let Some(summary) = summary_from_stored_thread(it, fallback_provider.as_str())
else {
continue;
};
if source_kind_filter
.as_ref()
.is_none_or(|filter| source_kind_matches(&summary.source, filter))
&& cwd_filters.as_ref().is_none_or(|expected_cwds| {
expected_cwds.iter().any(|expected_cwd| {
path_utils::paths_match_after_normalization(&summary.cwd, expected_cwd)
})
})
{
filtered.push(summary);
if filtered.len() >= remaining {
break;
}
}
}
items.extend(filtered);
remaining = requested_page_size.saturating_sub(items.len());
next_cursor = page.next_cursor;
if remaining == 0 {
break;
}
let Some(cursor_val) = next_cursor.clone() else {
break;
};
// Break if our pagination would reuse the same cursor again; this avoids
// an infinite loop when filtering drops everything on the page.
if last_cursor.as_ref() == Some(&cursor_val) {
next_cursor = None;
break;
}
last_cursor = Some(cursor_val.clone());
cursor_obj = Some(cursor_val);
}
Ok((items, next_cursor))
}
}