Add recency sorting to project/list (#41223)

## What changed

- Include `recencyAt`, derived from the newest non-archived assigned thread, in project responses.
- Let `project/list` sort by `position` or `recencyAt` in either direction, with recency as the default descending order for that key and projects without activity placed last.
- Encode the sort key, direction, and millisecond-precision anchor in new cursors while retaining existing ascending-position cursors.
- Add an index for project recency lookups and document the API's sorting and pagination behavior.

## Testing

- Cover recency updates when threads are assigned, archived, unarchived, or deleted.
- Cover ordering, null handling, tie-breaking, cursor pagination, invalid sort combinations, and cursor mismatches.

GitOrigin-RevId: 6165e2b9976b0b45cfa16bcd42978b67811f6df5
This commit is contained in:
Anton Panasenko
2026-08-27 23:43:46 +00:00
committed by copyberry
parent 2d929eb7c3
commit f6726e8986
22 changed files with 796 additions and 122 deletions
@@ -2619,6 +2619,13 @@
],
"type": "object"
},
"ProjectSortKey": {
"enum": [
"position",
"recencyAt"
],
"type": "string"
},
"RealtimeConversationVersion": {
"enum": [
"v1",
@@ -16945,6 +16945,14 @@
"format": "int64",
"type": "integer"
},
"recencyAt": {
"description": "Newest non-archived member thread's recency, in Unix seconds; null when none exist.",
"format": "int64",
"type": [
"integer",
"null"
]
},
"roots": {
"items": {
"$ref": "#/definitions/v2/ProjectRoot"
@@ -17003,6 +17011,13 @@
],
"type": "object"
},
"ProjectSortKey": {
"enum": [
"position",
"recencyAt"
],
"type": "string"
},
"QueuedSubmission": {
"properties": {
"clientUserMessageId": {
@@ -13078,6 +13078,14 @@
"format": "int64",
"type": "integer"
},
"recencyAt": {
"description": "Newest non-archived member thread's recency, in Unix seconds; null when none exist.",
"format": "int64",
"type": [
"integer",
"null"
]
},
"roots": {
"items": {
"$ref": "#/definitions/ProjectRoot"
@@ -13136,6 +13144,13 @@
],
"type": "object"
},
"ProjectSortKey": {
"enum": [
"position",
"recencyAt"
],
"type": "string"
},
"QueuedSubmission": {
"properties": {
"clientUserMessageId": {
@@ -3,4 +3,8 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { ProjectRoot } from "./ProjectRoot";
export type Project = { id: string, name: string, roots: Array<ProjectRoot>, metadata: { [key in string]?: string }, position: number, createdAt: number, updatedAt: number, };
export type Project = { id: string, name: string, roots: Array<ProjectRoot>, metadata: { [key in string]?: string }, position: number, createdAt: number, updatedAt: number,
/**
* Newest non-archived member thread's recency, in Unix seconds; null when none exist.
*/
recencyAt: number | null, };
@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type ProjectSortKey = "position" | "recencyAt";
@@ -388,6 +388,7 @@ export type { Project } from "./Project";
export type { ProjectChangeType } from "./ProjectChangeType";
export type { ProjectChangedNotification } from "./ProjectChangedNotification";
export type { ProjectRoot } from "./ProjectRoot";
export type { ProjectSortKey } from "./ProjectSortKey";
export type { QueuedSubmission } from "./QueuedSubmission";
export type { RateLimitReachedType } from "./RateLimitReachedType";
export type { RateLimitResetCredit } from "./RateLimitResetCredit";
@@ -7,6 +7,8 @@ use codex_utils_absolute_path::AbsolutePathBuf;
use serde::Deserialize;
use serde::Serialize;
use super::SortDirection;
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
@@ -28,6 +30,17 @@ pub struct Project {
pub created_at: i64,
#[ts(type = "number")]
pub updated_at: i64,
/// Newest non-archived member thread's recency, in Unix seconds; null when none exist.
#[ts(type = "number | null")]
pub recency_at: Option<i64>,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase", export_to = "v2/")]
pub enum ProjectSortKey {
Position,
RecencyAt,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS, ExperimentalApi)]
@@ -38,6 +51,12 @@ pub struct ProjectListParams {
pub cursor: Option<String>,
#[ts(optional = nullable)]
pub limit: Option<u32>,
/// Defaults to position. Recency sorting always places empty projects last.
#[ts(optional = nullable)]
pub sort_key: Option<ProjectSortKey>,
/// Requires sortKey. Defaults to asc for position and desc for recencyAt.
#[ts(optional = nullable)]
pub sort_direction: Option<SortDirection>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
@@ -1500,7 +1500,7 @@ pub enum ThreadSearchSortKey {
RecencyAt,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, JsonSchema, TS)]
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export_to = "v2/")]
pub enum SortDirection {
+22
View File
@@ -454,6 +454,28 @@ To branch from a stored session, call `thread/fork` with the `thread.id`. This c
Like `thread/resume`, full-history hydration is deprecated for paginated `thread/fork` and emits `deprecationNotice`. Clients should pass `excludeTurns: true` to return only thread metadata in `thread.turns` and page history with `thread/turns/list` and `thread/items/list`. Metadata-only forks do not replay restored `thread/tokenUsage/updated`. Ephemeral forks of paginated threads require `excludeTurns: true`.
### Listing projects
`project/list` accepts optional `sortKey` (`position` or `recencyAt`) and
`sortDirection` (`asc` or `desc`), alongside `limit` and the opaque `cursor`.
Omitting `sortKey` preserves manual position order. A non-null `sortDirection`
requires an explicit key; it defaults to `asc` for `position` and `desc` for `recencyAt`.
```json
{ "sortKey": "recencyAt", "sortDirection": "desc", "limit": 50, "cursor": null }
```
Every project response includes `recencyAt`: the newest non-archived, explicitly
assigned thread's recency in Unix seconds, across all sources, or `null` when none
exist. Like `thread/list`, thread recency starts at creation and advances at turn
start, not for background output. Removing or archiving members can lower project
recency. Task activity does not change project `updatedAt` or emit `project/changed`.
Nulls sort last in either direction; project IDs break ties in the same direction.
Cursor anchors retain millisecond precision. Continue with the same sort options;
existing position cursors remain supported. Pagination is a live view, so concurrent
activity can move projects across a cursor.
### Example: List threads (with pagination & filters)
`thread/list` lets you render a history UI. Results default to `createdAt` (newest first) descending.
@@ -19,14 +19,18 @@ use codex_app_server_protocol::ProjectMoveResponse;
use codex_app_server_protocol::ProjectReadParams;
use codex_app_server_protocol::ProjectReadResponse;
use codex_app_server_protocol::ProjectRoot;
use codex_app_server_protocol::ProjectSortKey;
use codex_app_server_protocol::ProjectUpdateParams;
use codex_app_server_protocol::ProjectUpdateResponse;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadProjectUpdatedNotification;
use codex_thread_store::CreateProjectParams as StoreCreateProjectParams;
use codex_thread_store::ListProjectsParams as StoreListProjectsParams;
use codex_thread_store::MoveProjectParams as StoreMoveProjectParams;
use codex_thread_store::ProjectMoveOutcome;
use codex_thread_store::ProjectSortKey as StoreProjectSortKey;
use codex_thread_store::SortDirection as StoreSortDirection;
use codex_thread_store::StoredProject;
use codex_thread_store::StoredProjectRoot;
use codex_thread_store::ThreadStore;
@@ -66,10 +70,27 @@ impl ProjectRequestProcessor {
&self,
params: ProjectListParams,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
if params.sort_key.is_none() && params.sort_direction.is_some() {
return Err(invalid_params("sortDirection requires sortKey"));
}
let sort_key = match params.sort_key.unwrap_or(ProjectSortKey::Position) {
ProjectSortKey::Position => StoreProjectSortKey::Position,
ProjectSortKey::RecencyAt => StoreProjectSortKey::RecencyAt,
};
let sort_direction = match params.sort_direction {
Some(SortDirection::Asc) => StoreSortDirection::Asc,
Some(SortDirection::Desc) => StoreSortDirection::Desc,
None => match sort_key {
StoreProjectSortKey::Position => StoreSortDirection::Asc,
StoreProjectSortKey::RecencyAt => StoreSortDirection::Desc,
},
};
let page = self
.thread_store
.list_projects(StoreListProjectsParams {
cursor: params.cursor,
sort_key,
sort_direction,
limit: params
.limit
.map(|limit| limit as usize)
@@ -359,6 +380,9 @@ fn api_project(project: StoredProject) -> Result<Project, JSONRPCErrorError> {
position: project.position,
created_at: project.created_at_ms / 1000,
updated_at: project.updated_at_ms / 1000,
recency_at: project
.recency_at_ms
.map(|timestamp| timestamp.div_euclid(/*rhs*/ 1000)),
})
}
@@ -22,9 +22,13 @@ use codex_app_server_protocol::ProjectMoveResponse;
use codex_app_server_protocol::ProjectReadParams;
use codex_app_server_protocol::ProjectReadResponse;
use codex_app_server_protocol::ProjectRoot;
use codex_app_server_protocol::ProjectSortKey;
use codex_app_server_protocol::ProjectUpdateParams;
use codex_app_server_protocol::ProjectUpdateResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadArchiveResponse;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadHistoryMode;
@@ -48,6 +52,155 @@ use pretty_assertions::assert_eq;
use tempfile::TempDir;
use uuid::Uuid;
#[tokio::test]
async fn projects_list_by_recency_and_preserve_metadata_timestamps() -> Result<()> {
let responses = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&responses.uri())
.enable_feature(Feature::Sqlite)
.write(codex_home.path())?;
let mut server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let mut projects = Vec::new();
let mut threads = Vec::new();
for name in ["first", "second", "empty"] {
let created: ProjectCreateResponse = server
.request(|request_id| ClientRequest::ProjectCreate {
request_id,
params: ProjectCreateParams {
name: name.to_string(),
roots: Vec::new(),
metadata: None,
idempotency_key: name.to_string(),
},
})
.await?;
assert_eq!(created.project.recency_at, None);
let mut expected = created.project;
if name != "empty" {
let started = server
.start_thread(ThreadStartParams {
project_id: Some(expected.id.clone()),
..Default::default()
})
.await?;
server
.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: started.thread.id.clone(),
input: vec![UserInput::Text {
text: "Hello".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let listed: ThreadListResponse = server
.request(|request_id| ClientRequest::ThreadList {
request_id,
params: ThreadListParams {
cursor: None,
limit: Some(10),
sort_key: None,
sort_direction: None,
model_providers: None,
source_kinds: None,
archived: None,
section_id: None,
project_id: Some(Some(expected.id.clone())),
cwd: None,
use_state_db_only: true,
search_term: None,
parent_thread_id: None,
ancestor_thread_id: None,
},
})
.await?;
expected.recency_at = listed.data[0].recency_at;
assert!(expected.recency_at.is_some());
threads.push(started.thread.id);
}
let read: ProjectReadResponse = server
.request(|request_id| ClientRequest::ProjectRead {
request_id,
params: ProjectReadParams {
project_id: expected.id.clone(),
},
})
.await?;
assert_eq!(read.project, expected);
projects.push(expected);
}
for (sort_key, sort_direction, order) in [
(None, None, [0, 1, 2]),
(Some(ProjectSortKey::RecencyAt), None, [1, 0, 2]),
(
Some(ProjectSortKey::RecencyAt),
Some(SortDirection::Asc),
[0, 1, 2],
),
] {
let mut cursor = None;
for index in order {
let page: ProjectListResponse = server
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
cursor: cursor.clone(),
limit: Some(1),
sort_key,
sort_direction,
},
})
.await?;
assert_eq!(page.data, vec![projects[index].clone()]);
cursor = page.next_cursor;
}
assert_eq!(cursor, None);
}
let _: ThreadArchiveResponse = server
.request(|request_id| ClientRequest::ThreadArchive {
request_id,
params: ThreadArchiveParams {
thread_id: threads[1].clone(),
},
})
.await?;
projects[1].recency_at = None;
let archived: ProjectReadResponse = server
.request(|request_id| ClientRequest::ProjectRead {
request_id,
params: ProjectReadParams {
project_id: projects[1].id.clone(),
},
})
.await?;
assert_eq!(archived.project, projects[1]);
for params in [
ProjectListParams {
cursor: None,
limit: None,
sort_key: None,
sort_direction: Some(SortDirection::Desc),
},
ProjectListParams {
cursor: Some(format!("0|{}", projects[0].id)),
limit: None,
sort_key: Some(ProjectSortKey::RecencyAt),
sort_direction: None,
},
] {
let id = server.send_project_list_request(params).await?;
let error = server
.read_stream_until_error_message(RequestId::Integer(id))
.await?;
assert_eq!(error.error.code, -32602);
}
Ok(())
}
#[tokio::test]
async fn projects_persist_and_assign_threads() -> Result<()> {
let responses = create_mock_responses_server_repeating_assistant("Done").await;
@@ -316,6 +469,10 @@ async fn projects_persist_and_assign_threads() -> Result<()> {
})
.await?;
assert_eq!(updated.project.name, "Renamed");
assert_eq!(
updated.project.recency_at,
reassigned_list.data[0].recency_at
);
assert!(updated.project.roots.is_empty());
assert!(updated.project.metadata.is_empty());
let updated_notification: ProjectChangedNotification =
@@ -351,6 +508,8 @@ async fn projects_persist_and_assign_threads() -> Result<()> {
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: None,
limit: Some(10),
},
@@ -385,6 +544,8 @@ async fn projects_persist_and_assign_threads() -> Result<()> {
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: None,
limit: Some(10),
},
@@ -436,6 +597,8 @@ async fn projects_persist_and_assign_threads() -> Result<()> {
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: None,
limit: Some(10),
},
@@ -641,6 +804,8 @@ async fn project_import_is_atomic_and_notifies_after_commit_in_order() -> Result
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: None,
limit: Some(10),
},
@@ -665,6 +830,8 @@ async fn project_import_is_atomic_and_notifies_after_commit_in_order() -> Result
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: None,
limit: Some(10),
},
@@ -695,6 +862,8 @@ async fn project_import_is_atomic_and_notifies_after_commit_in_order() -> Result
.request(|request_id| ClientRequest::ProjectList {
request_id,
params: ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: None,
limit: Some(10),
},
@@ -742,6 +911,8 @@ async fn projects_validate_filters_cursors_and_sqlite_less_assignment() -> Resul
}
let cursor_id = server
.send_project_list_request(ProjectListParams {
sort_key: None,
sort_direction: None,
cursor: Some("123|not-a-uuid|extra".to_string()),
limit: Some(10),
})
@@ -0,0 +1,3 @@
CREATE INDEX idx_threads_project_recency
ON threads(project_id, recency_at_ms DESC)
WHERE archived = 0 AND project_id IS NOT NULL;
+1
View File
@@ -26,6 +26,7 @@ pub use model::LogRow;
pub use model::Phase2JobClaimOutcome;
pub use model::Project;
pub use model::ProjectRoot;
pub use model::ProjectSortKey;
pub use model::ProjectsPage;
pub use model::QueuedUserSubmissionRecord;
pub use model::RolloutMigrationCursor;
+1
View File
@@ -23,6 +23,7 @@ pub use memories::Stage1StartupClaimParams;
pub use project::CreatedProject;
pub use project::Project;
pub use project::ProjectRoot;
pub use project::ProjectSortKey;
pub use project::ProjectsPage;
pub use queued_item::QueuedUserSubmissionRecord;
pub use rollout_migration_state::RolloutMigrationCursor;
+7
View File
@@ -17,6 +17,13 @@ pub struct Project {
pub position: i64,
pub created_at_ms: i64,
pub updated_at_ms: i64,
pub recency_at_ms: Option<i64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProjectSortKey {
Position,
RecencyAt,
}
#[derive(Debug, Clone, PartialEq, Eq)]
+195 -114
View File
@@ -1,13 +1,25 @@
//! Persists projects and assignments, and lists projects with computed member recency.
//! Listing pagination precedes root hydration; cursor anchors preserve milliseconds and nulls.
use std::collections::BTreeMap;
use sqlx::QueryBuilder;
use sqlx::Row;
use sqlx::Sqlite;
use uuid::Uuid;
use super::StateRuntime;
use crate::CreatedProject;
use crate::Project;
use crate::ProjectRoot;
use crate::ProjectSortKey;
use crate::ProjectsPage;
use crate::SortDirection;
const PROJECT_SELECT: &str = "SELECT projects.*,
(SELECT MAX(recency_at_ms) FROM threads
WHERE project_id = projects.id AND archived = 0) AS recency_at_ms
FROM projects";
impl StateRuntime {
pub async fn set_thread_project(
@@ -50,43 +62,27 @@ impl StateRuntime {
&self,
cursor: Option<&str>,
limit: usize,
sort_key: ProjectSortKey,
sort_direction: SortDirection,
) -> anyhow::Result<ProjectsPage> {
let anchor = cursor.map(parse_project_cursor).transpose()?;
let mut tx = self.pool.begin().await?;
let rows = if let Some((position, id)) = anchor {
sqlx::query(
"SELECT id, name, metadata, position, created_at_ms, updated_at_ms FROM projects WHERE position > ? OR (position = ? AND id > ?) ORDER BY position ASC, id ASC LIMIT ?",
)
.bind(position)
.bind(position)
.bind(id)
.bind((limit + 1) as i64)
.fetch_all(&mut *tx)
.await?
} else {
sqlx::query(
"SELECT id, name, metadata, position, created_at_ms, updated_at_ms FROM projects ORDER BY position ASC, id ASC LIMIT ?",
)
.bind((limit + 1) as i64)
.fetch_all(&mut *tx)
.await?
};
let project_ids = rows
.iter()
.map(|row| row.try_get("id"))
.collect::<Result<Vec<String>, _>>()?;
let mut roots_by_project_id = project_roots_by_id_in_tx(&mut tx, &project_ids).await?;
let mut projects = Vec::with_capacity(rows.len());
let mut query = project_list_query(cursor, limit, sort_key, sort_direction)?;
let rows = query.build().fetch_all(self.pool.as_ref()).await?;
let mut projects: Vec<Project> = Vec::new();
for row in rows {
let project_id: String = row.try_get("id")?;
let roots = roots_by_project_id.remove(&project_id).unwrap_or_default();
projects.push(project_from_row(&row, roots)?);
let id: String = row.try_get("id")?;
if projects.last().is_none_or(|project| project.id != id) {
projects.push(project_from_row(&row, Vec::new())?);
}
if let Some(path) = row.try_get::<Option<String>, _>("root_path")? {
let project = projects
.last_mut()
.ok_or_else(|| anyhow::anyhow!("project missing for root"))?;
project.roots.push(ProjectRoot { path });
}
}
let next_cursor = (projects.len() > limit)
.then(|| project_cursor(&projects[limit - 1]))
.flatten();
.then(|| project_cursor(&projects[limit - 1], sort_key, sort_direction));
projects.truncate(limit);
tx.commit().await?;
Ok(ProjectsPage {
projects,
next_cursor,
@@ -95,12 +91,12 @@ impl StateRuntime {
pub async fn get_project(&self, id: &str) -> anyhow::Result<Option<Project>> {
let mut tx = self.pool.begin().await?;
let row = sqlx::query(
"SELECT id, name, metadata, position, created_at_ms, updated_at_ms FROM projects WHERE id = ?",
)
.bind(id)
.fetch_optional(&mut *tx)
.await?;
let row = QueryBuilder::<Sqlite>::new(PROJECT_SELECT)
.push(" WHERE id = ")
.push_bind(id)
.build()
.fetch_optional(&mut *tx)
.await?;
let project = match row {
Some(row) => Some(project_from_row_in_tx(&mut tx, &row).await?),
None => None,
@@ -124,12 +120,12 @@ impl StateRuntime {
tx.commit().await?;
return Ok(None);
};
let row = sqlx::query(
"SELECT id, name, metadata, position, created_at_ms, updated_at_ms FROM projects WHERE id = ?",
)
.bind(&project_id)
.fetch_optional(&mut *tx)
.await?;
let row = QueryBuilder::<Sqlite>::new(PROJECT_SELECT)
.push(" WHERE id = ")
.push_bind(&project_id)
.build()
.fetch_optional(&mut *tx)
.await?;
let Some(row) = row else {
tx.rollback().await?;
anyhow::bail!("idempotency key refers to deleted project: {idempotency_key}");
@@ -155,12 +151,12 @@ impl StateRuntime {
.fetch_optional(&mut *tx)
.await?;
if let Some(existing_project_id) = existing_project_id {
let row = sqlx::query(
"SELECT id, name, metadata, position, created_at_ms, updated_at_ms FROM projects WHERE id = ?",
)
.bind(&existing_project_id)
.fetch_optional(&mut *tx)
.await?;
let row = QueryBuilder::<Sqlite>::new(PROJECT_SELECT)
.push(" WHERE id = ")
.push_bind(&existing_project_id)
.build()
.fetch_optional(&mut *tx)
.await?;
let Some(row) = row else {
tx.rollback().await?;
anyhow::bail!("idempotency key refers to deleted project: {idempotency_key}");
@@ -217,17 +213,16 @@ impl StateRuntime {
.bind(now)
.execute(&mut *tx)
.await?;
let row = QueryBuilder::<Sqlite>::new(PROJECT_SELECT)
.push(" WHERE id = ")
.push_bind(&id)
.build()
.fetch_one(&mut *tx)
.await?;
let project = project_from_row(&row, roots)?;
tx.commit().await?;
Ok(CreatedProject {
project: Project {
id,
name,
roots,
metadata,
position,
created_at_ms: now,
updated_at_ms: now,
},
project,
created: true,
})
}
@@ -240,12 +235,12 @@ impl StateRuntime {
metadata: Option<BTreeMap<String, String>>,
) -> anyhow::Result<Option<(Project, bool)>> {
let mut tx = self.pool.begin_with("BEGIN IMMEDIATE").await?;
let row = sqlx::query(
"SELECT id, name, metadata, position, created_at_ms, updated_at_ms FROM projects WHERE id = ?",
)
.bind(id)
.fetch_optional(&mut *tx)
.await?;
let row = QueryBuilder::<Sqlite>::new(PROJECT_SELECT)
.push(" WHERE id = ")
.push_bind(id)
.build()
.fetch_optional(&mut *tx)
.await?;
let Some(row) = row else {
tx.rollback().await?;
return Ok(None);
@@ -282,6 +277,7 @@ impl StateRuntime {
position: current.position,
created_at_ms: current.created_at_ms,
updated_at_ms: now,
recency_at_ms: current.recency_at_ms,
},
true,
)))
@@ -410,37 +406,10 @@ fn project_from_row(
position: row.try_get("position")?,
created_at_ms: row.try_get("created_at_ms")?,
updated_at_ms: row.try_get("updated_at_ms")?,
recency_at_ms: row.try_get("recency_at_ms")?,
})
}
async fn project_roots_by_id_in_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
project_ids: &[String],
) -> anyhow::Result<BTreeMap<String, Vec<ProjectRoot>>> {
if project_ids.is_empty() {
return Ok(BTreeMap::new());
}
let mut builder = sqlx::QueryBuilder::<sqlx::Sqlite>::new(
"SELECT project_id, path FROM project_roots WHERE project_id IN (",
);
let mut separated = builder.separated(", ");
for project_id in project_ids {
separated.push_bind(project_id);
}
builder.push(") ORDER BY project_id ASC, position ASC");
let rows = builder.build().fetch_all(&mut **tx).await?;
let mut roots_by_project_id = BTreeMap::<String, Vec<ProjectRoot>>::new();
for row in rows {
roots_by_project_id
.entry(row.try_get("project_id")?)
.or_default()
.push(ProjectRoot {
path: row.try_get("path")?,
});
}
Ok(roots_by_project_id)
}
async fn replace_roots(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
project_id: &str,
@@ -461,30 +430,142 @@ async fn replace_roots(
Ok(())
}
fn project_cursor(project: &Project) -> Option<String> {
Some(format!("{}|{}", project.position, project.id))
}
fn parse_project_cursor(cursor: &str) -> anyhow::Result<(i64, String)> {
let mut parts = cursor.split('|');
let position_component = parts.next().ok_or_else(|| invalid_project_cursor(cursor))?;
let position: i64 = position_component
.parse()
.map_err(|_| invalid_project_cursor(cursor))?;
let id = parts.next().ok_or_else(|| invalid_project_cursor(cursor))?;
let uuid = Uuid::parse_str(id).map_err(|_| invalid_project_cursor(cursor))?;
if parts.next().is_some()
|| position < 0
|| position.to_string() != position_component
|| uuid.to_string() != id
{
return Err(invalid_project_cursor(cursor));
fn project_list_query(
cursor: Option<&str>,
limit: usize,
sort_key: ProjectSortKey,
sort_direction: SortDirection,
) -> anyhow::Result<QueryBuilder<Sqlite>> {
anyhow::ensure!(limit > 0, "project limit must be positive");
let query_limit = i64::try_from(limit)?
.checked_add(/*rhs*/ 1)
.ok_or_else(|| anyhow::anyhow!("project limit overflow"))?;
let column = match sort_key {
ProjectSortKey::Position => "p.position",
ProjectSortKey::RecencyAt => "p.recency_at_ms",
};
let operator = match sort_direction {
SortDirection::Asc => ">",
SortDirection::Desc => "<",
};
let mut query = QueryBuilder::new(format!(
"WITH project_activity AS ({PROJECT_SELECT}), page AS (SELECT * FROM project_activity p"
));
if let Some(cursor) = cursor {
let (value, id) = parse_project_cursor(cursor, sort_key, sort_direction)?;
if let Some(value) = value {
query
.push(format!(" WHERE ({column} {operator} "))
.push_bind(value);
query.push(format!(" OR ({column} = ")).push_bind(value);
query
.push(format!(" AND p.id {operator} "))
.push_bind(id)
.push(")");
if sort_key == ProjectSortKey::RecencyAt {
query.push(" OR p.recency_at_ms IS NULL");
}
query.push(")");
} else {
query
.push(format!(
" WHERE p.recency_at_ms IS NULL AND p.id {operator} "
))
.push_bind(id);
}
}
Ok((position, id.to_string()))
push_project_order(&mut query, sort_key, sort_direction);
query.push(" LIMIT ").push_bind(query_limit);
query.push(
") SELECT p.*, roots.path AS root_path FROM page p \
LEFT JOIN project_roots roots ON roots.project_id = p.id",
);
push_project_order(&mut query, sort_key, sort_direction);
query.push(", roots.position ASC");
Ok(query)
}
fn invalid_project_cursor(cursor: &str) -> anyhow::Error {
anyhow::anyhow!("invalid project cursor: {cursor}")
fn push_project_order(
query: &mut QueryBuilder<Sqlite>,
sort_key: ProjectSortKey,
sort_direction: SortDirection,
) {
let direction = match sort_direction {
SortDirection::Asc => "ASC",
SortDirection::Desc => "DESC",
};
query.push(" ORDER BY ");
match sort_key {
ProjectSortKey::Position => query.push("p.position"),
ProjectSortKey::RecencyAt => query.push("p.recency_at_ms IS NULL ASC, p.recency_at_ms"),
};
query.push(format!(" {direction}, p.id {direction}"));
}
fn project_cursor(project: &Project, sort_key: ProjectSortKey, direction: SortDirection) -> String {
if sort_key == ProjectSortKey::Position && direction == SortDirection::Asc {
// Retain the existing format for clients reconnecting to an older server.
return format!("{}|{}", project.position, project.id);
}
let (key, value) = match sort_key {
ProjectSortKey::Position => ("position", project.position.to_string()),
ProjectSortKey::RecencyAt => (
"recencyAt",
project
.recency_at_ms
.map_or_else(|| "null".to_string(), |value| value.to_string()),
),
};
let direction = match direction {
SortDirection::Asc => "asc",
SortDirection::Desc => "desc",
};
format!("v1|{key}|{direction}|{value}|{}", project.id)
}
fn parse_project_cursor(
cursor: &str,
sort_key: ProjectSortKey,
direction: SortDirection,
) -> anyhow::Result<(Option<i64>, String)> {
let invalid = || anyhow::anyhow!("invalid project cursor: malformed or mismatched sort anchor");
if cursor.len() > 128 {
return Err(invalid());
}
let parts: Vec<_> = cursor.split('|').collect();
let key = match sort_key {
ProjectSortKey::Position => "position",
ProjectSortKey::RecencyAt => "recencyAt",
};
let order = match direction {
SortDirection::Asc => "asc",
SortDirection::Desc => "desc",
};
let (value, id) = match parts.as_slice() {
[value, id] if sort_key == ProjectSortKey::Position && direction == SortDirection::Asc => {
(*value, *id)
}
["v1", cursor_key, cursor_order, value, id]
if *cursor_key == key && *cursor_order == order =>
{
(*value, *id)
}
_ => return Err(invalid()),
};
let value = if value == "null" && sort_key == ProjectSortKey::RecencyAt {
None
} else {
let parsed: i64 = value.parse().map_err(|_| invalid())?;
if parsed.to_string() != value || sort_key == ProjectSortKey::Position && parsed < 0 {
return Err(invalid());
}
Some(parsed)
};
let uuid = Uuid::parse_str(id).map_err(|_| invalid())?;
if uuid.to_string() != id {
return Err(invalid());
}
Ok((value, id.to_string()))
}
#[cfg(test)]
+287 -5
View File
@@ -6,6 +6,8 @@ use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq;
use super::*;
use crate::ProjectSortKey;
use crate::SortDirection;
use crate::ThreadMetadataBuilder;
use crate::runtime::test_support::unique_temp_dir;
@@ -63,6 +65,13 @@ async fn project_lifecycle_preserves_order_and_clears_assignments() -> anyhow::R
.unwrap();
assert!(!unchanged.1);
assert_eq!(unchanged.0.updated_at_ms, project.updated_at_ms);
runtime
.touch_thread_updated_at(thread_id, chrono::Utc::now())
.await?;
assert_eq!(
runtime.get_project(&project.id).await?,
Some(project.clone())
);
runtime
.mark_archived(
@@ -71,6 +80,23 @@ async fn project_lifecycle_preserves_order_and_clears_assignments() -> anyhow::R
chrono::Utc::now(),
)
.await?;
assert_eq!(
runtime.get_project(&project.id).await?,
Some(Project {
recency_at_ms: None,
..project.clone()
})
);
runtime
.mark_unarchived(thread_id, &metadata.rollout_path)
.await?;
assert_eq!(
runtime.get_project(&project.id).await?,
Some(project.clone())
);
runtime
.mark_archived(thread_id, &metadata.rollout_path, chrono::Utc::now())
.await?;
let active_thread_id = ThreadId::default();
let active_metadata = ThreadMetadataBuilder::new(
active_thread_id,
@@ -84,6 +110,46 @@ async fn project_lifecycle_preserves_order_and_clears_assignments() -> anyhow::R
.set_thread_project(&active_thread_id.to_string(), Some(&project.id))
.await?
.expect("active thread exists");
let active_recency = runtime
.get_thread(active_thread_id)
.await?
.unwrap()
.recency_at
.timestamp_millis();
runtime
.mark_unarchived(thread_id, &metadata.rollout_path)
.await?;
let active_project = Project {
recency_at_ms: Some(active_recency),
..project.clone()
};
assert_eq!(
runtime.get_project(&project.id).await?,
Some(active_project)
);
runtime
.set_thread_project(&active_thread_id.to_string(), /*project_id*/ None)
.await?;
assert_eq!(
runtime.get_project(&project.id).await?,
Some(project.clone())
);
runtime
.set_thread_project(&active_thread_id.to_string(), Some(&project.id))
.await?;
runtime.delete_thread(active_thread_id).await?;
assert_eq!(
runtime.get_project(&project.id).await?,
Some(project.clone())
);
// Restore the active fixture for the project-deletion assertions below.
runtime.upsert_thread(&active_metadata).await?;
runtime
.set_thread_project(&active_thread_id.to_string(), Some(&project.id))
.await?;
runtime
.mark_archived(thread_id, &metadata.rollout_path, chrono::Utc::now())
.await?;
let (affected_active_thread_ids, affected_archived_thread_ids) =
runtime.delete_project(&project.id).await?.unwrap();
assert_eq!(
@@ -192,7 +258,12 @@ async fn project_import_rejects_unknown_thread_without_partial_project() -> anyh
assert!(error.to_string().contains("thread not found"));
assert!(
runtime
.list_projects(/*cursor*/ None, /*limit*/ 10)
.list_projects(
/*cursor*/ None,
/*limit*/ 10,
ProjectSortKey::Position,
SortDirection::Asc
)
.await?
.projects
.is_empty()
@@ -276,7 +347,12 @@ async fn project_list_rejects_malformed_cursors() -> anyhow::Result<()> {
"not-a-position|00000000-0000-0000-0000-000000000000",
] {
let error = runtime
.list_projects(Some(cursor), /*limit*/ 10)
.list_projects(
Some(cursor),
/*limit*/ 10,
ProjectSortKey::Position,
SortDirection::Asc,
)
.await
.expect_err("malformed cursor should fail");
assert!(error.to_string().starts_with("invalid project cursor:"));
@@ -298,16 +374,215 @@ async fn project_list_cursor_round_trips_across_pages() -> anyhow::Result<()> {
.await?;
}
let first = runtime.list_projects(/*cursor*/ None, /*limit*/ 1).await?;
let first = runtime
.list_projects(
/*cursor*/ None,
/*limit*/ 1,
ProjectSortKey::Position,
SortDirection::Asc,
)
.await?;
assert_eq!(first.projects.len(), 1);
let cursor = first.next_cursor.expect("next cursor");
let second = runtime.list_projects(Some(&cursor), /*limit*/ 1).await?;
let second = runtime
.list_projects(
Some(&cursor),
/*limit*/ 1,
ProjectSortKey::Position,
SortDirection::Asc,
)
.await?;
assert_eq!(second.projects.len(), 1);
assert_ne!(first.projects[0].id, second.projects[0].id);
assert_eq!(second.next_cursor, None);
Ok(())
}
#[tokio::test]
async fn project_list_orders_and_pages_recency_with_roots_and_nulls() -> anyhow::Result<()> {
let home = unique_temp_dir();
let runtime = StateRuntime::init(
crate::SqliteConfig::new_for_testing(home.as_path().abs()),
"test-provider".to_string(),
)
.await?;
let mut projects = Vec::new();
for (name, recency_at_ms, archived) in [
("older", Some(1_700_000_000_001), false),
("tie-a", Some(1_700_000_000_002), false),
("tie-b", Some(1_700_000_000_002), false),
("empty", None, false),
("archived", Some(1_700_000_010_000), true),
] {
let mut threads = Vec::new();
if let Some(timestamp) = recency_at_ms {
for timestamp in [timestamp - 1, timestamp] {
let id = ThreadId::default();
let mut metadata = ThreadMetadataBuilder::new(
id,
home.join(format!("{id}.jsonl")),
chrono::DateTime::from_timestamp_millis(timestamp).unwrap(),
SessionSource::Exec,
)
.build("test-provider");
metadata.archived_at = archived.then(chrono::Utc::now);
runtime.upsert_thread(&metadata).await?;
// Force exact ties despite the monotonic timestamp allocator.
sqlx::query("UPDATE threads SET recency_at_ms = ?, recency_at = ? WHERE id = ?")
.bind(timestamp)
.bind(timestamp / 1000)
.bind(id.to_string())
.execute(runtime.pool.as_ref())
.await?;
threads.push(id.to_string());
}
}
let project = runtime
.create_project(
name.to_string(),
vec![
ProjectRoot {
path: "/tmp/z".to_string(),
},
ProjectRoot {
path: "/tmp/a".to_string(),
},
],
BTreeMap::new(),
&threads,
name,
)
.await?
.project;
let expected = Project {
recency_at_ms: recency_at_ms.filter(|_| !archived),
..project.clone()
};
assert_eq!(project, expected);
assert_eq!(
runtime.get_project(&project.id).await?,
Some(expected.clone())
);
projects.push(expected);
}
let mut ties = projects[1..3].to_vec();
ties.sort_by(|a, b| a.id.cmp(&b.id));
let mut empty = projects[3..].to_vec();
empty.sort_by(|a, b| a.id.cmp(&b.id));
let ascending = [vec![projects[0].clone()], ties.clone(), empty.clone()].concat();
ties.reverse();
empty.reverse();
let descending = [ties, vec![projects[0].clone()], empty].concat();
for (key, direction, expected) in [
(
ProjectSortKey::Position,
SortDirection::Asc,
projects.clone(),
),
(
ProjectSortKey::Position,
SortDirection::Desc,
projects.into_iter().rev().collect(),
),
(ProjectSortKey::RecencyAt, SortDirection::Asc, ascending),
(
ProjectSortKey::RecencyAt,
SortDirection::Desc,
descending.clone(),
),
] {
for limit in [1, 2, 3, 10] {
let mut cursor = None;
let mut actual = Vec::new();
loop {
let page = runtime
.list_projects(cursor.as_deref(), limit, key, direction)
.await?;
actual.extend(page.projects);
assert!(
actual.len() <= expected.len(),
"pagination repeated a project"
);
cursor = page.next_cursor;
if cursor.is_none() {
break;
}
}
assert_eq!(actual, expected);
}
}
let first = runtime
.list_projects(
/*cursor*/ None,
/*limit*/ 1,
ProjectSortKey::RecencyAt,
SortDirection::Desc,
)
.await?;
runtime.delete_project(&first.projects[0].id).await?;
let rest = runtime
.list_projects(
first.next_cursor.as_deref(),
/*limit*/ 10,
ProjectSortKey::RecencyAt,
SortDirection::Desc,
)
.await?;
assert_eq!(
rest,
ProjectsPage {
projects: descending[1..].to_vec(),
next_cursor: None
}
);
let query = project_list_query(
/*cursor*/ None,
/*limit*/ 50,
ProjectSortKey::RecencyAt,
SortDirection::Desc,
)?;
let plan = QueryBuilder::<Sqlite>::new("EXPLAIN QUERY PLAN ")
.push(query.sql().as_str())
.build()
.bind(/*value*/ 51_i64)
.fetch_all(runtime.pool.as_ref())
.await?;
let details = plan
.iter()
.map(|row| row.get::<String, _>("detail"))
.collect::<Vec<_>>();
assert!(
details
.iter()
.any(|detail| detail.contains("USING COVERING INDEX idx_threads_project_recency")),
"{details:?}"
);
Ok(())
}
#[test]
fn project_list_rejects_cursor_sort_mismatches() {
for cursor in [
"1|00000000-0000-0000-0000-000000000000",
"v1|position|desc|1|00000000-0000-0000-0000-000000000000",
"v1|recencyAt|asc|1|00000000-0000-0000-0000-000000000000",
"v2|recencyAt|desc|1|00000000-0000-0000-0000-000000000000",
"v1|recencyAt|desc|01|00000000-0000-0000-0000-000000000000",
"v1|recencyAt|desc|null|not-a-uuid",
] {
assert!(
project_list_query(
Some(cursor),
/*limit*/ 10,
ProjectSortKey::RecencyAt,
SortDirection::Desc
)
.is_err()
);
}
}
#[tokio::test]
async fn project_move_reorders_projects_and_preserves_no_op_timestamp() -> anyhow::Result<()> {
let home = unique_temp_dir();
@@ -351,7 +626,14 @@ async fn project_move_reorders_projects_and_preserves_no_op_timestamp() -> anyho
runtime.move_project(&three.id, Some(&one.id)).await?,
Some(true)
);
let reordered = runtime.list_projects(/*cursor*/ None, /*limit*/ 10).await?;
let reordered = runtime
.list_projects(
/*cursor*/ None,
/*limit*/ 10,
ProjectSortKey::Position,
SortDirection::Asc,
)
.await?;
assert_eq!(
reordered
.projects
+1
View File
@@ -16,6 +16,7 @@ mod thread_sections;
mod types;
pub use codex_state::MAX_QUEUE_ITEMS;
pub use codex_state::ProjectSortKey;
pub use codex_state::QueuedUserSubmissionRecord;
pub use error::ThreadStoreError;
pub use error::ThreadStoreResult;
+11 -1
View File
@@ -5,6 +5,7 @@ use crate::DeletedProject;
use crate::ListProjectsParams;
use crate::MoveProjectParams;
use crate::ProjectMoveOutcome;
use crate::SortDirection;
use crate::StoredProject;
use crate::StoredProjectRoot;
use crate::StoredProjectsPage;
@@ -20,7 +21,15 @@ pub(super) async fn list_projects(
let state = state(store)?;
let has_cursor = params.cursor.is_some();
let page = state
.list_projects(params.cursor.as_deref(), params.limit)
.list_projects(
params.cursor.as_deref(),
params.limit,
params.sort_key,
match params.sort_direction {
SortDirection::Asc => codex_state::SortDirection::Asc,
SortDirection::Desc => codex_state::SortDirection::Desc,
},
)
.await
.map_err(|error| {
if has_cursor && error.to_string().starts_with("invalid project cursor:") {
@@ -178,5 +187,6 @@ fn stored_project(project: codex_state::Project) -> StoredProject {
position: project.position,
created_at_ms: project.created_at_ms,
updated_at_ms: project.updated_at_ms,
recency_at_ms: project.recency_at_ms,
}
}
+5
View File
@@ -1,5 +1,7 @@
use std::collections::BTreeMap;
use crate::SortDirection;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredProjectRoot {
pub path: String,
@@ -14,6 +16,7 @@ pub struct StoredProject {
pub position: i64,
pub created_at_ms: i64,
pub updated_at_ms: i64,
pub recency_at_ms: Option<i64>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -26,6 +29,8 @@ pub struct StoredProjectsPage {
pub struct ListProjectsParams {
pub cursor: Option<String>,
pub limit: usize,
pub sort_key: codex_state::ProjectSortKey,
pub sort_direction: SortDirection,
}
#[derive(Clone, Debug, PartialEq, Eq)]