Replay EventMsgs from Response Items when resuming a session with history. (#3123)
### Overview This PR introduces the following changes: 1. Adds a unified mechanism to convert ResponseItem into EventMsg. 2. Ensures that when a session is initialized with initial history, a vector of EventMsg is sent along with the session configuration. This allows clients to re-render the UI accordingly. 3. Added integration testing ### Caveats This implementation does not send every EventMsg that was previously dispatched to clients. The excluded events fall into two categories: • “Arguably” rolled-out events Examples include tool calls and apply-patch calls. While these events are conceptually rolled out, we currently only roll out ResponseItems. These events are already being handled elsewhere and transformed into EventMsg before being sent. • Non-rolled-out events Certain events such as TurnDiff, Error, and TokenCount are not rolled out at all. ### Future Directions At present, resuming a session involves maintaining two states: • UI State Clients can replay most of the important UI from the provided EventMsg history. • Model State The model receives the complete session history to reconstruct its internal state. This design provides a solid foundation. If, in the future, more precise UI reconstruction is needed, we have two potential paths: 1. Introduce a third data structure that allows us to derive both ResponseItems and EventMsgs. 2. Clearly divide responsibilities: the core system ensures the integrity of the model state, while clients are responsible for reconstructing the UI.
This commit is contained in:
Ahmed Ibrahim
GitHub
@@ -9,6 +9,7 @@ use std::sync::atomic::AtomicU64;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::AuthManager;
|
||||
use crate::event_mapping::map_response_item_to_event_messages;
|
||||
use async_channel::Receiver;
|
||||
use async_channel::Sender;
|
||||
use codex_apply_patch::ApplyPatchAction;
|
||||
@@ -75,9 +76,7 @@ use crate::project_doc::get_user_instructions;
|
||||
use crate::protocol::AgentMessageDeltaEvent;
|
||||
use crate::protocol::AgentMessageEvent;
|
||||
use crate::protocol::AgentReasoningDeltaEvent;
|
||||
use crate::protocol::AgentReasoningEvent;
|
||||
use crate::protocol::AgentReasoningRawContentDeltaEvent;
|
||||
use crate::protocol::AgentReasoningRawContentEvent;
|
||||
use crate::protocol::AgentReasoningSectionBreakEvent;
|
||||
use crate::protocol::ApplyPatchApprovalRequestEvent;
|
||||
use crate::protocol::AskForApproval;
|
||||
@@ -102,7 +101,6 @@ use crate::protocol::Submission;
|
||||
use crate::protocol::TaskCompleteEvent;
|
||||
use crate::protocol::TurnDiffEvent;
|
||||
use crate::protocol::WebSearchBeginEvent;
|
||||
use crate::protocol::WebSearchEndEvent;
|
||||
use crate::rollout::RolloutRecorder;
|
||||
use crate::safety::SafetyCheck;
|
||||
use crate::safety::assess_command_safety;
|
||||
@@ -117,12 +115,9 @@ use codex_protocol::custom_prompts::CustomPrompt;
|
||||
use codex_protocol::models::ContentItem;
|
||||
use codex_protocol::models::FunctionCallOutputPayload;
|
||||
use codex_protocol::models::LocalShellAction;
|
||||
use codex_protocol::models::ReasoningItemContent;
|
||||
use codex_protocol::models::ReasoningItemReasoningSummary;
|
||||
use codex_protocol::models::ResponseInputItem;
|
||||
use codex_protocol::models::ResponseItem;
|
||||
use codex_protocol::models::ShellToolCallParams;
|
||||
use codex_protocol::models::WebSearchAction;
|
||||
|
||||
// A convenience extension trait for acquiring mutex locks where poisoning is
|
||||
// unrecoverable and should abort the program. This avoids scattered `.unwrap()`
|
||||
@@ -199,6 +194,7 @@ impl Codex {
|
||||
config.clone(),
|
||||
auth_manager.clone(),
|
||||
tx_event.clone(),
|
||||
conversation_history.clone(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
@@ -361,6 +357,7 @@ impl Session {
|
||||
config: Arc<Config>,
|
||||
auth_manager: Arc<AuthManager>,
|
||||
tx_event: Sender<Event>,
|
||||
initial_history: InitialHistory,
|
||||
) -> anyhow::Result<(Arc<Self>, TurnContext)> {
|
||||
let session_id = Uuid::new_v4();
|
||||
let ConfigureSession {
|
||||
@@ -480,6 +477,11 @@ impl Session {
|
||||
});
|
||||
|
||||
// Dispatch the SessionConfiguredEvent first and then report any errors.
|
||||
// If resuming, include converted initial messages in the payload so UIs can render them immediately.
|
||||
let initial_messages = match &initial_history {
|
||||
InitialHistory::New => None,
|
||||
InitialHistory::Resumed(items) => Some(sess.build_initial_messages(items)),
|
||||
};
|
||||
let events = std::iter::once(Event {
|
||||
id: INITIAL_SUBMIT_ID.to_owned(),
|
||||
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
|
||||
@@ -487,6 +489,7 @@ impl Session {
|
||||
model,
|
||||
history_log_id,
|
||||
history_entry_count,
|
||||
initial_messages,
|
||||
}),
|
||||
})
|
||||
.chain(post_session_configured_error_events.into_iter());
|
||||
@@ -552,6 +555,17 @@ impl Session {
|
||||
self.record_conversation_items(&items).await;
|
||||
}
|
||||
|
||||
/// build the initial messages vector for SessionConfigured by converting
|
||||
/// ResponseItems into EventMsg.
|
||||
fn build_initial_messages(&self, items: &[ResponseItem]) -> Vec<EventMsg> {
|
||||
items
|
||||
.iter()
|
||||
.flat_map(|item| {
|
||||
map_response_item_to_event_messages(item, self.show_raw_agent_reasoning)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Sends the given event to the client and swallows the send event, if
|
||||
/// any, logging it as an error.
|
||||
pub(crate) async fn send_event(&self, event: Event) {
|
||||
@@ -1903,53 +1917,6 @@ async fn handle_response_item(
|
||||
) -> CodexResult<Option<ResponseInputItem>> {
|
||||
debug!(?item, "Output item");
|
||||
let output = match item {
|
||||
ResponseItem::Message { content, .. } => {
|
||||
for item in content {
|
||||
if let ContentItem::OutputText { text } = item {
|
||||
let event = Event {
|
||||
id: sub_id.to_string(),
|
||||
msg: EventMsg::AgentMessage(AgentMessageEvent { message: text }),
|
||||
};
|
||||
sess.tx_event.send(event).await.ok();
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
ResponseItem::Reasoning {
|
||||
id: _,
|
||||
summary,
|
||||
content,
|
||||
encrypted_content: _,
|
||||
} => {
|
||||
for item in summary {
|
||||
let text = match item {
|
||||
ReasoningItemReasoningSummary::SummaryText { text } => text,
|
||||
};
|
||||
let event = Event {
|
||||
id: sub_id.to_string(),
|
||||
msg: EventMsg::AgentReasoning(AgentReasoningEvent { text }),
|
||||
};
|
||||
sess.tx_event.send(event).await.ok();
|
||||
}
|
||||
if sess.show_raw_agent_reasoning
|
||||
&& let Some(content) = content
|
||||
{
|
||||
for item in content {
|
||||
let text = match item {
|
||||
ReasoningItemContent::ReasoningText { text } => text,
|
||||
ReasoningItemContent::Text { text } => text,
|
||||
};
|
||||
let event = Event {
|
||||
id: sub_id.to_string(),
|
||||
msg: EventMsg::AgentReasoningRawContent(AgentReasoningRawContentEvent {
|
||||
text,
|
||||
}),
|
||||
};
|
||||
sess.tx_event.send(event).await.ok();
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
ResponseItem::FunctionCall {
|
||||
name,
|
||||
arguments,
|
||||
@@ -2039,12 +2006,14 @@ async fn handle_response_item(
|
||||
debug!("unexpected CustomToolCallOutput from stream");
|
||||
None
|
||||
}
|
||||
ResponseItem::WebSearchCall { id, action, .. } => {
|
||||
if let WebSearchAction::Search { query } = action {
|
||||
let call_id = id.unwrap_or_else(|| "".to_string());
|
||||
ResponseItem::Message { .. }
|
||||
| ResponseItem::Reasoning { .. }
|
||||
| ResponseItem::WebSearchCall { .. } => {
|
||||
let msgs = map_response_item_to_event_messages(&item, sess.show_raw_agent_reasoning);
|
||||
for msg in msgs {
|
||||
let event = Event {
|
||||
id: sub_id.to_string(),
|
||||
msg: EventMsg::WebSearchEnd(WebSearchEndEvent { call_id, query }),
|
||||
msg,
|
||||
};
|
||||
sess.tx_event.send(event).await.ok();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user