github/codex
4
0 Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-03 08:22:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex] Remove unused event messages (#20511)

Browse Source 
## Why

Several legacy `EventMsg` variants were still emitted or mapped even
though clients either ignored them or had moved to item/lifecycle
events. `Op::Undo` had also degraded to an unavailable shim, so this
removes that dead task path instead of preserving a command that cannot
do useful work.

`McpStartupComplete`, `WebSearchBegin`, and `ImageGenerationBegin` are
intentionally kept because useful consumers still depend on them: MCP
startup completion drives readiness behavior, and the begin events let
app-server/core consumers surface in-progress web-search and
image-generation items before the final payload arrives.

## What Changed

- Removed weak legacy event variants and payloads from `codex-protocol`,
including legacy agent deltas, background events, and undo lifecycle
events.
- Kept/restored `EventMsg::McpStartupComplete`,
`EventMsg::WebSearchBegin`, and `EventMsg::ImageGenerationBegin` with
serializer and emission coverage.
- Updated core, rollout, MCP server, app-server thread history,
review/delegate filtering, and tests to rely on the useful replacement
events that remain.
- Removed `Op::Undo`, `UndoTask`, the undo test module, and stale TUI
slash-command comments.
- Stopped agent job/background progress and compaction retry notices
from emitting `BackgroundEvent` payloads.

## Verification

- `cargo check -p codex-protocol -p codex-app-server-protocol -p
codex-core -p codex-rollout -p codex-rollout-trace -p codex-mcp-server`
- `cargo test -p codex-protocol -p codex-app-server-protocol -p
codex-rollout -p codex-rollout-trace -p codex-mcp-server`
- `cargo test -p codex-core --test all suite::items`
- `just fix -p codex-protocol -p codex-app-server-protocol -p codex-core
-p codex-rollout -p codex-rollout-trace -p codex-mcp-server`
- Earlier coverage on this PR also included `codex-mcp`, `codex-tui`,
core library tests, MCP/plugin/delegate/review/agent job tests, and MCP
startup TUI tests.
This commit is contained in:
pakrym-oai authored and GitHub committed 2026-04-30 20:03:26 -07:00
1 parent bb60b78c46
commit f50c02d7bc
23 files changed
+7 -467
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

codex-rs/core/src/codex_delegate.rs
-5
View File
@@ -262,11 +262,6 @@ async fn forward_events(
Err(_) => break,
};
match event {
// ignore all legacy delta events
Event {
id: _,
msg: EventMsg::AgentMessageDelta(_) | EventMsg::AgentReasoningDelta(_),
} => {}
Event {
id: _,
msg: EventMsg::TokenCount(_),
codex-rs/core/src/compact.rs
-12
View File
@@ -165,8 +165,6 @@ async fn run_compact_task_inner_impl(
turn_context.truncation_policy,
);
let mut truncated_count = 0usize;
let max_retries = turn_context.provider.info().stream_max_retries();
let mut retries = 0;
let mut client_session = sess.services.model_client.new_session();
@@ -198,15 +196,6 @@ async fn run_compact_task_inner_impl(
match attempt_result {
Ok(()) => {
if truncated_count > 0 {
sess.notify_background_event(
turn_context.as_ref(),
format!(
"Trimmed {truncated_count} older thread item(s) before compacting so the prompt fits the model context window."
),
)
.await;
}
break;
}
Err(CodexErr::Interrupted) => {
@@ -219,7 +208,6 @@ async fn run_compact_task_inner_impl(
"Context window exceeded while compacting; removing oldest history item. Error: {e}"
);
history.remove_first_item();
truncated_count += 1;
retries = 0;
continue;
}
codex-rs/core/src/mcp_skill_dependencies.rs
-16
View File
@@ -136,14 +136,6 @@ pub(crate) async fn maybe_install_mcp_dependencies(
}
};
sess.notify_background_event(
turn_context,
format!(
"Authenticating MCP {name}... Follow instructions in your browser if prompted."
),
)
.await;
let resolved_scopes = resolve_oauth_scopes(
/*explicit_scopes*/ None,
server_config.scopes.clone(),
@@ -164,14 +156,6 @@ pub(crate) async fn maybe_install_mcp_dependencies(
if let Err(err) = first_attempt {
if should_retry_without_scopes(&resolved_scopes, &err) {
sess.notify_background_event(
turn_context,
format!(
"Retrying MCP {name} authentication without scopes after provider rejection."
),
)
.await;
if let Err(err) = perform_oauth_login(
&name,
&oauth_config.url,
codex-rs/core/src/session/handlers.rs
-11
View File
@@ -27,7 +27,6 @@ use codex_utils_absolute_path::AbsolutePathBuf;
use crate::review_prompts::resolve_review_request;
use crate::tasks::CompactTask;
use crate::tasks::UndoTask;
use crate::tasks::UserShellCommandMode;
use crate::tasks::UserShellCommandTask;
use crate::tasks::execute_user_shell_command;
@@ -649,12 +648,6 @@ pub async fn list_skills(sess: &Session, sub_id: String, cwds: Vec<PathBuf>, for
sess.send_event_raw(event).await;
}
pub async fn undo(sess: &Arc<Session>, sub_id: String) {
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
sess.spawn_task(turn_context, Vec::new(), UndoTask::new())
.await;
}
pub async fn compact(sess: &Arc<Session>, sub_id: String) {
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
@@ -1118,10 +1111,6 @@ pub(super) async fn submission_loop(
list_skills(&sess, sub.id.clone(), cwds, force_reload).await;
false
}
Op::Undo => {
undo(&sess, sub.id.clone()).await;
false
}
Op::Compact => {
compact(&sess, sub.id.clone()).await;
false
codex-rs/core/src/session/mod.rs
-12
View File
@@ -318,7 +318,6 @@ use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::BackgroundEventEvent;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::DeprecationNoticeEvent;
@@ -2938,17 +2937,6 @@ impl Session {
self.ensure_rollout_materialized().await;
}
pub(crate) async fn notify_background_event(
&self,
turn_context: &TurnContext,
message: impl Into<String>,
) {
let event = EventMsg::BackgroundEvent(BackgroundEventEvent {
message: message.into(),
});
self.send_event(turn_context, event).await;
}
pub(crate) async fn notify_stream_error(
&self,
turn_context: &TurnContext,
codex-rs/core/src/session/turn.rs
+1 -7
View File
@@ -336,7 +336,7 @@ pub(crate) async fn run_turn(
record_additional_contexts(&sess, &turn_context, additional_contexts).await;
if !input.is_empty() {
// Track the previous-turn baseline from the regular user-turn path only so
// standalone tasks (compact/shell/review/undo) cannot suppress future
// standalone tasks (compact/shell/review) cannot suppress future
// model/realtime injections.
sess.set_previous_turn_settings(Some(PreviousTurnSettings {
model: turn_context.model_info.slug.clone(),
@@ -1443,11 +1443,8 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
| EventMsg::TurnComplete(_)
| EventMsg::TokenCount(_)
| EventMsg::UserMessage(_)
| EventMsg::AgentMessageDelta(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningDelta(_)
| EventMsg::AgentReasoningRawContent(_)
| EventMsg::AgentReasoningRawContentDelta(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::SessionConfigured(_)
| EventMsg::ThreadNameUpdated(_)
@@ -1477,9 +1474,6 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
| EventMsg::ElicitationRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::DeprecationNotice(_)
| EventMsg::BackgroundEvent(_)
| EventMsg::UndoStarted(_)
| EventMsg::UndoCompleted(_)
| EventMsg::StreamError(_)
| EventMsg::TurnDiff(_)
| EventMsg::GetHistoryEntryResponse(_)
codex-rs/core/src/tasks/mod.rs
-2
View File
@@ -1,7 +1,6 @@
mod compact;
mod regular;
mod review;
mod undo;
mod user_shell;
use std::sync::Arc;
@@ -57,7 +56,6 @@ use codex_protocol::models::ContentItem;
pub(crate) use compact::CompactTask;
pub(crate) use regular::RegularTask;
pub(crate) use review::ReviewTask;
pub(crate) use undo::UndoTask;
pub(crate) use user_shell::UserShellCommandMode;
pub(crate) use user_shell::UserShellCommandTask;
pub(crate) use user_shell::execute_user_shell_command;
codex-rs/core/src/tasks/review.rs
-2
View File
@@ -6,7 +6,6 @@ use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AgentMessageContentDeltaEvent;
use codex_protocol::protocol::AgentMessageDeltaEvent;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
@@ -162,7 +161,6 @@ async fn process_review_events(
item: TurnItem::AgentMessage(_),
..
})
| EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { .. })
| EventMsg::AgentMessageContentDelta(AgentMessageContentDeltaEvent { .. }) => {}
EventMsg::TurnComplete(task_complete) => {
// Parse review output from the last agent message (if present).
codex-rs/core/src/tasks/undo.rs
-72
View File
@@ -1,72 +0,0 @@
use std::sync::Arc;
use crate::session::turn_context::TurnContext;
use crate::state::TaskKind;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::UndoCompletedEvent;
use codex_protocol::protocol::UndoStartedEvent;
use codex_protocol::user_input::UserInput;
use tokio_util::sync::CancellationToken;
pub(crate) struct UndoTask;
impl UndoTask {
pub(crate) fn new() -> Self {
Self
}
}
impl SessionTask for UndoTask {
fn kind(&self) -> TaskKind {
TaskKind::Regular
}
fn span_name(&self) -> &'static str {
"session_task.undo"
}
async fn run(
self: Arc<Self>,
session: Arc<SessionTaskContext>,
ctx: Arc<TurnContext>,
_input: Vec<UserInput>,
cancellation_token: CancellationToken,
) -> Option<String> {
session
.session
.services
.session_telemetry
.counter("codex.task.undo", /*inc*/ 1, &[]);
let sess = session.clone_session();
sess.send_event(
ctx.as_ref(),
EventMsg::UndoStarted(UndoStartedEvent {
message: Some("Undo in progress...".to_string()),
}),
)
.await;
if cancellation_token.is_cancelled() {
sess.send_event(
ctx.as_ref(),
EventMsg::UndoCompleted(UndoCompletedEvent {
success: false,
message: Some("Undo cancelled.".to_string()),
}),
)
.await;
return None;
}
let completed = UndoCompletedEvent {
success: false,
message: Some("Undo is no longer available.".to_string()),
};
sess.send_event(ctx.as_ref(), EventMsg::UndoCompleted(completed))
.await;
None
}
}
codex-rs/core/src/tools/handlers/agent_jobs.rs
-141
View File
@@ -41,7 +41,6 @@ pub struct BatchJobHandler;
const DEFAULT_AGENT_JOB_CONCURRENCY: usize = 16;
const MAX_AGENT_JOB_CONCURRENCY: usize = 64;
const STATUS_POLL_INTERVAL: Duration = Duration::from_millis(250);
const PROGRESS_EMIT_INTERVAL: Duration = Duration::from_secs(1);
const DEFAULT_AGENT_JOB_ITEM_TIMEOUT: Duration = Duration::from_secs(60 * 30);
#[derive(Debug, Deserialize)]
@@ -83,17 +82,6 @@ struct AgentJobFailureSummary {
last_error: String,
}
#[derive(Debug, Serialize)]
struct AgentJobProgressUpdate {
job_id: String,
total_items: usize,
pending_items: usize,
running_items: usize,
completed_items: usize,
failed_items: usize,
eta_seconds: Option<u64>,
}
#[derive(Debug, Serialize)]
struct ReportAgentJobResultToolResult {
accepted: bool,
@@ -112,73 +100,6 @@ struct ActiveJobItem {
status_rx: Option<Receiver<AgentStatus>>,
}
struct JobProgressEmitter {
started_at: Instant,
last_emit_at: Instant,
last_processed: usize,
last_failed: usize,
}
impl JobProgressEmitter {
fn new() -> Self {
let now = Instant::now();
let last_emit_at = now.checked_sub(PROGRESS_EMIT_INTERVAL).unwrap_or(now);
Self {
started_at: now,
last_emit_at,
last_processed: 0,
last_failed: 0,
}
}
async fn maybe_emit(
&mut self,
session: &Session,
turn: &TurnContext,
job_id: &str,
progress: &codex_state::AgentJobProgress,
force: bool,
) -> anyhow::Result<()> {
let processed = progress.completed_items + progress.failed_items;
let should_emit = force
|| processed != self.last_processed
|| progress.failed_items != self.last_failed
|| self.last_emit_at.elapsed() >= PROGRESS_EMIT_INTERVAL;
if !should_emit {
return Ok(());
}
let elapsed = self.started_at.elapsed().as_secs_f64();
let eta_seconds = if processed > 0 && elapsed > 0.0 {
let remaining = progress.total_items.saturating_sub(processed) as f64;
let rate = processed as f64 / elapsed;
if rate > 0.0 {
Some((remaining / rate).round() as u64)
} else {
None
}
} else {
None
};
let update = AgentJobProgressUpdate {
job_id: job_id.to_string(),
total_items: progress.total_items,
pending_items: progress.pending_items,
running_items: progress.running_items,
completed_items: progress.completed_items,
failed_items: progress.failed_items,
eta_seconds,
};
let payload = serde_json::to_string(&update)?;
session
.notify_background_event(turn, format!("agent_job_progress:{payload}"))
.await;
self.last_emit_at = Instant::now();
self.last_processed = processed;
self.last_failed = progress.failed_items;
Ok(())
}
}
impl ToolHandler for BatchJobHandler {
type Output = FunctionToolOutput;
@@ -358,12 +279,6 @@ mod spawn_agents_on_csv {
"failed to transition agent job {job_id} to running: {err}"
))
})?;
let max_threads = turn.config.agent_max_threads;
let effective_concurrency = options.max_concurrency;
let message = format!(
"agent job concurrency: job_id={job_id} requested={requested_concurrency:?} max_threads={max_threads:?} effective={effective_concurrency}"
);
let _ = session.notify_background_event(&turn, message).await;
if let Err(err) = run_agent_job_loop(
session.clone(),
turn.clone(),
@@ -584,7 +499,6 @@ async fn run_agent_job_loop(
.ok_or_else(|| anyhow::anyhow!("agent job {job_id} was not found"))?;
let runtime_timeout = job_runtime_timeout(&job);
let mut active_items: HashMap<ThreadId, ActiveJobItem> = HashMap::new();
let mut progress_emitter = JobProgressEmitter::new();
recover_running_items(
session.clone(),
db.clone(),
@@ -593,16 +507,6 @@ async fn run_agent_job_loop(
runtime_timeout,
)
.await?;
let initial_progress = db.get_agent_job_progress(job_id.as_str()).await?;
progress_emitter
.maybe_emit(
&session,
&turn,
job_id.as_str(),
&initial_progress,
/*force*/ true,
)
.await?;
let mut cancel_requested = db.is_agent_job_cancelled(job_id.as_str()).await?;
loop {
@@ -610,12 +514,6 @@ async fn run_agent_job_loop(
if !cancel_requested && db.is_agent_job_cancelled(job_id.as_str()).await? {
cancel_requested = true;
let _ = session
.notify_background_event(
&turn,
format!("agent job {job_id} cancellation requested; stopping new workers"),
)
.await;
}
if !cancel_requested && active_items.len() < options.max_concurrency {
@@ -749,20 +647,9 @@ async fn run_agent_job_loop(
)
.await?;
active_items.remove(&thread_id);
let progress = db.get_agent_job_progress(job_id.as_str()).await?;
progress_emitter
.maybe_emit(
&session,
&turn,
job_id.as_str(),
&progress,
/*force*/ false,
)
.await?;
}
}
let progress = db.get_agent_job_progress(job_id.as_str()).await?;
if let Err(err) = export_job_csv_snapshot(db.clone(), &job).await {
let message = format!("auto-export failed: {err}");
db.mark_agent_job_failed(job_id.as_str(), message.as_str())
@@ -771,37 +658,9 @@ async fn run_agent_job_loop(
}
let cancelled = cancel_requested || db.is_agent_job_cancelled(job_id.as_str()).await?;
if cancelled {
let pending_items = progress.pending_items;
let message =
format!("agent job {job_id} cancelled with {pending_items} unprocessed items");
let _ = session.notify_background_event(&turn, message).await;
progress_emitter
.maybe_emit(
&session,
&turn,
job_id.as_str(),
&progress,
/*force*/ true,
)
.await?;
return Ok(());
}
if progress.failed_items > 0 {
let failed_items = progress.failed_items;
let message = format!("agent job completed with {failed_items} failed items");
let _ = session.notify_background_event(&turn, message).await;
}
db.mark_agent_job_completed(job_id.as_str()).await?;
let progress = db.get_agent_job_progress(job_id.as_str()).await?;
progress_emitter
.maybe_emit(
&session,
&turn,
job_id.as_str(),
&progress,
/*force*/ true,
)
.await?;
Ok(())
}