feat: disable memory endpoint (#17626)

This commit is contained in:
jif-oai
2026-04-13 18:29:49 +01:00
committed by GitHub
parent ac82443d07
commit 46a266cd6a
16 changed files with 478 additions and 0 deletions
+11
View File
@@ -140,6 +140,7 @@ Example with notification opt-out:
- `thread/loaded/list` — list the thread ids currently loaded in memory.
- `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
- `thread/metadata/update` — patch stored thread metadata in sqlite; currently supports updating persisted `gitInfo` fields and returns the refreshed `thread`.
- `thread/memoryMode/set` — experimental; set a threads persisted memory eligibility to `"enabled"` or `"disabled"` for either a loaded thread or a stored rollout; returns `{}` on success.
- `thread/status/changed` — notification emitted when a loaded threads status changes (`threadId` + new `status`).
- `thread/archive` — move a threads rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
@@ -395,6 +396,16 @@ Use `thread/metadata/update` to patch sqlite-backed metadata for a thread withou
} }
```
Experimental: use `thread/memoryMode/set` to change whether a thread remains eligible for future memory generation.
```json
{ "method": "thread/memoryMode/set", "id": 26, "params": {
"threadId": "thr_123",
"mode": "disabled"
} }
{ "id": 26, "result": {} }
```
### Example: Archive a thread
Use `thread/archive` to move the persisted rollout (stored as a JSONL file on disk) into the archived sessions directory.
@@ -137,6 +137,8 @@ use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadLoadedListParams;
use codex_app_server_protocol::ThreadLoadedListResponse;
use codex_app_server_protocol::ThreadMemoryModeSetParams;
use codex_app_server_protocol::ThreadMemoryModeSetResponse;
use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams;
use codex_app_server_protocol::ThreadMetadataUpdateParams;
use codex_app_server_protocol::ThreadMetadataUpdateResponse;
@@ -773,6 +775,10 @@ impl CodexMessageProcessor {
self.thread_metadata_update(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadMemoryModeSet { request_id, params } => {
self.thread_memory_mode_set(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadUnarchive { request_id, params } => {
self.thread_unarchive(to_connection_request_id(request_id), params)
.await;
@@ -2790,6 +2796,118 @@ impl CodexMessageProcessor {
.await;
}
async fn thread_memory_mode_set(
&self,
request_id: ConnectionRequestId,
params: ThreadMemoryModeSetParams,
) {
let ThreadMemoryModeSetParams { thread_id, mode } = params;
let thread_id = match ThreadId::from_string(&thread_id) {
Ok(id) => id,
Err(err) => {
self.send_invalid_request_error(request_id, format!("invalid thread id: {err}"))
.await;
return;
}
};
if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
if thread.config_snapshot().await.ephemeral {
self.send_invalid_request_error(
request_id,
format!("ephemeral thread does not support memory mode updates: {thread_id}"),
)
.await;
return;
}
if let Err(err) = thread.set_thread_memory_mode(mode.to_core()).await {
self.send_internal_error(
request_id,
format!("failed to set thread memory mode: {err}"),
)
.await;
return;
}
self.outgoing
.send_response(request_id, ThreadMemoryModeSetResponse {})
.await;
return;
}
let rollout_path =
match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()).await
{
Ok(Some(path)) => Some(path),
Ok(None) => None,
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {thread_id}: {err}"),
)
.await;
return;
}
};
let Some(rollout_path) = rollout_path else {
self.send_invalid_request_error(request_id, format!("thread not found: {thread_id}"))
.await;
return;
};
let mut session_meta = match read_session_meta_line(rollout_path.as_path()).await {
Ok(session_meta) => session_meta,
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to set thread memory mode: {err}"),
)
.await;
return;
}
};
if session_meta.meta.id != thread_id {
self.send_internal_error(
request_id,
format!(
"failed to set thread memory mode: rollout session metadata id mismatch: expected {thread_id}, found {}",
session_meta.meta.id
),
)
.await;
return;
}
session_meta.meta.memory_mode = Some(mode.as_str().to_string());
let item = RolloutItem::SessionMeta(session_meta);
if let Err(err) = append_rollout_item_to_path(rollout_path.as_path(), &item).await {
self.send_internal_error(
request_id,
format!("failed to set thread memory mode: {err}"),
)
.await;
return;
}
let state_db_ctx = open_state_db_for_direct_thread_lookup(&self.config).await;
reconcile_rollout(
state_db_ctx.as_deref(),
rollout_path.as_path(),
self.config.model_provider_id.as_str(),
/*builder*/ None,
&[],
/*archived_only*/ None,
/*new_thread_memory_mode*/ None,
)
.await;
self.outgoing
.send_response(request_id, ThreadMemoryModeSetResponse {})
.await;
}
async fn thread_metadata_update(
&self,
request_id: ConnectionRequestId,
@@ -64,6 +64,7 @@ use codex_app_server_protocol::ThreadCompactStartParams;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadLoadedListParams;
use codex_app_server_protocol::ThreadMemoryModeSetParams;
use codex_app_server_protocol::ThreadMetadataUpdateParams;
use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
@@ -583,6 +584,15 @@ impl McpProcess {
self.send_request("mock/experimentalMethod", params).await
}
/// Send a `thread/memoryMode/set` JSON-RPC request (v2, experimental).
pub async fn send_thread_memory_mode_set_request(
&mut self,
params: ThreadMemoryModeSetParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("thread/memoryMode/set", params).await
}
/// Send a `turn/start` JSON-RPC request (v2).
pub async fn send_turn_start_request(
&mut self,
@@ -11,6 +11,8 @@ use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::MockExperimentalMethodParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadMemoryMode;
use codex_app_server_protocol::ThreadMemoryModeSetParams;
use codex_app_server_protocol::ThreadRealtimeStartParams;
use codex_app_server_protocol::ThreadRealtimeStartTransport;
use codex_app_server_protocol::ThreadStartParams;
@@ -89,6 +91,39 @@ async fn realtime_conversation_start_requires_experimental_api_capability() -> R
Ok(())
}
#[tokio::test]
async fn thread_memory_mode_set_requires_experimental_api_capability() -> Result<()> {
let codex_home = TempDir::new()?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
let init = mcp
.initialize_with_capabilities(
default_client_info(),
Some(InitializeCapabilities {
experimental_api: false,
opt_out_notification_methods: None,
}),
)
.await?;
let JSONRPCMessage::Response(_) = init else {
anyhow::bail!("expected initialize response, got {init:?}");
};
let request_id = mcp
.send_thread_memory_mode_set_request(ThreadMemoryModeSetParams {
thread_id: "thr_123".to_string(),
mode: ThreadMemoryMode::Disabled,
})
.await?;
let error = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
)
.await??;
assert_experimental_capability_error(error, "thread/memoryMode/set");
Ok(())
}
#[tokio::test]
async fn realtime_webrtc_start_requires_experimental_api_capability() -> Result<()> {
let codex_home = TempDir::new()?;
@@ -37,6 +37,7 @@ mod thread_archive;
mod thread_fork;
mod thread_list;
mod thread_loaded_list;
mod thread_memory_mode_set;
mod thread_metadata_update;
mod thread_name_websocket;
mod thread_read;
@@ -0,0 +1,138 @@
use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_fake_rollout;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadMemoryMode;
use codex_app_server_protocol::ThreadMemoryModeSetParams;
use codex_app_server_protocol::ThreadMemoryModeSetResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_protocol::ThreadId;
use codex_state::StateRuntime;
use pretty_assertions::assert_eq;
use std::path::Path;
use std::sync::Arc;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_memory_mode_set_updates_loaded_thread_state() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let state_db = init_state_db(codex_home.path()).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let thread_uuid = ThreadId::from_string(&thread.id)?;
let set_id = mcp
.send_thread_memory_mode_set_request(ThreadMemoryModeSetParams {
thread_id: thread.id,
mode: ThreadMemoryMode::Disabled,
})
.await?;
let set_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(set_id)),
)
.await??;
let _: ThreadMemoryModeSetResponse = to_response::<ThreadMemoryModeSetResponse>(set_resp)?;
let memory_mode = state_db.get_thread_memory_mode(thread_uuid).await?;
assert_eq!(memory_mode.as_deref(), Some("disabled"));
Ok(())
}
#[tokio::test]
async fn thread_memory_mode_set_updates_stored_thread_state() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let state_db = init_state_db(codex_home.path()).await?;
let thread_id = create_fake_rollout(
codex_home.path(),
"2025-01-06T08-30-00",
"2025-01-06T08:30:00Z",
"Stored thread preview",
Some("mock_provider"),
/*git_info*/ None,
)?;
let thread_uuid = ThreadId::from_string(&thread_id)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
for mode in [ThreadMemoryMode::Disabled, ThreadMemoryMode::Enabled] {
let set_id = mcp
.send_thread_memory_mode_set_request(ThreadMemoryModeSetParams {
thread_id: thread_id.clone(),
mode,
})
.await?;
let set_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(set_id)),
)
.await??;
let _: ThreadMemoryModeSetResponse = to_response::<ThreadMemoryModeSetResponse>(set_resp)?;
}
let memory_mode = state_db.get_thread_memory_mode(thread_uuid).await?;
assert_eq!(memory_mode.as_deref(), Some("enabled"));
Ok(())
}
async fn init_state_db(codex_home: &Path) -> Result<Arc<StateRuntime>> {
let state_db = StateRuntime::init(codex_home.to_path_buf(), "mock_provider".into()).await?;
state_db
.mark_backfill_complete(/*last_watermark*/ None)
.await?;
Ok(state_db)
}
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
model_provider = "mock_provider"
suppress_unstable_features_warning = true
[features]
sqlite = true
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}