agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

app-server: expose loaded thread status via read/list and notifications (#11786)

Browse Source 
Motivation
- Today, a newly connected client has no direct way to determine the
current runtime status of threads from read/list responses alone.
- This forces clients to infer state from transient events, which can
lead to stale or inconsistent UI when reconnecting or attaching late.

Changes
- Add `status` to `thread/read` responses.
- Add `statuses` to `thread/list` responses.
- Emit `thread/status/changed` notifications with `threadId` and the new
status.
- Track runtime status for all loaded threads and default unknown
threads to `idle`.
- Update protocol/docs/tests/schema fixtures for the revised API.

Testing
- Validated protocol API changes with automated protocol tests and
regenerated schema/type fixtures.
- Validated app-server behavior with unit and integration test suites,
including status transitions and notifications.
This commit is contained in:
Ruslan Nigmatullin
2026-02-18 15:20:03 -08:00
committed by GitHub
parent 216fe7f2ef
commit 1f54496c48
34 changed files with 2563 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+102 -18
View File
@@ -10,6 +10,7 @@ use crate::outgoing_message::ConnectionRequestId;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotification;
use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
use crate::thread_status::ThreadWatchManager;
use chrono::DateTime;
use chrono::SecondsFormat;
use chrono::Utc;
@@ -146,6 +147,7 @@ use codex_app_server_protocol::ThreadSourceKind;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnarchiveResponse;
use codex_app_server_protocol::ThreadUnarchivedNotification;
@@ -338,12 +340,13 @@ pub(crate) struct CodexMessageProcessor {
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
active_login: Arc<Mutex<Option<ActiveLogin>>>,
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
fuzzy_search_sessions: Arc<Mutex<HashMap<String, FuzzyFileSearchSession>>>,
feedback: CodexFeedback,
}
#[derive(Clone, Copy, Debug, Default)]
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(crate) enum ApiVersion {
V1,
#[default]
@@ -399,13 +402,14 @@ impl CodexMessageProcessor {
Self {
auth_manager,
thread_manager,
outgoing,
outgoing: outgoing.clone(),
codex_linux_sandbox_exe,
config,
cli_overrides,
cloud_requirements,
active_login: Arc::new(Mutex::new(None)),
thread_state_manager: ThreadStateManager::new(),
thread_watch_manager: ThreadWatchManager::new_with_outgoing(outgoing),
pending_fuzzy_searches: Arc::new(Mutex::new(HashMap::new())),
fuzzy_search_sessions: Arc::new(Mutex::new(HashMap::new())),
feedback,
@@ -2016,22 +2020,12 @@ impl CodexMessageProcessor {
..
} = new_conv;
let config_snapshot = thread.config_snapshot().await;
let thread = build_thread_from_snapshot(
let mut thread = build_thread_from_snapshot(
thread_id,
&config_snapshot,
session_configured.rollout_path.clone(),
);
let response = ThreadStartResponse {
thread: thread.clone(),
model: config_snapshot.model,
model_provider: config_snapshot.model_provider_id,
cwd: config_snapshot.cwd,
approval_policy: config_snapshot.approval_policy.into(),
sandbox: config_snapshot.sandbox_policy.into(),
reasoning_effort: config_snapshot.reasoning_effort,
};
// Auto-attach a thread listener when starting a thread.
// Use the same behavior as the v1 API, with opt-in support for raw item events.
if let Err(err) = self
@@ -2050,6 +2044,25 @@ impl CodexMessageProcessor {
);
}
self.thread_watch_manager
.upsert_thread(thread.clone())
.await;
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let response = ThreadStartResponse {
thread: thread.clone(),
model: config_snapshot.model,
model_provider: config_snapshot.model_provider_id,
cwd: config_snapshot.cwd,
approval_policy: config_snapshot.approval_policy.into(),
sandbox: config_snapshot.sandbox_policy.into(),
reasoning_effort: config_snapshot.reasoning_effort,
};
self.outgoing.send_response(request_id, response).await;
let notif = ThreadStartedNotification { thread };
@@ -2353,7 +2366,11 @@ impl CodexMessageProcessor {
.await;
match result {
Ok(thread) => {
Ok(mut thread) => {
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let thread_id = thread.id.clone();
let response = ThreadUnarchiveResponse { thread };
self.outgoing.send_response(request_id, response).await;
@@ -2525,7 +2542,23 @@ impl CodexMessageProcessor {
}
};
let data = summaries.into_iter().map(summary_to_thread).collect();
let data = summaries
.into_iter()
.map(summary_to_thread)
.collect::<Vec<_>>();
let statuses = self
.thread_watch_manager
.loaded_statuses_for_threads(data.iter().map(|thread| thread.id.clone()).collect())
.await;
let data = data
.into_iter()
.map(|mut thread| {
if let Some(status) = statuses.get(&thread.id) {
thread.status = status.clone();
}
thread
})
.collect();
let response = ThreadListResponse { data, next_cursor };
self.outgoing.send_response(request_id, response).await;
}
@@ -2711,6 +2744,10 @@ impl CodexMessageProcessor {
}
}
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let response = ThreadReadResponse { thread };
self.outgoing.send_response(request_id, response).await;
}
@@ -2731,6 +2768,13 @@ impl CodexMessageProcessor {
thread_id: ThreadId,
connection_ids: Vec<ConnectionId>,
) {
if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
let config_snapshot = thread.config_snapshot().await;
let loaded_thread =
build_thread_from_snapshot(thread_id, &config_snapshot, thread.rollout_path());
self.thread_watch_manager.upsert_thread(loaded_thread).await;
}
for connection_id in connection_ids {
if let Err(err) = self
.ensure_conversation_listener(thread_id, connection_id, false, ApiVersion::V2)
@@ -2864,7 +2908,7 @@ impl CodexMessageProcessor {
);
}
let Some(thread) = self
let Some(mut thread) = self
.load_thread_from_rollout_or_send_internal(
request_id.clone(),
thread_id,
@@ -2876,6 +2920,15 @@ impl CodexMessageProcessor {
return;
};
self.thread_watch_manager
.upsert_thread(thread.clone())
.await;
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let response = ThreadResumeResponse {
thread,
model: session_configured.model,
@@ -3015,7 +3068,7 @@ impl CodexMessageProcessor {
);
}
let Some(thread) = self
let Some(mut thread) = self
.load_thread_from_rollout_or_send_internal(
request_id.clone(),
existing_thread_id,
@@ -3036,6 +3089,10 @@ impl CodexMessageProcessor {
reasoning_effort,
..
} = config_snapshot;
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let response = ThreadResumeResponse {
thread,
model,
@@ -3385,6 +3442,15 @@ impl CodexMessageProcessor {
}
}
self.thread_watch_manager
.upsert_thread(thread.clone())
.await;
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let response = ThreadForkResponse {
thread: thread.clone(),
model: session_configured.model,
@@ -4658,6 +4724,10 @@ impl CodexMessageProcessor {
.await;
}
self.thread_watch_manager
.remove_thread(&thread_id.to_string())
.await;
if state_db_ctx.is_none() {
state_db_ctx = get_state_db(&self.config, None).await;
}
@@ -5512,7 +5582,14 @@ impl CodexMessageProcessor {
if let Some(rollout_path) = review_thread.rollout_path() {
match read_summary_from_rollout(rollout_path.as_path(), fallback_provider).await {
Ok(summary) => {
let thread = summary_to_thread(summary);
let mut thread = summary_to_thread(summary);
self.thread_watch_manager
.upsert_thread(thread.clone())
.await;
thread.status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
let notif = ThreadStartedNotification { thread };
self.outgoing
.send_server_notification(ServerNotification::ThreadStarted(notif))
@@ -5747,6 +5824,8 @@ impl CodexMessageProcessor {
thread_state.set_listener(cancel_tx, &conversation);
}
let outgoing_for_task = self.outgoing.clone();
let thread_manager = self.thread_manager.clone();
let thread_watch_manager = self.thread_watch_manager.clone();
let fallback_model_provider = self.config.model_provider_id.clone();
tokio::spawn(async move {
loop {
@@ -5819,8 +5898,10 @@ impl CodexMessageProcessor {
event.clone(),
conversation_id,
conversation.clone(),
thread_manager.clone(),
thread_outgoing,
thread_state.clone(),
thread_watch_manager.clone(),
api_version,
fallback_model_provider.clone(),
)
@@ -6733,6 +6814,7 @@ fn build_thread_from_snapshot(
model_provider: config_snapshot.model_provider_id.clone(),
created_at: now,
updated_at: now,
status: ThreadStatus::NotLoaded,
path,
cwd: config_snapshot.cwd.clone(),
cli_version: env!("CARGO_PKG_VERSION").to_string(),
@@ -6770,6 +6852,7 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread {
model_provider,
created_at: created_at.map(|dt| dt.timestamp()).unwrap_or(0),
updated_at: updated_at.map(|dt| dt.timestamp()).unwrap_or(0),
status: ThreadStatus::NotLoaded,
path: Some(path),
cwd,
cli_version,
@@ -6786,6 +6869,7 @@ mod tests {
use codex_protocol::protocol::SessionSource;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::path::PathBuf;
use tempfile::TempDir;
#[test]