openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 5de0d38e61 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-04-17 12:10:28 -07:00
committed by GitHub
parent 71dc1bfa20 5de0d38e61
commit 5f2bdf8cd1
65 changed files with 3709 additions and 379 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/CODEOWNERS
+5
View File
@@ -0,0 +1,5 @@
# Core crate ownership.
/codex-rs/core/ @openai/codex-core-agent-team
# Keep ownership changes reviewed by the same team.
/.github/CODEOWNERS @openai/codex-core-agent-team
codex-rs/app-server-protocol/schema/json/ClientRequest.json
+80
View File
@@ -2645,6 +2645,13 @@
},
"type": "object"
},
"SortDirection": {
"enum": [
"asc",
"desc"
],
"type": "string"
},
"TextElement": {
"properties": {
"byteRange": {
@@ -2857,6 +2864,17 @@
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional sort direction; defaults to descending (newest first)."
},
"sortKey": {
"anyOf": [
{
@@ -3361,6 +3379,44 @@
],
"type": "string"
},
"ThreadTurnsListParams": {
"properties": {
"cursor": {
"description": "Opaque cursor to pass to the next call to continue after the last turn.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
},
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"type": "object"
},
"ThreadUnarchiveParams": {
"properties": {
"threadId": {
@@ -4052,6 +4108,30 @@
"title": "Thread/readRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/turns/list"
],
"title": "Thread/turns/listRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadTurnsListParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/turns/listRequest",
"type": "object"
},
{
"description": "Append raw Responses API items to the thread history without starting a user turn.",
"properties": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
+119
View File
@@ -578,6 +578,30 @@
"title": "Thread/readRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/v2/RequestId"
},
"method": {
"enum": [
"thread/turns/list"
],
"title": "Thread/turns/listRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/ThreadTurnsListParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/turns/listRequest",
"type": "object"
},
{
"description": "Append raw Responses API items to the thread history without starting a user turn.",
"properties": {
@@ -12559,6 +12583,13 @@
"title": "SkillsListResponse",
"type": "object"
},
"SortDirection": {
"enum": [
"asc",
"desc"
],
"type": "string"
},
"SubAgentSource": {
"oneOf": [
{
@@ -13820,6 +13851,17 @@
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/v2/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional sort direction; defaults to descending (newest first)."
},
"sortKey": {
"anyOf": [
{
@@ -13848,6 +13890,13 @@
"ThreadListResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"backwardsCursor": {
"description": "Opaque cursor to pass as `cursor` when reversing `sortDirection`. This is only populated when the page contains at least one thread. Use it with the opposite `sortDirection`; for timestamp sorts it anchors at the start of the page timestamp so same-second updates are not skipped.",
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/v2/Thread"
@@ -14909,6 +14958,76 @@
"title": "ThreadTokenUsageUpdatedNotification",
"type": "object"
},
"ThreadTurnsListParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"cursor": {
"description": "Opaque cursor to pass to the next call to continue after the last turn.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/v2/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
},
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"title": "ThreadTurnsListParams",
"type": "object"
},
"ThreadTurnsListResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"backwardsCursor": {
"description": "Opaque cursor to pass as `cursor` when reversing `sortDirection`. This is only populated when the page contains at least one turn. Use it with the opposite `sortDirection` to include the anchor turn again and catch updates to that turn.",
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/v2/Turn"
},
"type": "array"
},
"nextCursor": {
"description": "Opaque cursor to pass to the next call to continue after the last turn. if None, there are no more turns to return.",
"type": [
"string",
"null"
]
}
},
"required": [
"data"
],
"title": "ThreadTurnsListResponse",
"type": "object"
},
"ThreadUnarchiveParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
+119
View File
@@ -1160,6 +1160,30 @@
"title": "Thread/readRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/turns/list"
],
"title": "Thread/turns/listRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadTurnsListParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/turns/listRequest",
"type": "object"
},
{
"description": "Append raw Responses API items to the thread history without starting a user turn.",
"properties": {
@@ -10403,6 +10427,13 @@
"title": "SkillsListResponse",
"type": "object"
},
"SortDirection": {
"enum": [
"asc",
"desc"
],
"type": "string"
},
"SubAgentSource": {
"oneOf": [
{
@@ -11664,6 +11695,17 @@
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional sort direction; defaults to descending (newest first)."
},
"sortKey": {
"anyOf": [
{
@@ -11692,6 +11734,13 @@
"ThreadListResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"backwardsCursor": {
"description": "Opaque cursor to pass as `cursor` when reversing `sortDirection`. This is only populated when the page contains at least one thread. Use it with the opposite `sortDirection`; for timestamp sorts it anchors at the start of the page timestamp so same-second updates are not skipped.",
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/Thread"
@@ -12753,6 +12802,76 @@
"title": "ThreadTokenUsageUpdatedNotification",
"type": "object"
},
"ThreadTurnsListParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"cursor": {
"description": "Opaque cursor to pass to the next call to continue after the last turn.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
},
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"title": "ThreadTurnsListParams",
"type": "object"
},
"ThreadTurnsListResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"backwardsCursor": {
"description": "Opaque cursor to pass as `cursor` when reversing `sortDirection`. This is only populated when the page contains at least one turn. Use it with the opposite `sortDirection` to include the anchor turn again and catch updates to that turn.",
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/Turn"
},
"type": "array"
},
"nextCursor": {
"description": "Opaque cursor to pass to the next call to continue after the last turn. if None, there are no more turns to return.",
"type": [
"string",
"null"
]
}
},
"required": [
"data"
],
"title": "ThreadTurnsListResponse",
"type": "object"
},
"ThreadUnarchiveParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
codex-rs/app-server-protocol/schema/json/v2/ThreadListParams.json
+18
View File
@@ -1,6 +1,13 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"SortDirection": {
"enum": [
"asc",
"desc"
],
"type": "string"
},
"ThreadSortKey": {
"enum": [
"created_at",
@@ -72,6 +79,17 @@
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional sort direction; defaults to descending (newest first)."
},
"sortKey": {
"anyOf": [
{
codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json
+7
View File
@@ -1953,6 +1953,13 @@
}
},
"properties": {
"backwardsCursor": {
"description": "Opaque cursor to pass as `cursor` when reversing `sortDirection`. This is only populated when the page contains at least one thread. Use it with the opposite `sortDirection`; for timestamp sorts it anchors at the start of the page timestamp so same-second updates are not skipped.",
"type": [
"string",
"null"
]
},
"data": {
"items": {
"$ref": "#/definitions/Thread"
codex-rs/app-server-protocol/schema/json/v2/ThreadTurnsListParams.json
+49
View File
@@ -0,0 +1,49 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"SortDirection": {
"enum": [
"asc",
"desc"
],
"type": "string"
}
},
"properties": {
"cursor": {
"description": "Opaque cursor to pass to the next call to continue after the last turn.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional turn page size.",
"format": "uint32",
"minimum": 0.0,
"type": [
"integer",
"null"
]
},
"sortDirection": {
"anyOf": [
{
"$ref": "#/definitions/SortDirection"
},
{
"type": "null"
}
],
"description": "Optional turn pagination direction; defaults to descending."
},
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"title": "ThreadTurnsListParams",
"type": "object"
}
codex-rs/app-server-protocol/schema/json/v2/ThreadTurnsListResponse.json
+1631
View File
File diff suppressed because it is too large Load Diff
codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/SortDirection.ts
+5
View File
@@ -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 SortDirection = "asc" | "desc";
codex-rs/app-server-protocol/schema/typescript/v2/ThreadListParams.ts
+5
View File
@@ -1,6 +1,7 @@
// 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.
import type { SortDirection } from "./SortDirection";
import type { ThreadSortKey } from "./ThreadSortKey";
import type { ThreadSourceKind } from "./ThreadSourceKind";
@@ -17,6 +18,10 @@ limit?: number | null,
* Optional sort key; defaults to created_at.
*/
sortKey?: ThreadSortKey | null,
/**
* Optional sort direction; defaults to descending (newest first).
*/
sortDirection?: SortDirection | null,
/**
* Optional provider filter; when set, only sessions recorded under these
* providers are returned. When present but empty, includes all providers.
codex-rs/app-server-protocol/schema/typescript/v2/ThreadListResponse.ts
+8 -1
View File
@@ -8,4 +8,11 @@ export type ThreadListResponse = { data: Array<Thread>,
* Opaque cursor to pass to the next call to continue after the last item.
* if None, there are no more items to return.
*/
nextCursor: string | null, };
nextCursor: string | null,
/**
* Opaque cursor to pass as `cursor` when reversing `sortDirection`.
* This is only populated when the page contains at least one thread.
* Use it with the opposite `sortDirection`; for timestamp sorts it anchors
* at the start of the page timestamp so same-second updates are not skipped.
*/
backwardsCursor: string | null, };
codex-rs/app-server-protocol/schema/typescript/v2/ThreadTurnsListParams.ts
+18
View File
@@ -0,0 +1,18 @@
// 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.
import type { SortDirection } from "./SortDirection";
export type ThreadTurnsListParams = { threadId: string,
/**
* Opaque cursor to pass to the next call to continue after the last turn.
*/
cursor?: string | null,
/**
* Optional turn page size.
*/
limit?: number | null,
/**
* Optional turn pagination direction; defaults to descending.
*/
sortDirection?: SortDirection | null, };
codex-rs/app-server-protocol/schema/typescript/v2/ThreadTurnsListResponse.ts
+18
View File
@@ -0,0 +1,18 @@
// 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.
import type { Turn } from "./Turn";
export type ThreadTurnsListResponse = { data: Array<Turn>,
/**
* Opaque cursor to pass to the next call to continue after the last turn.
* if None, there are no more turns to return.
*/
nextCursor: string | null,
/**
* Opaque cursor to pass as `cursor` when reversing `sortDirection`.
* This is only populated when the page contains at least one turn.
* Use it with the opposite `sortDirection` to include the anchor turn again
* and catch updates to that turn.
*/
backwardsCursor: string | null, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
+3
View File
@@ -274,6 +274,7 @@ export type { SkillsListEntry } from "./SkillsListEntry";
export type { SkillsListExtraRootsForCwd } from "./SkillsListExtraRootsForCwd";
export type { SkillsListParams } from "./SkillsListParams";
export type { SkillsListResponse } from "./SkillsListResponse";
export type { SortDirection } from "./SortDirection";
export type { TerminalInteractionNotification } from "./TerminalInteractionNotification";
export type { TextElement } from "./TextElement";
export type { TextPosition } from "./TextPosition";
@@ -329,6 +330,8 @@ export type { ThreadStatus } from "./ThreadStatus";
export type { ThreadStatusChangedNotification } from "./ThreadStatusChangedNotification";
export type { ThreadTokenUsage } from "./ThreadTokenUsage";
export type { ThreadTokenUsageUpdatedNotification } from "./ThreadTokenUsageUpdatedNotification";
export type { ThreadTurnsListParams } from "./ThreadTurnsListParams";
export type { ThreadTurnsListResponse } from "./ThreadTurnsListResponse";
export type { ThreadUnarchiveParams } from "./ThreadUnarchiveParams";
export type { ThreadUnarchiveResponse } from "./ThreadUnarchiveResponse";
export type { ThreadUnarchivedNotification } from "./ThreadUnarchivedNotification";
codex-rs/app-server-protocol/src/protocol/common.rs
+4
View File
@@ -327,6 +327,10 @@ client_request_definitions! {
params: v2::ThreadReadParams,
response: v2::ThreadReadResponse,
},
ThreadTurnsList => "thread/turns/list" {
params: v2::ThreadTurnsListParams,
response: v2::ThreadTurnsListResponse,
},
/// Append raw Responses API items to the thread history without starting a user turn.
ThreadInjectItems => "thread/inject_items" {
params: v2::ThreadInjectItemsParams,
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+10 -3
View File
@@ -991,8 +991,15 @@ impl ThreadHistoryBuilder {
}
fn new_turn(&mut self, id: Option<String>) -> PendingTurn {
let id = id.unwrap_or_else(|| {
if self.next_rollout_index == 0 {
Uuid::now_v7().to_string()
} else {
format!("rollout-{}", self.current_rollout_index)
}
});
PendingTurn {
id: id.unwrap_or_else(|| Uuid::now_v7().to_string()),
id,
items: Vec::new(),
error: None,
status: TurnStatus::Completed,
@@ -1640,8 +1647,8 @@ mod tests {
.collect::<Vec<_>>();
let turns = build_turns_from_rollout_items(&items);
assert_eq!(turns.len(), 2);
assert!(Uuid::parse_str(&turns[0].id).is_ok());
assert!(Uuid::parse_str(&turns[1].id).is_ok());
assert_eq!(turns[0].id, "rollout-0");
assert_eq!(turns[1].id, "rollout-5");
assert_ne!(turns[0].id, turns[1].id);
assert_eq!(turns[0].status, TurnStatus::Completed);
assert_eq!(turns[1].status, TurnStatus::Completed);
codex-rs/app-server-protocol/src/protocol/v2.rs
+47
View File
@@ -3230,6 +3230,9 @@ pub struct ThreadListParams {
/// Optional sort key; defaults to created_at.
#[ts(optional = nullable)]
pub sort_key: Option<ThreadSortKey>,
/// Optional sort direction; defaults to descending (newest first).
#[ts(optional = nullable)]
pub sort_direction: Option<SortDirection>,
/// Optional provider filter; when set, only sessions recorded under these
/// providers are returned. When present but empty, includes all providers.
#[ts(optional = nullable)]
@@ -3277,6 +3280,14 @@ pub enum ThreadSortKey {
UpdatedAt,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
#[ts(export_to = "v2/")]
pub enum SortDirection {
Asc,
Desc,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
@@ -3285,6 +3296,11 @@ pub struct ThreadListResponse {
/// Opaque cursor to pass to the next call to continue after the last item.
/// if None, there are no more items to return.
pub next_cursor: Option<String>,
/// Opaque cursor to pass as `cursor` when reversing `sortDirection`.
/// This is only populated when the page contains at least one thread.
/// Use it with the opposite `sortDirection`; for timestamp sorts it anchors
/// at the start of the page timestamp so same-second updates are not skipped.
pub backwards_cursor: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)]
@@ -3350,6 +3366,37 @@ pub struct ThreadReadResponse {
pub thread: Thread,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadTurnsListParams {
pub thread_id: String,
/// Opaque cursor to pass to the next call to continue after the last turn.
#[ts(optional = nullable)]
pub cursor: Option<String>,
/// Optional turn page size.
#[ts(optional = nullable)]
pub limit: Option<u32>,
/// Optional turn pagination direction; defaults to descending.
#[ts(optional = nullable)]
pub sort_direction: Option<SortDirection>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadTurnsListResponse {
pub data: Vec<Turn>,
/// Opaque cursor to pass to the next call to continue after the last turn.
/// if None, there are no more turns to return.
pub next_cursor: Option<String>,
/// Opaque cursor to pass as `cursor` when reversing `sortDirection`.
/// This is only populated when the page contains at least one turn.
/// Use it with the opposite `sortDirection` to include the anchor turn again
/// and catch updates to that turn.
pub backwards_cursor: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
codex-rs/app-server-test-client/src/lib.rs
+1
View File
@@ -1124,6 +1124,7 @@ async fn thread_list(endpoint: &Endpoint, config_overrides: &[String], limit: u3
cursor: None,
limit: Some(limit),
sort_key: None,
sort_direction: None,
model_providers: None,
source_kinds: None,
archived: None,
codex-rs/app-server/README.md
+23 -2
View File
@@ -142,6 +142,7 @@ Example with notification opt-out:
- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, `cwd`, and `searchTerm` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded.
- `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/turns/list` — page through a stored thread’s turn history without resuming it; supports cursor-based pagination with `sortDirection`, `nextCursor`, and `backwardsCursor`.
- `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 thread’s persisted memory eligibility to `"enabled"` or `"disabled"` for either a loaded thread or a stored rollout; returns `{}` on success.
- `memory/reset` — experimental; clear the current `CODEX_HOME/memories` directory and reset persisted memory stage data in sqlite while preserving existing thread memory modes; returns `{}` on success.
@@ -281,11 +282,13 @@ Experimental API: `thread/start`, `thread/resume`, and `thread/fork` accept `per
- `cursor` — opaque string from a prior response; omit for the first page.
- `limit` — server defaults to a reasonable page size if unset.
- `sortKey` — `created_at` (default) or `updated_at`.
- `sortDirection` — `desc` (default) or `asc`.
- `modelProviders` — restrict results to specific providers; unset, null, or an empty array will include all providers.
- `sourceKinds` — restrict results to specific sources; omit or pass `[]` for interactive sessions only (`cli`, `vscode`).
- `archived` — when `true`, list archived threads only. When `false` or `null`, list non-archived threads (default).
- `cwd` — restrict results to threads whose session cwd exactly matches this path. Relative paths are resolved against the app-server process cwd before matching.
- `searchTerm` — restrict results to threads whose extracted title contains this substring (case-sensitive).
- Responses include `nextCursor` to continue in the same direction and `backwardsCursor` to pass as `cursor` when reversing `sortDirection`.
- Responses include `agentNickname` and `agentRole` for AgentControl-spawned thread sub-agents when available.
Example:
@@ -301,7 +304,8 @@ Example:
{ "id": "thr_a", "preview": "Create a TUI", "modelProvider": "openai", "createdAt": 1730831111, "updatedAt": 1730831111, "status": { "type": "notLoaded" }, "agentNickname": "Atlas", "agentRole": "explorer" },
{ "id": "thr_b", "preview": "Fix tests", "modelProvider": "openai", "createdAt": 1730750000, "updatedAt": 1730750000, "status": { "type": "notLoaded" } }
],
"nextCursor": "opaque-token-or-null"
"nextCursor": "opaque-token-or-null",
"backwardsCursor": "opaque-token-or-null"
} }
```
@@ -363,7 +367,7 @@ Later, after the idle unload timeout:
### Example: Read a thread
Use `thread/read` to fetch a stored thread by id without resuming it. Pass `includeTurns` when you want the rollout history loaded into `thread.turns`. The returned thread includes `agentNickname` and `agentRole` for AgentControl-spawned thread sub-agents when available.
Use `thread/read` to fetch a stored thread by id without resuming it. Pass `includeTurns` when you want the full rollout history loaded into `thread.turns`. The returned thread includes `agentNickname` and `agentRole` for AgentControl-spawned thread sub-agents when available.
```json
{ "method": "thread/read", "id": 22, "params": { "threadId": "thr_123" } }
@@ -379,6 +383,23 @@ Use `thread/read` to fetch a stored thread by id without resuming it. Pass `incl
} }
```
### Example: List thread turns
Use `thread/turns/list` to page a stored thread’s turn history without resuming it. By default, results are sorted descending so clients can start at the present and fetch older turns with `nextCursor`. The response also includes `backwardsCursor`; pass it as `cursor` on a later request with `sortDirection: "asc"` to fetch turns newer than the first item from the earlier page.
```json
{ "method": "thread/turns/list", "id": 24, "params": {
"threadId": "thr_123",
"limit": 50,
"sortDirection": "desc"
} }
{ "id": 24, "result": {
"data": [ ... ],
"nextCursor": "older-turns-cursor-or-null",
"backwardsCursor": "newer-turns-cursor-or-null"
} }
```
### Example: Update stored thread metadata
Use `thread/metadata/update` to patch sqlite-backed metadata for a thread without resuming it. Today this supports persisted `gitInfo`; omitted fields are left unchanged, while explicit `null` clears a stored value.
codex-rs/app-server/src/bespoke_event_handling.rs
+9 -6
View File
@@ -2016,9 +2016,12 @@ async fn complete_file_change_item(
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
let mut state = thread_state.lock().await;
state.turn_summary.file_change_started.remove(&item_id);
drop(state);
thread_state
.lock()
.await
.turn_summary
.file_change_started
.remove(&item_id);
let notification = ItemCompletedNotification {
thread_id: conversation_id.to_string(),
@@ -2087,12 +2090,12 @@ async fn complete_command_execution_item(
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
let mut state = thread_state.lock().await;
let should_emit = state
let should_emit = thread_state
.lock()
.await
.turn_summary
.command_execution_started
.remove(&item_id);
drop(state);
if !should_emit {
return;
}
codex-rs/app-server/src/codex_message_processor.rs
+427 -20
View File
@@ -19,6 +19,7 @@ use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
use crate::thread_status::ThreadWatchManager;
use crate::thread_status::resolve_thread_status;
use chrono::DateTime;
use chrono::Duration as ChronoDuration;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_analytics::AnalyticsEventsClient;
@@ -123,6 +124,7 @@ use codex_app_server_protocol::SkillsConfigWriteParams;
use codex_app_server_protocol::SkillsConfigWriteResponse;
use codex_app_server_protocol::SkillsListParams;
use codex_app_server_protocol::SkillsListResponse;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadArchiveResponse;
@@ -177,6 +179,8 @@ 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::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnarchiveResponse;
use codex_app_server_protocol::ThreadUnarchivedNotification;
@@ -325,6 +329,7 @@ use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams;
use codex_thread_store::ListThreadsParams as StoreListThreadsParams;
use codex_thread_store::LocalThreadStore;
use codex_thread_store::ReadThreadParams as StoreReadThreadParams;
use codex_thread_store::SortDirection as StoreSortDirection;
use codex_thread_store::StoredThread;
use codex_thread_store::ThreadSortKey as StoreThreadSortKey;
use codex_thread_store::ThreadStore;
@@ -377,6 +382,8 @@ use token_usage_replay::send_thread_token_usage_update_to_connection;
const THREAD_LIST_DEFAULT_LIMIT: usize = 25;
const THREAD_LIST_MAX_LIMIT: usize = 100;
const THREAD_TURNS_DEFAULT_LIMIT: usize = 25;
const THREAD_TURNS_MAX_LIMIT: usize = 100;
struct ThreadListFilters {
model_providers: Option<Vec<String>>,
@@ -951,6 +958,10 @@ impl CodexMessageProcessor {
self.thread_read(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadTurnsList { request_id, params } => {
self.thread_turns_list(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadShellCommand { request_id, params } => {
self.thread_shell_command(to_connection_request_id(request_id), params)
.await;
@@ -3510,9 +3521,7 @@ impl CodexMessageProcessor {
// No ThreadRollback event will arrive if an error occurs.
// Clean up and reply immediately.
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
let mut thread_state = thread_state.lock().await;
thread_state.pending_rollbacks = None;
drop(thread_state);
thread_state.lock().await.pending_rollbacks = None;
self.send_internal_error(request, format!("failed to start rollback: {err}"))
.await;
@@ -3641,6 +3650,7 @@ impl CodexMessageProcessor {
cursor,
limit,
sort_key,
sort_direction,
model_providers,
source_kinds,
archived,
@@ -3663,11 +3673,13 @@ impl CodexMessageProcessor {
ThreadSortKey::CreatedAt => StoreThreadSortKey::CreatedAt,
ThreadSortKey::UpdatedAt => StoreThreadSortKey::UpdatedAt,
};
let (summaries, next_cursor) = match self
let sort_direction = sort_direction.unwrap_or(SortDirection::Desc);
let list_result = self
.list_threads_common(
requested_page_size,
cursor,
store_sort_key,
sort_direction,
ThreadListFilters {
model_providers,
source_kinds,
@@ -3676,14 +3688,17 @@ impl CodexMessageProcessor {
search_term,
},
)
.await
{
.await;
let (summaries, next_cursor) = match list_result {
Ok(r) => r,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let backwards_cursor = summaries.first().and_then(|summary| {
thread_backwards_cursor_for_sort_key(summary, store_sort_key, sort_direction)
});
let mut threads = Vec::with_capacity(summaries.len());
let mut thread_ids = HashSet::with_capacity(summaries.len());
let mut status_ids = Vec::with_capacity(summaries.len());
@@ -3704,7 +3719,7 @@ impl CodexMessageProcessor {
.loaded_statuses_for_threads(status_ids)
.await;
let data = threads
let data: Vec<_> = threads
.into_iter()
.map(|(conversation_id, mut thread)| {
if let Some(title) = names.get(&conversation_id).cloned() {
@@ -3716,7 +3731,11 @@ impl CodexMessageProcessor {
thread
})
.collect();
let response = ThreadListResponse { data, next_cursor };
let response = ThreadListResponse {
data,
next_cursor,
backwards_cursor,
};
self.outgoing.send_response(request_id, response).await;
}
@@ -3832,7 +3851,7 @@ impl CodexMessageProcessor {
)));
};
let has_live_in_progress_turn = if let Some(loaded_thread) = loaded_thread {
let has_live_in_progress_turn = if let Some(loaded_thread) = loaded_thread.as_ref() {
matches!(loaded_thread.agent_status().await, AgentStatus::Running)
} else {
false
@@ -3959,6 +3978,157 @@ impl CodexMessageProcessor {
Ok(())
}
async fn thread_turns_list(
&self,
request_id: ConnectionRequestId,
params: ThreadTurnsListParams,
) {
let ThreadTurnsListParams {
thread_id,
cursor,
limit,
sort_direction,
} = params;
let thread_uuid = 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;
}
};
let state_db_ctx = get_state_db(&self.config).await;
let mut rollout_path = self
.resolve_rollout_path(thread_uuid, state_db_ctx.as_ref())
.await;
if rollout_path.is_none() {
rollout_path =
match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string())
.await
{
Ok(Some(path)) => Some(path),
Ok(None) => match find_archived_thread_path_by_id_str(
&self.config.codex_home,
&thread_uuid.to_string(),
)
.await
{
Ok(path) => path,
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate archived thread id {thread_uuid}: {err}"),
)
.await;
return;
}
},
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {thread_uuid}: {err}"),
)
.await;
return;
}
};
}
if rollout_path.is_none() {
match self.thread_manager.get_thread(thread_uuid).await {
Ok(thread) => {
rollout_path = thread.rollout_path();
if rollout_path.is_none() {
self.send_invalid_request_error(
request_id,
"ephemeral threads do not support thread/turns/list".to_string(),
)
.await;
return;
}
}
Err(_) => {
self.send_invalid_request_error(
request_id,
format!("thread not loaded: {thread_uuid}"),
)
.await;
return;
}
}
}
let Some(rollout_path) = rollout_path.as_ref() else {
self.send_internal_error(
request_id,
format!("failed to locate rollout for thread {thread_uuid}"),
)
.await;
return;
};
match read_rollout_items_from_rollout(rollout_path).await {
Ok(items) => {
// This API optimizes network transfer by letting clients page through a
// thread's turns incrementally, but it still replays the entire rollout on
// every request. Rollback and compaction events can change earlier turns, so
// the server has to rebuild the full turn list until turn metadata is indexed
// separately.
let has_live_in_progress_turn =
match self.thread_manager.get_thread(thread_uuid).await {
Ok(thread) => matches!(thread.agent_status().await, AgentStatus::Running),
Err(_) => false,
};
let turns = reconstruct_thread_turns_from_rollout_items(
&items,
self.thread_watch_manager
.loaded_status_for_thread(&thread_uuid.to_string())
.await,
has_live_in_progress_turn,
);
let page = match paginate_thread_turns(
turns,
cursor.as_deref(),
limit,
sort_direction.unwrap_or(SortDirection::Desc),
) {
Ok(page) => page,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let response = ThreadTurnsListResponse {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
};
self.outgoing.send_response(request_id, response).await;
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
self.send_invalid_request_error(
request_id,
format!(
"thread {thread_uuid} is not materialized yet; thread/turns/list is unavailable before first user message"
),
)
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!(
"failed to load rollout `{}` for thread {thread_uuid}: {err}",
rollout_path.display()
),
)
.await;
}
}
}
pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> {
self.thread_manager.subscribe_thread_created()
}
@@ -4980,6 +5150,7 @@ impl CodexMessageProcessor {
requested_page_size: usize,
cursor: Option<String>,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
filters: ThreadListFilters,
) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
let ThreadListFilters {
@@ -5008,6 +5179,10 @@ impl CodexMessageProcessor {
let fallback_provider = self.config.model_provider_id.clone();
let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds);
let allowed_sources = allowed_sources_vec.as_slice();
let store_sort_direction = match sort_direction {
SortDirection::Asc => StoreSortDirection::Asc,
SortDirection::Desc => StoreSortDirection::Desc,
};
while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
@@ -5017,6 +5192,7 @@ impl CodexMessageProcessor {
page_size,
cursor: cursor_obj.clone(),
sort_key,
sort_direction: store_sort_direction,
allowed_sources: allowed_sources.to_vec(),
model_providers: model_provider_filter.clone(),
archived,
@@ -5025,12 +5201,17 @@ impl CodexMessageProcessor {
.await
.map_err(thread_store_list_error)?;
let mut filtered = Vec::with_capacity(page.items.len());
let mut candidate_summaries = Vec::with_capacity(page.items.len());
for it in page.items {
let Some(summary) = summary_from_stored_thread(it, fallback_provider.as_str())
else {
continue;
};
candidate_summaries.push(summary);
}
let mut filtered = Vec::with_capacity(candidate_summaries.len());
for summary in candidate_summaries {
if source_kind_filter
.as_ref()
.is_none_or(|filter| source_kind_matches(&summary.source, filter))
@@ -8051,8 +8232,10 @@ impl CodexMessageProcessor {
start_fuzzy_file_search_session(session_id.clone(), roots, self.outgoing.clone());
match session {
Ok(session) => {
let mut sessions = self.fuzzy_search_sessions.lock().await;
sessions.insert(session_id, session);
self.fuzzy_search_sessions
.lock()
.await
.insert(session_id, session);
self.outgoing
.send_response(request_id, FuzzyFileSearchSessionStartResponse {})
.await;
@@ -9339,8 +9522,18 @@ fn summary_from_stored_thread(
conversation_id: thread.thread_id,
path,
preview: thread.first_user_message.unwrap_or(thread.preview),
timestamp: Some(thread.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
updated_at: Some(thread.updated_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
// Preserve millisecond precision from the thread store so thread/list cursors
// round-trip the same ordering key used by pagination queries.
timestamp: Some(
thread
.created_at
.to_rfc3339_opts(SecondsFormat::Millis, true),
),
updated_at: Some(
thread
.updated_at
.to_rfc3339_opts(SecondsFormat::Millis, true),
),
model_provider: if thread.model_provider.is_empty() {
fallback_provider.to_string()
} else {
@@ -9764,17 +9957,188 @@ pub(crate) fn summary_to_thread(
}
}
fn thread_backwards_cursor_for_sort_key(
summary: &ConversationSummary,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
) -> Option<String> {
let timestamp = match sort_key {
StoreThreadSortKey::CreatedAt => summary.timestamp.as_deref(),
StoreThreadSortKey::UpdatedAt => summary
.updated_at
.as_deref()
.or(summary.timestamp.as_deref()),
};
let timestamp = parse_datetime(timestamp)?;
// The state DB stores unique millisecond timestamps. Offset the reverse cursor by one
// millisecond so the opposite-direction query includes the page anchor.
let timestamp = match sort_direction {
SortDirection::Asc => timestamp.checked_add_signed(ChronoDuration::milliseconds(1))?,
SortDirection::Desc => timestamp.checked_sub_signed(ChronoDuration::milliseconds(1))?,
};
Some(timestamp.to_rfc3339_opts(SecondsFormat::Millis, true))
}
struct ThreadTurnsPage {
turns: Vec<Turn>,
next_cursor: Option<String>,
backwards_cursor: Option<String>,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct ThreadTurnsCursor {
turn_id: String,
include_anchor: bool,
}
fn paginate_thread_turns(
turns: Vec<Turn>,
cursor: Option<&str>,
limit: Option<u32>,
sort_direction: SortDirection,
) -> Result<ThreadTurnsPage, JSONRPCErrorError> {
if turns.is_empty() {
return Ok(ThreadTurnsPage {
turns: Vec::new(),
next_cursor: None,
backwards_cursor: None,
});
}
let anchor = cursor.map(parse_thread_turns_cursor).transpose()?;
let page_size = limit
.map(|value| value as usize)
.unwrap_or(THREAD_TURNS_DEFAULT_LIMIT)
.clamp(1, THREAD_TURNS_MAX_LIMIT);
let anchor_index = anchor
.as_ref()
.and_then(|anchor| turns.iter().position(|turn| turn.id == anchor.turn_id));
if anchor.is_some() && anchor_index.is_none() {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "invalid cursor: anchor turn is no longer present".to_string(),
data: None,
});
}
let mut keyed_turns: Vec<_> = turns.into_iter().enumerate().collect();
match sort_direction {
SortDirection::Asc => {
if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) {
keyed_turns.retain(|(index, _)| {
if anchor.include_anchor {
*index >= anchor_index
} else {
*index > anchor_index
}
});
}
}
SortDirection::Desc => {
keyed_turns.reverse();
if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) {
keyed_turns.retain(|(index, _)| {
if anchor.include_anchor {
*index <= anchor_index
} else {
*index < anchor_index
}
});
}
}
}
let more_turns_available = keyed_turns.len() > page_size;
keyed_turns.truncate(page_size);
let backwards_cursor = keyed_turns
.first()
.map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, /*include_anchor*/ true))
.transpose()?;
let next_cursor = if more_turns_available {
keyed_turns
.last()
.map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, /*include_anchor*/ false))
.transpose()?
} else {
None
};
let turns = keyed_turns.into_iter().map(|(_, turn)| turn).collect();
Ok(ThreadTurnsPage {
turns,
next_cursor,
backwards_cursor,
})
}
fn serialize_thread_turns_cursor(
turn_id: &str,
include_anchor: bool,
) -> Result<String, JSONRPCErrorError> {
serde_json::to_string(&ThreadTurnsCursor {
turn_id: turn_id.to_string(),
include_anchor,
})
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to serialize cursor: {err}"),
data: None,
})
}
fn parse_thread_turns_cursor(cursor: &str) -> Result<ThreadTurnsCursor, JSONRPCErrorError> {
serde_json::from_str(cursor).map_err(|_| JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("invalid cursor: {cursor}"),
data: None,
})
}
fn reconstruct_thread_turns_from_rollout_items(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_in_progress_turn: bool,
) -> Vec<Turn> {
let mut turns = build_turns_from_rollout_items(items);
normalize_thread_turns_status(&mut turns, loaded_status, has_live_in_progress_turn);
turns
}
fn normalize_thread_turns_status(
turns: &mut [Turn],
loaded_status: ThreadStatus,
has_live_in_progress_turn: bool,
) {
let status = resolve_thread_status(loaded_status, has_live_in_progress_turn);
if matches!(status, ThreadStatus::Active { .. }) {
return;
}
for turn in turns {
if matches!(turn.status, TurnStatus::InProgress) {
turn.status = TurnStatus::Interrupted;
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::outgoing_message::OutgoingEnvelope;
use crate::outgoing_message::OutgoingMessage;
use anyhow::Result;
use chrono::DateTime;
use chrono::Utc;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::ToolRequestUserInputParams;
use codex_protocol::ThreadId;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_thread_store::StoredThread;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
@@ -9824,6 +10188,53 @@ mod tests {
validate_dynamic_tools(&tools).expect("valid schema");
}
#[test]
fn summary_from_stored_thread_preserves_millisecond_precision() {
let created_at =
DateTime::parse_from_rfc3339("2025-01-02T03:04:05.678Z").expect("valid timestamp");
let updated_at =
DateTime::parse_from_rfc3339("2025-01-02T03:04:06.789Z").expect("valid timestamp");
let thread_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread");
let stored_thread = StoredThread {
thread_id,
rollout_path: Some(PathBuf::from("/tmp/thread.jsonl")),
forked_from_id: None,
preview: "preview".to_string(),
name: None,
model_provider: "openai".to_string(),
model: None,
reasoning_effort: None,
created_at: created_at.with_timezone(&Utc),
updated_at: updated_at.with_timezone(&Utc),
archived_at: None,
cwd: PathBuf::from("/tmp"),
cli_version: "0.0.0".to_string(),
source: SessionSource::Cli,
agent_nickname: None,
agent_role: None,
agent_path: None,
git_info: None,
approval_mode: AskForApproval::OnRequest,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
token_usage: None,
first_user_message: Some("first user message".to_string()),
history: None,
};
let summary =
summary_from_stored_thread(stored_thread, "fallback").expect("summary should exist");
assert_eq!(
summary.timestamp.as_deref(),
Some("2025-01-02T03:04:05.678Z")
);
assert_eq!(
summary.updated_at.as_deref(),
Some("2025-01-02T03:04:06.789Z")
);
}
#[test]
fn config_load_error_marks_cloud_requirements_failures_for_relogin() {
let err = std::io::Error::other(CloudRequirementsLoadError::new(
@@ -10446,13 +10857,9 @@ mod tests {
assert_eq!(cancel_rx.await, Ok(()));
let state = manager.thread_state(thread_id).await;
let subscribed_connection_ids = manager.subscribed_connection_ids(thread_id).await;
let state = state.lock().await;
assert!(
manager
.subscribed_connection_ids(thread_id)
.await
.is_empty()
);
assert!(subscribed_connection_ids.is_empty());
assert!(state.cancel_tx.is_none());
assert!(state.active_turn_snapshot().is_none());
Ok(())
codex-rs/app-server/src/transport/remote_control/websocket.rs
+22 -14
View File
@@ -392,7 +392,14 @@ impl RemoteControlWebsocket {
ping_interval: std::time::Duration,
shutdown_token: CancellationToken,
) -> io::Result<()> {
for server_envelope in state.lock().await.outbound_buffer.server_envelopes() {
let server_envelopes = state
.lock()
.await
.outbound_buffer
.server_envelopes()
.cloned()
.collect::<Vec<_>>();
for server_envelope in server_envelopes {
let payload = match serde_json::to_string(&server_envelope) {
Ok(payload) => payload,
Err(err) => {
@@ -594,21 +601,22 @@ impl RemoteControlWebsocket {
}
};
let mut websocket_state = state.lock().await;
if let Some(cursor) = client_envelope.cursor.as_deref() {
websocket_state.subscribe_cursor = Some(cursor.to_string());
}
if let ClientEvent::Ack = &client_envelope.event
&& let Some(acked_seq_id) = client_envelope.seq_id
&& let Some(stream_id) = client_envelope.stream_id.as_ref()
{
websocket_state.outbound_buffer.ack(
&client_envelope.client_id,
stream_id,
acked_seq_id,
);
let mut websocket_state = state.lock().await;
if let Some(cursor) = client_envelope.cursor.as_deref() {
websocket_state.subscribe_cursor = Some(cursor.to_string());
}
if let ClientEvent::Ack = &client_envelope.event
&& let Some(acked_seq_id) = client_envelope.seq_id
&& let Some(stream_id) = client_envelope.stream_id.as_ref()
{
websocket_state.outbound_buffer.ack(
&client_envelope.client_id,
stream_id,
acked_seq_id,
);
}
}
drop(websocket_state);
if client_tracker
.handle_message(client_envelope)
codex-rs/app-server/tests/common/mcp_process.rs
+10
View File
@@ -79,6 +79,7 @@ use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadSetNameParams;
use codex_app_server_protocol::ThreadShellCommandParams;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnsubscribeParams;
use codex_app_server_protocol::TurnCompletedNotification;
@@ -464,6 +465,15 @@ impl McpProcess {
self.send_request("thread/read", params).await
}
/// Send a `thread/turns/list` JSON-RPC request.
pub async fn send_thread_turns_list_request(
&mut self,
params: ThreadTurnsListParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("thread/turns/list", params).await
}
/// Send a `model/list` JSON-RPC request.
pub async fn send_list_models_request(
&mut self,
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+1
View File
@@ -543,6 +543,7 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()
cursor: None,
limit: Some(10),
sort_key: None,
sort_direction: None,
model_providers: None,
source_kinds: None,
archived: None,
codex-rs/app-server/tests/suite/v2/thread_list.rs
+113
View File
@@ -14,6 +14,7 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadSortKey;
use codex_app_server_protocol::ThreadSourceKind;
@@ -84,6 +85,7 @@ async fn list_threads_with_sort(
cursor,
limit,
sort_key,
sort_direction: None,
model_providers: providers,
source_kinds,
archived,
@@ -357,6 +359,7 @@ async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> {
let ThreadListResponse {
data: data1,
next_cursor: cursor1,
..
} = list_threads(
&mut mcp,
/*cursor*/ None,
@@ -384,6 +387,7 @@ async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> {
let ThreadListResponse {
data: data2,
next_cursor: cursor2,
..
} = list_threads(
&mut mcp,
Some(cursor1),
@@ -498,6 +502,7 @@ async fn thread_list_respects_cwd_filter() -> Result<()> {
cursor: None,
limit: Some(10),
sort_key: None,
sort_direction: None,
model_providers: Some(vec!["mock_provider".to_string()]),
source_kinds: None,
archived: None,
@@ -584,6 +589,7 @@ sqlite = true
/*page_size*/ 10,
/*cursor*/ None,
codex_core::ThreadSortKey::CreatedAt,
codex_core::SortDirection::Desc,
&[],
/*model_providers*/ None,
"mock_provider",
@@ -598,6 +604,7 @@ sqlite = true
cursor: None,
limit: Some(10),
sort_key: None,
sort_direction: None,
model_providers: Some(vec!["mock_provider".to_string()]),
source_kinds: None,
archived: None,
@@ -1252,6 +1259,111 @@ async fn thread_list_updated_at_paginates_with_cursor() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_list_backwards_cursor_can_seed_forward_delta_sync() -> Result<()> {
let codex_home = TempDir::new()?;
create_minimal_config(codex_home.path())?;
let id_old = create_fake_rollout(
codex_home.path(),
"2025-02-01T10-00-00",
"2025-02-01T10:00:00Z",
"Hello",
Some("mock_provider"),
/*git_info*/ None,
)?;
let id_watermark = create_fake_rollout(
codex_home.path(),
"2025-02-01T11-00-00",
"2025-02-01T11:00:00Z",
"Hello",
Some("mock_provider"),
/*git_info*/ None,
)?;
set_rollout_mtime(
rollout_path(codex_home.path(), "2025-02-01T10-00-00", &id_old).as_path(),
"2025-02-02T00:00:00Z",
)?;
set_rollout_mtime(
rollout_path(codex_home.path(), "2025-02-01T11-00-00", &id_watermark).as_path(),
"2025-02-03T00:00:00Z",
)?;
let mut mcp = init_mcp(codex_home.path()).await?;
let ThreadListResponse {
data: page1,
backwards_cursor,
..
} = {
let request_id = mcp
.send_thread_list_request(codex_app_server_protocol::ThreadListParams {
cursor: None,
limit: Some(1),
sort_key: Some(ThreadSortKey::UpdatedAt),
sort_direction: Some(SortDirection::Desc),
model_providers: Some(vec!["mock_provider".to_string()]),
source_kinds: None,
archived: None,
cwd: None,
search_term: None,
})
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
to_response::<ThreadListResponse>(resp)?
};
let ids_page1: Vec<_> = page1.iter().map(|thread| thread.id.as_str()).collect();
assert_eq!(ids_page1, vec![id_watermark.as_str()]);
let backwards_cursor = backwards_cursor.expect("expected backwardsCursor on first page");
assert_eq!(backwards_cursor, "2025-02-02T23:59:59.999Z");
let id_new = create_fake_rollout(
codex_home.path(),
"2025-02-01T12-00-00",
"2025-02-01T12:00:00Z",
"Hello",
Some("mock_provider"),
/*git_info*/ None,
)?;
set_rollout_mtime(
rollout_path(codex_home.path(), "2025-02-01T12-00-00", &id_new).as_path(),
"2025-02-04T00:00:00Z",
)?;
let ThreadListResponse {
data: delta_page, ..
} = {
let request_id = mcp
.send_thread_list_request(codex_app_server_protocol::ThreadListParams {
cursor: Some(backwards_cursor),
limit: Some(10),
sort_key: Some(ThreadSortKey::UpdatedAt),
sort_direction: Some(SortDirection::Asc),
model_providers: Some(vec!["mock_provider".to_string()]),
source_kinds: None,
archived: None,
cwd: None,
search_term: None,
})
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
to_response::<ThreadListResponse>(resp)?
};
let ids_delta: Vec<_> = delta_page.iter().map(|thread| thread.id.as_str()).collect();
assert_eq!(ids_delta, vec![id_watermark.as_str(), id_new.as_str()]);
Ok(())
}
#[tokio::test]
async fn thread_list_created_at_tie_breaks_by_uuid() -> Result<()> {
let codex_home = TempDir::new()?;
@@ -1468,6 +1580,7 @@ async fn thread_list_invalid_cursor_returns_error() -> Result<()> {
cursor: Some("not-a-cursor".to_string()),
limit: Some(2),
sort_key: None,
sort_direction: None,
model_providers: Some(vec!["mock_provider".to_string()]),
source_kinds: None,
archived: None,
codex-rs/app-server/tests/suite/v2/thread_metadata_update.rs
+2 -2
View File
@@ -109,7 +109,7 @@ async fn thread_metadata_update_patches_git_branch_and_returns_updated_thread()
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread: read, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(
read.git_info,
@@ -421,7 +421,7 @@ async fn thread_metadata_update_can_clear_stored_git_fields() -> Result<()> {
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread: read, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(read.git_info, None);
codex-rs/app-server/tests/suite/v2/thread_read.rs
+213 -6
View File
@@ -9,6 +9,7 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadItem;
@@ -24,6 +25,8 @@ use codex_app_server_protocol::ThreadSetNameResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
@@ -34,6 +37,8 @@ use codex_protocol::user_input::TextElement;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::io::Write;
use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -81,7 +86,7 @@ async fn thread_read_returns_summary_without_turns() -> Result<()> {
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.id, conversation_id);
assert_eq!(thread.preview, preview);
@@ -136,7 +141,7 @@ async fn thread_read_can_include_turns() -> Result<()> {
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.turns.len(), 1);
let turn = &thread.turns[0];
@@ -159,6 +164,88 @@ async fn thread_read_can_include_turns() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_turns_list_can_page_backward_and_forward() -> 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 filename_ts = "2025-01-05T12-00-00";
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
filename_ts,
"2025-01-05T12:00:00Z",
"first",
vec![],
Some("mock_provider"),
/*git_info*/ None,
)?;
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?;
append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let read_id = mcp
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: conversation_id.clone(),
cursor: None,
limit: Some(2),
sort_direction: Some(SortDirection::Desc),
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadTurnsListResponse {
data,
next_cursor,
backwards_cursor,
} = to_response::<ThreadTurnsListResponse>(read_resp)?;
assert_eq!(turn_user_texts(&data), vec!["third", "second"]);
let next_cursor = next_cursor.expect("expected nextCursor for older turns");
let backwards_cursor = backwards_cursor.expect("expected backwardsCursor for newest turn");
let read_id = mcp
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: conversation_id.clone(),
cursor: Some(next_cursor),
limit: Some(10),
sort_direction: Some(SortDirection::Desc),
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadTurnsListResponse { data, .. } = to_response::<ThreadTurnsListResponse>(read_resp)?;
assert_eq!(turn_user_texts(&data), vec!["first"]);
append_user_message(rollout_path.as_path(), "2025-01-05T12:03:00Z", "fourth")?;
let read_id = mcp
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: conversation_id,
cursor: Some(backwards_cursor),
limit: Some(10),
sort_direction: Some(SortDirection::Asc),
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadTurnsListResponse { data, .. } = to_response::<ThreadTurnsListResponse>(read_resp)?;
assert_eq!(turn_user_texts(&data), vec!["third", "fourth"]);
Ok(())
}
#[tokio::test]
async fn thread_read_can_return_archived_threads_by_id() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -207,6 +294,75 @@ async fn thread_read_can_return_archived_threads_by_id() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_turns_list_rejects_cursor_when_anchor_turn_is_rolled_back() -> 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 filename_ts = "2025-01-05T12-00-00";
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
filename_ts,
"2025-01-05T12:00:00Z",
"first",
vec![],
Some("mock_provider"),
/*git_info*/ None,
)?;
let rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
append_user_message(rollout_path.as_path(), "2025-01-05T12:01:00Z", "second")?;
append_user_message(rollout_path.as_path(), "2025-01-05T12:02:00Z", "third")?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let read_id = mcp
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: conversation_id.clone(),
cursor: None,
limit: Some(2),
sort_direction: Some(SortDirection::Desc),
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadTurnsListResponse {
backwards_cursor, ..
} = to_response::<ThreadTurnsListResponse>(read_resp)?;
let backwards_cursor = backwards_cursor.expect("expected backwardsCursor for newest turn");
append_thread_rollback(
rollout_path.as_path(),
"2025-01-05T12:03:00Z",
/*num_turns*/ 1,
)?;
let read_id = mcp
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: conversation_id,
cursor: Some(backwards_cursor),
limit: Some(10),
sort_direction: Some(SortDirection::Asc),
})
.await?;
let read_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(read_id)),
)
.await??;
assert_eq!(
read_err.error.message,
"invalid cursor: anchor turn is no longer present"
);
Ok(())
}
#[tokio::test]
async fn thread_read_returns_forked_from_id_for_forked_threads() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -250,7 +406,7 @@ async fn thread_read_returns_forked_from_id_for_forked_threads() -> Result<()> {
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.forked_from_id, Some(conversation_id));
@@ -295,7 +451,7 @@ async fn thread_read_loaded_thread_returns_precomputed_path_before_materializati
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread: read } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread: read, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(read.id, thread.id);
assert_eq!(read.path, Some(thread_path));
@@ -363,7 +519,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
)
.await??;
let read_result = read_resp.result.clone();
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.id, conversation_id);
assert_eq!(thread.name.as_deref(), Some(new_name));
let thread_json = read_result
@@ -387,6 +543,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
cursor: None,
limit: Some(50),
sort_key: None,
sort_direction: None,
model_providers: Some(vec!["mock_provider".to_string()]),
source_kinds: None,
archived: None,
@@ -572,13 +729,63 @@ async fn thread_read_reports_system_error_idle_flag_after_failed_turn() -> Resul
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.status, ThreadStatus::SystemError,);
Ok(())
}
fn append_user_message(path: &Path, timestamp: &str, text: &str) -> std::io::Result<()> {
let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
writeln!(
file,
"{}",
json!({
"timestamp": timestamp,
"type":"event_msg",
"payload": {
"type":"user_message",
"message": text,
"text_elements": [],
"local_images": []
}
})
)
}
fn append_thread_rollback(path: &Path, timestamp: &str, num_turns: u32) -> std::io::Result<()> {
let mut file = std::fs::OpenOptions::new().append(true).open(path)?;
writeln!(
file,
"{}",
json!({
"timestamp": timestamp,
"type":"event_msg",
"payload": {
"type":"thread_rolled_back",
"num_turns": num_turns
}
})
)
}
fn turn_user_texts(turns: &[codex_app_server_protocol::Turn]) -> Vec<&str> {
turns
.iter()
.filter_map(|turn| match turn.items.first()? {
ThreadItem::UserMessage { content, .. } => match content.first()? {
UserInput::Text { text, .. } => Some(text.as_str()),
UserInput::Image { .. }
| UserInput::LocalImage { .. }
| UserInput::Skill { .. }
| UserInput::Mention { .. } => None,
},
_ => None,
})
.collect()
}
// Helper to create a config.toml pointing at the mock model server.
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+1
View File
@@ -844,6 +844,7 @@ async fn thread_resume_and_read_interrupt_incomplete_rollout_turn_when_thread_is
.await??;
let ThreadReadResponse {
thread: read_thread,
..
} = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(read_thread.status, ThreadStatus::Idle);
codex-rs/app-server/tests/suite/v2/thread_shell_command.rs
+2 -2
View File
@@ -135,7 +135,7 @@ async fn thread_shell_command_runs_as_standalone_turn_and_persists_history() ->
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.turns.len(), 1);
let ThreadItem::CommandExecution {
source,
@@ -305,7 +305,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> {
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.turns.len(), 1);
assert!(
thread.turns[0].items.iter().any(|item| {
codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs
+1 -1
View File
@@ -288,7 +288,7 @@ async fn thread_unsubscribe_preserves_cached_status_before_idle_unload() -> Resu
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread } = to_response::<ThreadReadResponse>(read_resp)?;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
assert_eq!(thread.status, ThreadStatus::SystemError);
let unsubscribe_id = mcp
codex-rs/core/src/lib.rs
+1
View File
@@ -157,6 +157,7 @@ pub use rollout::RolloutRecorder;
pub use rollout::RolloutRecorderParams;
pub use rollout::SESSIONS_SUBDIR;
pub use rollout::SessionMeta;
pub use rollout::SortDirection;
pub use rollout::ThreadItem;
pub use rollout::ThreadSortKey;
pub use rollout::ThreadsPage;
codex-rs/core/src/personality_migration.rs
+1
View File
@@ -79,6 +79,7 @@ async fn has_threads(store: &LocalThreadStore, archived: bool) -> io::Result<boo
page_size: 1,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: codex_thread_store::SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived,
codex-rs/core/src/realtime_context.rs
+2
View File
@@ -6,6 +6,7 @@ use codex_exec_server::LOCAL_FS;
use codex_git_utils::resolve_root_git_project_for_trust;
use codex_protocol::models::ResponseItem;
use codex_thread_store::ListThreadsParams;
use codex_thread_store::SortDirection;
use codex_thread_store::StoredThread;
use codex_thread_store::ThreadSortKey;
use codex_thread_store::ThreadStore;
@@ -132,6 +133,7 @@ async fn load_recent_threads(sess: &Session) -> Vec<StoredThread> {
page_size: MAX_RECENT_THREADS,
cursor: None,
sort_key: ThreadSortKey::UpdatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: false,
codex-rs/core/src/rollout.rs
+1
View File
@@ -7,6 +7,7 @@ pub use codex_rollout::RolloutRecorder;
pub use codex_rollout::RolloutRecorderParams;
pub use codex_rollout::SESSIONS_SUBDIR;
pub use codex_rollout::SessionMeta;
pub use codex_rollout::SortDirection;
pub use codex_rollout::ThreadItem;
pub use codex_rollout::ThreadSortKey;
pub use codex_rollout::ThreadsPage;
codex-rs/core/src/session/mod.rs
+35 -28
View File
@@ -1385,37 +1385,44 @@ impl Session {
&self,
updates: SessionSettingsUpdate,
) -> ConstraintResult<()> {
let mut state = self.state.lock().await;
match state.session_configuration.apply(&updates) {
Ok(updated) => {
let previous_cwd = state.session_configuration.cwd.clone();
let sandbox_policy_changed =
state.session_configuration.sandbox_policy != updated.sandbox_policy;
let next_cwd = updated.cwd.clone();
let codex_home = updated.codex_home.clone();
let session_source = updated.session_source.clone();
state.session_configuration = updated;
drop(state);
self.maybe_refresh_shell_snapshot_for_cwd(
&previous_cwd,
&next_cwd,
&codex_home,
&session_source,
);
if sandbox_policy_changed {
self.refresh_managed_network_proxy_for_current_sandbox_policy()
.await;
let (previous_cwd, sandbox_policy_changed, next_cwd, codex_home, session_source) = {
let mut state = self.state.lock().await;
let updated = match state.session_configuration.apply(&updates) {
Ok(updated) => updated,
Err(err) => {
warn!("rejected session settings update: {err}");
return Err(err);
}
};
Ok(())
}
Err(err) => {
warn!("rejected session settings update: {err}");
Err(err)
}
let previous_cwd = state.session_configuration.cwd.clone();
let sandbox_policy_changed =
state.session_configuration.sandbox_policy != updated.sandbox_policy;
let next_cwd = updated.cwd.clone();
let codex_home = updated.codex_home.clone();
let session_source = updated.session_source.clone();
state.session_configuration = updated;
(
previous_cwd,
sandbox_policy_changed,
next_cwd,
codex_home,
session_source,
)
};
self.maybe_refresh_shell_snapshot_for_cwd(
&previous_cwd,
&next_cwd,
&codex_home,
&session_source,
);
if sandbox_policy_changed {
self.refresh_managed_network_proxy_for_current_sandbox_policy()
.await;
}
Ok(())
}
pub(crate) async fn set_session_startup_prewarm(
codex-rs/core/src/session/turn_context.rs
+24 -21
View File
@@ -446,13 +446,7 @@ impl Session {
sub_id: String,
updates: SessionSettingsUpdate,
) -> ConstraintResult<Arc<TurnContext>> {
let (
session_configuration,
sandbox_policy_changed,
previous_cwd,
codex_home,
session_source,
) = {
let update_result = {
let mut state = self.state.lock().await;
match state.session_configuration.clone().apply(&updates) {
Ok(next) => {
@@ -462,26 +456,35 @@ impl Session {
let codex_home = next.codex_home.clone();
let session_source = next.session_source.clone();
state.session_configuration = next.clone();
(
Ok((
next,
sandbox_policy_changed,
previous_cwd,
codex_home,
session_source,
)
}
Err(err) => {
drop(state);
self.send_event_raw(Event {
id: sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: err.to_string(),
codex_error_info: Some(CodexErrorInfo::BadRequest),
}),
})
.await;
return Err(err);
))
}
Err(err) => Err(err),
}
};
let (
session_configuration,
sandbox_policy_changed,
previous_cwd,
codex_home,
session_source,
) = match update_result {
Ok(update) => update,
Err(err) => {
self.send_event_raw(Event {
id: sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: err.to_string(),
codex_error_info: Some(CodexErrorInfo::BadRequest),
}),
})
.await;
return Err(err);
}
};
codex-rs/core/src/tools/js_repl/mod.rs
+42 -40
View File
@@ -884,9 +884,8 @@ impl JsReplManager {
let (req_id, rx) = {
let req_id = Uuid::new_v4().to_string();
let mut pending = pending_execs.lock().await;
let (tx, rx) = tokio::sync::oneshot::channel();
pending.insert(req_id.clone(), tx);
pending_execs.lock().await.insert(req_id.clone(), tx);
exec_contexts.lock().await.insert(
req_id.clone(),
ExecContext {
@@ -956,9 +955,7 @@ impl JsReplManager {
let response = match tokio::time::timeout(Duration::from_millis(timeout_ms), rx).await {
Ok(Ok(msg)) => msg,
Ok(Err(_)) => {
let mut pending = pending_execs.lock().await;
let removed = pending.remove(&req_id).is_some();
drop(pending);
let removed = pending_execs.lock().await.remove(&req_id).is_some();
if removed {
self.clear_top_level_exec_if_matches(&req_id).await;
}
@@ -1340,40 +1337,40 @@ impl JsReplManager {
KernelToHost::EmitImage(req) => {
let exec_id = req.exec_id.clone();
let emit_id = req.id.clone();
let response =
if let Some(ctx) = exec_contexts.lock().await.get(&exec_id).cloned() {
match validate_emitted_image_url(&req.image_url) {
Ok(()) => {
let content_item = emitted_image_content_item(
ctx.turn.as_ref(),
req.image_url,
req.detail,
);
JsReplManager::record_exec_content_item(
&exec_tool_calls,
&exec_id,
content_item,
)
.await;
HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: true,
error: None,
})
}
Err(error) => HostToKernel::EmitImageResult(EmitImageResult {
let context = exec_contexts.lock().await.get(&exec_id).cloned();
let response = if let Some(ctx) = context {
match validate_emitted_image_url(&req.image_url) {
Ok(()) => {
let content_item = emitted_image_content_item(
ctx.turn.as_ref(),
req.image_url,
req.detail,
);
JsReplManager::record_exec_content_item(
&exec_tool_calls,
&exec_id,
content_item,
)
.await;
HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: false,
error: Some(error),
}),
ok: true,
error: None,
})
}
} else {
HostToKernel::EmitImageResult(EmitImageResult {
Err(error) => HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: false,
error: Some("js_repl exec context not found".to_string()),
})
};
error: Some(error),
}),
}
} else {
HostToKernel::EmitImageResult(EmitImageResult {
id: emit_id,
ok: false,
error: Some("js_repl exec context not found".to_string()),
})
};
if let Err(err) = JsReplManager::write_message(&stdin, &response).await {
let snapshot =
@@ -1424,7 +1421,7 @@ impl JsReplManager {
let exec_id = req.exec_id.clone();
let tool_call_id = req.id.clone();
let tool_name = req.tool_name.clone();
let context = { exec_contexts.lock().await.get(&exec_id).cloned() };
let context = exec_contexts.lock().await.get(&exec_id).cloned();
let result = match context {
Some(ctx) => {
tokio::select! {
@@ -1502,14 +1499,19 @@ impl JsReplManager {
}
}
let mut pending = pending_execs.lock().await;
let pending_exec_ids = pending.keys().cloned().collect::<Vec<_>>();
for (_id, tx) in pending.drain() {
let pending_execs_to_notify = {
let mut pending = pending_execs.lock().await;
pending.drain().collect::<Vec<_>>()
};
let pending_exec_ids = pending_execs_to_notify
.iter()
.map(|(id, _)| id.clone())
.collect::<Vec<_>>();
for (_id, tx) in pending_execs_to_notify {
let _ = tx.send(ExecResultMessage::Err {
message: kernel_exit_message.clone(),
});
}
drop(pending);
if !pending_exec_ids.is_empty() {
Self::clear_top_level_exec_if_matches_any_map(&manager_kernel, &pending_exec_ids).await;
}
codex-rs/core/src/tools/network_approval.rs
+4 -8
View File
@@ -223,8 +223,7 @@ impl NetworkApprovalService {
}
pub(crate) async fn unregister_call(&self, registration_id: &str) {
let mut active_calls = self.active_calls.lock().await;
active_calls.shift_remove(registration_id);
self.active_calls.lock().await.shift_remove(registration_id);
let mut call_outcomes = self.call_outcomes.lock().await;
call_outcomes.remove(registration_id);
}
@@ -344,8 +343,7 @@ impl NetworkApprovalService {
let Some(turn_context) = Self::active_turn_context(session.as_ref()).await else {
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
self.pending_host_approvals.lock().await.remove(&key);
self.record_outcome_for_single_active_call(NetworkApprovalOutcome::DeniedByPolicy(
policy_denial_message,
))
@@ -354,8 +352,7 @@ impl NetworkApprovalService {
};
if !sandbox_policy_allows_network_approval_flow(turn_context.sandbox_policy.get()) {
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
self.pending_host_approvals.lock().await.remove(&key);
self.record_outcome_for_single_active_call(NetworkApprovalOutcome::DeniedByPolicy(
policy_denial_message,
))
@@ -364,8 +361,7 @@ impl NetworkApprovalService {
}
if !allows_network_approval_flow(turn_context.approval_policy.value()) {
pending.set_decision(PendingApprovalDecision::Deny).await;
let mut pending_approvals = self.pending_host_approvals.lock().await;
pending_approvals.remove(&key);
self.pending_host_approvals.lock().await.remove(&key);
self.record_outcome_for_single_active_call(NetworkApprovalOutcome::DeniedByPolicy(
policy_denial_message,
))
codex-rs/debug-client/src/client.rs
+1
View File
@@ -178,6 +178,7 @@ impl AppServerClient {
cursor,
limit: None,
sort_key: None,
sort_direction: None,
model_providers: None,
source_kinds: None,
archived: None,
codex-rs/exec/src/lib.rs
+2
View File
@@ -1234,6 +1234,7 @@ async fn resolve_resume_thread_id(
cursor,
limit: Some(100),
sort_key: Some(ThreadSortKey::UpdatedAt),
sort_direction: None,
model_providers: model_providers.clone(),
source_kinds: Some(all_thread_source_kinds()),
archived: Some(false),
@@ -1296,6 +1297,7 @@ async fn resolve_resume_thread_id(
cursor,
limit: Some(100),
sort_key: Some(ThreadSortKey::UpdatedAt),
sort_direction: None,
model_providers: model_providers.clone(),
source_kinds: Some(all_thread_source_kinds()),
archived: Some(false),
codex-rs/rmcp-client/src/bin/test_streamable_http_server.rs
+14 -13
View File
@@ -418,22 +418,23 @@ async fn fail_session_post_when_armed(
return next.run(request).await;
}
let mut armed_failure = state.armed_failure.lock().await;
if let Some(failure) = armed_failure.as_mut()
&& failure.remaining > 0
{
failure.remaining -= 1;
let status = failure.status;
if failure.remaining == 0 {
*armed_failure = None;
let mut armed_failure = state.armed_failure.lock().await;
if let Some(failure) = armed_failure.as_mut()
&& failure.remaining > 0
{
failure.remaining -= 1;
let status = failure.status;
if failure.remaining == 0 {
*armed_failure = None;
}
let mut response = Response::new(Body::from(format!(
"forced session failure with status {status}"
)));
*response.status_mut() = status;
return response;
}
let mut response = Response::new(Body::from(format!(
"forced session failure with status {status}"
)));
*response.status_mut() = status;
return response;
}
drop(armed_failure);
next.run(request).await
}
codex-rs/rollout/src/lib.rs
+1
View File
@@ -34,6 +34,7 @@ pub use config::Config;
pub use config::RolloutConfig;
pub use config::RolloutConfigView;
pub use list::Cursor;
pub use list::SortDirection;
pub use list::ThreadItem;
pub use list::ThreadListConfig;
pub use list::ThreadListLayout;
codex-rs/rollout/src/list.rs
+27 -25
View File
@@ -111,6 +111,12 @@ pub enum ThreadSortKey {
UpdatedAt,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SortDirection {
Asc,
Desc,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ThreadListLayout {
NestedByDate,
@@ -124,26 +130,28 @@ pub struct ThreadListConfig<'a> {
pub layout: ThreadListLayout,
}
/// Pagination cursor identifying a file by timestamp and UUID.
/// Pagination cursor identifying the timestamp of the last item in a page.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Cursor {
ts: OffsetDateTime,
id: Uuid,
}
impl Cursor {
fn new(ts: OffsetDateTime, id: Uuid) -> Self {
Self { ts, id }
fn new(ts: OffsetDateTime) -> Self {
Self { ts }
}
pub(crate) fn timestamp(&self) -> OffsetDateTime {
self.ts
}
}
/// Keeps track of where a paginated listing left off. As the file scan goes newest -> oldest,
/// it ignores everything until it reaches the last seen item from the previous page, then
/// it ignores everything until it passes the last seen timestamp from the previous page, then
/// starts returning results after that. This makes paging stable even if new files show up during
/// pagination.
struct AnchorState {
ts: OffsetDateTime,
id: Uuid,
passed: bool,
}
@@ -152,22 +160,20 @@ impl AnchorState {
match anchor {
Some(cursor) => Self {
ts: cursor.ts,
id: cursor.id,
passed: false,
},
None => Self {
ts: OffsetDateTime::UNIX_EPOCH,
id: Uuid::nil(),
passed: true,
},
}
}
fn should_skip(&mut self, ts: OffsetDateTime, id: Uuid) -> bool {
fn should_skip(&mut self, ts: OffsetDateTime, _id: Uuid) -> bool {
if self.passed {
return false;
}
if ts < self.ts || (ts == self.ts && id < self.id) {
if ts < self.ts {
self.passed = true;
false
} else {
@@ -275,7 +281,7 @@ impl serde::Serialize for Cursor {
.ts
.format(&Rfc3339)
.map_err(|e| serde::ser::Error::custom(format!("format error: {e}")))?;
serializer.serialize_str(&format!("{ts_str}|{}", self.id))
serializer.serialize_str(&ts_str)
}
}
@@ -296,14 +302,14 @@ impl From<codex_state::Anchor> for Cursor {
.timestamp_nanos_opt()
.and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(nanos as i128).ok())
.unwrap_or(OffsetDateTime::UNIX_EPOCH);
Self::new(ts, anchor.id)
Self::new(ts)
}
}
/// Retrieve recorded thread file paths with token pagination. The returned `next_cursor`
/// can be supplied on the next call to resume after the last returned item, resilient to
/// concurrent new sessions being appended. Ordering is stable by the requested sort key
/// (timestamp desc, then UUID desc).
/// (timestamp desc).
pub async fn get_threads(
codex_home: &Path,
page_size: usize,
@@ -657,31 +663,27 @@ async fn traverse_flat_paths_updated(
})
}
/// Pagination cursor token format: "<ts>|<uuid>" where `ts` uses
/// YYYY-MM-DDThh-mm-ss (UTC, second precision).
/// The cursor orders files by the requested sort key (timestamp desc, then UUID desc).
/// Pagination cursor token format: an RFC3339 timestamp.
pub fn parse_cursor(token: &str) -> Option<Cursor> {
let (file_ts, uuid_str) = token.split_once('|')?;
let Ok(uuid) = Uuid::parse_str(uuid_str) else {
if token.contains('|') {
return None;
};
}
let ts = OffsetDateTime::parse(file_ts, &Rfc3339).ok().or_else(|| {
let ts = OffsetDateTime::parse(token, &Rfc3339).ok().or_else(|| {
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
PrimitiveDateTime::parse(file_ts, format)
PrimitiveDateTime::parse(token, format)
.ok()
.map(PrimitiveDateTime::assume_utc)
})?;
Some(Cursor::new(ts, uuid))
Some(Cursor::new(ts))
}
fn build_next_cursor(items: &[ThreadItem], sort_key: ThreadSortKey) -> Option<Cursor> {
let last = items.last()?;
let file_name = last.path.file_name()?.to_string_lossy();
let (created_ts, id) = parse_timestamp_uuid_from_filename(&file_name)?;
let (created_ts, _id) = parse_timestamp_uuid_from_filename(&file_name)?;
let ts = match sort_key {
ThreadSortKey::CreatedAt => created_ts,
ThreadSortKey::UpdatedAt => {
@@ -689,7 +691,7 @@ fn build_next_cursor(items: &[ThreadItem], sort_key: ThreadSortKey) -> Option<Cu
OffsetDateTime::parse(updated_at, &Rfc3339).ok()?
}
};
Some(Cursor::new(ts, id))
Some(Cursor::new(ts))
}
async fn build_thread_item(
codex-rs/rollout/src/recorder.rs
+284 -28
View File
@@ -1,5 +1,6 @@
//! Persist Codex session rollouts (.jsonl) so sessions can be replayed or inspected later.
use std::collections::HashSet;
use std::fs;
use std::fs::File;
use std::io::Error as IoError;
@@ -32,6 +33,7 @@ use tracing::warn;
use super::ARCHIVED_SESSIONS_SUBDIR;
use super::SESSIONS_SUBDIR;
use super::list::Cursor;
use super::list::SortDirection;
use super::list::ThreadItem;
use super::list::ThreadListConfig;
use super::list::ThreadListLayout;
@@ -44,6 +46,7 @@ use super::list::parse_timestamp_uuid_from_filename;
use super::metadata;
use super::policy::EventPersistenceMode;
use super::policy::is_persisted_response_item;
use super::session_index::find_thread_names_by_ids;
use crate::config::RolloutConfigView;
use crate::default_client::originator;
use crate::state_db;
@@ -219,6 +222,7 @@ impl RolloutRecorder {
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
@@ -229,6 +233,7 @@ impl RolloutRecorder {
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
default_provider,
@@ -245,6 +250,7 @@ impl RolloutRecorder {
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
@@ -255,6 +261,7 @@ impl RolloutRecorder {
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
default_provider,
@@ -270,6 +277,7 @@ impl RolloutRecorder {
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
@@ -290,6 +298,7 @@ impl RolloutRecorder {
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
archived,
@@ -304,38 +313,44 @@ impl RolloutRecorder {
// Filesystem-first listing intentionally overfetches so we can repair stale/missing
// SQLite rollout paths before the final DB-backed page is returned.
let fs_page_size = page_size.saturating_mul(2).max(page_size);
let fs_page = if archived {
let root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR);
get_threads_in_root(
root,
fs_page_size,
cursor,
sort_key,
ThreadListConfig {
let fs_page = match sort_direction {
SortDirection::Asc => {
list_threads_from_files_asc(
codex_home,
page_size,
cursor,
sort_key,
allowed_sources,
model_providers,
default_provider,
layout: ThreadListLayout::Flat,
},
)
.await?
} else {
get_threads(
codex_home,
fs_page_size,
cursor,
sort_key,
allowed_sources,
model_providers,
default_provider,
)
.await?
archived,
search_term,
)
.await?
}
SortDirection::Desc => {
list_threads_from_files_desc(
codex_home,
fs_page_size,
cursor,
sort_key,
allowed_sources,
model_providers,
default_provider,
archived,
search_term,
)
.await?
}
};
if state_db_ctx.is_none() {
// Keep legacy behavior when SQLite is unavailable: return filesystem results
// at the requested page size.
return Ok(truncate_fs_page(fs_page, page_size, sort_key));
return Ok(match sort_direction {
SortDirection::Asc => fs_page,
SortDirection::Desc => truncate_fs_page(fs_page, page_size, sort_key),
});
}
// Warm the DB by repairing every filesystem hit before querying SQLite.
@@ -355,6 +370,7 @@ impl RolloutRecorder {
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
archived,
@@ -367,7 +383,10 @@ impl RolloutRecorder {
// If SQLite listing still fails, return the filesystem page rather than failing the list.
tracing::error!("Falling back on rollout system");
tracing::warn!("state db discrepancy during list_threads_with_db_fallback: falling_back");
Ok(truncate_fs_page(fs_page, page_size, sort_key))
Ok(match sort_direction {
SortDirection::Asc => fs_page,
SortDirection::Desc => truncate_fs_page(fs_page, page_size, sort_key),
})
}
/// Find the newest recorded thread path, optionally filtering to a matching cwd.
@@ -393,6 +412,7 @@ impl RolloutRecorder {
page_size,
db_cursor.as_ref(),
sort_key,
SortDirection::Desc,
allowed_sources,
model_providers,
/*archived*/ false,
@@ -766,16 +786,252 @@ fn truncate_fs_page(
page.items.truncate(page_size);
page.next_cursor = page.items.last().and_then(|item| {
let file_name = item.path.file_name()?.to_str()?;
let (created_at, id) = parse_timestamp_uuid_from_filename(file_name)?;
let (created_at, _id) = parse_timestamp_uuid_from_filename(file_name)?;
let cursor_token = match sort_key {
ThreadSortKey::CreatedAt => format!("{}|{id}", created_at.format(&Rfc3339).ok()?),
ThreadSortKey::UpdatedAt => format!("{}|{id}", item.updated_at.as_deref()?),
ThreadSortKey::CreatedAt => created_at.format(&Rfc3339).ok()?,
ThreadSortKey::UpdatedAt => item.updated_at.as_deref()?.to_string(),
};
parse_cursor(cursor_token.as_str())
});
page
}
#[allow(clippy::too_many_arguments)]
async fn list_threads_from_files_desc(
codex_home: &Path,
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
archived: bool,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
if let Some(search_term) = search_term {
let mut matching_items = Vec::new();
let mut scanned_files = 0usize;
let mut reached_scan_cap = false;
let mut page_cursor = cursor.cloned();
let scan_page_size = page_size.saturating_mul(8).clamp(256, 2048);
loop {
let mut page = list_threads_from_files_desc_unfiltered(
codex_home,
scan_page_size,
page_cursor.as_ref(),
sort_key,
allowed_sources,
model_providers,
default_provider,
archived,
)
.await?;
scanned_files = scanned_files.saturating_add(page.num_scanned_files);
reached_scan_cap |= page.reached_scan_cap;
filter_thread_items_by_search_term(codex_home, &mut page.items, Some(search_term))
.await?;
matching_items.extend(page.items);
page_cursor = page.next_cursor;
if matching_items.len() > page_size || page_cursor.is_none() {
break;
}
}
let more_matches_available =
matching_items.len() > page_size || page_cursor.is_some() || reached_scan_cap;
matching_items.truncate(page_size);
let next_cursor = if more_matches_available {
matching_items
.last()
.and_then(|item| cursor_from_thread_item(item, sort_key))
} else {
None
};
return Ok(ThreadsPage {
items: matching_items,
next_cursor,
num_scanned_files: scanned_files,
reached_scan_cap,
});
}
list_threads_from_files_desc_unfiltered(
codex_home,
page_size,
cursor,
sort_key,
allowed_sources,
model_providers,
default_provider,
archived,
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn list_threads_from_files_desc_unfiltered(
codex_home: &Path,
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
archived: bool,
) -> std::io::Result<ThreadsPage> {
if archived {
let root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR);
get_threads_in_root(
root,
page_size,
cursor,
sort_key,
ThreadListConfig {
allowed_sources,
model_providers,
default_provider,
layout: ThreadListLayout::Flat,
},
)
.await
} else {
get_threads(
codex_home,
page_size,
cursor,
sort_key,
allowed_sources,
model_providers,
default_provider,
)
.await
}
}
#[allow(clippy::too_many_arguments)]
async fn list_threads_from_files_asc(
codex_home: &Path,
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
default_provider: &str,
archived: bool,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
let mut all_items = Vec::new();
let mut scanned_files = 0usize;
let mut reached_scan_cap = false;
let mut page_cursor = None;
let scan_page_size = page_size.saturating_mul(8).clamp(256, 2048);
loop {
let page = list_threads_from_files_desc(
codex_home,
scan_page_size,
page_cursor.as_ref(),
sort_key,
allowed_sources,
model_providers,
default_provider,
archived,
/*search_term*/ None,
)
.await?;
scanned_files = scanned_files.saturating_add(page.num_scanned_files);
reached_scan_cap |= page.reached_scan_cap;
all_items.extend(page.items);
page_cursor = page.next_cursor;
if page_cursor.is_none() {
break;
}
}
filter_thread_items_by_search_term(codex_home, &mut all_items, search_term).await?;
let mut keyed_items = all_items
.into_iter()
.filter_map(|item| thread_item_sort_key(&item, sort_key).map(|key| (key, item)))
.collect::<Vec<_>>();
keyed_items.sort_by_key(|(key, _)| *key);
let mut all_items = keyed_items
.into_iter()
.map(|(_, item)| item)
.collect::<Vec<_>>();
if let Some(cursor) = cursor {
let anchor = cursor.timestamp();
all_items
.retain(|item| thread_item_sort_key(item, sort_key).is_some_and(|key| key.0 > anchor));
}
let more_matches_available = all_items.len() > page_size || reached_scan_cap;
all_items.truncate(page_size);
let next_cursor = if more_matches_available {
all_items
.last()
.and_then(|item| cursor_from_thread_item(item, sort_key))
} else {
None
};
Ok(ThreadsPage {
items: all_items,
next_cursor,
num_scanned_files: scanned_files,
reached_scan_cap,
})
}
async fn filter_thread_items_by_search_term(
codex_home: &Path,
items: &mut Vec<ThreadItem>,
search_term: Option<&str>,
) -> std::io::Result<()> {
let Some(search_term) = search_term else {
return Ok(());
};
// The file-backed fallback only has the thread title in the sidecar session index.
// Match the SQLite path's title substring filter so search pagination behaves the same
// whether the state DB is available or not.
let thread_ids = items
.iter()
.filter_map(|item| item.thread_id)
.collect::<HashSet<_>>();
let thread_names = find_thread_names_by_ids(codex_home, &thread_ids).await?;
items.retain(|item| {
item.thread_id
.and_then(|thread_id| thread_names.get(&thread_id))
.is_some_and(|title| title.contains(search_term))
});
Ok(())
}
fn thread_item_sort_key(
item: &ThreadItem,
sort_key: ThreadSortKey,
) -> Option<(OffsetDateTime, uuid::Uuid)> {
let file_name = item.path.file_name()?.to_str()?;
let (created_at, id) = parse_timestamp_uuid_from_filename(file_name)?;
let timestamp = match sort_key {
ThreadSortKey::CreatedAt => created_at,
ThreadSortKey::UpdatedAt => {
let updated_at = item.updated_at.as_deref().or(item.created_at.as_deref())?;
OffsetDateTime::parse(updated_at, &Rfc3339).ok()?
}
};
Some((timestamp, id))
}
fn cursor_from_thread_item(item: &ThreadItem, sort_key: ThreadSortKey) -> Option<Cursor> {
let (timestamp, _id) = thread_item_sort_key(item, sort_key)?;
let cursor_token = timestamp.format(&Rfc3339).ok()?;
parse_cursor(cursor_token.as_str())
}
struct LogFileInfo {
/// Full path to the rollout file.
path: PathBuf,
codex-rs/rollout/src/recorder_tests.rs
+4
View File
@@ -372,6 +372,7 @@ async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Re
/*page_size*/ 1,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
default_provider.as_str(),
@@ -387,6 +388,7 @@ async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Re
/*page_size*/ 1,
Some(&cursor),
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
default_provider.as_str(),
@@ -444,6 +446,7 @@ async fn list_threads_db_enabled_drops_missing_rollout_paths() -> std::io::Resul
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
default_provider.as_str(),
@@ -506,6 +509,7 @@ async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Resul
/*page_size*/ 1,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
default_provider.as_str(),
codex-rs/rollout/src/state_db.rs
+20 -25
View File
@@ -1,10 +1,10 @@
use crate::config::RolloutConfig;
use crate::config::RolloutConfigView;
use crate::list::Cursor;
use crate::list::SortDirection;
use crate::list::ThreadSortKey;
use crate::metadata;
use chrono::DateTime;
use chrono::NaiveDateTime;
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::dynamic_tools::DynamicToolSpec;
@@ -18,7 +18,6 @@ use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use tracing::warn;
use uuid::Uuid;
/// Core-facing handle to the SQLite-backed state runtime.
pub type StateDbHandle = Arc<codex_state::StateRuntime>;
@@ -117,21 +116,10 @@ async fn require_backfill_complete(
fn cursor_to_anchor(cursor: Option<&Cursor>) -> Option<codex_state::Anchor> {
let cursor = cursor?;
let value = serde_json::to_value(cursor).ok()?;
let cursor_str = value.as_str()?;
let (ts_str, id_str) = cursor_str.split_once('|')?;
if id_str.contains('|') {
return None;
}
let id = Uuid::parse_str(id_str).ok()?;
let ts = if let Ok(naive) = NaiveDateTime::parse_from_str(ts_str, "%Y-%m-%dT%H-%M-%S") {
DateTime::<Utc>::from_naive_utc_and_offset(naive, Utc)
} else if let Ok(dt) = DateTime::parse_from_rfc3339(ts_str) {
dt.with_timezone(&Utc)
} else {
return None;
};
Some(codex_state::Anchor { ts, id })
let millis = cursor.timestamp().unix_timestamp_nanos() / 1_000_000;
let millis = i64::try_from(millis).ok()?;
let ts = chrono::DateTime::<Utc>::from_timestamp_millis(millis)?;
Some(codex_state::Anchor { ts })
}
pub fn normalize_cwd_for_state_db(cwd: &Path) -> PathBuf {
@@ -200,6 +188,7 @@ pub async fn list_threads_db(
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
archived: bool,
@@ -227,15 +216,21 @@ pub async fn list_threads_db(
match ctx
.list_threads(
page_size,
anchor.as_ref(),
match sort_key {
ThreadSortKey::CreatedAt => codex_state::SortKey::CreatedAt,
ThreadSortKey::UpdatedAt => codex_state::SortKey::UpdatedAt,
codex_state::ThreadFilterOptions {
archived_only: archived,
allowed_sources: allowed_sources.as_slice(),
model_providers: model_providers.as_deref(),
anchor: anchor.as_ref(),
sort_key: match sort_key {
ThreadSortKey::CreatedAt => codex_state::SortKey::CreatedAt,
ThreadSortKey::UpdatedAt => codex_state::SortKey::UpdatedAt,
},
sort_direction: match sort_direction {
SortDirection::Asc => codex_state::SortDirection::Asc,
SortDirection::Desc => codex_state::SortDirection::Desc,
},
search_term,
},
allowed_sources.as_slice(),
model_providers.as_deref(),
archived,
search_term,
)
.await
{
codex-rs/rollout/src/state_db_tests.rs
+1 -5
View File
@@ -7,14 +7,11 @@ use chrono::NaiveDateTime;
use chrono::Timelike;
use chrono::Utc;
use pretty_assertions::assert_eq;
use uuid::Uuid;
#[test]
fn cursor_to_anchor_normalizes_timestamp_format() {
let uuid = Uuid::new_v4();
let ts_str = "2026-01-27T12-34-56";
let token = format!("{ts_str}|{uuid}");
let cursor = parse_cursor(token.as_str()).expect("cursor should parse");
let cursor = parse_cursor(ts_str).expect("cursor should parse");
let anchor = cursor_to_anchor(Some(&cursor)).expect("anchor should parse");
let naive =
@@ -23,6 +20,5 @@ fn cursor_to_anchor_normalizes_timestamp_format() {
.with_nanosecond(0)
.expect("nanosecond");
assert_eq!(anchor.id, uuid);
assert_eq!(anchor.ts, expected_ts);
}
codex-rs/rollout/src/tests.rs
+8 -31
View File
@@ -707,8 +707,7 @@ async fn test_pagination_cursor() {
.join(format!("rollout-2025-03-04T09-00-00-{u4}.jsonl"));
let updated_page1: Vec<Option<String>> =
page1.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_cursor1: Cursor =
serde_json::from_str(&format!("\"2025-03-04T09-00-00|{u4}\"")).unwrap();
let expected_cursor1: Cursor = serde_json::from_str("\"2025-03-04T09-00-00\"").unwrap();
let expected_page1 = ThreadsPage {
items: vec![
ThreadItem {
@@ -775,8 +774,7 @@ async fn test_pagination_cursor() {
.join(format!("rollout-2025-03-02T09-00-00-{u2}.jsonl"));
let updated_page2: Vec<Option<String>> =
page2.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_cursor2: Cursor =
serde_json::from_str(&format!("\"2025-03-02T09-00-00|{u2}\"")).unwrap();
let expected_cursor2: Cursor = serde_json::from_str("\"2025-03-02T09-00-00\"").unwrap();
let expected_page2 = ThreadsPage {
items: vec![
ThreadItem {
@@ -1207,7 +1205,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
}
#[tokio::test]
async fn test_stable_ordering_same_second_pagination() {
async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() {
let temp = TempDir::new().unwrap();
let home = temp.path();
@@ -1268,7 +1266,7 @@ async fn test_stable_ordering_same_second_pagination() {
.join(format!("rollout-2025-07-01T00-00-00-{u2}.jsonl"));
let updated_page1: Vec<Option<String>> =
page1.items.iter().map(|i| i.updated_at.clone()).collect();
let expected_cursor1: Cursor = serde_json::from_str(&format!("\"{ts}|{u2}\"")).unwrap();
let expected_cursor1: Cursor = serde_json::from_str(&format!("\"{ts}\"")).unwrap();
let expected_page1 = ThreadsPage {
items: vec![
ThreadItem {
@@ -1321,33 +1319,12 @@ async fn test_stable_ordering_same_second_pagination() {
)
.await
.unwrap();
let p1 = home
.join("sessions")
.join("2025")
.join("07")
.join("01")
.join(format!("rollout-2025-07-01T00-00-00-{u1}.jsonl"));
let updated_page2: Vec<Option<String>> =
page2.items.iter().map(|i| i.updated_at.clone()).collect();
// The filesystem fallback only has second-precision timestamps in filenames. The primary
// SQLite-backed listing uses unique millisecond timestamps and does not have this tie.
let expected_page2 = ThreadsPage {
items: vec![ThreadItem {
path: p1,
thread_id: Some(thread_id_from_uuid(u1)),
first_user_message: Some("Hello from user".to_string()),
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
git_origin_url: None,
source: Some(SessionSource::VSCode),
agent_nickname: None,
agent_role: None,
model_provider: Some(TEST_PROVIDER.to_string()),
cli_version: Some("test_version".to_string()),
created_at: Some(ts.to_string()),
updated_at: updated_page2.first().cloned().flatten(),
}],
items: Vec::new(),
next_cursor: None,
num_scanned_files: 3, // scanned u3, u2 (anchor), u1
num_scanned_files: 3,
reached_scan_cap: false,
};
assert_eq!(page2, expected_page2);
codex-rs/shell-escalation/src/unix/escalate_server.rs
+10 -10
View File
@@ -391,11 +391,11 @@ mod tests {
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use tempfile::TempDir;
use tokio::sync::Semaphore;
use tokio::time::Instant;
use tokio::time::sleep;
static ESCALATE_SERVER_TEST_LOCK: LazyLock<tokio::sync::Mutex<()>> =
LazyLock::new(|| tokio::sync::Mutex::new(()));
static ESCALATE_SERVER_TEST_LOCK: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(1));
struct DeterministicEscalationPolicy {
decision: EscalationDecision,
@@ -596,7 +596,7 @@ mod tests {
/// until `close_client_socket()` is called.
#[tokio::test]
async fn start_session_exposes_wrapper_env_overlay() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let execve_wrapper = PathBuf::from("/tmp/codex-execve-wrapper");
let execve_wrapper_str = execve_wrapper.to_string_lossy().to_string();
let server = EscalateServer::new(
@@ -638,7 +638,7 @@ mod tests {
#[tokio::test]
async fn exec_closes_parent_socket_after_shell_spawn() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let after_spawn_invoked = Arc::new(AtomicBool::new(false));
let server = EscalateServer::new(
PathBuf::from("/bin/bash"),
@@ -672,7 +672,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_respects_run_in_sandbox_decision() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let server_task = tokio::spawn(handle_escalate_session_with_policy(
server,
@@ -712,7 +712,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_resolves_relative_file_against_request_workdir()
-> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let tmp = tempfile::TempDir::new()?;
let workdir = tmp.path().join("workspace");
@@ -751,7 +751,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_executes_escalated_command() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let server_task = tokio::spawn(handle_escalate_session_with_policy(
server,
@@ -844,7 +844,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_accepts_received_fds_that_overlap_destinations()
-> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let mut pipe_fds = [0; 2];
if unsafe { libc::pipe(pipe_fds.as_mut_ptr()) } == -1 {
return Err(std::io::Error::last_os_error().into());
@@ -916,7 +916,7 @@ mod tests {
#[tokio::test]
async fn handle_escalate_session_passes_permissions_to_executor() -> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let (server, client) = AsyncSocket::pair()?;
let server_task = tokio::spawn(handle_escalate_session_with_policy(
server,
@@ -972,7 +972,7 @@ mod tests {
#[tokio::test]
async fn dropping_session_aborts_intercept_workers_and_kills_spawned_child()
-> anyhow::Result<()> {
let _guard = ESCALATE_SERVER_TEST_LOCK.lock().await;
let _guard = ESCALATE_SERVER_TEST_LOCK.acquire().await?;
let tmp = TempDir::new()?;
let pid_file = tmp.path().join("escalated-child.pid");
let pid_file_display = pid_file.display().to_string();
codex-rs/state/src/lib.rs
+2
View File
@@ -37,6 +37,7 @@ pub use model::BackfillStats;
pub use model::BackfillStatus;
pub use model::DirectionalThreadSpawnEdgeStatus;
pub use model::ExtractionOutcome;
pub use model::SortDirection;
pub use model::SortKey;
pub use model::Stage1JobClaim;
pub use model::Stage1JobClaimOutcome;
@@ -47,6 +48,7 @@ pub use model::ThreadMetadata;
pub use model::ThreadMetadataBuilder;
pub use model::ThreadsPage;
pub use runtime::RemoteControlEnrollmentRecord;
pub use runtime::ThreadFilterOptions;
pub use runtime::logs_db_filename;
pub use runtime::logs_db_path;
pub use runtime::state_db_filename;
codex-rs/state/src/model/mod.rs
+1
View File
@@ -28,6 +28,7 @@ pub use memories::Stage1StartupClaimParams;
pub use thread_metadata::Anchor;
pub use thread_metadata::BackfillStats;
pub use thread_metadata::ExtractionOutcome;
pub use thread_metadata::SortDirection;
pub use thread_metadata::SortKey;
pub use thread_metadata::ThreadMetadata;
pub use thread_metadata::ThreadMetadataBuilder;
codex-rs/state/src/model/thread_metadata.rs
+8 -5
View File
@@ -9,7 +9,6 @@ use codex_protocol::protocol::SessionSource;
use sqlx::Row;
use sqlx::sqlite::SqliteRow;
use std::path::PathBuf;
use uuid::Uuid;
/// The sort key to use when listing threads.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -20,13 +19,18 @@ pub enum SortKey {
UpdatedAt,
}
/// Sort direction to use when listing threads.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SortDirection {
Asc,
Desc,
}
/// A pagination anchor used for keyset pagination.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Anchor {
/// The timestamp component of the anchor.
pub ts: DateTime<Utc>,
/// The UUID component of the anchor.
pub id: Uuid,
}
/// A single page of thread metadata results.
@@ -414,12 +418,11 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
}
pub(crate) fn anchor_from_item(item: &ThreadMetadata, sort_key: SortKey) -> Option<Anchor> {
let id = Uuid::parse_str(&item.id.to_string()).ok()?;
let ts = match sort_key {
SortKey::CreatedAt => item.created_at,
SortKey::UpdatedAt => item.updated_at,
};
Some(Anchor { ts, id })
Some(Anchor { ts })
}
pub(crate) fn datetime_to_epoch_millis(dt: DateTime<Utc>) -> i64 {
codex-rs/state/src/runtime.rs
+1
View File
@@ -63,6 +63,7 @@ mod test_support;
mod threads;
pub use remote_control::RemoteControlEnrollmentRecord;
pub use threads::ThreadFilterOptions;
// "Partition" is the retained-log-content bucket we cap at 10 MiB:
// - one bucket per non-null thread_id
codex-rs/state/src/runtime/memories.rs
+18 -8
View File
@@ -1,6 +1,8 @@
use super::threads::ThreadFilterOptions;
use super::threads::push_thread_filters;
use super::threads::push_thread_order_and_limit;
use super::*;
use crate::SortDirection;
use crate::model::Phase2InputSelection;
use crate::model::Phase2JobClaimOutcome;
use crate::model::Stage1JobClaim;
@@ -101,7 +103,7 @@ WHERE thread_id = ?
/// - excludes the current thread id
/// - keeps only threads whose millisecond `updated_at` is in the age window
/// - keeps only threads whose memory is stale compared to millisecond `updated_at`
/// - orders by `updated_at_ms DESC, id DESC` and applies `scan_limit`
/// - orders by `updated_at_ms DESC` and applies `scan_limit`
///
/// For each selected thread, this function calls [`Self::try_claim_stage1_job`]
/// with `source_updated_at = thread.updated_at.timestamp()` and returns up to
@@ -169,12 +171,15 @@ LEFT JOIN jobs
);
push_thread_filters(
&mut builder,
/*archived_only*/ false,
allowed_sources,
/*model_providers*/ None,
/*anchor*/ None,
SortKey::UpdatedAt,
/*search_term*/ None,
ThreadFilterOptions {
archived_only: false,
allowed_sources,
model_providers: None,
anchor: None,
sort_key: SortKey::UpdatedAt,
sort_direction: SortDirection::Desc,
search_term: None,
},
);
builder.push(" AND threads.memory_mode = 'enabled'");
builder
@@ -195,7 +200,12 @@ LEFT JOIN jobs
builder.push(updated_at);
builder.push(" AND ((COALESCE(jobs.last_success_watermark, -1) + 1) * 1000) <= ");
builder.push(updated_at);
push_thread_order_and_limit(&mut builder, SortKey::UpdatedAt, scan_limit);
push_thread_order_and_limit(
&mut builder,
SortKey::UpdatedAt,
SortDirection::Desc,
scan_limit,
);
let items = builder
.build()
codex-rs/state/src/runtime/threads.rs
+150 -46
View File
@@ -1,4 +1,5 @@
use super::*;
use crate::SortDirection;
use codex_protocol::protocol::SessionSource;
use std::sync::atomic::Ordering;
@@ -342,12 +343,15 @@ ON CONFLICT(child_thread_id) DO NOTHING
builder.push(" FROM threads");
push_thread_filters(
&mut builder,
archived_only,
allowed_sources,
model_providers,
/*anchor*/ None,
crate::SortKey::UpdatedAt,
/*search_term*/ None,
ThreadFilterOptions {
archived_only,
allowed_sources,
model_providers,
anchor: None,
sort_key: crate::SortKey::UpdatedAt,
sort_direction: SortDirection::Desc,
search_term: None,
},
);
builder.push(" AND threads.title = ");
builder.push_bind(title);
@@ -355,7 +359,12 @@ ON CONFLICT(child_thread_id) DO NOTHING
builder.push(" AND threads.cwd = ");
builder.push_bind(cwd.display().to_string());
}
push_thread_order_and_limit(&mut builder, crate::SortKey::UpdatedAt, /*limit*/ 1);
push_thread_order_and_limit(
&mut builder,
crate::SortKey::UpdatedAt,
SortDirection::Desc,
/*limit*/ 1,
);
let row = builder.build().fetch_optional(self.pool.as_ref()).await?;
row.map(|row| ThreadRow::try_from_row(&row).and_then(crate::ThreadMetadata::try_from))
@@ -363,32 +372,20 @@ ON CONFLICT(child_thread_id) DO NOTHING
}
/// List threads using the underlying database.
#[allow(clippy::too_many_arguments)]
pub async fn list_threads(
&self,
page_size: usize,
anchor: Option<&crate::Anchor>,
sort_key: crate::SortKey,
allowed_sources: &[String],
model_providers: Option<&[String]>,
archived_only: bool,
search_term: Option<&str>,
filters: ThreadFilterOptions<'_>,
) -> anyhow::Result<crate::ThreadsPage> {
let limit = page_size.saturating_add(1);
let sort_key = filters.sort_key;
let sort_direction = filters.sort_direction;
let mut builder = QueryBuilder::<Sqlite>::new("");
push_thread_select_columns(&mut builder);
builder.push(" FROM threads");
push_thread_filters(
&mut builder,
archived_only,
allowed_sources,
model_providers,
anchor,
sort_key,
search_term,
);
push_thread_order_and_limit(&mut builder, sort_key, limit);
push_thread_filters(&mut builder, filters);
push_thread_order_and_limit(&mut builder, sort_key, sort_direction, limit);
let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
let mut items = rows
@@ -424,14 +421,17 @@ ON CONFLICT(child_thread_id) DO NOTHING
let mut builder = QueryBuilder::<Sqlite>::new("SELECT threads.id FROM threads");
push_thread_filters(
&mut builder,
archived_only,
allowed_sources,
model_providers,
anchor,
sort_key,
/*search_term*/ None,
ThreadFilterOptions {
archived_only,
allowed_sources,
model_providers,
anchor,
sort_key,
sort_direction: SortDirection::Desc,
search_term: None,
},
);
push_thread_order_and_limit(&mut builder, sort_key, limit);
push_thread_order_and_limit(&mut builder, sort_key, SortDirection::Desc, limit);
let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
rows.into_iter()
@@ -983,15 +983,30 @@ fn thread_spawn_parent_thread_id_from_source_str(source: &str) -> Option<ThreadI
}
}
#[derive(Clone, Copy)]
pub struct ThreadFilterOptions<'a> {
pub archived_only: bool,
pub allowed_sources: &'a [String],
pub model_providers: Option<&'a [String]>,
pub anchor: Option<&'a crate::Anchor>,
pub sort_key: SortKey,
pub sort_direction: SortDirection,
pub search_term: Option<&'a str>,
}
pub(super) fn push_thread_filters<'a>(
builder: &mut QueryBuilder<'a, Sqlite>,
archived_only: bool,
allowed_sources: &'a [String],
model_providers: Option<&'a [String]>,
anchor: Option<&crate::Anchor>,
sort_key: SortKey,
search_term: Option<&'a str>,
options: ThreadFilterOptions<'a>,
) {
let ThreadFilterOptions {
archived_only,
allowed_sources,
model_providers,
anchor,
sort_key,
sort_direction,
search_term,
} = options;
builder.push(" WHERE 1 = 1");
if archived_only {
builder.push(" AND threads.archived = 1");
@@ -1028,32 +1043,38 @@ pub(super) fn push_thread_filters<'a>(
SortKey::CreatedAt => "threads.created_at_ms",
SortKey::UpdatedAt => "threads.updated_at_ms",
};
let operator = match sort_direction {
SortDirection::Asc => ">",
SortDirection::Desc => "<",
};
builder.push(" AND (");
builder.push(column);
builder.push(" < ");
builder.push(" ");
builder.push(operator);
builder.push(" ");
builder.push_bind(anchor_ts);
builder.push(" OR (");
builder.push(column);
builder.push(" = ");
builder.push_bind(anchor_ts);
builder.push(" AND id < ");
builder.push_bind(anchor.id.to_string());
builder.push("))");
builder.push(")");
}
}
pub(super) fn push_thread_order_and_limit(
builder: &mut QueryBuilder<'_, Sqlite>,
sort_key: SortKey,
sort_direction: SortDirection,
limit: usize,
) {
let order_column = match sort_key {
SortKey::CreatedAt => "threads.created_at_ms",
SortKey::UpdatedAt => "threads.updated_at_ms",
};
let order_direction = match sort_direction {
SortDirection::Asc => "ASC",
SortDirection::Desc => "DESC",
};
builder.push(" ORDER BY ");
builder.push(order_column);
builder.push(" DESC, id DESC");
builder.push(" ");
builder.push(order_direction);
builder.push(" LIMIT ");
builder.push_bind(limit as i64);
}
@@ -1061,6 +1082,7 @@ pub(super) fn push_thread_order_and_limit(
#[cfg(test)]
mod tests {
use super::*;
use crate::Anchor;
use crate::DirectionalThreadSpawnEdgeStatus;
use crate::runtime::test_support::test_thread_metadata;
use crate::runtime::test_support::unique_temp_dir;
@@ -1110,6 +1132,88 @@ mod tests {
assert_eq!(memory_mode, "disabled");
}
#[tokio::test]
async fn list_threads_updated_after_returns_oldest_changes_first() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
.expect("state db should initialize");
let older_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000001").expect("valid thread id");
let middle_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000002").expect("valid thread id");
let newer_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000003").expect("valid thread id");
let older_updated_at =
DateTime::<Utc>::from_timestamp(1_700_000_100, 0).expect("valid older timestamp");
let newer_updated_at =
DateTime::<Utc>::from_timestamp(1_700_000_200, 0).expect("valid newer timestamp");
for (thread_id, updated_at) in [
(older_id, older_updated_at),
(newer_id, newer_updated_at),
(middle_id, newer_updated_at),
] {
let mut metadata = test_thread_metadata(&codex_home, thread_id, codex_home.clone());
metadata.updated_at = updated_at;
metadata.first_user_message = Some("hello".to_string());
runtime
.upsert_thread(&metadata)
.await
.expect("thread insert should succeed");
}
let anchor = Anchor {
ts: older_updated_at,
};
let model_providers = ["test-provider".to_string()];
let page = runtime
.list_threads(
/*page_size*/ 1,
ThreadFilterOptions {
archived_only: false,
allowed_sources: &[],
model_providers: Some(&model_providers),
anchor: Some(&anchor),
sort_key: SortKey::UpdatedAt,
sort_direction: SortDirection::Asc,
search_term: None,
},
)
.await
.expect("list should succeed");
let ids = page.items.iter().map(|item| item.id).collect::<Vec<_>>();
assert_eq!(ids, vec![newer_id]);
assert_eq!(
page.next_anchor,
Some(Anchor {
ts: DateTime::<Utc>::from_timestamp_millis(1_700_000_200_000)
.expect("valid timestamp"),
})
);
let page = runtime
.list_threads(
/*page_size*/ 1,
ThreadFilterOptions {
archived_only: false,
allowed_sources: &[],
model_providers: Some(&model_providers),
anchor: page.next_anchor.as_ref(),
sort_key: SortKey::UpdatedAt,
sort_direction: SortDirection::Asc,
search_term: None,
},
)
.await
.expect("second page should succeed");
let ids = page.items.iter().map(|item| item.id).collect::<Vec<_>>();
assert_eq!(ids, vec![middle_id]);
assert_eq!(page.next_anchor, None);
}
#[tokio::test]
async fn apply_rollout_items_restores_memory_mode_from_session_meta() {
let codex_home = unique_temp_dir();
codex-rs/thread-store/src/lib.rs
+1
View File
@@ -27,6 +27,7 @@ pub use types::OptionalStringPatch;
pub use types::ReadThreadParams;
pub use types::ResumeThreadRecorderParams;
pub use types::SetThreadNameParams;
pub use types::SortDirection;
pub use types::StoredThread;
pub use types::StoredThreadHistory;
pub use types::ThreadEventPersistenceMode;
codex-rs/thread-store/src/local/archive_thread.rs
+1
View File
@@ -100,6 +100,7 @@ mod tests {
page_size: 10,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: crate::SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: true,
codex-rs/thread-store/src/local/list_threads.rs
+22 -1
View File
@@ -5,6 +5,7 @@ use codex_rollout::parse_cursor;
use super::LocalThreadStore;
use super::helpers::stored_thread_from_rollout_item;
use crate::ListThreadsParams;
use crate::SortDirection;
use crate::ThreadPage;
use crate::ThreadSortKey;
use crate::ThreadStoreError;
@@ -27,7 +28,18 @@ pub(super) async fn list_threads(
ThreadSortKey::CreatedAt => codex_rollout::ThreadSortKey::CreatedAt,
ThreadSortKey::UpdatedAt => codex_rollout::ThreadSortKey::UpdatedAt,
};
let page = list_rollout_threads(&store.config, &params, cursor.as_ref(), sort_key).await?;
let sort_direction = match params.sort_direction {
SortDirection::Asc => codex_rollout::SortDirection::Asc,
SortDirection::Desc => codex_rollout::SortDirection::Desc,
};
let page = list_rollout_threads(
&store.config,
&params,
cursor.as_ref(),
sort_key,
sort_direction,
)
.await?;
let next_cursor = page
.next_cursor
@@ -54,6 +66,7 @@ async fn list_rollout_threads(
params: &ListThreadsParams,
cursor: Option<&codex_rollout::Cursor>,
sort_key: codex_rollout::ThreadSortKey,
sort_direction: codex_rollout::SortDirection,
) -> ThreadStoreResult<codex_rollout::ThreadsPage> {
let page = if params.archived {
RolloutRecorder::list_archived_threads(
@@ -61,6 +74,7 @@ async fn list_rollout_threads(
params.page_size,
cursor,
sort_key,
sort_direction,
params.allowed_sources.as_slice(),
params.model_providers.as_deref(),
config.model_provider_id.as_str(),
@@ -73,6 +87,7 @@ async fn list_rollout_threads(
params.page_size,
cursor,
sort_key,
sort_direction,
params.allowed_sources.as_slice(),
params.model_providers.as_deref(),
config.model_provider_id.as_str(),
@@ -122,6 +137,7 @@ mod tests {
page_size: 10,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: false,
@@ -177,6 +193,7 @@ mod tests {
page_size: 10,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: false,
@@ -213,6 +230,7 @@ mod tests {
page_size: 10,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: false,
@@ -225,6 +243,7 @@ mod tests {
page_size: 10,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: true,
@@ -273,6 +292,7 @@ mod tests {
page_size: 10,
cursor: None,
sort_key: ThreadSortKey::CreatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: vec![SessionSource::Cli],
model_providers: Some(vec!["test-provider".to_string()]),
archived: false,
@@ -306,6 +326,7 @@ mod tests {
page_size: 10,
cursor: Some("not-a-cursor".to_string()),
sort_key: ThreadSortKey::CreatedAt,
sort_direction: SortDirection::Desc,
allowed_sources: Vec::new(),
model_providers: None,
archived: false,
codex-rs/thread-store/src/remote/list_threads.rs
+1
View File
@@ -161,6 +161,7 @@ mod tests {
page_size: 2,
cursor: Some("cursor-1".to_string()),
sort_key: ThreadSortKey::UpdatedAt,
sort_direction: crate::SortDirection::Desc,
allowed_sources: vec![SessionSource::Cli],
model_providers: Some(vec!["openai".to_string()]),
archived: true,
codex-rs/thread-store/src/types.rs
+12
View File
@@ -101,6 +101,16 @@ pub enum ThreadSortKey {
UpdatedAt,
}
/// The direction to use when listing stored threads.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum SortDirection {
/// Return older threads before newer threads.
Asc,
/// Return newer threads before older threads.
#[default]
Desc,
}
/// Parameters for listing threads.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ListThreadsParams {
@@ -110,6 +120,8 @@ pub struct ListThreadsParams {
pub cursor: Option<String>,
/// Sort order requested by the caller.
pub sort_key: ThreadSortKey,
/// Sort direction requested by the caller.
pub sort_direction: SortDirection,
/// Allowed session sources. Empty means implementation default.
pub allowed_sources: Vec<SessionSource>,
/// Optional model provider filter. `None` means implementation default, while an empty vector
codex-rs/tui/src/lib.rs
+2
View File
@@ -517,6 +517,7 @@ async fn lookup_session_target_by_name_with_app_server(
cursor: cursor.clone(),
limit: Some(100),
sort_key: Some(AppServerThreadSortKey::UpdatedAt),
sort_direction: None,
model_providers: None,
source_kinds: Some(vec![ThreadSourceKind::Cli, ThreadSourceKind::VsCode]),
archived: Some(false),
@@ -612,6 +613,7 @@ fn latest_session_lookup_params(
cursor: None,
limit: Some(1),
sort_key: Some(AppServerThreadSortKey::UpdatedAt),
sort_direction: None,
model_providers: if is_remote {
None
} else {
codex-rs/tui/src/resume_picker.rs
+8 -18
View File
@@ -336,6 +336,7 @@ fn spawn_rollout_page_loader(
PAGE_SIZE,
cursor,
request.sort_key,
codex_rollout::SortDirection::Desc,
INTERACTIVE_SESSION_SOURCES.as_slice(),
default_provider.as_ref().map(std::slice::from_ref),
default_provider.as_deref().unwrap_or_default(),
@@ -1179,6 +1180,7 @@ fn thread_list_params(
ThreadSortKey::CreatedAt => AppServerThreadSortKey::CreatedAt,
ThreadSortKey::UpdatedAt => AppServerThreadSortKey::UpdatedAt,
}),
sort_direction: None,
model_providers: match provider_filter {
ProviderFilter::Any => None,
ProviderFilter::MatchDefault(default_provider) => Some(vec![default_provider]),
@@ -2420,9 +2422,7 @@ mod tests {
make_item("/tmp/a.jsonl", "2025-01-03T00:00:00Z", "third"),
make_item("/tmp/b.jsonl", "2025-01-02T00:00:00Z", "second"),
],
Some(cursor_from_str(
"2025-01-02T00-00-00|00000000-0000-0000-0000-000000000000",
)),
Some(cursor_from_str("2025-01-02T00:00:00Z")),
/*num_scanned_files*/ 2,
/*reached_scan_cap*/ false,
));
@@ -2432,9 +2432,7 @@ mod tests {
make_item("/tmp/a.jsonl", "2025-01-03T00:00:00Z", "duplicate"),
make_item("/tmp/c.jsonl", "2025-01-01T00:00:00Z", "first"),
],
Some(cursor_from_str(
"2025-01-01T00-00-00|00000000-0000-0000-0000-000000000001",
)),
Some(cursor_from_str("2025-01-01T00:00:00Z")),
/*num_scanned_files*/ 2,
/*reached_scan_cap*/ false,
));
@@ -2488,9 +2486,7 @@ mod tests {
make_item("/tmp/a.jsonl", "2025-01-01T00:00:00Z", "one"),
make_item("/tmp/b.jsonl", "2025-01-02T00:00:00Z", "two"),
],
Some(cursor_from_str(
"2025-01-03T00-00-00|00000000-0000-0000-0000-000000000000",
)),
Some(cursor_from_str("2025-01-03T00:00:00Z")),
/*num_scanned_files*/ 2,
/*reached_scan_cap*/ false,
));
@@ -2806,9 +2802,7 @@ mod tests {
"2025-01-01T00:00:00Z",
"alpha",
)],
Some(cursor_from_str(
"2025-01-02T00-00-00|00000000-0000-0000-0000-000000000000",
)),
Some(cursor_from_str("2025-01-02T00:00:00Z")),
/*num_scanned_files*/ 1,
/*reached_scan_cap*/ false,
));
@@ -2827,9 +2821,7 @@ mod tests {
search_token: first_request.search_token,
page: Ok(page(
vec![make_item("/tmp/beta.jsonl", "2025-01-02T00:00:00Z", "beta")],
Some(cursor_from_str(
"2025-01-03T00-00-00|00000000-0000-0000-0000-000000000001",
)),
Some(cursor_from_str("2025-01-03T00:00:00Z")),
/*num_scanned_files*/ 5,
/*reached_scan_cap*/ false,
)),
@@ -2855,9 +2847,7 @@ mod tests {
"2025-01-03T00:00:00Z",
"target log",
)],
Some(cursor_from_str(
"2025-01-04T00-00-00|00000000-0000-0000-0000-000000000002",
)),
Some(cursor_from_str("2025-01-04T00:00:00Z")),
/*num_scanned_files*/ 7,
/*reached_scan_cap*/ false,
)),
codex-rs/utils/pty/src/pipe.rs
+3 -4
View File
@@ -162,12 +162,11 @@ async fn spawn_process_with_stdin_mode(
let (stdout_tx, stdout_rx) = mpsc::channel::<Vec<u8>>(128);
let (stderr_tx, stderr_rx) = mpsc::channel::<Vec<u8>>(128);
let writer_handle = if let Some(stdin) = stdin {
let writer = Arc::new(tokio::sync::Mutex::new(stdin));
tokio::spawn(async move {
let mut writer = stdin;
while let Some(bytes) = writer_rx.recv().await {
let mut guard = writer.lock().await;
let _ = guard.write_all(&bytes).await;
let _ = guard.flush().await;
let _ = writer.write_all(&bytes).await;
let _ = writer.flush().await;
}
})
} else {
codex-rs/utils/readiness/src/lib.rs
+24 -5
View File
@@ -277,17 +277,36 @@ mod tests {
#[tokio::test]
async fn subscribe_returns_error_when_lock_is_held() {
let flag = ReadinessFlag::new();
let _guard = flag
.tokens
.try_lock()
.expect("initial lock acquisition should succeed");
let flag = Arc::new(ReadinessFlag::new());
let (locked_tx, locked_rx) = std::sync::mpsc::channel();
let (release_tx, release_rx) = std::sync::mpsc::channel();
let lock_thread = {
let flag = Arc::clone(&flag);
std::thread::spawn(move || {
let _guard = flag.tokens.blocking_lock();
locked_tx
.send(())
.expect("test should receive lock acquisition notification");
release_rx
.recv()
.expect("test should release held readiness lock");
})
};
locked_rx
.recv()
.expect("test should observe held readiness lock");
let err = flag
.subscribe()
.await
.expect_err("contended subscribe should report a lock failure");
assert_matches!(err, ReadinessError::TokenLockFailed);
release_tx
.send(())
.expect("test should release readiness lock thread");
lock_thread
.join()
.expect("readiness lock thread should not panic");
}
#[tokio::test]