mirror of
https://github.com/openai/codex.git
synced 2026-11-13 14:46:14 +00:00
Fix goal-first live threads missing from thread/list (#28808)
Fixes #28263. ## Why When a thread starts with `/goal`, the goal extension can update SQLite goal state before the thread has any user-turn rollout items. `thread/list` and `thread/search` rely on persisted listing metadata, so a goal-first live thread could be absent from app-server listings after restart even though the goal itself existed. This regressed when goal handling moved out of core: the core path wrote the goal update through the live thread rollout path, while the extension-backed app-server path only updated goal state and emitted the live notification. ## What - Add `GoalSetOutcome::thread_goal_updated_item()` so the goal extension owns the canonical `ThreadGoalUpdated` rollout item shape. - Expose a narrow `CodexThread::append_rollout_items()` helper that appends through the live thread and keeps derived SQLite metadata in sync. - When app-server sets a goal on an active live thread, persist the goal update through that live-thread path. - Add an app-server regression test that starts a live thread with `thread/goal/set` and verifies it appears in state-DB-only `thread/list`. ## Verification - `env -u CODEX_SQLITE_HOME just test -p codex-app-server goal_first_live_thread_appears_in_state_db_thread_list`
This commit is contained in:
1 parent
9684ec25be
commit
e8dd1b45cb
4 files changed
+129
No files matched your search
@@ -135,6 +135,21 @@ impl ThreadGoalRequestProcessor {
|
|||||||
.await
|
.await
|
||||||
.map_err(goal_service_error)?;
|
.map_err(goal_service_error)?;
|
||||||
let goal = ThreadGoal::from(outcome.goal.clone());
|
let goal = ThreadGoal::from(outcome.goal.clone());
|
||||||
|
|
||||||
|
let persist_result = match self.thread_manager.get_thread(thread_id).await {
|
||||||
|
Ok(thread) => {
|
||||||
|
// Live goal-first threads can be listed before any user turn is written.
|
||||||
|
// Use the live path so JSONL and SQLite preview metadata stay in sync.
|
||||||
|
thread
|
||||||
|
.append_rollout_items(&[outcome.thread_goal_updated_item()])
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
Err(_) => Ok(()),
|
||||||
|
};
|
||||||
|
if let Err(err) = persist_result {
|
||||||
|
warn!("failed to persist goal update for live thread {thread_id}: {err}");
|
||||||
|
}
|
||||||
|
|
||||||
self.outgoing
|
self.outgoing
|
||||||
.send_response(
|
.send_response(
|
||||||
request_id.clone(),
|
request_id.clone(),
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ use codex_app_server_protocol::ThreadGoalClearResponse;
|
|||||||
use codex_app_server_protocol::ThreadGoalSetResponse;
|
use codex_app_server_protocol::ThreadGoalSetResponse;
|
||||||
use codex_app_server_protocol::ThreadGoalStatus;
|
use codex_app_server_protocol::ThreadGoalStatus;
|
||||||
use codex_app_server_protocol::ThreadItem;
|
use codex_app_server_protocol::ThreadItem;
|
||||||
|
use codex_app_server_protocol::ThreadListResponse;
|
||||||
use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams;
|
use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams;
|
||||||
use codex_app_server_protocol::ThreadMetadataUpdateParams;
|
use codex_app_server_protocol::ThreadMetadataUpdateParams;
|
||||||
use codex_app_server_protocol::ThreadReadParams;
|
use codex_app_server_protocol::ThreadReadParams;
|
||||||
@@ -464,6 +465,95 @@ async fn thread_goal_get_rejects_unmaterialized_thread() -> Result<()> {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn goal_first_live_thread_appears_in_state_db_thread_list() -> Result<()> {
|
||||||
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||||
|
let codex_home = TempDir::new()?;
|
||||||
|
let codex_home_path = normalized_existing_path(codex_home.path())?;
|
||||||
|
create_config_toml(&codex_home_path, &server.uri())?;
|
||||||
|
let config_path = codex_home_path.join("config.toml");
|
||||||
|
let config = std::fs::read_to_string(&config_path)?;
|
||||||
|
std::fs::write(
|
||||||
|
&config_path,
|
||||||
|
config.replace("personality = true\n", "personality = true\ngoals = true\n"),
|
||||||
|
)?;
|
||||||
|
|
||||||
|
let sqlite_home = codex_home_path
|
||||||
|
.as_path()
|
||||||
|
.to_str()
|
||||||
|
.expect("test codex home should be utf-8");
|
||||||
|
let mut mcp = TestAppServer::new_without_managed_config_with_env(
|
||||||
|
&codex_home_path,
|
||||||
|
&[("CODEX_SQLITE_HOME", Some(sqlite_home))],
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||||
|
|
||||||
|
let start_id = mcp
|
||||||
|
.send_thread_start_request(ThreadStartParams {
|
||||||
|
model: Some("gpt-5.2-codex".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, cwd, .. } = to_response::<ThreadStartResponse>(start_resp)?;
|
||||||
|
|
||||||
|
let goal_id = mcp
|
||||||
|
.send_raw_request(
|
||||||
|
"thread/goal/set",
|
||||||
|
Some(json!({
|
||||||
|
"threadId": thread.id.clone(),
|
||||||
|
"objective": "keep the goal-first thread visible",
|
||||||
|
"status": "paused",
|
||||||
|
})),
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
let goal_resp: JSONRPCResponse = timeout(
|
||||||
|
DEFAULT_READ_TIMEOUT,
|
||||||
|
mcp.read_stream_until_response_message(RequestId::Integer(goal_id)),
|
||||||
|
)
|
||||||
|
.await??;
|
||||||
|
let _goal: ThreadGoalSetResponse = to_response(goal_resp)?;
|
||||||
|
timeout(
|
||||||
|
DEFAULT_READ_TIMEOUT,
|
||||||
|
mcp.read_stream_until_notification_message("thread/goal/updated"),
|
||||||
|
)
|
||||||
|
.await??;
|
||||||
|
|
||||||
|
let list_id = mcp
|
||||||
|
.send_raw_request(
|
||||||
|
"thread/list",
|
||||||
|
Some(json!({
|
||||||
|
"limit": 10,
|
||||||
|
"modelProviders": ["mock_provider"],
|
||||||
|
"sourceKinds": ["vscode"],
|
||||||
|
"archived": false,
|
||||||
|
"cwd": cwd.as_path().to_string_lossy().to_string(),
|
||||||
|
"useStateDbOnly": true,
|
||||||
|
})),
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
let list_resp: JSONRPCResponse = timeout(
|
||||||
|
DEFAULT_READ_TIMEOUT,
|
||||||
|
mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
|
||||||
|
)
|
||||||
|
.await??;
|
||||||
|
let list: ThreadListResponse = to_response(list_resp)?;
|
||||||
|
assert_eq!(
|
||||||
|
list.data
|
||||||
|
.iter()
|
||||||
|
.map(|thread| &thread.id)
|
||||||
|
.collect::<Vec<_>>(),
|
||||||
|
vec![&thread.id]
|
||||||
|
);
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
|
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
|
||||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ use codex_protocol::protocol::AskForApproval;
|
|||||||
use codex_protocol::protocol::Event;
|
use codex_protocol::protocol::Event;
|
||||||
use codex_protocol::protocol::MultiAgentVersion;
|
use codex_protocol::protocol::MultiAgentVersion;
|
||||||
use codex_protocol::protocol::Op;
|
use codex_protocol::protocol::Op;
|
||||||
|
use codex_protocol::protocol::RolloutItem;
|
||||||
use codex_protocol::protocol::SandboxPolicy;
|
use codex_protocol::protocol::SandboxPolicy;
|
||||||
use codex_protocol::protocol::SessionConfiguredEvent;
|
use codex_protocol::protocol::SessionConfiguredEvent;
|
||||||
use codex_protocol::protocol::SessionSource;
|
use codex_protocol::protocol::SessionSource;
|
||||||
@@ -535,6 +536,18 @@ impl CodexThread {
|
|||||||
live_thread.update_metadata(patch, include_archived).await
|
live_thread.update_metadata(patch, include_archived).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Appends rollout items through the live thread so derived metadata stays in sync.
|
||||||
|
pub async fn append_rollout_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()> {
|
||||||
|
let live_thread = self
|
||||||
|
.codex
|
||||||
|
.session
|
||||||
|
.live_thread_for_persistence("append rollout items")
|
||||||
|
.map_err(|err| ThreadStoreError::Internal {
|
||||||
|
message: err.to_string(),
|
||||||
|
})?;
|
||||||
|
live_thread.append_items(items).await
|
||||||
|
}
|
||||||
|
|
||||||
pub fn state_db(&self) -> Option<StateDbHandle> {
|
pub fn state_db(&self) -> Option<StateDbHandle> {
|
||||||
self.codex.state_db()
|
self.codex.state_db()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,8 +6,11 @@ use std::sync::PoisonError;
|
|||||||
use std::sync::Weak;
|
use std::sync::Weak;
|
||||||
|
|
||||||
use codex_protocol::ThreadId;
|
use codex_protocol::ThreadId;
|
||||||
|
use codex_protocol::protocol::EventMsg;
|
||||||
|
use codex_protocol::protocol::RolloutItem;
|
||||||
use codex_protocol::protocol::ThreadGoal;
|
use codex_protocol::protocol::ThreadGoal;
|
||||||
use codex_protocol::protocol::ThreadGoalStatus;
|
use codex_protocol::protocol::ThreadGoalStatus;
|
||||||
|
use codex_protocol::protocol::ThreadGoalUpdatedEvent;
|
||||||
use codex_protocol::protocol::validate_thread_goal_objective;
|
use codex_protocol::protocol::validate_thread_goal_objective;
|
||||||
|
|
||||||
use crate::runtime::GoalRuntimeHandle;
|
use crate::runtime::GoalRuntimeHandle;
|
||||||
@@ -61,6 +64,14 @@ pub struct GoalSetOutcome {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl GoalSetOutcome {
|
impl GoalSetOutcome {
|
||||||
|
pub fn thread_goal_updated_item(&self) -> RolloutItem {
|
||||||
|
RolloutItem::EventMsg(EventMsg::ThreadGoalUpdated(ThreadGoalUpdatedEvent {
|
||||||
|
thread_id: self.goal.thread_id,
|
||||||
|
turn_id: None,
|
||||||
|
goal: self.goal.clone(),
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
pub async fn apply_runtime_effects(&self, goal_service: &GoalService) {
|
pub async fn apply_runtime_effects(&self, goal_service: &GoalService) {
|
||||||
if let Some(runtime) = goal_service.runtime_for_thread(self.goal.thread_id)
|
if let Some(runtime) = goal_service.runtime_for_thread(self.goal.thread_id)
|
||||||
&& let Err(err) = runtime
|
&& let Err(err) = runtime
|
||||||
|
|||||||
Reference in new issue
Block a user