This commit is contained in:
Ahmed Ibrahim
2025-09-10 10:22:06 -07:00
parent 39db113cc9
commit 70dc8db746
8 changed files with 184 additions and 20 deletions
+17 -9
View File
@@ -9,6 +9,7 @@ use std::sync::atomic::AtomicU64;
use std::time::Duration;
use crate::AuthManager;
use crate::conversation_history::EventMsgsHistory;
use crate::event_mapping::map_response_item_to_event_messages;
use async_channel::Receiver;
use async_channel::Sender;
@@ -44,7 +45,7 @@ use crate::client_common::Prompt;
use crate::client_common::ResponseEvent;
use crate::config::Config;
use crate::config_types::ShellEnvironmentPolicy;
use crate::conversation_history::ConversationHistory;
use crate::conversation_history::ResponseItemsHistory;
use crate::environment_context::EnvironmentContext;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
@@ -263,7 +264,8 @@ struct State {
current_task: Option<AgentTask>,
pending_approvals: HashMap<String, oneshot::Sender<ReviewDecision>>,
pending_input: Vec<ResponseInputItem>,
history: ConversationHistory,
response_items: ResponseItemsHistory,
event_msgs: EventMsgsHistory,
token_info: Option<TokenUsageInfo>,
}
@@ -417,7 +419,7 @@ impl Session {
let rollout_path = rollout_recorder.rollout_path.clone();
// Create the mutable state for the Session.
let state = State {
history: ConversationHistory::new(),
response_items: ResponseItemsHistory::new(),
..Default::default()
};
@@ -666,7 +668,7 @@ impl Session {
fn record_into_history(&self, items: &[ResponseItem]) {
self.state
.lock_unchecked()
.history
.response_items
.record_items(items.iter());
}
@@ -908,7 +910,7 @@ impl Session {
/// Build the full turn input by concatenating the current conversation
/// history with additional items for this turn.
pub fn turn_input_with_history(&self, extra: Vec<ResponseItem>) -> Vec<ResponseItem> {
[self.state.lock_unchecked().history.contents(), extra].concat()
[self.state.lock_unchecked().response_items.contents(), extra].concat()
}
/// Returns the input if there was no task running to inject into
@@ -1380,12 +1382,16 @@ async fn submission_loop(
}
Op::GetHistory => {
let sub_id = sub.id.clone();
let entries = {
let state = sess.state.lock_unchecked();
let rolled: Vec<RolloutItem> = (&state.response_items).into();
rolled
};
let event = Event {
id: sub_id.clone(),
msg: EventMsg::ConversationHistory(ConversationHistoryResponseEvent {
conversation_id: sess.conversation_id,
entries: sess.state.lock_unchecked().history.contents(),
history: InitialHistory::Forked(entries),
}),
};
sess.send_event(event).await;
@@ -1928,7 +1934,7 @@ async fn run_compact_task(
{
let mut state = sess.state.lock_unchecked();
state.history.keep_last_messages(1);
state.response_items.keep_last_messages(1);
}
let event = Event {
@@ -2882,7 +2888,9 @@ async fn drain_to_completed(
Ok(ResponseEvent::OutputItemDone(item)) => {
// Record only to in-memory conversation history; avoid state snapshot.
let mut state = sess.state.lock_unchecked();
state.history.record_items(std::slice::from_ref(&item));
state
.response_items
.record_items(std::slice::from_ref(&item));
}
Ok(ResponseEvent::Completed {
response_id: _,
+44 -3
View File
@@ -1,13 +1,15 @@
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
/// Transcript of conversation history
#[derive(Debug, Clone, Default)]
pub(crate) struct ConversationHistory {
pub(crate) struct ResponseItemsHistory {
/// The oldest items are at the beginning of the vector.
items: Vec<ResponseItem>,
}
impl ConversationHistory {
impl ResponseItemsHistory {
pub(crate) fn new() -> Self {
Self { items: Vec::new() }
}
@@ -61,6 +63,45 @@ impl ConversationHistory {
}
}
#[derive(Debug, Clone, Default)]
pub(crate) struct EventMsgsHistory {
items: Vec<EventMsg>,
}
impl EventMsgsHistory {
pub(crate) fn record_items<I>(&mut self, items: I)
where
I: IntoIterator,
I::Item: std::ops::Deref<Target = EventMsg>,
{
for item in items {
self.items.push(item.clone());
}
}
}
impl From<&ResponseItemsHistory> for Vec<RolloutItem> {
fn from(history: &ResponseItemsHistory) -> Self {
history
.items
.iter()
.cloned()
.map(RolloutItem::ResponseItem)
.collect()
}
}
impl From<&EventMsgsHistory> for Vec<RolloutItem> {
fn from(history: &EventMsgsHistory) -> Self {
history
.items
.iter()
.cloned()
.map(RolloutItem::EventMsg)
.collect()
}
}
/// Anything that is not a system message or "reasoning" message is considered
/// an API message.
fn is_api_message(message: &ResponseItem) -> bool {
@@ -103,7 +144,7 @@ mod tests {
#[test]
fn filters_non_api_messages() {
let mut h = ConversationHistory::default();
let mut h = ResponseItemsHistory::default();
// System message is not an API message; Other is ignored.
let system = ResponseItem::Message {
id: None,
+1
View File
@@ -45,6 +45,7 @@ mod event_mapping;
pub use codex_protocol::protocol::InitialHistory;
pub use conversation_manager::ConversationManager;
pub use conversation_manager::NewConversation;
pub use protocol::InitialHistory;
// Re-export common auth types for workspace consumers
pub use auth::AuthManager;
pub use auth::CodexAuth;
+1 -1
View File
@@ -10,7 +10,7 @@ use time::macros::format_description;
use uuid::Uuid;
use super::SESSIONS_SUBDIR;
use crate::protocol::EventMsg;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
+1 -1
View File
@@ -1,6 +1,6 @@
use crate::protocol::EventMsg;
use crate::protocol::RolloutItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::RolloutItem;
/// Whether a rollout `item` should be persisted in rollout files.
#[inline]
@@ -78,7 +78,7 @@ async fn fork_conversation_twice_drops_to_first_message() {
// Capture entries from the base history and compute expected prefixes after each fork.
let entries_after_three = match &base_history {
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { entries, .. }) => {
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { history, .. }) => {
entries.clone()
}
_ => panic!("expected ConversationHistory event"),
@@ -123,10 +123,10 @@ async fn fork_conversation_twice_drops_to_first_message() {
})
.await;
let entries_after_first_fork = match &fork1_history {
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { entries, .. }) => {
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { history, .. }) => {
assert!(matches!(
fork1_history,
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { ref entries, .. }) if *entries == expected_after_first
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { ref history, .. }) if *entries == expected_after_first
));
entries.clone()
}
@@ -149,6 +149,6 @@ async fn fork_conversation_twice_drops_to_first_message() {
.await;
assert!(matches!(
fork2_history,
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { ref entries, .. }) if *entries == expected_after_second
EventMsg::ConversationHistory(ConversationHistoryResponseEvent { ref history, .. }) if *entries == expected_after_second
));
}
+115 -1
View File
@@ -803,7 +803,121 @@ pub struct WebSearchEndEvent {
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub struct ConversationHistoryResponseEvent {
pub conversation_id: ConversationId,
pub entries: Vec<ResponseItem>,
pub history: InitialHistory,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub struct ResumedHistory {
pub conversation_id: ConversationId,
pub history: Vec<RolloutItem>,
pub rollout_path: PathBuf,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
pub enum InitialHistory {
New,
Resumed(ResumedHistory),
Forked(Vec<RolloutItem>),
}
impl InitialHistory {
pub fn get_rollout_items(&self) -> Vec<RolloutItem> {
match self {
InitialHistory::New => Vec::new(),
InitialHistory::Resumed(resumed) => resumed.history.clone(),
InitialHistory::Forked(items) => items.clone(),
}
}
pub fn get_response_items(&self) -> Vec<ResponseItem> {
match self {
InitialHistory::New => Vec::new(),
InitialHistory::Resumed(resumed) => resumed
.history
.iter()
.filter_map(|ri| match ri {
RolloutItem::ResponseItem(item) => Some(item.clone()),
_ => None,
})
.collect(),
InitialHistory::Forked(items) => items
.iter()
.filter_map(|ri| match ri {
RolloutItem::ResponseItem(item) => Some(item.clone()),
_ => None,
})
.collect(),
}
}
pub fn get_event_msgs(&self) -> Option<Vec<EventMsg>> {
match self {
InitialHistory::New => None,
InitialHistory::Resumed(resumed) => Some(
resumed
.history
.iter()
.filter_map(|ri| match ri {
RolloutItem::EventMsg(ev) => Some(ev.clone()),
_ => None,
})
.collect(),
),
InitialHistory::Forked(items) => Some(
items
.iter()
.filter_map(|ri| match ri {
RolloutItem::EventMsg(ev) => Some(ev.clone()),
_ => None,
})
.collect(),
),
}
}
}
#[derive(Serialize, Deserialize, Clone, Default, Debug, TS)]
pub struct SessionMeta {
pub id: ConversationId,
pub timestamp: String,
pub cwd: PathBuf,
pub originator: String,
pub cli_version: String,
pub instructions: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, TS)]
pub struct SessionMetaLine {
#[serde(flatten)]
pub meta: SessionMeta,
#[serde(skip_serializing_if = "Option::is_none")]
pub git: Option<GitInfo>,
}
#[derive(Serialize, Deserialize, Debug, Clone, TS)]
#[serde(tag = "type", content = "payload", rename_all = "snake_case")]
pub enum RolloutItem {
SessionMeta(SessionMetaLine),
ResponseItem(ResponseItem),
EventMsg(EventMsg),
}
#[derive(Serialize, Deserialize, Clone)]
pub struct RolloutLine {
pub timestamp: String,
#[serde(flatten)]
pub item: RolloutItem,
}
#[derive(Serialize, Deserialize, Clone, Debug, TS)]
pub struct GitInfo {
/// Current commit hash (SHA)
#[serde(skip_serializing_if = "Option::is_none")]
pub commit_hash: Option<String>,
/// Current branch name
#[serde(skip_serializing_if = "Option::is_none")]
pub branch: Option<String>,
/// Repository URL (if available from remote)
#[serde(skip_serializing_if = "Option::is_none")]
pub repository_url: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize, TS)]
+1 -1
View File
@@ -288,7 +288,7 @@ impl App {
let cfg = self.chat_widget.config_ref().clone();
// Perform the fork via a thin wrapper for clarity/testability.
let result = self
.perform_fork(ev.entries.clone(), drop_count, cfg.clone())
.perform_fork(ev.history.clone(), drop_count, cfg.clone())
.await;
match result {
Ok(new_conv) => {