openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-01 19:33:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

other approoach

Browse Source
This commit is contained in:
jif-oai
2026-03-30 19:09:53 +02:00
parent 90cabecba9
commit 0257728856
10 changed files with 174 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+31 -6
View File
@@ -44,6 +44,7 @@ use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnOutcome;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::protocol::ViewImageToolCallEvent;
@@ -920,9 +921,19 @@ impl ThreadHistoryBuilder {
}
fn handle_turn_complete(&mut self, payload: &TurnCompleteEvent) {
let mark_completed = |status: &mut TurnStatus| {
if matches!(*status, TurnStatus::Completed | TurnStatus::InProgress) {
*status = TurnStatus::Completed;
let mark_finished = |turn: &mut PendingTurn| match &payload.outcome {
TurnOutcome::Succeeded { .. } => {
if matches!(turn.status, TurnStatus::Completed | TurnStatus::InProgress) {
turn.status = TurnStatus::Completed;
}
}
TurnOutcome::Failed { error } => {
turn.status = TurnStatus::Failed;
turn.error = Some(V2TurnError {
message: error.message.clone(),
codex_error_info: error.codex_error_info.clone().map(Into::into),
additional_details: None,
});
}
};
@@ -932,7 +943,7 @@ impl ThreadHistoryBuilder {
.as_mut()
.filter(|turn| turn.id == payload.turn_id)
{
mark_completed(&mut current_turn.status);
mark_finished(current_turn);
self.finish_current_turn();
return;
}
@@ -942,13 +953,27 @@ impl ThreadHistoryBuilder {
.iter_mut()
.find(|turn| turn.id == payload.turn_id)
{
mark_completed(&mut turn.status);
match &payload.outcome {
TurnOutcome::Succeeded { .. } => {
if matches!(turn.status, TurnStatus::Completed | TurnStatus::InProgress) {
turn.status = TurnStatus::Completed;
}
}
TurnOutcome::Failed { error } => {
turn.status = TurnStatus::Failed;
turn.error = Some(V2TurnError {
message: error.message.clone(),
codex_error_info: error.codex_error_info.clone().map(Into::into),
additional_details: None,
});
}
}
return;
}
// If the completion event cannot be matched, apply it to the active turn.
if let Some(current_turn) = self.current_turn.as_mut() {
mark_completed(&mut current_turn.status);
mark_finished(current_turn);
self.finish_current_turn();
}
}
codex-rs/app-server/src/bespoke_event_handling.rs
+26 -3
View File
@@ -128,6 +128,7 @@ use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::ReviewOutputEvent;
use codex_protocol::protocol::TokenCountEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnDiffEvent;
use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile;
@@ -297,11 +298,22 @@ pub(crate) async fn apply_bespoke_event_handling(
EventMsg::TurnComplete(_ev) => {
// All per-thread requests are bound to a turn, so abort them.
outgoing.abort_pending_server_requests().await;
let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some();
let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some()
|| matches!(
_ev.error(),
Some(error) if error.affects_turn_status()
);
thread_watch_manager
.note_turn_completed(&conversation_id.to_string(), turn_failed)
.await;
handle_turn_complete(conversation_id, event_turn_id, &outgoing, &thread_state).await;
handle_turn_complete(
conversation_id,
event_turn_id,
&_ev,
&outgoing,
&thread_state,
)
.await;
}
EventMsg::SkillsUpdateAvailable => {
if let ApiVersion::V2 = api_version {
@@ -2100,6 +2112,7 @@ async fn find_and_remove_turn_summary(
async fn handle_turn_complete(
conversation_id: ThreadId,
event_turn_id: String,
event: &TurnCompleteEvent,
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
@@ -2107,7 +2120,17 @@ async fn handle_turn_complete(
let (status, error) = match turn_summary.last_error {
Some(error) => (TurnStatus::Failed, Some(error)),
None => (TurnStatus::Completed, None),
None => match event.error() {
Some(error) if error.affects_turn_status() => (
TurnStatus::Failed,
Some(TurnError {
message: error.message.clone(),
codex_error_info: error.codex_error_info.clone().map(Into::into),
additional_details: None,
}),
),
_ => (TurnStatus::Completed, None),
},
};
emit_turn_completed_with_status(conversation_id, event_turn_id, status, error, outgoing).await;
codex-rs/core/src/agent/status.rs
+8 -2
View File
@@ -1,19 +1,25 @@
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnOutcome;
/// Derive the next agent status from a single emitted event.
/// Returns `None` when the event does not affect status tracking.
pub(crate) fn agent_status_from_event(msg: &EventMsg) -> Option<AgentStatus> {
match msg {
EventMsg::TurnStarted(_) => Some(AgentStatus::Running),
EventMsg::TurnComplete(ev) => Some(AgentStatus::Completed(ev.last_agent_message.clone())),
EventMsg::TurnComplete(ev) => Some(match &ev.outcome {
TurnOutcome::Succeeded { last_agent_message } => {
AgentStatus::Completed(last_agent_message.clone())
}
TurnOutcome::Failed { error } => AgentStatus::Errored(error.message.clone()),
}),
EventMsg::TurnAborted(ev) => match ev.reason {
codex_protocol::protocol::TurnAbortReason::Interrupted => {
Some(AgentStatus::Interrupted)
}
_ => Some(AgentStatus::Errored(format!("{:?}", ev.reason))),
},
EventMsg::Error(ev) => Some(AgentStatus::Errored(ev.message.clone())),
EventMsg::Error(_) => None,
EventMsg::ShutdownComplete => Some(AgentStatus::Shutdown),
_ => None,
}
codex-rs/core/src/guardian/review_session.rs
+10 -16
View File
@@ -18,6 +18,7 @@ use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TurnOutcome;
use codex_protocol::user_input::UserInput;
use serde_json::Value;
use tokio::sync::Mutex;
@@ -578,8 +579,6 @@ async fn wait_for_guardian_review(
) -> (GuardianReviewSessionOutcome, bool) {
let timeout = tokio::time::sleep_until(deadline);
tokio::pin!(timeout);
let mut last_error_message: Option<String> = None;
loop {
tokio::select! {
_ = &mut timeout => {
@@ -600,21 +599,16 @@ async fn wait_for_guardian_review(
match event {
Ok(event) => match event.msg {
EventMsg::TurnComplete(turn_complete) => {
if turn_complete.last_agent_message.is_none()
&& let Some(error_message) = last_error_message
{
return (
GuardianReviewSessionOutcome::Completed(Err(anyhow!(error_message))),
return match turn_complete.outcome {
TurnOutcome::Succeeded { last_agent_message } => (
GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)),
true,
);
}
return (
GuardianReviewSessionOutcome::Completed(Ok(turn_complete.last_agent_message)),
true,
);
}
EventMsg::Error(error) => {
last_error_message = Some(error.message);
),
TurnOutcome::Failed { error } => (
GuardianReviewSessionOutcome::Completed(Err(anyhow!(error.message))),
true,
),
};
}
EventMsg::TurnAborted(_) => {
return (GuardianReviewSessionOutcome::Aborted, true);
codex-rs/core/src/tasks/mod.rs
+7 -7
View File
@@ -438,13 +438,13 @@ impl Session {
);
}
let event = match completion {
TaskCompletion::Completed(last_agent_message) => {
EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: turn_context.sub_id.clone(),
last_agent_message,
})
}
TaskCompletion::Failed(error) => EventMsg::Error(error),
TaskCompletion::Completed(last_agent_message) => EventMsg::TurnComplete(
TurnCompleteEvent::succeeded(turn_context.sub_id.clone(), last_agent_message),
),
TaskCompletion::Failed(error) => EventMsg::TurnComplete(TurnCompleteEvent::failed(
turn_context.sub_id.clone(),
error,
)),
};
self.send_event(turn_context.as_ref(), event).await;
}
codex-rs/core/src/tasks/review.rs
+13 -6
View File
@@ -15,6 +15,7 @@ use codex_protocol::protocol::ExitedReviewModeEvent;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::ReviewOutputEvent;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TurnOutcome;
use codex_utils_template::Template;
use tokio_util::sync::CancellationToken;
@@ -167,12 +168,18 @@ async fn process_review_events(
| EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { .. })
| EventMsg::AgentMessageContentDelta(AgentMessageContentDeltaEvent { .. }) => {}
EventMsg::TurnComplete(task_complete) => {
// Parse review output from the last agent message (if present).
let out = task_complete
.last_agent_message
.as_deref()
.map(parse_review_output_event);
return out;
return match task_complete.outcome {
TurnOutcome::Succeeded { last_agent_message } => {
last_agent_message.as_deref().map(parse_review_output_event)
}
TurnOutcome::Failed { error } => {
session
.clone_session()
.send_event(ctx.as_ref(), EventMsg::Error(error))
.await;
None
}
};
}
EventMsg::TurnAborted(_) => {
// Cancellation or abort: consumer will finalize with None.
codex-rs/mcp-server/src/codex_tool_runner.rs
+9 -9
View File
@@ -22,6 +22,7 @@ use codex_protocol::protocol::ExecApprovalRequestEvent;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::Submission;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnOutcome;
use codex_protocol::user_input::UserInput;
use rmcp::model::CallToolResult;
use rmcp::model::Content;
@@ -291,16 +292,15 @@ async fn run_codex_tool_session_inner(
.await;
continue;
}
EventMsg::TurnComplete(TurnCompleteEvent {
last_agent_message, ..
}) => {
let text = match last_agent_message {
Some(msg) => msg,
None => "".to_string(),
EventMsg::TurnComplete(TurnCompleteEvent { outcome, .. }) => {
let (text, is_error) = match outcome {
TurnOutcome::Succeeded { last_agent_message } => {
(last_agent_message.unwrap_or_default(), None)
}
TurnOutcome::Failed { error } => (error.message, Some(true)),
};
let result = create_call_tool_result_with_thread_id(
thread_id, text, /*is_error*/ None,
);
let result =
create_call_tool_result_with_thread_id(thread_id, text, is_error);
outgoing.send_response(request_id.clone(), result).await;
// unregister the id so we don't keep it in the map
running_requests_id_to_codex_uuid
codex-rs/protocol/src/protocol.rs
+39 -1
View File
@@ -1858,10 +1858,48 @@ pub struct ModelRerouteEvent {
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub struct ContextCompactedEvent;
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
#[serde(untagged)]
pub enum TurnOutcome {
Succeeded { last_agent_message: Option<String> },
Failed { error: ErrorEvent },
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
pub struct TurnCompleteEvent {
pub turn_id: String,
pub last_agent_message: Option<String>,
#[serde(flatten)]
pub outcome: TurnOutcome,
}
impl TurnCompleteEvent {
pub fn succeeded(turn_id: String, last_agent_message: Option<String>) -> Self {
Self {
turn_id,
outcome: TurnOutcome::Succeeded { last_agent_message },
}
}
pub fn failed(turn_id: String, error: ErrorEvent) -> Self {
Self {
turn_id,
outcome: TurnOutcome::Failed { error },
}
}
pub fn last_agent_message(&self) -> Option<&str> {
match &self.outcome {
TurnOutcome::Succeeded { last_agent_message } => last_agent_message.as_deref(),
TurnOutcome::Failed { .. } => None,
}
}
pub fn error(&self) -> Option<&ErrorEvent> {
match &self.outcome {
TurnOutcome::Succeeded { .. } => None,
TurnOutcome::Failed { error } => Some(error),
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
codex-rs/tui/src/app/app_server_adapter.rs
+22 -21
View File
@@ -731,17 +731,17 @@ fn turn_snapshot_events(
///
/// - `Completed` → `TurnComplete`
/// - `Interrupted` → `TurnAborted { reason: Interrupted }`
/// - `Failed` → `Error` (if present) then `TurnComplete`
/// - `Failed` → `TurnComplete` with failed outcome
/// - `InProgress` → no events (the turn is still running)
#[cfg(test)]
fn append_terminal_turn_events(events: &mut Vec<Event>, turn: &Turn, include_failed_error: bool) {
match turn.status {
TurnStatus::Completed => events.push(Event {
id: String::new(),
msg: EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: turn.id.clone(),
last_agent_message: None,
}),
msg: EventMsg::TurnComplete(TurnCompleteEvent::succeeded(
turn.id.clone(),
/*last_agent_message*/ None,
)),
}),
TurnStatus::Interrupted => events.push(Event {
id: String::new(),
@@ -751,24 +751,25 @@ fn append_terminal_turn_events(events: &mut Vec<Event>, turn: &Turn, include_fai
}),
}),
TurnStatus::Failed => {
if include_failed_error && let Some(error) = &turn.error {
events.push(Event {
id: String::new(),
msg: EventMsg::Error(ErrorEvent {
message: error.message.clone(),
codex_error_info: error
.codex_error_info
.clone()
.and_then(app_server_codex_error_info_to_core),
}),
});
}
let _ = include_failed_error;
events.push(Event {
id: String::new(),
msg: EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: turn.id.clone(),
last_agent_message: None,
}),
msg: EventMsg::TurnComplete(TurnCompleteEvent::failed(
turn.id.clone(),
ErrorEvent {
message: turn
.error
.as_ref()
.map(|error| error.message.clone())
.unwrap_or_default(),
codex_error_info: turn.error.as_ref().and_then(|error| {
error
.codex_error_info
.clone()
.and_then(app_server_codex_error_info_to_core)
}),
},
)),
});
}
TurnStatus::InProgress => {
codex-rs/tui/src/chatwidget.rs
+9 -6
View File
@@ -193,10 +193,10 @@ use codex_protocol::protocol::TerminalInteractionEvent;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use codex_protocol::protocol::TurnAbortReason;
#[cfg(test)]
use codex_protocol::protocol::TurnCompleteEvent;
#[cfg(test)]
use codex_protocol::protocol::TurnDiffEvent;
use codex_protocol::protocol::TurnOutcome;
#[cfg(test)]
use codex_protocol::protocol::UndoCompletedEvent;
#[cfg(test)]
@@ -6927,11 +6927,14 @@ impl ChatWidget {
self.on_task_started();
}
}
EventMsg::TurnComplete(TurnCompleteEvent {
last_agent_message, ..
}) => {
self.on_task_complete(last_agent_message, from_replay);
}
EventMsg::TurnComplete(TurnCompleteEvent { outcome, .. }) => match outcome {
TurnOutcome::Succeeded { last_agent_message } => {
self.on_task_complete(last_agent_message, from_replay);
}
TurnOutcome::Failed { error } => {
self.on_error(error.message);
}
},
EventMsg::TokenCount(ev) => {
self.set_token_info(ev.info);
self.on_rate_limit_snapshot(ev.rate_limits);