openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-10 21:00:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

reflections

Browse Source
This commit is contained in:
Itay Golan
2026-04-16 10:00:17 -07:00
parent 8b7d0e9201
commit 4bf6575725
30 changed files with 2103 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/facts.rs
+2
View File
@@ -216,6 +216,7 @@ pub enum CompactionReason {
pub enum CompactionImplementation {
Responses,
ResponsesCompact,
Reflections,
}
#[derive(Clone, Copy, Debug, Serialize)]
@@ -231,6 +232,7 @@ pub enum CompactionPhase {
pub enum CompactionStrategy {
Memento,
PrefixCompaction,
Reflections,
}
#[derive(Clone, Copy, Debug, Serialize)]
codex-rs/config/src/config_toml.rs
+4
View File
@@ -36,6 +36,7 @@ use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::OLLAMA_CHAT_PROVIDER_REMOVED_ERROR;
use codex_model_provider_info::OLLAMA_OSS_PROVIDER_ID;
use codex_model_provider_info::OPENAI_PROVIDER_ID;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::config_types::ForcedLoginMethod;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary;
@@ -144,6 +145,9 @@ pub struct ConfigToml {
/// Compact prompt used for history compaction.
pub compact_prompt: Option<String>,
/// Strategy used when compacting conversation history.
pub compaction_strategy: Option<CompactionStrategyConfig>,
/// Optional commit attribution text for commit message co-author trailers.
///
/// Set to an empty string to disable automatic commit attribution.
codex-rs/config/src/profile_toml.rs
+2
View File
@@ -9,6 +9,7 @@ use crate::types::ApprovalsReviewer;
use crate::types::Personality;
use crate::types::WindowsToml;
use codex_features::FeaturesToml;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::SandboxMode;
use codex_protocol::config_types::ServiceTier;
@@ -40,6 +41,7 @@ pub struct ConfigProfile {
pub chatgpt_base_url: Option<String>,
/// Optional path to a file containing model instructions.
pub model_instructions_file: Option<AbsolutePathBuf>,
pub compaction_strategy: Option<CompactionStrategyConfig>,
pub js_repl_node_path: Option<AbsolutePathBuf>,
/// Ordered list of directories to search for Node modules in `js_repl`.
pub js_repl_node_module_dirs: Option<Vec<AbsolutePathBuf>>,
codex-rs/config/src/schema.rs
+4
View File
@@ -38,6 +38,10 @@ pub fn features_schema(schema_gen: &mut SchemaGenerator) -> Schema {
.properties
.insert(feature.key.to_string(), schema_gen.subschema_for::<bool>());
}
validation.properties.insert(
"reflections".to_string(),
schema_gen.subschema_for::<codex_features::ReflectionsConfigToml>(),
);
for legacy_key in legacy_feature_keys() {
validation
.properties
codex-rs/core/config.schema.json
+36
View File
@@ -302,6 +302,13 @@
},
"type": "object"
},
"CompactionStrategyConfig": {
"enum": [
"default",
"reflections"
],
"type": "string"
},
"ConfigProfile": {
"additionalProperties": false,
"description": "Collection of common configuration options that a user can define as a unit in `config.toml`.",
@@ -318,6 +325,9 @@
"chatgpt_base_url": {
"type": "string"
},
"compaction_strategy": {
"$ref": "#/definitions/CompactionStrategyConfig"
},
"experimental_compact_prompt_file": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -437,6 +447,9 @@
"realtime_conversation": {
"type": "boolean"
},
"reflections": {
"$ref": "#/definitions/ReflectionsConfigToml"
},
"remote_control": {
"type": "boolean"
},
@@ -1654,6 +1667,18 @@
}
]
},
"ReflectionsConfigToml": {
"additionalProperties": false,
"properties": {
"usage_hint_enabled": {
"type": "boolean"
},
"usage_hint_text": {
"type": "string"
}
},
"type": "object"
},
"SandboxMode": {
"enum": [
"read-only",
@@ -2141,6 +2166,14 @@
"description": "Compact prompt used for history compaction.",
"type": "string"
},
"compaction_strategy": {
"allOf": [
{
"$ref": "#/definitions/CompactionStrategyConfig"
}
],
"description": "Strategy used when compacting conversation history."
},
"default_permissions": {
"description": "Default named permissions profile to apply from the `[permissions]` table.",
"type": "string"
@@ -2293,6 +2326,9 @@
"realtime_conversation": {
"type": "boolean"
},
"reflections": {
"$ref": "#/definitions/ReflectionsConfigToml"
},
"remote_control": {
"type": "boolean"
},
codex-rs/core/src/codex.rs
+205 -3
View File
@@ -3,7 +3,9 @@ use std::collections::HashSet;
use std::fmt::Debug;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
@@ -52,6 +54,7 @@ use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppInvocation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionTrigger;
use codex_analytics::InvocationType;
use codex_analytics::SubAgentThreadStartedInput;
use codex_analytics::TurnResolvedConfigFact;
@@ -96,6 +99,7 @@ use codex_protocol::approvals::ExecPolicyAmendment;
use codex_protocol::approvals::NetworkPolicyAmendment;
use codex_protocol::approvals::NetworkPolicyRuleAction;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::config_types::WebSearchMode;
@@ -107,6 +111,7 @@ use codex_protocol::items::UserMessageItem;
use codex_protocol::items::build_hook_prompt_message;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::FileSystemPermissions;
use codex_protocol::models::PermissionProfile;
use codex_protocol::models::format_allow_prefixes;
use codex_protocol::openai_models::ModelInfo;
@@ -660,6 +665,7 @@ impl Codex {
persist_extended_history,
inherited_shell_snapshot,
user_shell_override,
reflections_sidecar_path: None,
};
// Generate a unique ID for the lifetime of this Codex session.
@@ -856,6 +862,8 @@ pub(crate) struct Session {
pub(crate) services: SessionServices,
js_repl: Arc<JsReplHandle>,
next_internal_sub_id: AtomicU64,
reflections_context_window_reset_requested: AtomicBool,
reflections_near_limit_reminder_recorded: AtomicBool,
}
#[derive(Clone, Debug)]
@@ -1204,6 +1212,7 @@ pub(crate) struct SessionConfiguration {
persist_extended_history: bool,
inherited_shell_snapshot: Option<Arc<ShellSnapshot>>,
user_shell_override: Option<shell::Shell>,
reflections_sidecar_path: Option<AbsolutePathBuf>,
}
impl SessionConfiguration {
@@ -1294,8 +1303,37 @@ impl SessionConfiguration {
if let Some(app_server_client_version) = updates.app_server_client_version.clone() {
next_configuration.app_server_client_version = Some(app_server_client_version);
}
next_configuration.add_reflections_sidecar_to_sandbox_policies();
Ok(next_configuration)
}
fn add_reflections_sidecar_to_sandbox_policies(&mut self) {
let Some(sidecar_path) = self.reflections_sidecar_path.clone() else {
return;
};
self.file_system_sandbox_policy = self
.file_system_sandbox_policy
.clone()
.with_additional_writable_roots(
self.cwd.as_path(),
std::slice::from_ref(&sidecar_path),
);
let mut sandbox_policy = self.sandbox_policy.get().clone();
let mut sandbox_policy_changed = false;
if let SandboxPolicy::WorkspaceWrite { writable_roots, .. } = &mut sandbox_policy
&& !writable_roots
.iter()
.any(|existing| existing == &sidecar_path)
{
writable_roots.push(sidecar_path);
sandbox_policy_changed = true;
}
if sandbox_policy_changed && let Err(err) = self.sandbox_policy.set(sandbox_policy) {
warn!("failed to add Reflections sidecar to sandbox policy: {err}");
}
}
}
#[derive(Default, Clone)]
@@ -1319,6 +1357,12 @@ pub(crate) struct AppServerClientMetadata {
pub(crate) client_version: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReflectionsSidecarPreparation {
Prepare,
Skip,
}
impl Session {
pub(crate) async fn app_server_client_metadata(&self) -> AppServerClientMetadata {
let state = self.state.lock().await;
@@ -1539,6 +1583,29 @@ impl Session {
let auth_manager_for_context = auth_manager;
let provider_for_context = provider;
let session_telemetry_for_context = session_telemetry;
let reflections_enabled = per_turn_config.compaction_strategy
== CompactionStrategyConfig::Reflections
&& session_configuration.reflections_sidecar_path.is_some();
let reflections_usage_hint_text = if reflections_enabled
&& per_turn_config.reflections.usage_hint_enabled
{
session_configuration
.reflections_sidecar_path
.as_ref()
.map(|sidecar_path| {
let context_window = model_info.context_window.map(|context_window| {
context_window.saturating_mul(model_info.effective_context_window_percent)
/ 100
});
crate::reflections::usage_hint(
context_window,
sidecar_path.as_path(),
per_turn_config.reflections.usage_hint_text.as_deref(),
)
})
} else {
None
};
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
available_models: &models_manager.try_list_models().unwrap_or_default(),
@@ -1560,6 +1627,7 @@ impl Session {
.with_spawn_agent_usage_hint(per_turn_config.multi_agent_v2.usage_hint_enabled)
.with_spawn_agent_usage_hint_text(per_turn_config.multi_agent_v2.usage_hint_text.clone())
.with_hide_spawn_agent_metadata(per_turn_config.multi_agent_v2.hide_spawn_agent_metadata)
.with_reflections(reflections_enabled, reflections_usage_hint_text)
.with_agent_type_description(crate::agent::role::spawn_tool_spec::build(
&per_turn_config.agent_roles,
));
@@ -1776,6 +1844,15 @@ impl Session {
let rollout_path = rollout_recorder
.as_ref()
.map(|rec| rec.rollout_path().to_path_buf());
if config.compaction_strategy == CompactionStrategyConfig::Reflections
&& let Some(sidecar_path) = rollout_path
.as_ref()
.map(|rollout_path| crate::reflections::sidecar_path_for_rollout(rollout_path))
.and_then(|path| AbsolutePathBuf::from_absolute_path(path).ok())
{
session_configuration.reflections_sidecar_path = Some(sidecar_path);
session_configuration.add_reflections_sidecar_to_sandbox_policies();
}
let mut post_session_configured_events = Vec::<Event>::new();
@@ -1946,7 +2023,16 @@ impl Session {
))
.await;
session_configuration.thread_name = thread_name.clone();
let state = SessionState::new(session_configuration.clone());
let mut state = SessionState::new(session_configuration.clone());
if let Some(sidecar_path) = session_configuration.reflections_sidecar_path.clone() {
state.record_granted_permissions(PermissionProfile {
file_system: Some(FileSystemPermissions {
read: Some(vec![]),
write: Some(vec![sidecar_path]),
}),
..Default::default()
});
}
let managed_network_requirements_enabled = config.managed_network_requirements_enabled();
let network_approval = Arc::new(NetworkApprovalService::default());
// The managed proxy can call back into core for allowlist-miss decisions.
@@ -2112,6 +2198,8 @@ impl Session {
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
reflections_context_window_reset_requested: AtomicBool::new(false),
reflections_near_limit_reminder_recorded: AtomicBool::new(false),
});
if let Some(network_policy_decider_session) = network_policy_decider_session {
let mut guard = network_policy_decider_session.write().await;
@@ -2295,6 +2383,68 @@ impl Session {
format!("auto-compact-{id}")
}
pub(crate) fn request_reflections_context_window_reset(&self) {
self.reflections_context_window_reset_requested
.store(true, Ordering::SeqCst);
}
fn take_reflections_context_window_reset_request(&self) -> bool {
self.reflections_context_window_reset_requested
.swap(false, Ordering::SeqCst)
}
pub(crate) fn reset_reflections_near_limit_reminder(&self) {
self.reflections_near_limit_reminder_recorded
.store(false, Ordering::SeqCst);
}
async fn maybe_record_reflections_near_limit_reminder(
&self,
turn_context: &TurnContext,
total_usage_tokens: i64,
auto_compact_limit: i64,
) {
if turn_context.config.compaction_strategy != CompactionStrategyConfig::Reflections
|| auto_compact_limit == i64::MAX
|| total_usage_tokens >= auto_compact_limit
{
return;
}
let reminder_threshold =
crate::reflections::near_limit_reminder_threshold(auto_compact_limit);
if total_usage_tokens < reminder_threshold
|| self
.reflections_near_limit_reminder_recorded
.load(Ordering::SeqCst)
{
return;
}
let history = self.clone_history().await;
if history
.raw_items()
.iter()
.any(crate::reflections::is_near_limit_reminder)
{
self.reflections_near_limit_reminder_recorded
.store(true, Ordering::SeqCst);
return;
}
if self
.reflections_near_limit_reminder_recorded
.swap(true, Ordering::SeqCst)
{
return;
}
let remaining_tokens = auto_compact_limit.saturating_sub(total_usage_tokens).max(0);
let reminder = crate::reflections::near_limit_reminder(Some(remaining_tokens));
self.record_conversation_items(turn_context, std::slice::from_ref(&reminder))
.await;
}
pub(crate) async fn route_realtime_text_input(self: &Arc<Self>, text: String) {
handlers::user_input_or_turn_inner(
self,
@@ -2615,6 +2765,7 @@ impl Session {
session_configuration,
updates.final_output_json_schema,
sandbox_policy_changed,
ReflectionsSidecarPreparation::Prepare,
)
.await)
}
@@ -2622,11 +2773,30 @@ impl Session {
async fn new_turn_from_configuration(
&self,
sub_id: String,
session_configuration: SessionConfiguration,
mut session_configuration: SessionConfiguration,
final_output_json_schema: Option<Option<Value>>,
sandbox_policy_changed: bool,
reflections_sidecar_preparation: ReflectionsSidecarPreparation,
) -> Arc<TurnContext> {
let per_turn_config = Self::build_per_turn_config(&session_configuration);
if matches!(
reflections_sidecar_preparation,
ReflectionsSidecarPreparation::Prepare
) && per_turn_config.compaction_strategy == CompactionStrategyConfig::Reflections
&& let Some(sidecar_path) = session_configuration.reflections_sidecar_path.clone()
{
let ensure_result = match self.try_ensure_rollout_materialized().await {
Ok(()) => crate::reflections::ensure_sidecar_dirs(sidecar_path.as_path()).await,
Err(err) => Err(err),
};
if let Err(err) = ensure_result {
warn!(
"failed to prepare Reflections sidecar {}: {err}",
sidecar_path.as_path().display()
);
session_configuration.reflections_sidecar_path = None;
}
}
{
let mcp_connection_manager = self.services.mcp_connection_manager.read().await;
mcp_connection_manager.set_approval_policy(&session_configuration.approval_policy);
@@ -2798,6 +2968,7 @@ impl Session {
session_configuration,
/*final_output_json_schema*/ None,
/*sandbox_policy_changed*/ false,
ReflectionsSidecarPreparation::Skip,
)
.await
}
@@ -6406,6 +6577,13 @@ pub(crate) async fn run_turn(
}
// Construct the input that we will send to the model.
let total_usage_tokens = sess.get_total_token_usage().await;
sess.maybe_record_reflections_near_limit_reminder(
turn_context.as_ref(),
total_usage_tokens,
auto_compact_limit,
)
.await;
let sampling_request_input: Vec<ResponseItem> = {
sess.clone_history()
.await
@@ -6765,7 +6943,16 @@ async fn run_auto_compact(
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
if should_use_remote_compact_task(&turn_context.provider) {
if turn_context.config.compaction_strategy == CompactionStrategyConfig::Reflections {
crate::reflections::run_inline_reflections_auto_compact_task(
Arc::clone(sess),
Arc::clone(turn_context),
initial_context_injection,
reason,
phase,
)
.await?;
} else if should_use_remote_compact_task(&turn_context.provider) {
run_inline_remote_auto_compact_task(
Arc::clone(sess),
Arc::clone(turn_context),
@@ -7747,17 +7934,32 @@ async fn drain_in_flight(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
) -> CodexResult<()> {
let mut reset_reflections_context_window = false;
while let Some(res) = in_flight.next().await {
match res {
Ok(response_input) => {
sess.record_conversation_items(&turn_context, &[response_input.into()])
.await;
if sess.take_reflections_context_window_reset_request() {
reset_reflections_context_window = true;
}
}
Err(err) => {
error_or_panic(format!("in-flight tool future failed during drain: {err}"));
}
}
}
if reset_reflections_context_window {
crate::reflections::run_reflections_compact_task_inner(
Arc::clone(&sess),
Arc::clone(&turn_context),
InitialContextInjection::DoNotInject,
CompactionTrigger::Manual,
CompactionReason::UserRequested,
CompactionPhase::MidTurn,
)
.await?;
}
Ok(())
}
codex-rs/core/src/codex_tests.rs
+57
View File
@@ -1930,6 +1930,7 @@ async fn set_rate_limits_retains_previous_credits() {
persist_extended_history: false,
inherited_shell_snapshot: None,
user_shell_override: None,
reflections_sidecar_path: None,
};
let mut state = SessionState::new(session_configuration);
@@ -2032,6 +2033,7 @@ async fn set_rate_limits_updates_plan_type_when_present() {
persist_extended_history: false,
inherited_shell_snapshot: None,
user_shell_override: None,
reflections_sidecar_path: None,
};
let mut state = SessionState::new(session_configuration);
@@ -2384,6 +2386,7 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati
persist_extended_history: false,
inherited_shell_snapshot: None,
user_shell_override: None,
reflections_sidecar_path: None,
}
}
@@ -2558,6 +2561,53 @@ async fn session_configuration_apply_rederives_legacy_file_system_policy_on_cwd_
);
}
#[tokio::test]
async fn session_configuration_apply_preserves_reflections_sidecar_write_root() {
let mut session_configuration = make_session_configuration_for_tests().await;
let workspace = tempfile::tempdir().expect("create temp dir");
let project_root = workspace.path().join("project");
let original_cwd = project_root.join("subdir");
let sidecar_path = workspace.path().join("rollout.reflections").abs();
std::fs::create_dir_all(&original_cwd).expect("create cwd");
session_configuration.cwd = original_cwd.abs();
session_configuration.reflections_sidecar_path = Some(sidecar_path.clone());
session_configuration.sandbox_policy =
codex_config::Constrained::allow_any(SandboxPolicy::WorkspaceWrite {
writable_roots: Vec::new(),
read_only_access: ReadOnlyAccess::Restricted {
include_platform_defaults: true,
readable_roots: Vec::new(),
},
network_access: false,
exclude_tmpdir_env_var: true,
exclude_slash_tmp: true,
});
session_configuration.file_system_sandbox_policy =
FileSystemSandboxPolicy::from_legacy_sandbox_policy(
session_configuration.sandbox_policy.get(),
&session_configuration.cwd,
);
let updated = session_configuration
.apply(&SessionSettingsUpdate {
cwd: Some(project_root),
..Default::default()
})
.expect("cwd-only update should succeed");
assert_eq!(
updated
.file_system_sandbox_policy
.resolve_access_with_cwd(sidecar_path.as_path(), updated.cwd.as_path()),
FileSystemAccessMode::Write
);
let SandboxPolicy::WorkspaceWrite { writable_roots, .. } = updated.sandbox_policy.get() else {
panic!("expected workspace-write sandbox policy");
};
assert!(writable_roots.contains(&sidecar_path));
}
#[tokio::test]
async fn session_update_settings_keeps_runtime_cwds_absolute() {
let (session, turn_context) = make_session_and_context().await;
@@ -2647,6 +2697,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_zsh_path() {
persist_extended_history: false,
inherited_shell_snapshot: None,
user_shell_override: None,
reflections_sidecar_path: None,
};
let (tx_event, _rx_event) = async_channel::unbounded();
@@ -2751,6 +2802,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
persist_extended_history: false,
inherited_shell_snapshot: None,
user_shell_override: None,
reflections_sidecar_path: None,
};
let per_turn_config = Session::build_per_turn_config(&session_configuration);
let model_info = ModelsManager::construct_model_info_offline_for_tests(
@@ -2884,6 +2936,8 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
reflections_context_window_reset_requested: AtomicBool::new(false),
reflections_near_limit_reminder_recorded: AtomicBool::new(false),
};
(session, turn_context)
@@ -3597,6 +3651,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
persist_extended_history: false,
inherited_shell_snapshot: None,
user_shell_override: None,
reflections_sidecar_path: None,
};
let per_turn_config = Session::build_per_turn_config(&session_configuration);
let model_info = ModelsManager::construct_model_info_offline_for_tests(
@@ -3730,6 +3785,8 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
services,
js_repl,
next_internal_sub_id: AtomicU64::new(0),
reflections_context_window_reset_requested: AtomicBool::new(false),
reflections_near_limit_reminder_recorded: AtomicBool::new(false),
});
(session, turn_context, rx_event)
codex-rs/core/src/compact.rs
+24 -1
View File
@@ -300,6 +300,7 @@ pub(crate) struct CompactionAnalyticsAttempt {
reason: CompactionReason,
implementation: CompactionImplementation,
phase: CompactionPhase,
strategy: CompactionStrategy,
active_context_tokens_before: i64,
started_at: u64,
start_instant: Instant,
@@ -313,6 +314,27 @@ impl CompactionAnalyticsAttempt {
reason: CompactionReason,
implementation: CompactionImplementation,
phase: CompactionPhase,
) -> Self {
Self::begin_with_strategy(
sess,
turn_context,
trigger,
reason,
implementation,
phase,
CompactionStrategy::Memento,
)
.await
}
pub(crate) async fn begin_with_strategy(
sess: &Session,
turn_context: &TurnContext,
trigger: CompactionTrigger,
reason: CompactionReason,
implementation: CompactionImplementation,
phase: CompactionPhase,
strategy: CompactionStrategy,
) -> Self {
let enabled = sess.enabled(Feature::GeneralAnalytics);
let active_context_tokens_before = sess.get_total_token_usage().await;
@@ -324,6 +346,7 @@ impl CompactionAnalyticsAttempt {
reason,
implementation,
phase,
strategy,
active_context_tokens_before,
started_at: now_unix_seconds(),
start_instant: Instant::now(),
@@ -349,7 +372,7 @@ impl CompactionAnalyticsAttempt {
reason: self.reason,
implementation: self.implementation,
phase: self.phase,
strategy: CompactionStrategy::Memento,
strategy: self.strategy,
status,
error,
active_context_tokens_before: self.active_context_tokens_before,
codex-rs/core/src/config/config_tests.rs
+81
View File
@@ -48,6 +48,7 @@ use codex_model_provider_info::LMSTUDIO_OSS_PROVIDER_ID;
use codex_model_provider_info::OLLAMA_OSS_PROVIDER_ID;
use codex_model_provider_info::WireApi;
use codex_models_manager::bundled_models_response;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::permissions::FileSystemAccessMode;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSandboxEntry;
@@ -4635,6 +4636,7 @@ fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> {
include_apps_instructions: true,
include_environment_context: true,
compact_prompt: None,
compaction_strategy: CompactionStrategyConfig::Default,
commit_attribution: None,
forced_chatgpt_workspace_id: None,
forced_login_method: None,
@@ -4645,6 +4647,7 @@ fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> {
background_terminal_max_timeout: DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS,
ghost_snapshot: GhostSnapshotConfig::default(),
multi_agent_v2: MultiAgentV2Config::default(),
reflections: ReflectionsConfig::default(),
features: Features::with_defaults().into(),
suppress_unstable_features_warning: false,
active_profile: Some("o3".to_string()),
@@ -4784,6 +4787,7 @@ fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> {
include_apps_instructions: true,
include_environment_context: true,
compact_prompt: None,
compaction_strategy: CompactionStrategyConfig::Default,
commit_attribution: None,
forced_chatgpt_workspace_id: None,
forced_login_method: None,
@@ -4794,6 +4798,7 @@ fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> {
background_terminal_max_timeout: DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS,
ghost_snapshot: GhostSnapshotConfig::default(),
multi_agent_v2: MultiAgentV2Config::default(),
reflections: ReflectionsConfig::default(),
features: Features::with_defaults().into(),
suppress_unstable_features_warning: false,
active_profile: Some("gpt3".to_string()),
@@ -4931,6 +4936,7 @@ fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> {
include_apps_instructions: true,
include_environment_context: true,
compact_prompt: None,
compaction_strategy: CompactionStrategyConfig::Default,
commit_attribution: None,
forced_chatgpt_workspace_id: None,
forced_login_method: None,
@@ -4941,6 +4947,7 @@ fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> {
background_terminal_max_timeout: DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS,
ghost_snapshot: GhostSnapshotConfig::default(),
multi_agent_v2: MultiAgentV2Config::default(),
reflections: ReflectionsConfig::default(),
features: Features::with_defaults().into(),
suppress_unstable_features_warning: false,
active_profile: Some("zdr".to_string()),
@@ -5064,6 +5071,7 @@ fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> {
include_apps_instructions: true,
include_environment_context: true,
compact_prompt: None,
compaction_strategy: CompactionStrategyConfig::Default,
commit_attribution: None,
forced_chatgpt_workspace_id: None,
forced_login_method: None,
@@ -5074,6 +5082,7 @@ fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> {
background_terminal_max_timeout: DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS,
ghost_snapshot: GhostSnapshotConfig::default(),
multi_agent_v2: MultiAgentV2Config::default(),
reflections: ReflectionsConfig::default(),
features: Features::with_defaults().into(),
suppress_unstable_features_warning: false,
active_profile: Some("gpt5".to_string()),
@@ -6297,6 +6306,78 @@ hide_spawn_agent_metadata = false
Ok(())
}
#[tokio::test]
async fn reflections_config_from_strategy_and_feature_table() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"compaction_strategy = "reflections"
[features.reflections]
usage_hint_enabled = false
usage_hint_text = "Custom recovery guidance."
"#,
)?;
let config = ConfigBuilder::without_managed_config_for_tests()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
assert_eq!(
config.compaction_strategy,
CompactionStrategyConfig::Reflections
);
assert!(!config.reflections.usage_hint_enabled);
assert_eq!(
config.reflections.usage_hint_text.as_deref(),
Some("Custom recovery guidance.")
);
Ok(())
}
#[tokio::test]
async fn profile_reflections_config_overrides_base() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"profile = "default_compaction"
compaction_strategy = "reflections"
[features.reflections]
usage_hint_enabled = true
usage_hint_text = "base hint"
[profiles.default_compaction]
compaction_strategy = "default"
[profiles.default_compaction.features.reflections]
usage_hint_enabled = false
usage_hint_text = "profile hint"
"#,
)?;
let config = ConfigBuilder::without_managed_config_for_tests()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
assert_eq!(
config.compaction_strategy,
CompactionStrategyConfig::Default
);
assert!(!config.reflections.usage_hint_enabled);
assert_eq!(
config.reflections.usage_hint_text.as_deref(),
Some("profile hint")
);
Ok(())
}
#[tokio::test]
async fn feature_requirements_normalize_runtime_feature_mutations() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
codex-rs/core/src/config/mod.rs
+58
View File
@@ -55,6 +55,7 @@ use codex_features::FeatureToml;
use codex_features::Features;
use codex_features::FeaturesToml;
use codex_features::MultiAgentV2ConfigToml;
use codex_features::ReflectionsConfigToml;
use codex_login::AuthManagerConfig;
use codex_mcp::McpConfig;
use codex_model_provider_info::LEGACY_OLLAMA_CHAT_PROVIDER_ID;
@@ -63,6 +64,7 @@ use codex_model_provider_info::OLLAMA_CHAT_PROVIDER_REMOVED_ERROR;
use codex_model_provider_info::built_in_model_providers;
use codex_models_manager::ModelsManagerConfig;
use codex_protocol::config_types::AltScreenMode;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::config_types::ForcedLoginMethod;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary;
@@ -550,9 +552,15 @@ pub struct Config {
/// Settings for ghost snapshots (used for undo).
pub ghost_snapshot: GhostSnapshotConfig,
/// Strategy used when compacting conversation history.
pub compaction_strategy: CompactionStrategyConfig,
/// Settings specific to the task-path-based multi-agent tool surface.
pub multi_agent_v2: MultiAgentV2Config,
/// Settings for the Reflections compaction strategy.
pub reflections: ReflectionsConfig,
/// Centralized feature flags; source of truth for feature gating.
pub features: ManagedFeatures,
@@ -614,6 +622,21 @@ impl Default for MultiAgentV2Config {
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReflectionsConfig {
pub usage_hint_enabled: bool,
pub usage_hint_text: Option<String>,
}
impl Default for ReflectionsConfig {
fn default() -> Self {
Self {
usage_hint_enabled: true,
usage_hint_text: None,
}
}
}
impl AuthManagerConfig for Config {
fn codex_home(&self) -> PathBuf {
self.codex_home.to_path_buf()
@@ -1375,6 +1398,34 @@ fn multi_agent_v2_toml_config(features: Option<&FeaturesToml>) -> Option<&MultiA
}
}
fn resolve_reflections_config(
config_toml: &ConfigToml,
config_profile: &ConfigProfile,
) -> ReflectionsConfig {
let base = reflections_toml_config(config_toml.features.as_ref());
let profile = reflections_toml_config(config_profile.features.as_ref());
let default = ReflectionsConfig::default();
let usage_hint_enabled = profile
.and_then(|config| config.usage_hint_enabled)
.or_else(|| base.and_then(|config| config.usage_hint_enabled))
.unwrap_or(default.usage_hint_enabled);
let usage_hint_text = profile
.and_then(|config| config.usage_hint_text.as_ref())
.or_else(|| base.and_then(|config| config.usage_hint_text.as_ref()))
.cloned()
.or(default.usage_hint_text);
ReflectionsConfig {
usage_hint_enabled,
usage_hint_text,
}
}
fn reflections_toml_config(features: Option<&FeaturesToml>) -> Option<&ReflectionsConfigToml> {
features?.reflections.as_ref()
}
pub(crate) fn resolve_web_search_mode_for_turn(
web_search_mode: &Constrained<WebSearchMode>,
sandbox_policy: &SandboxPolicy,
@@ -1706,6 +1757,11 @@ impl Config {
.unwrap_or(WebSearchMode::Cached);
let web_search_config = resolve_web_search_config(&cfg, &config_profile);
let multi_agent_v2 = resolve_multi_agent_v2_config(&cfg, &config_profile);
let reflections = resolve_reflections_config(&cfg, &config_profile);
let compaction_strategy = config_profile
.compaction_strategy
.or(cfg.compaction_strategy)
.unwrap_or_default();
let agent_roles =
agent_roles::load_agent_roles(&cfg, &config_layer_stack, &mut startup_warnings)?;
@@ -2146,7 +2202,9 @@ impl Config {
use_experimental_unified_exec_tool,
background_terminal_max_timeout,
ghost_snapshot,
compaction_strategy,
multi_agent_v2,
reflections,
features,
suppress_unstable_features_warning: cfg
.suppress_unstable_features_warning
codex-rs/core/src/lib.rs
+1
View File
@@ -199,6 +199,7 @@ pub use file_watcher::FileWatcherEvent;
pub use turn_metadata::build_turn_metadata_header;
pub mod compact;
pub(crate) mod memory_trace;
pub(crate) mod reflections;
pub use memory_trace::BuiltMemory;
pub use memory_trace::build_memories_from_trace_files;
pub mod otel_init;
codex-rs/core/src/reflections/mod.rs
+211
View File
@@ -0,0 +1,211 @@
mod prompt;
mod storage;
mod transcript;
use std::sync::Arc;
use crate::RolloutRecorder;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::compact::CompactionAnalyticsAttempt;
use crate::compact::InitialContextInjection;
use crate::compact::compaction_status_from_result;
use crate::compact::insert_initial_context_before_last_real_user_or_summary;
use codex_analytics::CompactionImplementation;
use codex_analytics::CompactionPhase;
use codex_analytics::CompactionReason;
use codex_analytics::CompactionStrategy;
use codex_analytics::CompactionTrigger;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::items::ContextCompactionItem;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::TurnStartedEvent;
pub(crate) use prompt::is_near_limit_reminder;
pub(crate) use prompt::near_limit_reminder;
pub(crate) use prompt::near_limit_reminder_threshold;
pub(crate) use prompt::usage_hint;
pub(crate) use storage::ensure_sidecar_dirs;
pub(crate) use storage::sidecar_path_for_rollout;
pub(crate) async fn run_reflections_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
) -> CodexResult<()> {
let start_event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_context.sub_id.clone(),
started_at: turn_context.turn_timing_state.started_at_unix_secs().await,
model_context_window: turn_context.model_context_window(),
collaboration_mode_kind: turn_context.collaboration_mode.mode,
});
sess.send_event(&turn_context, start_event).await;
run_reflections_compact_task_inner(
sess,
turn_context,
InitialContextInjection::DoNotInject,
CompactionTrigger::Manual,
CompactionReason::UserRequested,
CompactionPhase::StandaloneTurn,
)
.await
}
pub(crate) async fn run_inline_reflections_auto_compact_task(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
run_reflections_compact_task_inner(
sess,
turn_context,
initial_context_injection,
CompactionTrigger::Auto,
reason,
phase,
)
.await
}
pub(crate) async fn run_reflections_compact_task_inner(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
trigger: CompactionTrigger,
reason: CompactionReason,
phase: CompactionPhase,
) -> CodexResult<()> {
let attempt = CompactionAnalyticsAttempt::begin_with_strategy(
sess.as_ref(),
turn_context.as_ref(),
trigger,
reason,
CompactionImplementation::Reflections,
phase,
CompactionStrategy::Reflections,
)
.await;
let result = run_reflections_compact_task_inner_impl(
Arc::clone(&sess),
Arc::clone(&turn_context),
initial_context_injection,
trigger,
)
.await;
attempt
.track(
sess.as_ref(),
compaction_status_from_result(&result),
result.as_ref().err().map(ToString::to_string),
)
.await;
if let Err(err) = &result {
let event = EventMsg::Error(
err.to_error_event(Some("Error running reflections compaction".to_string())),
);
sess.send_event(&turn_context, event).await;
}
result
}
async fn run_reflections_compact_task_inner_impl(
sess: Arc<Session>,
turn_context: Arc<TurnContext>,
initial_context_injection: InitialContextInjection,
trigger: CompactionTrigger,
) -> CodexResult<()> {
let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new());
sess.emit_turn_item_started(&turn_context, &compaction_item)
.await;
let window = write_current_window(&sess, &turn_context, trigger).await?;
let handoff = prompt::post_compaction_handoff(
turn_context.model_context_window(),
&window.logs_path,
&window.notes_path,
);
let mut new_history = vec![ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: handoff.clone(),
}],
end_turn: None,
phase: None,
}];
if matches!(
initial_context_injection,
InitialContextInjection::BeforeLastUserMessage
) {
let initial_context = sess.build_initial_context(turn_context.as_ref()).await;
new_history =
insert_initial_context_before_last_real_user_or_summary(new_history, initial_context);
}
let history_snapshot = sess.clone_history().await;
let ghost_snapshots: Vec<ResponseItem> = history_snapshot
.raw_items()
.iter()
.filter(|item| matches!(item, ResponseItem::GhostSnapshot { .. }))
.cloned()
.collect();
new_history.extend(ghost_snapshots);
let reference_context_item = match initial_context_injection {
InitialContextInjection::DoNotInject => None,
InitialContextInjection::BeforeLastUserMessage => Some(turn_context.to_turn_context_item()),
};
let compacted_item = CompactedItem {
message: handoff,
replacement_history: Some(new_history.clone()),
};
sess.replace_compacted_history(new_history, reference_context_item, compacted_item)
.await;
sess.reset_reflections_near_limit_reminder();
sess.recompute_token_usage(&turn_context).await;
sess.emit_turn_item_completed(&turn_context, compaction_item)
.await;
Ok(())
}
pub(crate) async fn write_current_window(
sess: &Session,
turn_context: &TurnContext,
trigger: CompactionTrigger,
) -> CodexResult<storage::WrittenWindow> {
sess.try_ensure_rollout_materialized().await?;
sess.flush_rollout().await?;
let Some(rollout_path) = sess.current_rollout_path().await else {
return Err(CodexErr::InvalidRequest(
"Reflections requires a persisted rollout path and is unavailable for ephemeral sessions"
.to_string(),
));
};
let sidecar_path = storage::sidecar_path_for_rollout(&rollout_path);
let (rollout_items, _, _) = RolloutRecorder::load_rollout_items(&rollout_path).await?;
let events = transcript::events_since_last_compaction(&rollout_items);
let transcript = transcript::render(transcript::TranscriptInput {
events: &events,
trigger,
context_window_size: turn_context.model_context_window(),
rollout_path: &rollout_path,
});
storage::write_window(
&sidecar_path,
&rollout_path,
trigger,
turn_context.model_context_window(),
transcript,
)
.await
.map_err(Into::into)
}
codex-rs/core/src/reflections/prompt.rs
+177
View File
@@ -0,0 +1,177 @@
use std::path::Path;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use super::storage::WINDOW_DIR_PATTERN;
const NEAR_LIMIT_REMINDER_HEADROOM_TOKENS: i64 = 4096;
const NEAR_LIMIT_REMINDER_PREFIX: &str = "Your context window is nearly exhausted";
pub(crate) fn usage_hint(
context_window_size: Option<i64>,
sidecar_path: &Path,
custom_text: Option<&str>,
) -> String {
if let Some(custom_text) = custom_text {
return custom_text.to_string();
}
format!(
"{context_window_text}\n\n\
Reflections is enabled. Codex automatically records visible messages and tool events from each context window under:\n\n\
{logs_path}/{window_dir_pattern}/transcript.md\n\n\
Use this directory for durable recovery notes:\n\n\
{notes_path}\n\n\
You may want to keep concise notes about your progress incrementally so you can more easily resume after the context window resets. Having things in context is useful, but when details may be needed later, prefer storing references to specific messages, files, commands, findings, or important decisions in notes rather than repeatedly reading complete files or full transcript logs.\n\n\
Future context windows will not automatically include the full previous transcript. If the current task, current user request, important tool result, or relevant instruction detail may matter later, record a concise note and reference the relevant transcript path and message heading.",
context_window_text = context_window_text(context_window_size),
logs_path = sidecar_path.join("logs").display(),
notes_path = sidecar_path.join("notes").display(),
window_dir_pattern = WINDOW_DIR_PATTERN,
)
}
pub(crate) fn near_limit_reminder(remaining_tokens: Option<i64>) -> ResponseItem {
let opening = match remaining_tokens {
Some(remaining_tokens) => format!(
"{NEAR_LIMIT_REMINDER_PREFIX} ({remaining_tokens} tokens remain before your context will be reset)."
),
None => format!("{NEAR_LIMIT_REMINDER_PREFIX} and will be reset soon."),
};
let text = format!(
"{opening}\n\n\
You may want to pause task work and write concise recovery notes under the Reflections notes directory before continuing. Include the current task, progress made, important files, commands, findings, decisions, and next steps. If you have not finished the user's task, you are advised NOT to send a final answer and to use this time to write or clean up notes so you can resume work in the next context window after the context reset.\n\n\
After saving notes, you may call `reflections_new_context_window` to start a fresh context window. If you continue without calling it, your context will automatically reset once the compaction limit is reached."
);
ResponseItem::Message {
id: None,
role: "developer".to_string(),
content: vec![ContentItem::InputText { text }],
end_turn: None,
phase: None,
}
}
pub(crate) fn near_limit_reminder_threshold(auto_compact_limit: i64) -> i64 {
auto_compact_limit
.saturating_sub(NEAR_LIMIT_REMINDER_HEADROOM_TOKENS)
.max(0)
}
pub(crate) fn is_near_limit_reminder(item: &ResponseItem) -> bool {
let ResponseItem::Message { role, content, .. } = item else {
return false;
};
role == "developer"
&& content.iter().any(|item| {
matches!(
item,
ContentItem::InputText { text } if text.starts_with(NEAR_LIMIT_REMINDER_PREFIX)
)
})
}
pub(crate) fn post_compaction_handoff(
context_window_size: Option<i64>,
logs_path: &Path,
notes_path: &Path,
) -> String {
format!(
"{context_window_text}\n\n\
Reflections is enabled. Codex automatically recorded visible messages and tool events from previous context windows here:\n\n\
{logs_path}\n\n\
Use this directory for durable recovery notes:\n\n\
{notes_path}\n\n\
Your context window was reset, and you are continuing in a fresh context window. The current task may only be available in the transcript logs from previous context windows. There may be no new user message in this context window, so do not assume one exists.\n\n\
To recover context, first inspect `notes/`. If `notes/` is empty, missing, or does not clearly identify the current task and status, inspect or search the full conversation context logs under `logs/` for the previous context windows. The logs are organized by context window as `logs/{window_dir_pattern}/transcript.md`. Prefer recovering only the details you need rather than rereading every transcript file.",
context_window_text = context_window_text(context_window_size),
logs_path = logs_path.display(),
notes_path = notes_path.display(),
window_dir_pattern = WINDOW_DIR_PATTERN,
)
}
fn context_window_text(context_window_size: Option<i64>) -> String {
match context_window_size {
Some(context_window_size) => {
format!("Your context window size is {context_window_size} tokens.")
}
None => "Your context window size is not available for this model.".to_string(),
}
}
#[cfg(test)]
mod tests {
use super::is_near_limit_reminder;
use super::near_limit_reminder;
use super::post_compaction_handoff;
use super::usage_hint;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use std::path::Path;
#[test]
fn usage_hint_mentions_sidecar_paths() {
let hint = usage_hint(Some(98304), Path::new("/tmp/rollout.reflections"), None);
assert!(hint.contains("Your context window size is 98304 tokens."));
assert!(hint.contains("/tmp/rollout.reflections/logs/cwNNNNN/transcript.md"));
assert!(hint.contains("/tmp/rollout.reflections/notes"));
}
#[test]
fn handoff_mentions_logs_and_notes() {
let handoff = post_compaction_handoff(
None,
Path::new("/tmp/rollout.reflections/logs"),
Path::new("/tmp/rollout.reflections/notes"),
);
assert!(handoff.contains("Your context window size is not available for this model."));
assert!(handoff.contains("previous context windows here:"));
assert!(handoff.contains("/tmp/rollout.reflections/logs"));
assert!(handoff.contains("logs/cwNNNNN/transcript.md"));
assert!(!handoff.contains("/tmp/rollout.reflections/logs/cw00000/transcript.md"));
assert!(handoff.contains("/tmp/rollout.reflections/notes"));
}
#[test]
fn near_limit_reminder_is_developer_message() {
let reminder = near_limit_reminder(Some(4058));
let ResponseItem::Message { role, content, .. } = &reminder else {
panic!("near-limit reminder should be a message");
};
assert_eq!(role, "developer");
let [ContentItem::InputText { text }] = content.as_slice() else {
panic!("near-limit reminder should contain exactly one text item");
};
assert!(text.contains(
"Your context window is nearly exhausted (4058 tokens remain before your context will be reset)."
));
assert!(text.contains("advised NOT to send a final answer"));
assert!(text.contains("reflections_new_context_window"));
assert!(is_near_limit_reminder(&reminder));
}
#[test]
fn near_limit_reminder_supports_unknown_remaining_tokens() {
let reminder = near_limit_reminder(None);
let ResponseItem::Message { content, .. } = &reminder else {
panic!("near-limit reminder should be a message");
};
let [ContentItem::InputText { text }] = content.as_slice() else {
panic!("near-limit reminder should contain exactly one text item");
};
assert!(
text.starts_with("Your context window is nearly exhausted and will be reset soon.")
);
}
#[test]
fn near_limit_reminder_threshold_never_goes_negative() {
assert_eq!(super::near_limit_reminder_threshold(10_000), 5904);
assert_eq!(super::near_limit_reminder_threshold(100), 0);
}
}
codex-rs/core/src/reflections/storage.rs
+195
View File
@@ -0,0 +1,195 @@
use std::path::Path;
use std::path::PathBuf;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_analytics::CompactionTrigger;
use serde::Deserialize;
use serde::Serialize;
use tokio::io::AsyncWriteExt;
const STATE_SCHEMA: &str = "reflections.state.v1";
const TRANSCRIPT_FILE: &str = "transcript.md";
const WINDOW_DIR_WIDTH: usize = 5;
pub(crate) const WINDOW_DIR_PATTERN: &str = "cwNNNNN";
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct WrittenWindow {
pub(crate) window: String,
pub(crate) logs_path: PathBuf,
pub(crate) transcript_path: PathBuf,
pub(crate) notes_path: PathBuf,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ReflectionsState {
schema: String,
next_window_index: u64,
latest_window: Option<String>,
rollout_path: PathBuf,
windows: Vec<WindowState>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct WindowState {
window: String,
trigger: String,
created_at: String,
transcript_path: PathBuf,
context_window_size: Option<i64>,
}
pub(crate) fn sidecar_path_for_rollout(rollout_path: &Path) -> PathBuf {
rollout_path.with_extension("reflections")
}
pub(crate) async fn ensure_sidecar_dirs(sidecar_path: &Path) -> std::io::Result<()> {
tokio::fs::create_dir_all(sidecar_path).await?;
tokio::fs::create_dir_all(sidecar_path.join("notes")).await?;
tokio::fs::create_dir_all(sidecar_path.join("logs")).await
}
pub(crate) async fn write_window(
sidecar_path: &Path,
rollout_path: &Path,
trigger: CompactionTrigger,
context_window_size: Option<i64>,
transcript: String,
) -> std::io::Result<WrittenWindow> {
ensure_sidecar_dirs(sidecar_path).await?;
let notes_path = sidecar_path.join("notes");
let logs_path = sidecar_path.join("logs");
let mut state = read_state(sidecar_path, rollout_path).await?;
let window_index = allocate_window_index(&state, &logs_path).await;
let window = window_dir_name(window_index);
let window_path = logs_path.join(&window);
tokio::fs::create_dir_all(&window_path).await?;
let transcript_path = window_path.join(TRANSCRIPT_FILE);
write_atomic(&transcript_path, transcript.as_bytes()).await?;
let created_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
state.next_window_index = window_index.saturating_add(1);
state.latest_window = Some(window.clone());
state.windows.push(WindowState {
window: window.clone(),
trigger: trigger_label(trigger).to_string(),
created_at,
transcript_path: PathBuf::from("logs").join(&window).join(TRANSCRIPT_FILE),
context_window_size,
});
write_state(sidecar_path, &state).await?;
Ok(WrittenWindow {
window,
logs_path,
transcript_path,
notes_path,
})
}
async fn read_state(sidecar_path: &Path, rollout_path: &Path) -> std::io::Result<ReflectionsState> {
let state_path = sidecar_path.join("state.json");
match tokio::fs::read_to_string(&state_path).await {
Ok(contents) => {
let mut state: ReflectionsState = serde_json::from_str(&contents)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
if state.schema != STATE_SCHEMA {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("unsupported Reflections state schema `{}`", state.schema),
));
}
state.rollout_path = rollout_path.to_path_buf();
Ok(state)
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(ReflectionsState {
schema: STATE_SCHEMA.to_string(),
next_window_index: 0,
latest_window: None,
rollout_path: rollout_path.to_path_buf(),
windows: Vec::new(),
}),
Err(err) => Err(err),
}
}
async fn allocate_window_index(state: &ReflectionsState, logs_path: &Path) -> u64 {
let mut candidate = state.next_window_index;
while tokio::fs::try_exists(logs_path.join(window_dir_name(candidate)))
.await
.unwrap_or(false)
{
candidate = candidate.saturating_add(1);
}
candidate
}
fn window_dir_name(index: u64) -> String {
format!("cw{index:0WINDOW_DIR_WIDTH$}")
}
async fn write_state(sidecar_path: &Path, state: &ReflectionsState) -> std::io::Result<()> {
let state_json = serde_json::to_vec_pretty(state)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
write_atomic(&sidecar_path.join("state.json"), &state_json).await
}
async fn write_atomic(path: &Path, contents: &[u8]) -> std::io::Result<()> {
let tmp_path = path.with_extension("tmp");
let mut file = tokio::fs::File::create(&tmp_path).await?;
file.write_all(contents).await?;
file.flush().await?;
drop(file);
tokio::fs::rename(&tmp_path, path).await
}
fn trigger_label(trigger: CompactionTrigger) -> &'static str {
match trigger {
CompactionTrigger::Manual => "manual_compact",
CompactionTrigger::Auto => "auto_compact",
}
}
#[cfg(test)]
mod tests {
use super::sidecar_path_for_rollout;
use super::write_window;
use codex_analytics::CompactionTrigger;
#[tokio::test]
async fn write_window_allocates_transcript_and_state() -> std::io::Result<()> {
let temp = tempfile::tempdir()?;
let rollout = temp.path().join("rollout-2026-04-14T00-00-00-thread.jsonl");
let sidecar = sidecar_path_for_rollout(&rollout);
let first = write_window(
&sidecar,
&rollout,
CompactionTrigger::Manual,
Some(98304),
"first".to_string(),
)
.await?;
let second = write_window(
&sidecar,
&rollout,
CompactionTrigger::Auto,
Some(98304),
"second".to_string(),
)
.await?;
assert_eq!(first.window, "cw00000");
assert_eq!(second.window, "cw00001");
assert_eq!(first.logs_path, sidecar.join("logs"));
assert_eq!(
tokio::fs::read_to_string(first.transcript_path).await?,
"first"
);
assert!(sidecar.join("notes").is_dir());
let state = tokio::fs::read_to_string(sidecar.join("state.json")).await?;
assert!(state.contains("\"latest_window\": \"cw00001\""));
Ok(())
}
}
codex-rs/core/src/reflections/transcript.rs
+559
View File
@@ -0,0 +1,559 @@
use std::path::Path;
use codex_analytics::CompactionTrigger;
use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use serde::Serialize;
use serde_json::json;
pub(crate) struct TranscriptInput<'a> {
pub(crate) events: &'a [EventMsg],
pub(crate) trigger: CompactionTrigger,
pub(crate) context_window_size: Option<i64>,
pub(crate) rollout_path: &'a Path,
}
pub(crate) fn events_since_last_compaction(items: &[RolloutItem]) -> Vec<EventMsg> {
let start_index = items
.iter()
.rposition(|item| matches!(item, RolloutItem::Compacted(_)))
.map_or(0, |index| index + 1);
items[start_index..]
.iter()
.filter_map(|item| match item {
RolloutItem::EventMsg(event) => Some(event.clone()),
RolloutItem::SessionMeta(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_) => None,
})
.collect()
}
pub(crate) fn render(input: TranscriptInput<'_>) -> String {
let mut out = String::new();
out.push_str("# Reflections Log\n\n");
out.push_str("- schema: reflections.transcript.v1\n");
out.push_str(&format!("- trigger: {}\n", trigger_label(input.trigger)));
out.push_str(&format!(
"- context_window_size: {}\n",
input
.context_window_size
.map(|value| value.to_string())
.unwrap_or_else(|| "unavailable".to_string())
));
out.push_str(&format!(
"- source_rollout: {}\n",
input.rollout_path.display()
));
out.push('\n');
let mut index = 1usize;
for event in input.events {
if push_event(&mut out, index, event) {
index += 1;
}
}
out
}
fn push_event(out: &mut String, index: usize, event: &EventMsg) -> bool {
match event {
EventMsg::UserMessage(event) => {
let mut text = event.message.clone();
if let Some(images) = event.images.as_ref().filter(|images| !images.is_empty()) {
push_blank_line_if_needed(&mut text);
text.push_str("images:\n");
for image in images {
text.push_str(&format!("- {image}\n"));
}
}
if !event.local_images.is_empty() {
push_blank_line_if_needed(&mut text);
text.push_str("local_images:\n");
for image in &event.local_images {
text.push_str(&format!("- {}\n", image.display()));
}
}
push_message(out, index, "user", &text);
true
}
EventMsg::AgentMessage(event) => {
let heading = event.phase.as_ref().map_or_else(
|| "assistant".to_string(),
|phase| format!("assistant {phase:?}").to_lowercase(),
);
push_message(out, index, &heading, &event.message);
true
}
EventMsg::McpToolCallBegin(event) => {
push_json(
out,
index,
&format!(
"tool_call mcp.{}.{}",
event.invocation.server, event.invocation.tool
),
&json!({
"call_id": event.call_id,
"arguments": event.invocation.arguments,
}),
);
true
}
EventMsg::McpToolCallEnd(event) => {
push_json(
out,
index,
&format!(
"tool_result mcp.{}.{}",
event.invocation.server, event.invocation.tool
),
&json!({
"call_id": event.call_id,
"arguments": event.invocation.arguments,
"success": event.is_success(),
"result": event.result,
}),
);
true
}
EventMsg::WebSearchBegin(event) => {
push_json(out, index, "tool_call web_search", event);
true
}
EventMsg::WebSearchEnd(event) => {
push_json(
out,
index,
"tool_result web_search",
&json!({
"call_id": event.call_id,
"query": event.query,
"action": event.action,
}),
);
true
}
EventMsg::ImageGenerationBegin(event) => {
push_json(out, index, "tool_call image_generation", event);
true
}
EventMsg::ImageGenerationEnd(event) => {
push_json(
out,
index,
"tool_result image_generation",
&json!({
"call_id": event.call_id,
"status": event.status,
"revised_prompt": event.revised_prompt,
"result": event.result,
"saved_path": event.saved_path.as_ref().map(|path| path.display().to_string()),
}),
);
true
}
EventMsg::ExecCommandBegin(event) => {
let mut text = String::new();
text.push_str(&format!("call_id: {}\n", event.call_id));
text.push_str(&format!("cwd: {}\n", event.cwd.display()));
text.push_str(&format!(
"command: {}\n",
serde_json::to_string(&event.command).unwrap_or_else(|_| "[]".to_string())
));
if let Some(input) = event.interaction_input.as_deref() {
text.push_str(&format!("interaction_input: {input}\n"));
}
push_fenced(out, index, "tool_call exec_command", "text", &text);
true
}
EventMsg::ExecCommandEnd(event) => {
let mut text = String::new();
text.push_str(&format!("call_id: {}\n", event.call_id));
text.push_str(&format!("status: {:?}\n", event.status));
text.push_str(&format!("exit_code: {}\n", event.exit_code));
text.push_str(&format!("cwd: {}\n", event.cwd.display()));
text.push_str(&format!(
"command: {}\n",
serde_json::to_string(&event.command).unwrap_or_else(|_| "[]".to_string())
));
let output = exec_output(event);
if !output.is_empty() {
text.push_str("\noutput:\n");
text.push_str(&output);
if !output.ends_with('\n') {
text.push('\n');
}
}
push_fenced(out, index, "tool_result exec_command", "text", &text);
true
}
EventMsg::ViewImageToolCall(event) => {
push_json(
out,
index,
"tool_call view_image",
&json!({
"call_id": event.call_id,
"path": event.path.display().to_string(),
}),
);
true
}
EventMsg::DynamicToolCallRequest(event) => {
push_json(
out,
index,
&format!("tool_call {}", event.tool),
&json!({
"call_id": event.call_id,
"arguments": event.arguments,
}),
);
true
}
EventMsg::DynamicToolCallResponse(event) => {
let text = format!(
"call_id: {}\nsuccess: {}\nerror: {}\n\n{}",
event.call_id,
event.success,
event.error.as_deref().unwrap_or(""),
dynamic_tool_output_items_to_text(&event.content_items)
);
push_fenced(
out,
index,
&format!("tool_result {}", event.tool),
"text",
&text,
);
true
}
EventMsg::PatchApplyBegin(event) => {
push_json(
out,
index,
"tool_call apply_patch",
&json!({
"call_id": event.call_id,
"auto_approved": event.auto_approved,
"changes": event.changes,
}),
);
true
}
EventMsg::PatchApplyEnd(event) => {
push_json(
out,
index,
"tool_result apply_patch",
&json!({
"call_id": event.call_id,
"success": event.success,
"status": event.status,
"stdout": event.stdout,
"stderr": event.stderr,
"changes": event.changes,
}),
);
true
}
EventMsg::Error(_)
| EventMsg::Warning(_)
| EventMsg::RealtimeConversationStarted(_)
| EventMsg::RealtimeConversationRealtime(_)
| EventMsg::RealtimeConversationClosed(_)
| EventMsg::RealtimeConversationSdp(_)
| EventMsg::ModelReroute(_)
| EventMsg::ContextCompacted(_)
| EventMsg::ThreadRolledBack(_)
| EventMsg::TurnStarted(_)
| EventMsg::TurnComplete(_)
| EventMsg::TokenCount(_)
| EventMsg::AgentMessageDelta(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningDelta(_)
| EventMsg::AgentReasoningRawContent(_)
| EventMsg::AgentReasoningRawContentDelta(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::SessionConfigured(_)
| EventMsg::ThreadNameUpdated(_)
| EventMsg::McpStartupUpdate(_)
| EventMsg::McpStartupComplete(_)
| EventMsg::ExecCommandOutputDelta(_)
| EventMsg::TerminalInteraction(_)
| EventMsg::ExecApprovalRequest(_)
| EventMsg::RequestPermissions(_)
| EventMsg::RequestUserInput(_)
| EventMsg::ElicitationRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::GuardianAssessment(_)
| EventMsg::DeprecationNotice(_)
| EventMsg::BackgroundEvent(_)
| EventMsg::UndoStarted(_)
| EventMsg::UndoCompleted(_)
| EventMsg::StreamError(_)
| EventMsg::TurnDiff(_)
| EventMsg::GetHistoryEntryResponse(_)
| EventMsg::McpListToolsResponse(_)
| EventMsg::ListSkillsResponse(_)
| EventMsg::RealtimeConversationListVoicesResponse(_)
| EventMsg::SkillsUpdateAvailable
| EventMsg::PlanUpdate(_)
| EventMsg::TurnAborted(_)
| EventMsg::ShutdownComplete
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::RawResponseItem(_)
| EventMsg::ItemStarted(_)
| EventMsg::ItemCompleted(_)
| EventMsg::HookStarted(_)
| EventMsg::HookCompleted(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::PlanDelta(_)
| EventMsg::ReasoningContentDelta(_)
| EventMsg::ReasoningRawContentDelta(_)
| EventMsg::CollabAgentSpawnBegin(_)
| EventMsg::CollabAgentSpawnEnd(_)
| EventMsg::CollabAgentInteractionBegin(_)
| EventMsg::CollabAgentInteractionEnd(_)
| EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::CollabCloseBegin(_)
| EventMsg::CollabCloseEnd(_)
| EventMsg::CollabResumeBegin(_)
| EventMsg::CollabResumeEnd(_) => false,
}
}
fn push_message(out: &mut String, index: usize, heading: &str, text: &str) {
push_fenced(out, index, heading, "text", text);
}
fn push_fenced(out: &mut String, index: usize, heading: &str, language: &str, text: &str) {
out.push_str(&format!("## msg-{index:06} {heading}\n\n"));
let fence = fence_for(text);
out.push_str(&format!("{fence}{language}\n"));
out.push_str(text);
if !text.ends_with('\n') {
out.push('\n');
}
out.push_str(&format!("{fence}\n\n"));
}
fn push_json<T: Serialize>(out: &mut String, index: usize, heading: &str, value: &T) {
let text = serde_json::to_string_pretty(value)
.unwrap_or_else(|err| format!("<failed to serialize visible event: {err}>"));
push_fenced(out, index, heading, "json", &text);
}
fn push_blank_line_if_needed(text: &mut String) {
if !text.is_empty() && !text.ends_with('\n') {
text.push('\n');
}
if !text.is_empty() {
text.push('\n');
}
}
fn exec_output(event: &codex_protocol::protocol::ExecCommandEndEvent) -> String {
if !event.aggregated_output.is_empty() {
return event.aggregated_output.clone();
}
let mut output = String::new();
if !event.stdout.is_empty() {
output.push_str("stdout:\n");
output.push_str(&event.stdout);
if !event.stdout.ends_with('\n') {
output.push('\n');
}
}
if !event.stderr.is_empty() {
if !output.is_empty() {
output.push('\n');
}
output.push_str("stderr:\n");
output.push_str(&event.stderr);
if !event.stderr.ends_with('\n') {
output.push('\n');
}
}
output
}
fn dynamic_tool_output_items_to_text(items: &[DynamicToolCallOutputContentItem]) -> String {
let mut pieces = Vec::new();
for item in items {
match item {
DynamicToolCallOutputContentItem::InputText { text } => pieces.push(text.clone()),
DynamicToolCallOutputContentItem::InputImage { .. } => {
pieces.push("[image omitted from Reflections transcript]".to_string());
}
}
}
pieces.join("\n")
}
fn fence_for(text: &str) -> String {
let longest_run = longest_backtick_run(text);
"`".repeat(longest_run.max(2) + 1)
}
fn longest_backtick_run(text: &str) -> usize {
let mut longest = 0usize;
let mut current = 0usize;
for ch in text.chars() {
if ch == '`' {
current += 1;
longest = longest.max(current);
} else {
current = 0;
}
}
longest
}
fn trigger_label(trigger: CompactionTrigger) -> &'static str {
match trigger {
CompactionTrigger::Manual => "manual_compact",
CompactionTrigger::Auto => "auto_compact",
}
}
#[cfg(test)]
mod tests {
use super::TranscriptInput;
use super::events_since_last_compaction;
use super::render;
use codex_analytics::CompactionTrigger;
use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem;
use codex_protocol::dynamic_tools::DynamicToolCallRequest;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ReasoningItemContent;
use codex_protocol::models::ReasoningItemReasoningSummary;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::AgentReasoningEvent;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::DynamicToolCallResponseEvent;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::UserMessageEvent;
use std::path::Path;
use std::time::Duration;
#[test]
fn transcript_renders_visible_events_only() {
let items = vec![
RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: "AGENTS.md instructions should be omitted".to_string(),
}],
end_turn: None,
phase: None,
}),
RolloutItem::ResponseItem(ResponseItem::Reasoning {
id: "rs_1".to_string(),
summary: vec![ReasoningItemReasoningSummary::SummaryText {
text: "checked tests".to_string(),
}],
content: Some(vec![ReasoningItemContent::ReasoningText {
text: "hidden chain of thought".to_string(),
}]),
encrypted_content: None,
}),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "please test".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
})),
RolloutItem::EventMsg(EventMsg::DynamicToolCallRequest(DynamicToolCallRequest {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
tool: "lookup_ticket".to_string(),
arguments: serde_json::json!({"id": "T-1"}),
})),
RolloutItem::EventMsg(EventMsg::DynamicToolCallResponse(
DynamicToolCallResponseEvent {
call_id: "call-1".to_string(),
turn_id: "turn-1".to_string(),
tool: "lookup_ticket".to_string(),
arguments: serde_json::json!({"id": "T-1"}),
content_items: vec![DynamicToolCallOutputContentItem::InputText {
text: "ok".to_string(),
}],
success: true,
error: None,
duration: Duration::from_millis(10),
},
)),
RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
message: "done".to_string(),
phase: None,
memory_citation: None,
})),
RolloutItem::EventMsg(EventMsg::AgentReasoning(AgentReasoningEvent {
text: "reasoning summary should be omitted".to_string(),
})),
];
let events = events_since_last_compaction(&items);
let transcript = render(TranscriptInput {
events: &events,
trigger: CompactionTrigger::Manual,
context_window_size: Some(98304),
rollout_path: Path::new("/tmp/rollout.jsonl"),
});
assert!(transcript.contains("## msg-000001 user"));
assert!(transcript.contains("## msg-000002 tool_call lookup_ticket"));
assert!(transcript.contains("## msg-000003 tool_result lookup_ticket"));
assert!(transcript.contains("## msg-000004 assistant"));
assert!(transcript.contains("done"));
assert!(!transcript.contains("AGENTS.md instructions should be omitted"));
assert!(!transcript.contains("checked tests"));
assert!(!transcript.contains("hidden chain of thought"));
assert!(!transcript.contains("reasoning summary should be omitted"));
}
#[test]
fn transcript_uses_events_after_last_compaction() {
let items = vec![
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "old window".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
})),
RolloutItem::Compacted(CompactedItem {
message: "handoff".to_string(),
replacement_history: None,
}),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "current window".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
})),
];
let events = events_since_last_compaction(&items);
let transcript = render(TranscriptInput {
events: &events,
trigger: CompactionTrigger::Auto,
context_window_size: None,
rollout_path: Path::new("/tmp/rollout.jsonl"),
});
assert!(transcript.contains("current window"));
assert!(!transcript.contains("old window"));
}
}
codex-rs/core/src/tasks/compact.rs
+9 -1
View File
@@ -4,6 +4,7 @@ use super::SessionTask;
use super::SessionTaskContext;
use crate::codex::TurnContext;
use crate::state::TaskKind;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::user_input::UserInput;
use tokio_util::sync::CancellationToken;
@@ -27,7 +28,14 @@ impl SessionTask for CompactTask {
_cancellation_token: CancellationToken,
) -> Option<String> {
let session = session.clone_session();
let _ = if crate::compact::should_use_remote_compact_task(&ctx.provider) {
let _ = if ctx.config.compaction_strategy == CompactionStrategyConfig::Reflections {
session.services.session_telemetry.counter(
"codex.task.compact",
/*inc*/ 1,
&[("type", "reflections")],
);
crate::reflections::run_reflections_compact_task(session.clone(), ctx).await
} else if crate::compact::should_use_remote_compact_task(&ctx.provider) {
session.services.session_telemetry.counter(
"codex.task.compact",
/*inc*/ 1,
codex-rs/core/src/tools/handlers/mod.rs
+3
View File
@@ -9,6 +9,7 @@ pub(crate) mod multi_agents;
pub(crate) mod multi_agents_common;
pub(crate) mod multi_agents_v2;
mod plan;
mod reflections;
mod request_permissions;
mod request_user_input;
mod shell;
@@ -42,6 +43,8 @@ pub use list_dir::ListDirHandler;
pub use mcp::McpHandler;
pub use mcp_resource::McpResourceHandler;
pub use plan::PlanHandler;
pub use reflections::ReflectionsGetContextRemainingHandler;
pub use reflections::ReflectionsNewContextWindowHandler;
pub use request_permissions::RequestPermissionsHandler;
pub use request_user_input::RequestUserInputHandler;
pub use shell::ShellCommandHandler;
codex-rs/core/src/tools/handlers/reflections.rs
+71
View File
@@ -0,0 +1,71 @@
use crate::function_tool::FunctionCallError;
use crate::tools::context::FunctionToolOutput;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolPayload;
use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
use codex_tools::REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME;
use codex_tools::REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME;
use serde_json::json;
pub struct ReflectionsNewContextWindowHandler;
impl ToolHandler for ReflectionsNewContextWindowHandler {
type Output = FunctionToolOutput;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
async fn is_mutating(&self, _invocation: &ToolInvocation) -> bool {
true
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
if !matches!(invocation.payload, ToolPayload::Function { .. }) {
return Err(FunctionCallError::RespondToModel(format!(
"{REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME} handler received unsupported payload"
)));
}
invocation
.session
.request_reflections_context_window_reset();
Ok(FunctionToolOutput::from_text(
"A fresh Reflections context window will start after this tool result is recorded."
.to_string(),
Some(true),
))
}
}
pub struct ReflectionsGetContextRemainingHandler;
impl ToolHandler for ReflectionsGetContextRemainingHandler {
type Output = FunctionToolOutput;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
if !matches!(invocation.payload, ToolPayload::Function { .. }) {
return Err(FunctionCallError::RespondToModel(format!(
"{REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME} handler received unsupported payload"
)));
}
let used_tokens = invocation.session.get_total_token_usage().await;
let context_window_size = invocation.turn.model_context_window();
let remaining_tokens =
context_window_size.map(|size| size.saturating_sub(used_tokens).max(0));
let content = json!({
"context_window_size": context_window_size,
"used_tokens": used_tokens,
"remaining_tokens": remaining_tokens,
})
.to_string();
Ok(FunctionToolOutput::from_text(content, Some(true)))
}
}
codex-rs/core/src/tools/spec.rs
+12
View File
@@ -72,6 +72,8 @@ pub(crate) fn build_specs_with_discoverable_tools(
use crate::tools::handlers::McpHandler;
use crate::tools::handlers::McpResourceHandler;
use crate::tools::handlers::PlanHandler;
use crate::tools::handlers::ReflectionsGetContextRemainingHandler;
use crate::tools::handlers::ReflectionsNewContextWindowHandler;
use crate::tools::handlers::RequestPermissionsHandler;
use crate::tools::handlers::RequestUserInputHandler;
use crate::tools::handlers::ShellCommandHandler;
@@ -206,6 +208,16 @@ pub(crate) fn build_specs_with_discoverable_tools(
ToolHandlerKind::Plan => {
builder.register_handler(handler.name, plan_handler.clone());
}
ToolHandlerKind::ReflectionsGetContextRemaining => {
builder.register_handler(
handler.name,
Arc::new(ReflectionsGetContextRemainingHandler),
);
}
ToolHandlerKind::ReflectionsNewContextWindow => {
builder
.register_handler(handler.name, Arc::new(ReflectionsNewContextWindowHandler));
}
ToolHandlerKind::RequestPermissions => {
builder.register_handler(handler.name, request_permissions_handler.clone());
}
codex-rs/core/tests/suite/compact.rs
+204
View File
@@ -7,6 +7,7 @@ use codex_login::CodexAuth;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::built_in_model_providers;
use codex_models_manager::bundled_models_response;
use codex_protocol::config_types::CompactionStrategyConfig;
use codex_protocol::items::TurnItem;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelsResponse;
@@ -410,6 +411,209 @@ async fn summarize_context_three_requests_and_instructions() {
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reflections_manual_compact_writes_sidecar_without_summarizer_request() {
skip_if_no_network!();
let server = start_mock_server().await;
let first_turn = sse(vec![
ev_assistant_message("m1", FIRST_REPLY),
ev_completed_with_tokens("r1", /*total_tokens*/ 80),
]);
let request_log = mount_sse_once(&server, first_turn).await;
let model_provider = non_openai_model_provider(&server);
let test = test_codex()
.with_config(move |config| {
config.model_provider = model_provider;
config.compaction_strategy = CompactionStrategyConfig::Reflections;
})
.build(&server)
.await
.unwrap();
let codex = test.codex.clone();
let rollout_path = test.session_configured.rollout_path.expect("rollout path");
let sidecar_path = rollout_path.with_extension("reflections");
assert!(!sidecar_path.exists());
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "remember this detail".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
assert!(sidecar_path.is_dir());
assert!(sidecar_path.join("notes").is_dir());
assert!(sidecar_path.join("logs").is_dir());
assert!(!sidecar_path.join("logs/cw00000").exists());
codex.submit(Op::Compact).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
assert_eq!(
request_log.requests().len(),
1,
"Reflections /compact should not call the summarizer"
);
codex.submit(Op::Shutdown).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
let transcript_path = sidecar_path.join("logs/cw00000/transcript.md");
let transcript = std::fs::read_to_string(&transcript_path).unwrap_or_else(|err| {
panic!(
"failed to read Reflections transcript {}: {err}",
transcript_path.display()
)
});
assert!(sidecar_path.join("notes").is_dir());
assert!(transcript.contains("schema: reflections.transcript.v1"));
assert!(transcript.contains("remember this detail"));
assert!(transcript.contains(FIRST_REPLY));
assert!(transcript.contains("source_rollout:"));
assert!(!transcript.contains("Reflections is enabled"));
let rollout = std::fs::read_to_string(&rollout_path).unwrap_or_else(|err| {
panic!(
"failed to read rollout file {}: {err}",
rollout_path.display()
)
});
assert!(rollout.contains("Reflections is enabled"));
assert!(rollout.contains(&format!("{}/logs", sidecar_path.display())));
assert!(!rollout.contains("logs/cw00000/transcript.md"));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reflections_near_limit_reminder_is_recorded_once_and_dropped_after_compaction() {
skip_if_no_network!();
let server = start_mock_server().await;
let first_turn = sse(vec![
ev_function_call(DUMMY_CALL_ID, DUMMY_FUNCTION_NAME, "{}"),
ev_completed_with_tokens("r1", /*total_tokens*/ 6_000),
]);
let reminder_turn = sse(vec![
ev_assistant_message("m2", FINAL_REPLY),
ev_completed_with_tokens("r2", /*total_tokens*/ 6_500),
]);
let post_compaction_turn = sse(vec![
ev_assistant_message("m3", "AFTER_REFLECTIONS_COMPACT"),
ev_completed_with_tokens("r3", /*total_tokens*/ 100),
]);
let request_log = mount_sse_sequence(
&server,
vec![first_turn, reminder_turn, post_compaction_turn],
)
.await;
let model_provider = non_openai_model_provider(&server);
let test = test_codex()
.with_config(move |config| {
config.model_provider = model_provider;
config.compaction_strategy = CompactionStrategyConfig::Reflections;
config.model_context_window = Some(20_000);
config.model_auto_compact_token_limit = Some(10_000);
})
.build(&server)
.await
.unwrap();
let codex = test.codex.clone();
let rollout_path = test.session_configured.rollout_path.expect("rollout path");
let sidecar_path = rollout_path.with_extension("reflections");
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "do work near the context limit".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
codex.submit(Op::Compact).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "continue after reflections compact".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await
.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
codex.submit(Op::Shutdown).await.unwrap();
wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
let reminder_prefix = "Your context window is nearly exhausted (";
let reminder_suffix = "tokens remain before your context will be reset).";
let request_bodies: Vec<String> = request_log
.requests()
.into_iter()
.map(|request| request.body_json().to_string())
.collect();
assert_eq!(request_bodies.len(), 3);
assert!(
!body_contains_text(&request_bodies[0], reminder_prefix),
"first request should not include the reminder before the threshold is crossed"
);
assert!(
body_contains_text(&request_bodies[1], reminder_prefix)
&& body_contains_text(&request_bodies[1], reminder_suffix),
"second request should include the near-limit developer reminder"
);
assert!(
body_contains_text(&request_bodies[1], "advised NOT to send a final answer"),
"near-limit reminder should include the stronger note-writing guidance"
);
assert!(
request_bodies[1].contains(&format!("unsupported call: {DUMMY_FUNCTION_NAME}")),
"tool output should still be sent with the reminder"
);
assert!(
!body_contains_text(&request_bodies[2], reminder_prefix),
"reminder should not survive into the fresh context after Reflections compaction"
);
let rollout = std::fs::read_to_string(&rollout_path).unwrap_or_else(|err| {
panic!(
"failed to read rollout file {}: {err}",
rollout_path.display()
)
});
assert!(
rollout.contains(reminder_prefix),
"near-limit reminder should be persisted in the rollout like other developer messages"
);
let transcript_path = sidecar_path.join("logs/cw00000/transcript.md");
let transcript = std::fs::read_to_string(&transcript_path).unwrap_or_else(|err| {
panic!(
"failed to read Reflections transcript {}: {err}",
transcript_path.display()
)
});
assert!(
!transcript.contains(reminder_prefix),
"near-limit reminder should not be copied into visible Reflections transcripts"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn manual_compact_uses_custom_prompt() {
skip_if_no_network!();
codex-rs/features/src/feature_configs.rs
+9
View File
@@ -21,3 +21,12 @@ impl FeatureConfig for MultiAgentV2ConfigToml {
self.enabled
}
}
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ReflectionsConfigToml {
#[serde(skip_serializing_if = "Option::is_none")]
pub usage_hint_enabled: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub usage_hint_text: Option<String>,
}
codex-rs/features/src/lib.rs
+3
View File
@@ -17,6 +17,7 @@ use toml::Table;
mod feature_configs;
mod legacy;
pub use feature_configs::MultiAgentV2ConfigToml;
pub use feature_configs::ReflectionsConfigToml;
use legacy::LegacyFeatureToggles;
pub use legacy::legacy_feature_keys;
@@ -502,6 +503,8 @@ pub fn is_known_feature_key(key: &str) -> bool {
pub struct FeaturesToml {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub multi_agent_v2: Option<FeatureToml<MultiAgentV2ConfigToml>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reflections: Option<ReflectionsConfigToml>,
/// Boolean feature toggles keyed by canonical or legacy feature name.
#[serde(flatten)]
entries: BTreeMap<String, bool>,
codex-rs/features/src/tests.rs
+21
View File
@@ -359,6 +359,27 @@ usage_hint_enabled = false
);
}
#[test]
fn reflections_feature_config_deserializes_table_without_feature_toggle() {
let features_toml: FeaturesToml = toml::from_str(
r#"
[reflections]
usage_hint_enabled = false
usage_hint_text = "Custom recovery guidance."
"#,
)
.expect("features table should deserialize");
assert_eq!(features_toml.entries(), BTreeMap::new());
assert_eq!(
features_toml.reflections,
Some(crate::ReflectionsConfigToml {
usage_hint_enabled: Some(false),
usage_hint_text: Some("Custom recovery guidance.".to_string()),
})
);
}
#[test]
fn unstable_warning_event_only_mentions_enabled_under_development_features() {
let mut configured_features = Table::new();
codex-rs/protocol/src/config_types.rs
+11
View File
@@ -52,6 +52,17 @@ pub enum Verbosity {
High,
}
#[derive(
Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Display, JsonSchema, TS, Default,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum CompactionStrategyConfig {
#[default]
Default,
Reflections,
}
#[derive(
Deserialize, Debug, Clone, Copy, PartialEq, Default, Serialize, Display, JsonSchema, TS,
)]
codex-rs/tools/src/lib.rs
+5
View File
@@ -13,6 +13,7 @@ mod local_tool;
mod mcp_resource_tool;
mod mcp_tool;
mod plan_tool;
mod reflections_tool;
mod request_user_input_tool;
mod responses_api;
mod tool_config;
@@ -74,6 +75,10 @@ pub use mcp_resource_tool::create_read_mcp_resource_tool;
pub use mcp_tool::mcp_call_tool_result_output_schema;
pub use mcp_tool::parse_mcp_tool;
pub use plan_tool::create_update_plan_tool;
pub use reflections_tool::REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME;
pub use reflections_tool::REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME;
pub use reflections_tool::create_reflections_get_context_remaining_tool;
pub use reflections_tool::create_reflections_new_context_window_tool;
pub use request_user_input_tool::REQUEST_USER_INPUT_TOOL_NAME;
pub use request_user_input_tool::create_request_user_input_tool;
pub use request_user_input_tool::normalize_request_user_input_args;
codex-rs/tools/src/reflections_tool.rs
+40
View File
@@ -0,0 +1,40 @@
use crate::JsonSchema;
use crate::ResponsesApiTool;
use crate::ToolSpec;
use std::collections::BTreeMap;
pub const REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME: &str = "reflections_new_context_window";
pub const REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME: &str = "reflections_get_context_remaining";
pub fn create_reflections_new_context_window_tool(usage_hint: Option<&str>) -> ToolSpec {
let mut description = "Starts a fresh context window for the same task. Use this after you have saved concise recovery notes under the Reflections notes directory when the current context is large or the next steps should continue from durable logs.".to_string();
if let Some(usage_hint) = usage_hint {
description.push_str("\n\n");
description.push_str(usage_hint);
}
ToolSpec::Function(ResponsesApiTool {
name: REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME.to_string(),
description,
strict: false,
defer_loading: None,
parameters: empty_parameters(),
output_schema: None,
})
}
pub fn create_reflections_get_context_remaining_tool() -> ToolSpec {
ToolSpec::Function(ResponsesApiTool {
name: REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME.to_string(),
description: "Returns the estimated context window size, used tokens, and remaining tokens for the current thread."
.to_string(),
strict: false,
defer_loading: None,
parameters: empty_parameters(),
output_schema: None,
})
}
fn empty_parameters() -> JsonSchema {
JsonSchema::object(BTreeMap::new(), Some(Vec::new()), Some(false.into()))
}
codex-rs/tools/src/tool_config.rs
+14
View File
@@ -104,6 +104,8 @@ pub struct ToolsConfig {
pub can_request_original_image_detail: bool,
pub collab_tools: bool,
pub multi_agent_v2: bool,
pub reflections: bool,
pub reflections_usage_hint_text: Option<String>,
pub hide_spawn_agent_metadata: bool,
pub spawn_agent_usage_hint: bool,
pub spawn_agent_usage_hint_text: Option<String>,
@@ -225,6 +227,8 @@ impl ToolsConfig {
can_request_original_image_detail: include_original_image_detail,
collab_tools: include_collab_tools,
multi_agent_v2: include_multi_agent_v2,
reflections: false,
reflections_usage_hint_text: None,
hide_spawn_agent_metadata: false,
spawn_agent_usage_hint: true,
spawn_agent_usage_hint_text: None,
@@ -254,6 +258,16 @@ impl ToolsConfig {
self
}
pub fn with_reflections(
mut self,
reflections: bool,
reflections_usage_hint_text: Option<String>,
) -> Self {
self.reflections = reflections;
self.reflections_usage_hint_text = reflections_usage_hint_text;
self
}
pub fn with_hide_spawn_agent_metadata(mut self, hide_spawn_agent_metadata: bool) -> Self {
self.hide_spawn_agent_metadata = hide_spawn_agent_metadata;
self
codex-rs/tools/src/tool_registry_plan.rs
+27
View File
@@ -1,4 +1,6 @@
use crate::CommandToolOptions;
use crate::REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME;
use crate::REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME;
use crate::REQUEST_USER_INPUT_TOOL_NAME;
use crate::ShellToolOptions;
use crate::SpawnAgentToolOptions;
@@ -33,6 +35,8 @@ use crate::create_list_mcp_resource_templates_tool;
use crate::create_list_mcp_resources_tool;
use crate::create_local_shell_tool;
use crate::create_read_mcp_resource_tool;
use crate::create_reflections_get_context_remaining_tool;
use crate::create_reflections_new_context_window_tool;
use crate::create_report_agent_job_result_tool;
use crate::create_request_permissions_tool;
use crate::create_request_user_input_tool;
@@ -247,6 +251,29 @@ pub fn build_tool_registry_plan(
plan.register_handler("request_permissions", ToolHandlerKind::RequestPermissions);
}
if config.reflections {
plan.push_spec(
create_reflections_new_context_window_tool(
config.reflections_usage_hint_text.as_deref(),
),
/*supports_parallel_tool_calls*/ false,
config.code_mode_enabled,
);
plan.push_spec(
create_reflections_get_context_remaining_tool(),
/*supports_parallel_tool_calls*/ true,
config.code_mode_enabled,
);
plan.register_handler(
REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME,
ToolHandlerKind::ReflectionsNewContextWindow,
);
plan.register_handler(
REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME,
ToolHandlerKind::ReflectionsGetContextRemaining,
);
}
if config.search_tool
&& let Some(deferred_mcp_tools) = params.deferred_mcp_tools
{
codex-rs/tools/src/tool_registry_plan_tests.rs
+56
View File
@@ -11,6 +11,7 @@ use crate::ResponsesApiTool;
use crate::ResponsesApiWebSearchFilters;
use crate::ResponsesApiWebSearchUserLocation;
use crate::ToolHandlerSpec;
use crate::ToolName;
use crate::ToolNamespace;
use crate::ToolRegistryPlanDeferredTool;
use crate::ToolsConfigParams;
@@ -632,6 +633,61 @@ fn request_permissions_tool_is_independent_from_additional_permissions() {
assert_lacks_tool_name(&tools, "request_permissions");
}
#[test]
fn reflections_tools_require_config_flag() {
let model_info = model_info();
let features = Features::with_defaults();
let available_models = Vec::new();
let default_tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
available_models: &available_models,
features: &features,
image_generation_tool_auth_allowed: true,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
sandbox_policy: &SandboxPolicy::DangerFullAccess,
windows_sandbox_level: WindowsSandboxLevel::Disabled,
});
let (tools, _) = build_specs(
&default_tools_config,
/*mcp_tools*/ None,
/*deferred_mcp_tools*/ None,
&[],
);
assert_lacks_tool_name(&tools, REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME);
assert_lacks_tool_name(&tools, REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME);
let reflections_tools_config =
default_tools_config.with_reflections(true, Some("Custom Reflections hint.".to_string()));
let (tools, handlers) = build_specs(
&reflections_tools_config,
/*mcp_tools*/ None,
/*deferred_mcp_tools*/ None,
&[],
);
assert_contains_tool_names(
&tools,
&[
REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME,
REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME,
],
);
assert!(handlers.contains(&ToolHandlerSpec {
name: ToolName::plain(REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME),
kind: ToolHandlerKind::ReflectionsNewContextWindow,
}));
assert!(handlers.contains(&ToolHandlerSpec {
name: ToolName::plain(REFLECTIONS_GET_CONTEXT_REMAINING_TOOL_NAME),
kind: ToolHandlerKind::ReflectionsGetContextRemaining,
}));
let new_context_tool = find_tool(&tools, REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME);
let ToolSpec::Function(ResponsesApiTool { description, .. }) = &new_context_tool.spec else {
panic!("{REFLECTIONS_NEW_CONTEXT_WINDOW_TOOL_NAME} should be a function tool");
};
assert!(description.contains("Custom Reflections hint."));
}
#[test]
fn js_repl_requires_feature_flag() {
let model_info = model_info();
codex-rs/tools/src/tool_registry_plan_types.rs
+2
View File
@@ -26,6 +26,8 @@ pub enum ToolHandlerKind {
Mcp,
McpResource,
Plan,
ReflectionsGetContextRemaining,
ReflectionsNewContextWindow,
RequestPermissions,
RequestUserInput,
ResumeAgentV1,