github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Prepare automatic TUI recap generation (#40697)

Browse Source 
## What changed

- Schedule recap eligibility checks when an unfocused thread changes, and cancel them when focus returns or recap state is replaced.
- Add bounded structured recap requests with stale-result rejection, normalized output, and one retry per turn revision.
- Harden temporary structured threads by reading the effective configuration and disabling all discovered MCP servers, tools, skills, and environment access.

## Testing

- Cover recap timers, focus changes, failed turns, output bounds, stale results, and retry behavior.

GitOrigin-RevId: eebf50059763928e9a2444b14d51648e0bdff4a8
This commit is contained in:
Felipe Coury
2026-08-25 20:56:03 +00:00
committed by copyberry
parent 7c1e36c23f
commit 40ba7da7b4
8 changed files with 788 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+9 -1
View File
@@ -775,7 +775,15 @@ impl App {
match &event {
TuiEvent::FocusLost => {
self.recap.note_focus_lost(Instant::now());
let now = Instant::now();
let thread_id = self.current_displayed_thread_id();
self.recap.note_focus_lost(now);
if let Some(thread_id) = thread_id {
self.recap
.schedule_check(thread_id, self.app_event_tx.clone(), now);
}
}
TuiEvent::FocusGained => {
self.recap.note_focus_gained();
codex-rs/tui/src/app/event_dispatch.rs
+5
View File
@@ -2897,6 +2897,11 @@ impl App {
AppEvent::KeymapCleared { context, action } => {
self.apply_keymap_clear(context, action).await;
}
AppEvent::CheckRecap { thread_id } => {
// The inference layer stacked on this change consumes the scheduled check.
let _ = thread_id;
}
AppEvent::RecapStarted { .. } | AppEvent::RecapGenerated { .. } => {}
}
Ok(AppRunControl::Continue)
}
codex-rs/tui/src/app/recap.rs
+345 -14
View File
@@ -1,22 +1,41 @@
//! Tracks when an unfocused conversation becomes eligible for an automatic recap.
//! Determines when an unfocused conversation is ready for an automatic recap.
#![cfg_attr(not(test), allow(dead_code))]
// The stacked event-dispatch change activates the generation methods.
#![allow(dead_code)]
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use super::App;
use crate::app_event::AppEvent;
use crate::app_event_sender::AppEventSender;
use crate::app_server_session::AppServerSession;
use crate::history_cell::AgentMarkdownCell;
use crate::history_cell::AgentMessageCell;
use crate::history_cell::HistoryCell;
use crate::history_cell::ThreadRecapHistoryCell;
use crate::history_cell::UserHistoryCell;
use crate::temporary_structured_request::TemporaryStructuredThreadOptions;
use crate::temporary_structured_request::run_temporary_structured_turn;
use crate::temporary_structured_request::start_temporary_thread;
use crate::temporary_structured_request::unsubscribe_temporary_thread;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnStatus;
use codex_protocol::ThreadId;
use serde::Deserialize;
use serde_json::Value;
use serde_json::json;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use uuid::Uuid;
const MIN_COMPLETED_TURNS: usize = 3;
const MIN_TURNS_BETWEEN_RECAPS: usize = 2;
const RECAP_DELAY: Duration = Duration::from_secs(/*secs*/ 3 * 60);
pub(super) const RECAP_DELAY: Duration = Duration::from_secs(/*secs*/ 3 * 60);
const RECAP_HISTORY_MAX_TURNS: usize = 8;
const RECAP_MAX_CHARS: usize = 320;
const RECAP_RETRY_DELAY: Duration = Duration::from_secs(/*secs*/ 30);
const RECAP_PROMPT_PREFIX: &str = concat!(
"Write a brief catch-up for a user returning to this Codex task. ",
"In at most 40 words and one or two plain-text sentences, explain the ",
@@ -36,6 +55,11 @@ fn render_recap_message(role: &str, content: &str, max_bytes: usize) -> Option<S
Some(format!("{prefix}{}", &content[..end]))
}
#[derive(Deserialize)]
struct GeneratedRecap {
recap: String,
}
fn recap_history(cells: &[Arc<dyn HistoryCell>]) -> String {
let mut messages = Vec::new();
let mut user_turns = 0;
@@ -113,6 +137,241 @@ fn recap_prompt(history: &str) -> String {
format!("{RECAP_PROMPT_PREFIX}{}", history.trim())
}
fn recap_output_schema() -> Value {
json!({
"type": "object",
"properties": {
"recap": {
"type": "string",
"minLength": 1,
"maxLength": RECAP_MAX_CHARS,
},
},
"required": ["recap"],
"additionalProperties": false,
})
}
fn parse_recap(response: &str) -> Option<String> {
let recap = serde_json::from_str::<GeneratedRecap>(response).ok()?.recap;
let recap = recap.trim();
if recap.is_empty() {
return None;
}
Some(recap.chars().take(RECAP_MAX_CHARS).collect())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct RecapRequest {
pub(super) thread_id: ThreadId,
pub(super) request_id: Uuid,
pub(super) completed_turn_count: usize,
pub(super) turn_revision: usize,
}
impl App {
pub(super) fn request_recap(&mut self, app_server: &AppServerSession, thread_id: ThreadId) {
if self.recap.in_flight_request.is_some() {
return;
}
let history = recap_history(&self.transcript_cells);
if history.is_empty() {
return;
}
let request = RecapRequest {
thread_id,
request_id: Uuid::new_v4(),
completed_turn_count: self.recap.completed_turns,
turn_revision: self.recap.turn_revision,
};
let request_handle = app_server.request_handle();
let model = self.chat_widget.current_model().to_string();
let config = self.chat_widget.config_ref();
let cwd = if app_server.uses_remote_workspace() {
app_server
.remote_cwd_override()
.map(|cwd| cwd.to_string_lossy().into_owned())
.unwrap_or_else(|| config.cwd.display().to_string())
} else {
config.cwd.display().to_string()
};
let options = TemporaryStructuredThreadOptions {
model,
model_provider: config.model_provider_id.clone(),
cwd,
active_permission_profile: config
.permissions
.active_permission_profile()
.map(|profile| profile.id),
mcp_server_names: config.mcp_servers.get().keys().cloned().collect(),
};
let event_sender = self.app_event_tx.clone();
let task = tokio::spawn(async move {
let result = start_temporary_thread(&request_handle, options)
.await
.map(|thread| thread.thread.id)
.map_err(|error| error.to_string());
event_sender.send(AppEvent::RecapStarted {
thread_id: request.thread_id,
request_id: request.request_id,
completed_turn_count: request.completed_turn_count,
turn_revision: request.turn_revision,
history,
result,
});
});
self.recap.in_flight_request_id = Some(request.request_id);
self.recap.in_flight_request = Some(task);
}
pub(super) fn handle_recap_started(
&mut self,
app_server: &AppServerSession,
request: RecapRequest,
history: String,
result: Result<String, String>,
) {
let RecapRequest {
thread_id,
request_id,
completed_turn_count,
turn_revision,
} = request;
let is_current_request = self.recap.in_flight_request_id == Some(request_id);
if is_current_request {
self.recap.in_flight_request.take();
}
let temporary_thread_id_text = match result {
Ok(temporary_thread_id) => temporary_thread_id,
Err(error) => {
if is_current_request {
self.recap.in_flight_request_id = None;
tracing::warn!(%thread_id, %error, "failed to start thread recap request");
self.recap
.schedule_retry(thread_id, self.app_event_tx.clone(), turn_revision);
}
return;
}
};
let request_is_fresh = is_current_request
&& self.current_displayed_thread_id() == Some(thread_id)
&& !self.chat_widget.is_user_turn_pending_or_running()
&& self.recap.completed_turns == completed_turn_count
&& self.recap.turn_revision == turn_revision
&& self.recap.should_generate(Instant::now());
let Ok(temporary_thread_id) = ThreadId::from_string(&temporary_thread_id_text) else {
if is_current_request {
self.recap.in_flight_request_id = None;
tracing::warn!(%thread_id, "thread recap request returned an invalid thread ID");
self.recap
.schedule_retry(thread_id, self.app_event_tx.clone(), turn_revision);
}
return;
};
if !request_is_fresh {
if is_current_request {
self.recap.in_flight_request_id = None;
}
let request_handle = app_server.request_handle();
tokio::spawn(async move {
unsubscribe_temporary_thread(&request_handle, temporary_thread_id_text).await;
});
return;
}
let (sender, receiver) = mpsc::unbounded_channel();
self.temporary_structured_requests
.insert(temporary_thread_id, sender);
let request_handle = app_server.request_handle();
let event_sender = self.app_event_tx.clone();
let task = tokio::spawn(async move {
let result = run_temporary_structured_turn(
request_handle,
temporary_thread_id_text,
recap_prompt(&history),
recap_output_schema(),
/*effort*/ None,
receiver,
)
.await
.map_err(|error| error.to_string());
event_sender.send(AppEvent::RecapGenerated {
thread_id,
request_id,
temporary_thread_id,
completed_turn_count,
turn_revision,
result,
});
});
self.recap.in_flight_thread_id = Some(temporary_thread_id);
self.recap.in_flight_request = Some(task);
}
pub(super) fn handle_generated_recap(
&mut self,
request: RecapRequest,
temporary_thread_id: ThreadId,
result: Result<String, String>,
) -> Option<ThreadRecapHistoryCell> {
let RecapRequest {
thread_id,
request_id,
completed_turn_count,
turn_revision,
} = request;
if self.recap.in_flight_request_id != Some(request_id)
|| self.recap.in_flight_thread_id != Some(temporary_thread_id)
{
return None;
}
self.recap.in_flight_request_id = None;
self.recap.in_flight_thread_id = None;
self.recap.in_flight_request.take();
if self.current_displayed_thread_id() != Some(thread_id)
|| self.chat_widget.is_user_turn_pending_or_running()
|| self.recap.completed_turns != completed_turn_count
|| self.recap.turn_revision != turn_revision
{
return None;
}
match result {
Ok(response) => {
let Some(recap) = parse_recap(&response) else {
tracing::warn!(%thread_id, "generated thread recap was invalid");
self.recap
.schedule_retry(thread_id, self.app_event_tx.clone(), turn_revision);
return None;
};
self.recap.mark_recapped(completed_turn_count);
Some(ThreadRecapHistoryCell::new(recap))
}
Err(error) => {
tracing::warn!(%thread_id, %error, "failed to generate thread recap");
self.recap
.schedule_retry(thread_id, self.app_event_tx.clone(), turn_revision);
None
}
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(super) struct RecapProgress {
pub(super) completed_turns: usize,
@@ -141,9 +400,15 @@ impl RecapProgress {
#[derive(Debug, Default)]
pub(super) struct RecapState {
unfocused_since: Option<Instant>,
last_completed_turn_at: Option<Instant>,
last_turn_finished_at: Option<Instant>,
completed_turns: usize,
last_recapped_turn_count: Option<usize>,
turn_revision: usize,
scheduled_check: Option<JoinHandle<()>>,
retry_revision: Option<usize>,
in_flight_request_id: Option<Uuid>,
in_flight_thread_id: Option<ThreadId>,
in_flight_request: Option<JoinHandle<()>>,
}
impl RecapState {
@@ -158,7 +423,7 @@ impl RecapState {
.max(progress.last_recapped_turn_count);
if progress.completed_turns > 0 {
self.last_completed_turn_at.get_or_insert(now);
self.last_turn_finished_at.get_or_insert(now);
}
}
@@ -171,23 +436,79 @@ impl RecapState {
pub(super) fn reset_for_new_thread(&mut self, now: Instant) {
let unfocused_since = self.unfocused_since.map(|_| now);
*self = Self {
unfocused_since,
..Self::default()
};
let mut replacement = Self::default();
replacement.unfocused_since = unfocused_since;
std::mem::swap(self, &mut replacement);
}
pub(super) fn note_focus_lost(&mut self, now: Instant) {
if self.unfocused_since.is_none() {
self.retry_revision = None;
}
self.unfocused_since.get_or_insert(now);
}
pub(super) fn note_focus_gained(&mut self) {
self.unfocused_since = None;
if let Some(task) = self.scheduled_check.take() {
task.abort();
}
self.retry_revision = None;
self.in_flight_request_id = None;
self.in_flight_thread_id = None;
self.in_flight_request.take();
}
pub(super) fn note_turn_completed(&mut self, now: Instant) {
self.completed_turns += 1;
self.last_completed_turn_at = Some(now);
pub(super) fn note_turn_finished(&mut self, status: &TurnStatus, now: Instant) {
if matches!(status, TurnStatus::Completed) {
self.completed_turns += 1;
}
self.turn_revision += 1;
self.last_turn_finished_at = Some(now);
}
pub(super) fn schedule_check(
&mut self,
thread_id: ThreadId,
app_event_tx: AppEventSender,
now: Instant,
) {
if let Some(task) = self.scheduled_check.take() {
task.abort();
}
let Some(deadline) = self.next_check_deadline() else {
return;
};
let delay = deadline.saturating_duration_since(now);
self.scheduled_check = Some(tokio::spawn(async move {
tokio::time::sleep(delay).await;
app_event_tx.send(AppEvent::CheckRecap { thread_id });
}));
}
pub(super) fn schedule_retry(
&mut self,
thread_id: ThreadId,
app_event_tx: AppEventSender,
turn_revision: usize,
) {
if self.turn_revision != turn_revision || self.retry_revision == Some(turn_revision) {
return;
}
self.retry_revision = Some(turn_revision);
if let Some(task) = self.scheduled_check.take() {
task.abort();
}
self.scheduled_check = Some(tokio::spawn(async move {
tokio::time::sleep(RECAP_RETRY_DELAY).await;
app_event_tx.send(AppEvent::CheckRecap { thread_id });
}));
}
pub(super) fn should_generate(&self, now: Instant) -> bool {
@@ -212,14 +533,24 @@ impl RecapState {
return None;
}
let last_completed_turn_at = self.last_completed_turn_at?;
let last_turn_finished_at = self.last_turn_finished_at?;
unfocused_since
.max(last_completed_turn_at)
.max(last_turn_finished_at)
.checked_add(RECAP_DELAY)
}
}
impl Drop for RecapState {
fn drop(&mut self) {
if let Some(task) = self.scheduled_check.take() {
task.abort();
}
// Let an in-flight request finish so it can unsubscribe its temporary thread.
}
}
#[cfg(test)]
#[path = "recap_tests.rs"]
mod tests;
codex-rs/tui/src/app/recap_tests.rs
+288 -11
View File
@@ -1,10 +1,18 @@
use super::App;
use super::RECAP_DELAY;
use super::RECAP_HISTORY_MAX_TURNS;
use super::RECAP_MAX_CHARS;
use super::RECAP_PROMPT_MAX_BYTES;
use super::RECAP_RETRY_DELAY;
use super::RecapProgress;
use super::RecapRequest;
use super::RecapState;
use super::parse_recap;
use super::recap_history;
use super::recap_prompt;
use crate::app::test_support::make_test_app;
use crate::app_event::AppEvent;
use crate::app_event_sender::AppEventSender;
use crate::history_cell::AgentMarkdownCell;
use crate::history_cell::AgentMessageCell;
use crate::history_cell::HistoryCell;
@@ -13,11 +21,14 @@ use crate::history_cell::ThreadRecapHistoryCell;
use crate::history_cell::UserHistoryCell;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnStatus;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::mpsc::error::TryRecvError;
use uuid::Uuid;
fn turn(status: TurnStatus) -> Turn {
Turn {
@@ -122,19 +133,40 @@ fn recap_history_caps_utf8_bytes_without_splitting_characters() {
assert!(history.starts_with("User: 最新の進捗🦀"));
}
#[test]
fn generated_recap_is_normalized_and_bounded() {
let expected = "🚀".repeat(RECAP_MAX_CHARS);
let cases = [
(
serde_json::json!({ "recap": " Fixed the parser. \n" }).to_string(),
Some("Fixed the parser.".to_string()),
),
(
serde_json::json!({ "recap": format!("{expected}discarded") }).to_string(),
Some(expected),
),
("not json".to_string(), None),
(r#"{"recap":" \t "}"#.to_string(), None),
];
for (response, expected) in cases {
assert_eq!(parse_recap(&response), expected, "response: {response}");
}
}
#[test]
fn recap_requires_focus_loss_and_three_completed_turns() {
let now = Instant::now();
let mut state = RecapState::default();
state.note_turn_completed(now);
state.note_turn_completed(now);
state.note_turn_finished(&TurnStatus::Completed, now);
state.note_turn_finished(&TurnStatus::Completed, now);
assert!(!state.should_generate(now + RECAP_DELAY));
state.note_focus_lost(now);
assert!(!state.should_generate(now + RECAP_DELAY));
state.note_turn_completed(now);
state.note_turn_finished(&TurnStatus::Completed, now);
assert!(state.should_generate(now + RECAP_DELAY));
}
@@ -144,7 +176,7 @@ fn recap_waits_after_focus_loss_even_if_turn_completed_earlier() {
let mut state = RecapState::default();
for _ in 0..3 {
state.note_turn_completed(started);
state.note_turn_finished(&TurnStatus::Completed, started);
}
let focus_lost = started + RECAP_DELAY;
@@ -162,16 +194,33 @@ fn completed_turn_resets_recap_deadline_while_unfocused() {
state.note_focus_lost(focus_lost);
for _ in 0..3 {
state.note_turn_completed(focus_lost);
state.note_turn_finished(&TurnStatus::Completed, focus_lost);
}
let last_completed_turn_at = focus_lost + Duration::from_secs(/*secs*/ 30);
state.note_turn_completed(last_completed_turn_at);
state.note_turn_finished(&TurnStatus::Completed, last_completed_turn_at);
assert!(!state.should_generate(focus_lost + RECAP_DELAY));
assert!(state.should_generate(last_completed_turn_at + RECAP_DELAY));
}
#[test]
fn failed_turn_invalidates_ready_recap_without_counting_as_completed() {
let focus_lost = Instant::now();
let mut state = RecapState::default();
state.note_focus_lost(focus_lost);
for _ in 0..3 {
state.note_turn_finished(&TurnStatus::Completed, focus_lost);
}
let failed_at = focus_lost + RECAP_DELAY;
state.note_turn_finished(&TurnStatus::Failed, failed_at);
assert_eq!(state.completed_turns, 3);
assert_eq!(state.turn_revision, 4);
assert!(!state.should_generate(failed_at));
assert!(state.should_generate(failed_at + RECAP_DELAY));
}
#[test]
fn repeated_focus_loss_does_not_restart_recap_delay() {
let focus_lost = Instant::now();
@@ -179,7 +228,7 @@ fn repeated_focus_loss_does_not_restart_recap_delay() {
state.note_focus_lost(focus_lost);
for _ in 0..3 {
state.note_turn_completed(focus_lost);
state.note_turn_finished(&TurnStatus::Completed, focus_lost);
}
state.note_focus_lost(focus_lost + Duration::from_secs(/*secs*/ 30));
@@ -194,7 +243,7 @@ fn regaining_focus_prevents_recap_generation() {
state.note_focus_lost(now);
for _ in 0..3 {
state.note_turn_completed(now);
state.note_turn_finished(&TurnStatus::Completed, now);
}
state.note_focus_gained();
@@ -202,6 +251,115 @@ fn regaining_focus_prevents_recap_generation() {
assert!(!state.should_generate(now + RECAP_DELAY));
}
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn regaining_focus_cancels_scheduled_check() {
let mut state = RecapState::default();
state.note_focus_lost(Instant::now());
let (task_tx, mut task_rx) = tokio::sync::mpsc::unbounded_channel();
state.scheduled_check = Some(tokio::spawn(async move {
tokio::time::sleep(RECAP_DELAY).await;
task_tx.send(()).expect("scheduled recap event");
}));
tokio::task::yield_now().await;
state.note_focus_gained();
tokio::time::advance(RECAP_DELAY).await;
tokio::task::yield_now().await;
assert!(state.scheduled_check.is_none());
assert!(matches!(
task_rx.try_recv(),
Err(tokio::sync::mpsc::error::TryRecvError::Disconnected)
));
}
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn scheduled_check_fires_at_recap_deadline() {
let thread_id = ThreadId::new();
let now = Instant::now();
let mut state = RecapState::default();
let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
state.note_focus_lost(now);
for _ in 0..3 {
state.note_turn_finished(&TurnStatus::Completed, now);
}
state.schedule_check(thread_id, AppEventSender::new(event_tx), now);
tokio::task::yield_now().await;
tokio::time::advance(RECAP_DELAY - Duration::from_secs(/*secs*/ 1)).await;
tokio::task::yield_now().await;
assert!(matches!(event_rx.try_recv(), Err(TryRecvError::Empty)));
tokio::time::advance(Duration::from_secs(/*secs*/ 1)).await;
tokio::task::yield_now().await;
assert!(matches!(
event_rx.try_recv(),
Ok(AppEvent::CheckRecap {
thread_id: event_thread_id,
}) if event_thread_id == thread_id
));
}
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn rescheduling_check_cancels_the_earlier_timer() {
let thread_id = ThreadId::new();
let first_turn = Instant::now();
let mut state = RecapState::default();
let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
let app_event_tx = AppEventSender::new(event_tx);
state.note_focus_lost(first_turn);
for _ in 0..3 {
state.note_turn_finished(&TurnStatus::Completed, first_turn);
}
state.schedule_check(thread_id, app_event_tx.clone(), first_turn);
tokio::task::yield_now().await;
let elapsed = Duration::from_secs(/*secs*/ 1);
tokio::time::advance(elapsed).await;
let later_turn = first_turn + elapsed;
state.note_turn_finished(&TurnStatus::Completed, later_turn);
state.schedule_check(thread_id, app_event_tx, later_turn);
tokio::task::yield_now().await;
tokio::time::advance(RECAP_DELAY - elapsed).await;
tokio::task::yield_now().await;
assert!(matches!(event_rx.try_recv(), Err(TryRecvError::Empty)));
tokio::time::advance(elapsed).await;
tokio::task::yield_now().await;
assert!(matches!(
event_rx.try_recv(),
Ok(AppEvent::CheckRecap {
thread_id: event_thread_id,
}) if event_thread_id == thread_id
));
}
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn dropping_recap_state_cancels_timer() {
let mut state = RecapState::default();
let (scheduled_tx, mut scheduled_rx) = tokio::sync::mpsc::unbounded_channel();
state.scheduled_check = Some(tokio::spawn(async move {
tokio::time::sleep(RECAP_DELAY).await;
scheduled_tx.send(()).expect("scheduled check");
}));
tokio::task::yield_now().await;
drop(state);
tokio::time::advance(RECAP_DELAY).await;
tokio::task::yield_now().await;
assert!(matches!(
scheduled_rx.try_recv(),
Err(TryRecvError::Disconnected)
));
}
#[test]
fn another_recap_requires_two_additional_completed_turns() {
let now = Instant::now();
@@ -218,11 +376,11 @@ fn another_recap_requires_two_additional_completed_turns() {
assert!(!state.should_generate(now + RECAP_DELAY));
let fourth_turn = now + RECAP_DELAY;
state.note_turn_completed(fourth_turn);
state.note_turn_finished(&TurnStatus::Completed, fourth_turn);
assert!(!state.should_generate(fourth_turn + RECAP_DELAY));
let fifth_turn = fourth_turn + Duration::from_secs(/*secs*/ 1);
state.note_turn_completed(fifth_turn);
state.note_turn_finished(&TurnStatus::Completed, fifth_turn);
assert!(state.should_generate(fifth_turn + RECAP_DELAY));
}
@@ -275,7 +433,7 @@ fn restored_history_never_reduces_observed_completed_turns() {
let mut state = RecapState::default();
for _ in 0..4 {
state.note_turn_completed(now);
state.note_turn_finished(&TurnStatus::Completed, now);
}
state.seed_from_turns(
@@ -374,3 +532,122 @@ fn recap_history_cell_preserves_explicit_line_breaks() {
Next: run the focused tests.
");
}
fn track_in_flight_recap(app: &mut App, thread_id: ThreadId) -> (RecapRequest, ThreadId) {
let request = RecapRequest {
thread_id,
request_id: Uuid::new_v4(),
completed_turn_count: 3,
turn_revision: 3,
};
let temporary_thread_id = ThreadId::new();
app.recap.completed_turns = request.completed_turn_count;
app.recap.turn_revision = request.turn_revision;
app.recap.in_flight_request_id = Some(request.request_id);
app.recap.in_flight_thread_id = Some(temporary_thread_id);
app.recap.in_flight_request = Some(tokio::spawn(async {}));
(request, temporary_thread_id)
}
async fn app_with_visible_thread(thread_id: ThreadId) -> App {
let mut app = make_test_app().await;
app.active_thread_id = Some(thread_id);
app
}
#[tokio::test]
async fn generated_recap_is_returned_for_synchronous_insertion() {
let thread_id = ThreadId::new();
let mut app = app_with_visible_thread(thread_id).await;
let (request, temporary_thread_id) = track_in_flight_recap(&mut app, thread_id);
let cell = app
.handle_generated_recap(
request,
temporary_thread_id,
Ok(serde_json::json!({ "recap": " Continue with focused tests. " }).to_string()),
)
.expect("fresh recap");
assert_eq!(app.recap.last_recapped_turn_count, Some(3));
assert_eq!(
cell.raw_lines()
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>(),
vec!["Conversation recap", "Continue with focused tests."]
);
}
#[tokio::test]
async fn obsolete_recap_result_does_not_clear_the_current_request() {
let thread_id = ThreadId::new();
let mut app = app_with_visible_thread(thread_id).await;
let (request, temporary_thread_id) = track_in_flight_recap(&mut app, thread_id);
let cell = app.handle_generated_recap(
RecapRequest {
request_id: Uuid::new_v4(),
..request
},
temporary_thread_id,
Ok(serde_json::json!({ "recap": "obsolete" }).to_string()),
);
assert!(cell.is_none());
assert_eq!(app.recap.in_flight_request_id, Some(request.request_id));
assert_eq!(app.recap.in_flight_thread_id, Some(temporary_thread_id));
}
#[tokio::test]
async fn newer_terminal_turn_invalidates_generated_recap() {
let thread_id = ThreadId::new();
let mut app = app_with_visible_thread(thread_id).await;
let (request, temporary_thread_id) = track_in_flight_recap(&mut app, thread_id);
app.recap
.note_turn_finished(&TurnStatus::Failed, Instant::now());
let cell = app.handle_generated_recap(
request,
temporary_thread_id,
Ok(serde_json::json!({ "recap": "missing the failure" }).to_string()),
);
assert!(cell.is_none());
assert_eq!(app.recap.last_recapped_turn_count, None);
assert!(app.recap.in_flight_request.is_none());
}
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn recap_failure_retries_once_for_the_same_turn_revision() {
let thread_id = ThreadId::new();
let mut app = app_with_visible_thread(thread_id).await;
let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
app.app_event_tx = AppEventSender::new(event_tx);
for attempt in 0..2 {
let (request, temporary_thread_id) = track_in_flight_recap(&mut app, thread_id);
assert!(
app.handle_generated_recap(
request,
temporary_thread_id,
Err("temporary failure".to_string()),
)
.is_none()
);
tokio::task::yield_now().await;
tokio::time::advance(RECAP_RETRY_DELAY).await;
tokio::task::yield_now().await;
if attempt == 0 {
assert!(matches!(
event_rx.try_recv(),
Ok(AppEvent::CheckRecap {
thread_id: event_thread_id,
}) if event_thread_id == thread_id
));
} else {
assert!(matches!(event_rx.try_recv(), Err(TryRecvError::Empty)));
}
}
}
codex-rs/tui/src/app/session_lifecycle.rs
+4 -1
View File
@@ -546,13 +546,16 @@ impl App {
self.active_thread_id = Some(thread_id);
self.active_thread_rx = Some(receiver);
self.recap.note_focus_gained();
self.recap = recap::RecapState::default();
if !tui.is_terminal_focused() {
self.recap.note_focus_lost(Instant::now());
}
let now = Instant::now();
self.recap.seed_from_progress(recap_progress, now);
self.recap
.seed_from_progress(recap_progress, Instant::now());
.schedule_check(thread_id, self.app_event_tx.clone(), now);
let init = self.chatwidget_init_for_forked_or_resumed_thread(
tui,
codex-rs/tui/src/app/thread_routing.rs
+12 -3
View File
@@ -963,9 +963,13 @@ impl App {
}
if self.current_displayed_thread_id() == Some(thread_id)
&& let ServerNotification::TurnCompleted(notification) = &notification
&& matches!(notification.turn.status, TurnStatus::Completed)
{
self.recap.note_turn_completed(Instant::now());
let now = Instant::now();
let app_event_tx = self.app_event_tx.clone();
self.recap
.note_turn_finished(&notification.turn.status, now);
self.recap.schedule_check(thread_id, app_event_tx, now);
}
let misalignment_policy_violation =
match &notification {
@@ -1327,7 +1331,12 @@ impl App {
self.app_event_tx
.send(AppEvent::BeginInitialHistoryReplayBuffer);
}
self.recap.seed_from_turns(&turns, Instant::now());
let now = Instant::now();
let app_event_tx = self.app_event_tx.clone();
self.recap.seed_from_turns(&turns, now);
self.recap.schedule_check(thread_id, app_event_tx, now);
self.chat_widget
.replay_thread_turns(turns, ReplayKind::ResumeInitialMessages);
if should_buffer_initial_replay {
codex-rs/tui/src/app_event.rs
+27
View File
@@ -1325,6 +1325,33 @@ pub(crate) enum AppEvent {
context: String,
action: String,
},
/// Recheck whether an unfocused thread is ready for an automatic recap.
CheckRecap {
thread_id: ThreadId,
},
/// Deliver the result of starting an automatic recap's temporary thread.
#[allow(dead_code)]
RecapStarted {
thread_id: ThreadId,
request_id: Uuid,
completed_turn_count: usize,
turn_revision: usize,
history: String,
result: Result<String, String>,
},
/// Deliver the generated recap from a temporary structured turn.
#[allow(dead_code)]
RecapGenerated {
thread_id: ThreadId,
request_id: Uuid,
temporary_thread_id: ThreadId,
completed_turn_count: usize,
turn_revision: usize,
result: Result<String, String>,
},
}
/// Named profile selection to apply after any required UI guardrails complete.
codex-rs/tui/src/temporary_structured_request.rs
+98 -50
View File
@@ -4,6 +4,8 @@ use codex_app_server_client::AppServerRequestHandle;
use codex_app_server_client::TypedRequestError;
use codex_app_server_protocol::AskForApproval;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ConfigReadParams;
use codex_app_server_protocol::ConfigReadResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SandboxMode;
use codex_app_server_protocol::SandboxPolicy;
@@ -37,7 +39,10 @@ pub(crate) struct TemporaryStructuredThreadOptions {
pub(crate) mcp_server_names: Vec<String>,
}
/// Start an ephemeral, non-interactive thread without widening its parent's permissions.
/// Start an ephemeral thread without widening permissions or exposing tools and environment access.
///
/// Structured prompts can contain untrusted transcript text, so the effective app-server config is
/// read first and every MCP server is explicitly disabled alongside built-in and extension tools.
pub(crate) async fn start_temporary_thread(
request_handle: &AppServerRequestHandle,
options: TemporaryStructuredThreadOptions,
@@ -72,9 +77,16 @@ pub(crate) async fn start_temporary_thread(
("features.shell_snapshot".to_string(), false.into()),
("features.shell_tool".to_string(), false.into()),
("features.standalone_web_search".to_string(), false.into()),
("features.token_budget".to_string(), false.into()),
("features.tool_suggest".to_string(), false.into()),
("features.unified_exec".to_string(), false.into()),
("features.view_image".to_string(), false.into()),
("orchestrator.skills.enabled".to_string(), false.into()),
("skills.include_instructions".to_string(), false.into()),
(
"token_budget.use_history_notes_extension".to_string(),
false.into(),
),
(
"tools.experimental_request_user_input.enabled".to_string(),
false.into(),
@@ -82,34 +94,62 @@ pub(crate) async fn start_temporary_thread(
("tools.update_plan.enabled".to_string(), false.into()),
("web_search".to_string(), "disabled".into()),
]);
for server_name in mcp_server_names {
config.insert(format!("mcp_servers.{server_name}.enabled"), false.into());
}
let response: ThreadStartResponse = tokio::time::timeout(STRUCTURED_TURN_TIMEOUT, async {
// Fail closed if the remote-effective MCP configuration cannot be read.
let effective_config: ConfigReadResponse = request_handle
.request_typed(ClientRequest::ConfigRead {
request_id: RequestId::String(format!("temporary-config-{}", Uuid::new_v4())),
params: ConfigReadParams {
include_layers: false,
cwd: Some(cwd.clone()),
},
})
.await?;
let mut mcp_server_names = mcp_server_names
.into_iter()
.collect::<std::collections::BTreeSet<_>>();
if let Some(effective_mcp_servers) = effective_config
.config
.additional
.get("mcp_servers")
.and_then(Value::as_object)
{
mcp_server_names.extend(effective_mcp_servers.keys().cloned());
}
config.insert(
"mcp_servers".to_string(),
Value::Object(
mcp_server_names
.into_iter()
.map(|name| (name, serde_json::json!({ "enabled": false })))
.collect(),
),
);
let response: ThreadStartResponse = tokio::time::timeout(
STRUCTURED_TURN_TIMEOUT,
request_handle.request_typed(ClientRequest::ThreadStart {
request_id: RequestId::String(format!("temporary-structured-{}", Uuid::new_v4())),
params: ThreadStartParams {
model: Some(model),
model_provider: Some(model_provider),
cwd: Some(cwd),
approval_policy: Some(AskForApproval::Never),
sandbox: custom_permission_profile
.is_none()
.then_some(SandboxMode::ReadOnly),
permissions: custom_permission_profile.clone(),
runtime_workspace_roots: Some(Vec::new()),
ephemeral: Some(true),
thread_source: Some(ThreadSource::Feature("system".to_string())),
environments: Some(Vec::new()),
dynamic_tools: Some(Vec::new()),
selected_capability_roots: Some(Vec::new()),
config: Some(config),
..ThreadStartParams::default()
},
}),
)
request_handle
.request_typed(ClientRequest::ThreadStart {
request_id: RequestId::String(format!("temporary-structured-{}", Uuid::new_v4())),
params: ThreadStartParams {
model: Some(model),
model_provider: Some(model_provider),
cwd: Some(cwd),
approval_policy: Some(AskForApproval::Never),
sandbox: custom_permission_profile
.is_none()
.then_some(SandboxMode::ReadOnly),
permissions: custom_permission_profile.clone(),
runtime_workspace_roots: Some(Vec::new()),
ephemeral: Some(true),
thread_source: Some(ThreadSource::Feature("system".to_string())),
environments: Some(Vec::new()),
dynamic_tools: Some(Vec::new()),
selected_capability_roots: Some(Vec::new()),
config: Some(config),
..ThreadStartParams::default()
},
})
.await
})
.await
.map_err(|_| eyre!("temporary structured thread start timed out"))??;
@@ -197,6 +237,35 @@ pub(crate) async fn collect_structured_response(
))
}
/// Make a bounded best-effort attempt to detach an ephemeral thread.
pub(crate) async fn unsubscribe_temporary_thread(
request_handle: &AppServerRequestHandle,
thread_id: String,
) {
match tokio::time::timeout(
STRUCTURED_TURN_TIMEOUT,
request_handle.request_typed::<ThreadUnsubscribeResponse>(
ClientRequest::ThreadUnsubscribe {
request_id: RequestId::String(format!(
"temporary-structured-unsubscribe-{}",
Uuid::new_v4()
)),
params: ThreadUnsubscribeParams { thread_id },
},
),
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(error)) => {
tracing::debug!(%error, "failed to unsubscribe from temporary structured thread");
}
Err(_) => {
tracing::debug!("temporary structured thread unsubscribe timed out");
}
}
}
/// Run a bounded structured turn and make a bounded temporary-thread cleanup attempt.
pub(crate) async fn run_temporary_structured_turn(
request_handle: AppServerRequestHandle,
@@ -221,28 +290,7 @@ pub(crate) async fn run_temporary_structured_turn(
.await
.unwrap_or_else(|_| Err(eyre!("temporary structured turn timed out")));
match tokio::time::timeout(
STRUCTURED_TURN_TIMEOUT,
request_handle.request_typed::<ThreadUnsubscribeResponse>(
ClientRequest::ThreadUnsubscribe {
request_id: RequestId::String(format!(
"temporary-structured-unsubscribe-{}",
Uuid::new_v4()
)),
params: ThreadUnsubscribeParams { thread_id },
},
),
)
.await
{
Ok(Ok(_)) => {}
Ok(Err(error)) => {
tracing::debug!(%error, "failed to unsubscribe from temporary structured thread");
}
Err(_) => {
tracing::debug!("temporary structured thread unsubscribe timed out");
}
}
unsubscribe_temporary_thread(&request_handle, thread_id).await;
result
}