Extract thread turns route
This commit is contained in:
Meta Vens
parent
55cfe87730
commit
4d14ff0721
6 changed files with 160 additions and 156 deletions
|
|
@ -60,6 +60,7 @@ remembers doctrine; the branch remembers the blade.
|
|||
- Turn control routes now live in `codex_message_processor/turn_routes.rs`: turn start, thread item injection, app-server client info setting, turn steer, turn interrupt, mode normalization, and input-limit validation with module-local unit tests.
|
||||
- Thread projection now lives in `codex_message_processor/thread_projection.rs`: state-db/thread-store/rollout summary projection, `ConversationSummary` to API `Thread` mapping, thread-title attachment helpers, turn pagination, and rollout turn reconstruction. `codex_message_processor.rs` is about 1,009 lines after this extraction and mostly owns JSON-RPC dispatch, processor construction, shared thread lookup/submit/error helpers, and lifecycle shutdown.
|
||||
- Turn paging/reconstruction now lives in `codex_message_processor/thread_turn_projection.rs`; `thread_projection.rs` keeps state-db/thread-store/rollout summary projection and `ConversationSummary` to API `Thread` mapping. Focused `thread_turns_list` cursor tests and stale in-progress turn interruption coverage pass.
|
||||
- The `thread/turns/list` route handler now lives in `codex_message_processor/thread_turn_routes.rs`, beside the turn projection helpers it consumes. `thread_read_routes.rs` is back to persisted/live thread listing, `thread/read`, loaded-thread listing, and summary reads.
|
||||
- Listener/subscription lifecycle now lives in `codex_message_processor/listener_lifecycle.rs`: connection attach/detach, listener startup, idle unload, unsubscribe, teardown, archive preparation, listener task context, idle-unload state, shutdown result handling, and listener command routing.
|
||||
- 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.
|
||||
|
|
|
|||
Binary file not shown.
File diff suppressed because one or more lines are too long
|
|
@ -19,6 +19,7 @@ mod thread_read_routes;
|
|||
mod thread_resume_routes;
|
||||
mod thread_start_routes;
|
||||
mod thread_turn_projection;
|
||||
mod thread_turn_routes;
|
||||
mod turn_routes;
|
||||
|
||||
use self::auth_routes::ActiveLogin;
|
||||
|
|
|
|||
|
|
@ -408,156 +408,6 @@ impl CodexMessageProcessor {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) async fn thread_turns_list(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: ThreadTurnsListParams,
|
||||
) {
|
||||
let ThreadTurnsListParams {
|
||||
thread_id,
|
||||
cursor,
|
||||
limit,
|
||||
sort_direction,
|
||||
} = params;
|
||||
|
||||
let thread_uuid = match ThreadId::from_string(&thread_id) {
|
||||
Ok(id) => id,
|
||||
Err(err) => {
|
||||
self.send_invalid_request_error(request_id, format!("invalid thread id: {err}"))
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let state_db_ctx = get_state_db(&self.config).await;
|
||||
let mut rollout_path = self
|
||||
.resolve_rollout_path(thread_uuid, state_db_ctx.as_ref())
|
||||
.await;
|
||||
if rollout_path.is_none() {
|
||||
rollout_path =
|
||||
match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string())
|
||||
.await
|
||||
{
|
||||
Ok(Some(path)) => Some(path),
|
||||
Ok(None) => match find_archived_thread_path_by_id_str(
|
||||
&self.config.codex_home,
|
||||
&thread_uuid.to_string(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(path) => path,
|
||||
Err(err) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("failed to locate archived thread id {thread_uuid}: {err}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("failed to locate thread id {thread_uuid}: {err}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if rollout_path.is_none() {
|
||||
match self.thread_manager.get_thread(thread_uuid).await {
|
||||
Ok(thread) => {
|
||||
rollout_path = thread.rollout_path();
|
||||
if rollout_path.is_none() {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
"ephemeral threads do not support thread/turns/list".to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("thread not loaded: {thread_uuid}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let Some(rollout_path) = rollout_path.as_ref() else {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!("failed to locate rollout for thread {thread_uuid}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
};
|
||||
|
||||
match read_rollout_items_from_rollout(rollout_path).await {
|
||||
Ok(items) => {
|
||||
// This API optimizes network transfer by letting clients page through a
|
||||
// thread's turns incrementally, but it still replays the entire rollout on
|
||||
// every request. Rollback and compaction events can change earlier turns, so
|
||||
// the server has to rebuild the full turn list until turn metadata is indexed
|
||||
// separately.
|
||||
let has_live_in_progress_turn =
|
||||
match self.thread_manager.get_thread(thread_uuid).await {
|
||||
Ok(thread) => matches!(thread.agent_status().await, AgentStatus::Running),
|
||||
Err(_) => false,
|
||||
};
|
||||
let turns = reconstruct_thread_turns_from_rollout_items(
|
||||
&items,
|
||||
self.thread_watch_manager
|
||||
.loaded_status_for_thread(&thread_uuid.to_string())
|
||||
.await,
|
||||
has_live_in_progress_turn,
|
||||
);
|
||||
let page = match paginate_thread_turns(
|
||||
turns,
|
||||
cursor.as_deref(),
|
||||
limit,
|
||||
sort_direction.unwrap_or(SortDirection::Desc),
|
||||
) {
|
||||
Ok(page) => page,
|
||||
Err(error) => {
|
||||
self.outgoing.send_error(request_id, error).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
let response = ThreadTurnsListResponse {
|
||||
data: page.turns,
|
||||
next_cursor: page.next_cursor,
|
||||
backwards_cursor: page.backwards_cursor,
|
||||
};
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
}
|
||||
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!(
|
||||
"thread {thread_uuid} is not materialized yet; thread/turns/list is unavailable before first user message"
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Err(err) => {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!(
|
||||
"failed to load rollout `{}` for thread {thread_uuid}: {err}",
|
||||
rollout_path.display()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
pub(super) async fn get_thread_summary(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
|
|
|
|||
154
vendor/codex/codex-rs/app-server/src/codex_message_processor/thread_turn_routes.rs
vendored
Normal file
154
vendor/codex/codex-rs/app-server/src/codex_message_processor/thread_turn_routes.rs
vendored
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
use super::*;
|
||||
|
||||
impl CodexMessageProcessor {
|
||||
pub(super) async fn thread_turns_list(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: ThreadTurnsListParams,
|
||||
) {
|
||||
let ThreadTurnsListParams {
|
||||
thread_id,
|
||||
cursor,
|
||||
limit,
|
||||
sort_direction,
|
||||
} = params;
|
||||
|
||||
let thread_uuid = match ThreadId::from_string(&thread_id) {
|
||||
Ok(id) => id,
|
||||
Err(err) => {
|
||||
self.send_invalid_request_error(request_id, format!("invalid thread id: {err}"))
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let state_db_ctx = get_state_db(&self.config).await;
|
||||
let mut rollout_path = self
|
||||
.resolve_rollout_path(thread_uuid, state_db_ctx.as_ref())
|
||||
.await;
|
||||
if rollout_path.is_none() {
|
||||
rollout_path =
|
||||
match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string())
|
||||
.await
|
||||
{
|
||||
Ok(Some(path)) => Some(path),
|
||||
Ok(None) => match find_archived_thread_path_by_id_str(
|
||||
&self.config.codex_home,
|
||||
&thread_uuid.to_string(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(path) => path,
|
||||
Err(err) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("failed to locate archived thread id {thread_uuid}: {err}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("failed to locate thread id {thread_uuid}: {err}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if rollout_path.is_none() {
|
||||
match self.thread_manager.get_thread(thread_uuid).await {
|
||||
Ok(thread) => {
|
||||
rollout_path = thread.rollout_path();
|
||||
if rollout_path.is_none() {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
"ephemeral threads do not support thread/turns/list".to_string(),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("thread not loaded: {thread_uuid}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let Some(rollout_path) = rollout_path.as_ref() else {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!("failed to locate rollout for thread {thread_uuid}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
};
|
||||
|
||||
match read_rollout_items_from_rollout(rollout_path).await {
|
||||
Ok(items) => {
|
||||
// This API optimizes network transfer by letting clients page through a
|
||||
// thread's turns incrementally, but it still replays the entire rollout on
|
||||
// every request. Rollback and compaction events can change earlier turns, so
|
||||
// the server has to rebuild the full turn list until turn metadata is indexed
|
||||
// separately.
|
||||
let has_live_in_progress_turn =
|
||||
match self.thread_manager.get_thread(thread_uuid).await {
|
||||
Ok(thread) => matches!(thread.agent_status().await, AgentStatus::Running),
|
||||
Err(_) => false,
|
||||
};
|
||||
let turns = reconstruct_thread_turns_from_rollout_items(
|
||||
&items,
|
||||
self.thread_watch_manager
|
||||
.loaded_status_for_thread(&thread_uuid.to_string())
|
||||
.await,
|
||||
has_live_in_progress_turn,
|
||||
);
|
||||
let page = match paginate_thread_turns(
|
||||
turns,
|
||||
cursor.as_deref(),
|
||||
limit,
|
||||
sort_direction.unwrap_or(SortDirection::Desc),
|
||||
) {
|
||||
Ok(page) => page,
|
||||
Err(error) => {
|
||||
self.outgoing.send_error(request_id, error).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
let response = ThreadTurnsListResponse {
|
||||
data: page.turns,
|
||||
next_cursor: page.next_cursor,
|
||||
backwards_cursor: page.backwards_cursor,
|
||||
};
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
}
|
||||
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!(
|
||||
"thread {thread_uuid} is not materialized yet; thread/turns/list is unavailable before first user message"
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Err(err) => {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!(
|
||||
"failed to load rollout `{}` for thread {thread_uuid}: {err}",
|
||||
rollout_path.display()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in a new issue