valknar/llmx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: annotate conversations with model_provider for filtering (#5658)

Browse Source 
Because conversations that use the Responses API can have encrypted
reasoning messages, trying to resume a conversation with a different
provider could lead to confusing "failed to decrypt" errors. (This is
reproducible by starting a conversation using ChatGPT login and resuming
it as a conversation that uses OpenAI models via Azure.)

This changes `ListConversationsParams` to take a `model_providers:
Option<Vec<String>>` and adds `model_provider` on each
`ConversationSummary` it returns so these cases can be disambiguated.

Note this ended up making changes to
`codex-rs/core/src/rollout/tests.rs` because it had a number of cases
where it expected `Some` for the value of `next_cursor`, but the list of
rollouts was complete, so according to this docstring:


bcd64c7e72/codex-rs/app-server-protocol/src/protocol.rs (L334-L337)

If there are no more items to return, then `next_cursor` should be
`None`. This PR updates that logic.






---
[//]: # (BEGIN SAPLING FOOTER)
Stack created with [Sapling](https://sapling-scm.com). Best reviewed
with [ReviewStack](https://reviewstack.dev/openai/codex/pull/5658).
* #5803
* #5793
* __->__ #5658
This commit is contained in:
Michael Bolin
2025-10-27 02:03:30 -07:00
committed by GitHub
parent f178805252
commit 5907422d65
14 changed files with 560 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
codex-rs/core/src/rollout/list.rs
View File

@@ -54,6 +54,7 @@ struct HeadTailSummary {
saw_session_meta: bool,
saw_user_event: bool,
source: Option<SessionSource>,
model_provider: Option<String>,
created_at: Option<String>,
updated_at: Option<String>,
}
@@ -109,6 +110,8 @@ pub(crate) async fn get_conversations(
page_size: usize,
cursor: Option<&Cursor>,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
) -> io::Result<ConversationsPage> {
let mut root = codex_home.to_path_buf();
root.push(SESSIONS_SUBDIR);
@@ -124,8 +127,17 @@ pub(crate) async fn get_conversations(
let anchor = cursor.cloned();
let result =
traverse_directories_for_paths(root.clone(), page_size, anchor, allowed_sources).await?;
let provider_matcher =
model_providers.and_then(|filters| ProviderMatcher::new(filters, default_provider));
let result = traverse_directories_for_paths(
root.clone(),
page_size,
anchor,
allowed_sources,
provider_matcher.as_ref(),
)
.await?;
Ok(result)
}
@@ -145,6 +157,7 @@ async fn traverse_directories_for_paths(
page_size: usize,
anchor: Option<Cursor>,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
) -> io::Result<ConversationsPage> {
let mut items: Vec<ConversationItem> = Vec::with_capacity(page_size);
let mut scanned_files = 0usize;
@@ -153,6 +166,7 @@ async fn traverse_directories_for_paths(
Some(c) => (c.ts, c.id),
None => (OffsetDateTime::UNIX_EPOCH, Uuid::nil()),
};
let mut more_matches_available = false;
let year_dirs = collect_dirs_desc(&root, |s| s.parse::<u16>().ok()).await?;
@@ -184,6 +198,7 @@ async fn traverse_directories_for_paths(
for (ts, sid, _name_str, path) in day_files.into_iter() {
scanned_files += 1;
if scanned_files >= MAX_SCAN_FILES && items.len() >= page_size {
more_matches_available = true;
break 'outer;
}
if !anchor_passed {
@@ -194,6 +209,7 @@ async fn traverse_directories_for_paths(
}
}
if items.len() == page_size {
more_matches_available = true;
break 'outer;
}
// Read head and simultaneously detect message events within the same
@@ -208,6 +224,11 @@ async fn traverse_directories_for_paths(
{
continue;
}
if let Some(matcher) = provider_matcher
&& !matcher.matches(summary.model_provider.as_deref())
{
continue;
}
// Apply filters: must have session meta and at least one user message event
if summary.saw_session_meta && summary.saw_user_event {
let HeadTailSummary {
@@ -231,12 +252,21 @@ async fn traverse_directories_for_paths(
}
}
let next = build_next_cursor(&items);
let reached_scan_cap = scanned_files >= MAX_SCAN_FILES;
if reached_scan_cap && !items.is_empty() {
more_matches_available = true;
}
let next = if more_matches_available {
build_next_cursor(&items)
} else {
None
};
Ok(ConversationsPage {
items,
next_cursor: next,
num_scanned_files: scanned_files,
reached_scan_cap: scanned_files >= MAX_SCAN_FILES,
reached_scan_cap,
})
}
@@ -328,6 +358,32 @@ fn parse_timestamp_uuid_from_filename(name: &str) -> Option<(OffsetDateTime, Uui
Some((ts, uuid))
}
struct ProviderMatcher<'a> {
filters: &'a [String],
matches_default_provider: bool,
}
impl<'a> ProviderMatcher<'a> {
fn new(filters: &'a [String], default_provider: &'a str) -> Option<Self> {
if filters.is_empty() {
return None;
}
let matches_default_provider = filters.iter().any(|provider| provider == default_provider);
Some(Self {
filters,
matches_default_provider,
})
}
fn matches(&self, session_provider: Option<&str>) -> bool {
match session_provider {
Some(provider) => self.filters.iter().any(|candidate| candidate == provider),
None => self.matches_default_provider,
}
}
}
async fn read_head_and_tail(
path: &Path,
head_limit: usize,
@@ -354,6 +410,7 @@ async fn read_head_and_tail(
match rollout_line.item {
RolloutItem::SessionMeta(session_meta_line) => {
summary.source = Some(session_meta_line.meta.source);
summary.model_provider = session_meta_line.meta.model_provider.clone();
summary.created_at = summary
.created_at
.clone()

13
codex-rs/core/src/rollout/recorder.rs
View File

@@ -97,8 +97,18 @@ impl RolloutRecorder {
page_size: usize,
cursor: Option<&Cursor>,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
) -> std::io::Result<ConversationsPage> {
get_conversations(codex_home, page_size, cursor, allowed_sources).await
get_conversations(
codex_home,
page_size,
cursor,
allowed_sources,
model_providers,
default_provider,
)
.await
}
/// Attempt to create a new [`RolloutRecorder`]. If the sessions directory
@@ -137,6 +147,7 @@ impl RolloutRecorder {
cli_version: env!("CARGO_PKG_VERSION").to_string(),
instructions,
source,
model_provider: Some(config.model_provider_id.clone()),
}),
)
}

268
codex-rs/core/src/rollout/tests.rs
View File

@@ -32,6 +32,14 @@ use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::UserMessageEvent;
const NO_SOURCE_FILTER: &[SessionSource] = &[];
const TEST_PROVIDER: &str = "test-provider";
fn provider_vec(providers: &[&str]) -> Vec<String> {
providers
.iter()
.map(std::string::ToString::to_string)
.collect()
}
fn write_session_file(
root: &Path,
@@ -39,6 +47,24 @@ fn write_session_file(
uuid: Uuid,
num_records: usize,
source: Option<SessionSource>,
) -> std::io::Result<(OffsetDateTime, Uuid)> {
write_session_file_with_provider(
root,
ts_str,
uuid,
num_records,
source,
Some("test-provider"),
)
}
fn write_session_file_with_provider(
root: &Path,
ts_str: &str,
uuid: Uuid,
num_records: usize,
source: Option<SessionSource>,
model_provider: Option<&str>,
) -> std::io::Result<(OffsetDateTime, Uuid)> {
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
@@ -68,6 +94,9 @@ fn write_session_file(
if let Some(source) = source {
payload["source"] = serde_json::to_value(source).unwrap();
}
if let Some(provider) = model_provider {
payload["model_provider"] = serde_json::Value::String(provider.to_string());
}
let meta = serde_json::json!({
"timestamp": ts_str,
@@ -134,9 +163,17 @@ async fn test_list_conversations_latest_first() {
)
.unwrap();
let page = get_conversations(home, 10, None, INTERACTIVE_SESSION_SOURCES)
.await
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
home,
10,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
// Build expected objects
let p1 = home
@@ -166,6 +203,7 @@ async fn test_list_conversations_latest_first() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let head_2 = vec![serde_json::json!({
"id": u2,
@@ -175,6 +213,7 @@ async fn test_list_conversations_latest_first() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let head_1 = vec![serde_json::json!({
"id": u1,
@@ -184,11 +223,9 @@ async fn test_list_conversations_latest_first() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let expected_cursor: Cursor =
serde_json::from_str(&format!("\"2025-01-01T12-00-00|{u1}\"")).unwrap();
let expected = ConversationsPage {
items: vec![
ConversationItem {
@@ -213,7 +250,7 @@ async fn test_list_conversations_latest_first() {
updated_at: Some("2025-01-01T12-00-00".into()),
},
],
next_cursor: Some(expected_cursor),
next_cursor: None,
num_scanned_files: 3,
reached_scan_cap: false,
};
@@ -275,9 +312,17 @@ async fn test_pagination_cursor() {
)
.unwrap();
let page1 = get_conversations(home, 2, None, INTERACTIVE_SESSION_SOURCES)
.await
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page1 = get_conversations(
home,
2,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
let p5 = home
.join("sessions")
.join("2025")
@@ -298,6 +343,7 @@ async fn test_pagination_cursor() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let head_4 = vec![serde_json::json!({
"id": u4,
@@ -307,6 +353,7 @@ async fn test_pagination_cursor() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let expected_cursor1: Cursor =
serde_json::from_str(&format!("\"2025-03-04T09-00-00|{u4}\"")).unwrap();
@@ -338,6 +385,8 @@ async fn test_pagination_cursor() {
2,
page1.next_cursor.as_ref(),
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
@@ -361,6 +410,7 @@ async fn test_pagination_cursor() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let head_2 = vec![serde_json::json!({
"id": u2,
@@ -370,6 +420,7 @@ async fn test_pagination_cursor() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let expected_cursor2: Cursor =
serde_json::from_str(&format!("\"2025-03-02T09-00-00|{u2}\"")).unwrap();
@@ -401,6 +452,8 @@ async fn test_pagination_cursor() {
2,
page2.next_cursor.as_ref(),
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
@@ -418,9 +471,8 @@ async fn test_pagination_cursor() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let expected_cursor3: Cursor =
serde_json::from_str(&format!("\"2025-03-01T09-00-00|{u1}\"")).unwrap();
let expected_page3 = ConversationsPage {
items: vec![ConversationItem {
path: p1,
@@ -429,7 +481,7 @@ async fn test_pagination_cursor() {
created_at: Some("2025-03-01T09-00-00".into()),
updated_at: Some("2025-03-01T09-00-00".into()),
}],
next_cursor: Some(expected_cursor3),
next_cursor: None,
num_scanned_files: 5, // scanned 05, 04 (anchor), 03, 02 (anchor), 01
reached_scan_cap: false,
};
@@ -445,9 +497,17 @@ async fn test_get_conversation_contents() {
let ts = "2025-04-01T10-30-00";
write_session_file(home, ts, uuid, 2, Some(SessionSource::VSCode)).unwrap();
let page = get_conversations(home, 1, None, INTERACTIVE_SESSION_SOURCES)
.await
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
home,
1,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
let path = &page.items[0].path;
let content = get_conversation(path).await.unwrap();
@@ -467,8 +527,8 @@ async fn test_get_conversation_contents() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})];
let expected_cursor: Cursor = serde_json::from_str(&format!("\"{ts}|{uuid}\"")).unwrap();
let expected_page = ConversationsPage {
items: vec![ConversationItem {
path: expected_path,
@@ -477,7 +537,7 @@ async fn test_get_conversation_contents() {
created_at: Some(ts.into()),
updated_at: Some(ts.into()),
}],
next_cursor: Some(expected_cursor),
next_cursor: None,
num_scanned_files: 1,
reached_scan_cap: false,
};
@@ -495,6 +555,7 @@ async fn test_get_conversation_contents() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
}
});
let user_event = serde_json::json!({
@@ -532,6 +593,7 @@ async fn test_tail_includes_last_response_items() -> Result<()> {
originator: "test_originator".into(),
cli_version: "test_version".into(),
source: SessionSource::VSCode,
model_provider: Some("test-provider".into()),
},
git: None,
}),
@@ -563,7 +625,16 @@ async fn test_tail_includes_last_response_items() -> Result<()> {
}
drop(file);
let page = get_conversations(home, 1, None, INTERACTIVE_SESSION_SOURCES).await?;
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
home,
1,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await?;
let item = page.items.first().expect("conversation item");
let tail_len = item.tail.len();
assert_eq!(tail_len, 10usize.min(total_messages));
@@ -615,6 +686,7 @@ async fn test_tail_handles_short_sessions() -> Result<()> {
originator: "test_originator".into(),
cli_version: "test_version".into(),
source: SessionSource::VSCode,
model_provider: Some("test-provider".into()),
},
git: None,
}),
@@ -645,7 +717,16 @@ async fn test_tail_handles_short_sessions() -> Result<()> {
}
drop(file);
let page = get_conversations(home, 1, None, INTERACTIVE_SESSION_SOURCES).await?;
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
home,
1,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await?;
let tail = &page.items.first().expect("conversation item").tail;
assert_eq!(tail.len(), 3);
@@ -699,6 +780,7 @@ async fn test_tail_skips_trailing_non_responses() -> Result<()> {
originator: "test_originator".into(),
cli_version: "test_version".into(),
source: SessionSource::VSCode,
model_provider: Some("test-provider".into()),
},
git: None,
}),
@@ -743,7 +825,16 @@ async fn test_tail_skips_trailing_non_responses() -> Result<()> {
writeln!(file, "{}", serde_json::to_string(&shutdown_event)?)?;
drop(file);
let page = get_conversations(home, 1, None, INTERACTIVE_SESSION_SOURCES).await?;
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page = get_conversations(
home,
1,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await?;
let tail = &page.items.first().expect("conversation item").tail;
let expected: Vec<serde_json::Value> = (0..4)
@@ -785,9 +876,17 @@ async fn test_stable_ordering_same_second_pagination() {
write_session_file(home, ts, u2, 0, Some(SessionSource::VSCode)).unwrap();
write_session_file(home, ts, u3, 0, Some(SessionSource::VSCode)).unwrap();
let page1 = get_conversations(home, 2, None, INTERACTIVE_SESSION_SOURCES)
.await
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let page1 = get_conversations(
home,
2,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
let p3 = home
.join("sessions")
@@ -810,6 +909,7 @@ async fn test_stable_ordering_same_second_pagination() {
"originator": "test_originator",
"cli_version": "test_version",
"source": "vscode",
"model_provider": "test-provider",
})]
};
let expected_cursor1: Cursor = serde_json::from_str(&format!("\"{ts}|{u2}\"")).unwrap();
@@ -841,6 +941,8 @@ async fn test_stable_ordering_same_second_pagination() {
2,
page1.next_cursor.as_ref(),
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
@@ -850,7 +952,6 @@ async fn test_stable_ordering_same_second_pagination() {
.join("07")
.join("01")
.join(format!("rollout-2025-07-01T00-00-00-{u1}.jsonl"));
let expected_cursor2: Cursor = serde_json::from_str(&format!("\"{ts}|{u1}\"")).unwrap();
let expected_page2 = ConversationsPage {
items: vec![ConversationItem {
path: p1,
@@ -859,7 +960,7 @@ async fn test_stable_ordering_same_second_pagination() {
created_at: Some(ts.to_string()),
updated_at: Some(ts.to_string()),
}],
next_cursor: Some(expected_cursor2),
next_cursor: None,
num_scanned_files: 3, // scanned u3, u2 (anchor), u1
reached_scan_cap: false,
};
@@ -891,9 +992,17 @@ async fn test_source_filter_excludes_non_matching_sessions() {
)
.unwrap();
let interactive_only = get_conversations(home, 10, None, INTERACTIVE_SESSION_SOURCES)
.await
.unwrap();
let provider_filter = provider_vec(&[TEST_PROVIDER]);
let interactive_only = get_conversations(
home,
10,
None,
INTERACTIVE_SESSION_SOURCES,
Some(provider_filter.as_slice()),
TEST_PROVIDER,
)
.await
.unwrap();
let paths: Vec<_> = interactive_only
.items
.iter()
@@ -905,7 +1014,7 @@ async fn test_source_filter_excludes_non_matching_sessions() {
path.ends_with("rollout-2025-08-02T10-00-00-00000000-0000-0000-0000-00000000002a.jsonl")
}));
let all_sessions = get_conversations(home, 10, None, NO_SOURCE_FILTER)
let all_sessions = get_conversations(home, 10, None, NO_SOURCE_FILTER, None, TEST_PROVIDER)
.await
.unwrap();
let all_paths: Vec<_> = all_sessions
@@ -921,3 +1030,102 @@ async fn test_source_filter_excludes_non_matching_sessions() {
path.ends_with("rollout-2025-08-01T10-00-00-00000000-0000-0000-0000-00000000004d.jsonl")
}));
}
#[tokio::test]
async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<()> {
let temp = TempDir::new().unwrap();
let home = temp.path();
let openai_id = Uuid::from_u128(1);
let beta_id = Uuid::from_u128(2);
let none_id = Uuid::from_u128(3);
write_session_file_with_provider(
home,
"2025-09-01T12-00-00",
openai_id,
1,
Some(SessionSource::VSCode),
Some("openai"),
)?;
write_session_file_with_provider(
home,
"2025-09-01T11-00-00",
beta_id,
1,
Some(SessionSource::VSCode),
Some("beta"),
)?;
write_session_file_with_provider(
home,
"2025-09-01T10-00-00",
none_id,
1,
Some(SessionSource::VSCode),
None,
)?;
let openai_id_str = openai_id.to_string();
let none_id_str = none_id.to_string();
let openai_filter = provider_vec(&["openai"]);
let openai_sessions = get_conversations(
home,
10,
None,
NO_SOURCE_FILTER,
Some(openai_filter.as_slice()),
"openai",
)
.await?;
assert_eq!(openai_sessions.items.len(), 2);
let openai_ids: Vec<_> = openai_sessions
.items
.iter()
.filter_map(|item| {
item.head
.first()
.and_then(|value| value.get("id"))
.and_then(serde_json::Value::as_str)
.map(str::to_string)
})
.collect();
assert!(openai_ids.contains(&openai_id_str));
assert!(openai_ids.contains(&none_id_str));
let beta_filter = provider_vec(&["beta"]);
let beta_sessions = get_conversations(
home,
10,
None,
NO_SOURCE_FILTER,
Some(beta_filter.as_slice()),
"openai",
)
.await?;
assert_eq!(beta_sessions.items.len(), 1);
let beta_id_str = beta_id.to_string();
let beta_head = beta_sessions
.items
.first()
.and_then(|item| item.head.first())
.and_then(|value| value.get("id"))
.and_then(serde_json::Value::as_str);
assert_eq!(beta_head, Some(beta_id_str.as_str()));
let unknown_filter = provider_vec(&["unknown"]);
let unknown_sessions = get_conversations(
home,
10,
None,
NO_SOURCE_FILTER,
Some(unknown_filter.as_slice()),
"openai",
)
.await?;
assert!(unknown_sessions.items.is_empty());
let all_sessions = get_conversations(home, 10, None, NO_SOURCE_FILTER, None, "openai").await?;
assert_eq!(all_sessions.items.len(), 3);
Ok(())
}