mirror of
https://github.com/openai/codex.git
synced 2026-10-24 18:18:41 +00:00
codex: propagate external message rename (#17580)
This commit is contained in:
@@ -20,8 +20,8 @@ use codex_app_server_protocol::TurnStartResponse;
|
||||
use codex_app_server_protocol::UserInput;
|
||||
use codex_core::ARCHIVED_SESSIONS_SUBDIR;
|
||||
use codex_core::find_thread_path_by_id_str;
|
||||
use codex_state::ExternalMessageCreateParams;
|
||||
use codex_state::StateRuntime;
|
||||
use codex_state::ThreadMessageCreateParams;
|
||||
use codex_state::ThreadTimerCreateParams;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::path::Path;
|
||||
@@ -125,11 +125,11 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> {
|
||||
.await
|
||||
.expect("initialize state db");
|
||||
state_db
|
||||
.create_thread_message(&message_params("message-1", &thread.id))
|
||||
.create_external_message(&message_params("message-1", &thread.id))
|
||||
.await
|
||||
.expect("create archived thread message");
|
||||
state_db
|
||||
.create_thread_message(&message_params("message-2", "other-thread"))
|
||||
.create_external_message(&message_params("message-2", "other-thread"))
|
||||
.await
|
||||
.expect("create other thread message");
|
||||
state_db
|
||||
@@ -181,7 +181,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> {
|
||||
);
|
||||
assert_eq!(
|
||||
state_db
|
||||
.list_thread_messages(&thread.id)
|
||||
.list_external_messages(&thread.id)
|
||||
.await
|
||||
.expect("list archived thread messages"),
|
||||
Vec::new()
|
||||
@@ -195,7 +195,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> {
|
||||
);
|
||||
assert_eq!(
|
||||
state_db
|
||||
.list_thread_messages("other-thread")
|
||||
.list_external_messages("other-thread")
|
||||
.await
|
||||
.expect("list other thread messages")
|
||||
.into_iter()
|
||||
@@ -381,8 +381,8 @@ fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn message_params(id: &str, thread_id: &str) -> ThreadMessageCreateParams {
|
||||
ThreadMessageCreateParams {
|
||||
fn message_params(id: &str, thread_id: &str) -> ExternalMessageCreateParams {
|
||||
ExternalMessageCreateParams {
|
||||
id: id.to_string(),
|
||||
thread_id: thread_id.to_string(),
|
||||
source: "external".to_string(),
|
||||
|
||||
@@ -47,7 +47,7 @@ pub(crate) async fn run_queue_command(
|
||||
StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()).await?;
|
||||
let delivery = TimerDelivery::AfterTurn;
|
||||
|
||||
let message_params = codex_state::ThreadMessageCreateParams::new(
|
||||
let message_params = codex_state::ExternalMessageCreateParams::new(
|
||||
thread_id,
|
||||
"external".to_string(),
|
||||
cmd.message,
|
||||
@@ -56,7 +56,7 @@ pub(crate) async fn run_queue_command(
|
||||
delivery.as_str().to_string(),
|
||||
unix_timestamp_now()?,
|
||||
);
|
||||
state_db.create_thread_message(&message_params).await?;
|
||||
state_db.create_external_message(&message_params).await?;
|
||||
remove_queued_message_if_thread_missing(
|
||||
config.codex_home.as_path(),
|
||||
&state_db,
|
||||
@@ -85,7 +85,7 @@ async fn remove_queued_message_if_thread_missing(
|
||||
}
|
||||
|
||||
state_db
|
||||
.delete_thread_message(thread_id, message_id)
|
||||
.delete_external_message(thread_id, message_id)
|
||||
.await?;
|
||||
anyhow::bail!("thread `{thread_id}` was archived before queued work could be created");
|
||||
}
|
||||
|
||||
@@ -348,7 +348,7 @@ impl Session {
|
||||
|
||||
loop {
|
||||
let claim = match state_db
|
||||
.claim_next_thread_message(
|
||||
.claim_next_external_message(
|
||||
&self.thread_id_string(),
|
||||
can_after_turn,
|
||||
can_steer_current_turn,
|
||||
@@ -363,7 +363,7 @@ impl Session {
|
||||
}
|
||||
};
|
||||
match claim {
|
||||
Some(codex_state::ThreadMessageClaim::Claimed(row)) => {
|
||||
Some(codex_state::ExternalMessageClaim::Claimed(row)) => {
|
||||
let (message, delivery) = match InjectedMessage::from_external_row(row) {
|
||||
Ok(parsed) => parsed,
|
||||
Err(err) => {
|
||||
@@ -375,11 +375,11 @@ impl Session {
|
||||
PendingInputItem::injected(message.prompt_input_item(), message.event());
|
||||
return Some(PendingMessageClaim::Claimed(Box::new(input_item), delivery));
|
||||
}
|
||||
Some(codex_state::ThreadMessageClaim::Invalid { id, reason }) => {
|
||||
Some(codex_state::ExternalMessageClaim::Invalid { id, reason }) => {
|
||||
warn!("dropped invalid queued message {id}: {reason}");
|
||||
continue;
|
||||
}
|
||||
Some(codex_state::ThreadMessageClaim::NotReady) | None => {
|
||||
Some(codex_state::ExternalMessageClaim::NotReady) | None => {
|
||||
*self.timer_start_in_progress.lock().await = false;
|
||||
return claim.map(|_| PendingMessageClaim::NotReady);
|
||||
}
|
||||
|
||||
@@ -45,7 +45,7 @@ pub(crate) enum InjectedMessage {
|
||||
|
||||
impl InjectedMessage {
|
||||
pub(crate) fn from_external_row(
|
||||
row: codex_state::ThreadMessage,
|
||||
row: codex_state::ExternalMessage,
|
||||
) -> Result<(Self, TimerDelivery), String> {
|
||||
let delivery = serde_json::from_value::<TimerDelivery>(serde_json::Value::String(
|
||||
row.delivery.clone(),
|
||||
|
||||
@@ -408,7 +408,7 @@ async fn queued_messages_feature_consumes_messages_without_timers() -> Result<()
|
||||
let test = builder.build(&server).await?;
|
||||
let db = test.codex.state_db().expect("state db enabled");
|
||||
let thread_id = test.session_configured.session_id.to_string();
|
||||
db.create_thread_message(&codex_state::ThreadMessageCreateParams::new(
|
||||
db.create_external_message(&codex_state::ExternalMessageCreateParams::new(
|
||||
thread_id,
|
||||
"external".to_string(),
|
||||
"queued hello".to_string(),
|
||||
@@ -515,7 +515,7 @@ async fn queued_message_runs_after_idle_recurring_timer() -> Result<()> {
|
||||
)
|
||||
.await;
|
||||
let thread_id = test.session_configured.session_id.to_string();
|
||||
db.create_thread_message(&codex_state::ThreadMessageCreateParams::new(
|
||||
db.create_external_message(&codex_state::ExternalMessageCreateParams::new(
|
||||
thread_id,
|
||||
"external".to_string(),
|
||||
"queued hello".to_string(),
|
||||
@@ -602,7 +602,7 @@ async fn queued_messages_feature_disabled_leaves_messages_queued() -> Result<()>
|
||||
let test = builder.build(&server).await?;
|
||||
let db = test.codex.state_db().expect("state db enabled");
|
||||
let thread_id = test.session_configured.session_id.to_string();
|
||||
db.create_thread_message(&codex_state::ThreadMessageCreateParams::new(
|
||||
db.create_external_message(&codex_state::ExternalMessageCreateParams::new(
|
||||
thread_id.clone(),
|
||||
"external".to_string(),
|
||||
"queued hello".to_string(),
|
||||
@@ -618,7 +618,7 @@ async fn queued_messages_feature_disabled_leaves_messages_queued() -> Result<()>
|
||||
|
||||
assert_eq!(mock.requests().len(), 1);
|
||||
assert!(
|
||||
db.claim_next_thread_message(
|
||||
db.claim_next_external_message(
|
||||
&thread_id, /*can_after_turn*/ true, /*can_steer_current_turn*/ true,
|
||||
)
|
||||
.await?
|
||||
|
||||
Reference in New Issue
Block a user