From fcfb6ecf5e769b494a34c22bd46e10b86159de9a Mon Sep 17 00:00:00 2001 From: Brent Traut Date: Thu, 11 Jun 2026 18:18:34 -0700 Subject: [PATCH] [codex] Add thread catalog subscription API --- .../schema/json/ClientRequest.json | 46 +++ .../schema/json/ServerNotification.json | 143 +++++++++ .../codex_app_server_protocol.schemas.json | 201 +++++++++++++ .../codex_app_server_protocol.v2.schemas.json | 201 +++++++++++++ .../v2/ThreadCatalogChangedNotification.json | 284 ++++++++++++++++++ .../v2/ThreadCatalogSubscribeResponse.json | 5 + .../v2/ThreadCatalogUnsubscribeResponse.json | 5 + .../schema/typescript/ServerNotification.ts | 1 + .../v2/ThreadCatalogChangedNotification.ts | 6 + .../v2/ThreadCatalogSubscribeResponse.ts | 5 + .../v2/ThreadCatalogUnsubscribeResponse.ts | 5 + .../schema/typescript/v2/ThreadSummary.ts | 9 + .../schema/typescript/v2/index.ts | 4 + .../src/protocol/common.rs | 11 + .../src/protocol/v2/thread.rs | 18 ++ .../src/protocol/v2/thread_data.rs | 28 ++ codex-rs/app-server/README.md | 12 + codex-rs/app-server/src/message_processor.rs | 15 + codex-rs/app-server/src/request_processors.rs | 6 + .../thread_catalog_subscription.rs | 84 ++++++ .../request_processors/thread_lifecycle.rs | 27 ++ .../request_processors/thread_processor.rs | 77 ++++- .../src/request_processors/thread_summary.rs | 30 ++ .../src/request_processors/turn_processor.rs | 9 + codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../suite/v2/thread_catalog_subscription.rs | 125 ++++++++ 26 files changed, 1350 insertions(+), 8 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogSubscribeResponse.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogUnsubscribeResponse.json create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogChangedNotification.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogSubscribeResponse.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogUnsubscribeResponse.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts create mode 100644 codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs create mode 100644 codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 68c242504f..4f414c1b37 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -5197,6 +5197,52 @@ "title": "Thread/listRequest", "type": "object" }, + { + "properties": { + "id": { + "$ref": "#/definitions/RequestId" + }, + "method": { + "enum": [ + "thread/catalog/subscribe" + ], + "title": "Thread/catalog/subscribeRequestMethod", + "type": "string" + }, + "params": { + "type": "null" + } + }, + "required": [ + "id", + "method" + ], + "title": "Thread/catalog/subscribeRequest", + "type": "object" + }, + { + "properties": { + "id": { + "$ref": "#/definitions/RequestId" + }, + "method": { + "enum": [ + "thread/catalog/unsubscribe" + ], + "title": "Thread/catalog/unsubscribeRequestMethod", + "type": "string" + }, + "params": { + "type": "null" + } + }, + "required": [ + "id", + "method" + ], + "title": "Thread/catalog/unsubscribeRequest", + "type": "object" + }, { "properties": { "id": { diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 31fe2f31c7..c504ce13b3 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -3744,6 +3744,17 @@ ], "type": "object" }, + "ThreadCatalogChangedNotification": { + "properties": { + "thread": { + "$ref": "#/definitions/ThreadSummary" + } + }, + "required": [ + "thread" + ], + "type": "object" + }, "ThreadClosedNotification": { "properties": { "threadId": { @@ -5002,6 +5013,118 @@ ], "type": "object" }, + "ThreadSummary": { + "properties": { + "agentNickname": { + "type": [ + "string", + "null" + ] + }, + "agentRole": { + "type": [ + "string", + "null" + ] + }, + "archivedAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "cliVersion": { + "type": "string" + }, + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cwd": { + "$ref": "#/definitions/AbsolutePathBuf" + }, + "ephemeral": { + "type": "boolean" + }, + "forkedFromId": { + "type": [ + "string", + "null" + ] + }, + "gitInfo": { + "anyOf": [ + { + "$ref": "#/definitions/GitInfo" + }, + { + "type": "null" + } + ] + }, + "id": { + "type": "string" + }, + "modelProvider": { + "type": "string" + }, + "name": { + "type": [ + "string", + "null" + ] + }, + "parentThreadId": { + "type": [ + "string", + "null" + ] + }, + "path": { + "type": [ + "string", + "null" + ] + }, + "preview": { + "type": "string" + }, + "sessionId": { + "type": "string" + }, + "source": { + "$ref": "#/definitions/SessionSource" + }, + "threadSource": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadSource" + }, + { + "type": "null" + } + ] + }, + "updatedAt": { + "format": "int64", + "type": "integer" + } + }, + "required": [ + "cliVersion", + "createdAt", + "cwd", + "ephemeral", + "id", + "modelProvider", + "preview", + "sessionId", + "source", + "updatedAt" + ], + "type": "object" + }, "ThreadTokenUsage": { "properties": { "last": { @@ -5712,6 +5835,26 @@ "title": "Thread/status/changedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/catalog/changed" + ], + "title": "Thread/catalog/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadCatalogChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/catalog/changedNotification", + "type": "object" + }, { "properties": { "method": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 4bbf58a12e..918d6117dd 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -636,6 +636,52 @@ "title": "Thread/listRequest", "type": "object" }, + { + "properties": { + "id": { + "$ref": "#/definitions/v2/RequestId" + }, + "method": { + "enum": [ + "thread/catalog/subscribe" + ], + "title": "Thread/catalog/subscribeRequestMethod", + "type": "string" + }, + "params": { + "type": "null" + } + }, + "required": [ + "id", + "method" + ], + "title": "Thread/catalog/subscribeRequest", + "type": "object" + }, + { + "properties": { + "id": { + "$ref": "#/definitions/v2/RequestId" + }, + "method": { + "enum": [ + "thread/catalog/unsubscribe" + ], + "title": "Thread/catalog/unsubscribeRequestMethod", + "type": "string" + }, + "params": { + "type": "null" + } + }, + "required": [ + "id", + "method" + ], + "title": "Thread/catalog/unsubscribeRequest", + "type": "object" + }, { "properties": { "id": { @@ -4168,6 +4214,26 @@ "title": "Thread/status/changedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/catalog/changed" + ], + "title": "Thread/catalog/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadCatalogChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/catalog/changedNotification", + "type": "object" + }, { "properties": { "method": { @@ -17010,6 +17076,29 @@ "title": "ThreadArchivedNotification", "type": "object" }, + "ThreadCatalogChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "thread": { + "$ref": "#/definitions/v2/ThreadSummary" + } + }, + "required": [ + "thread" + ], + "title": "ThreadCatalogChangedNotification", + "type": "object" + }, + "ThreadCatalogSubscribeResponse": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadCatalogSubscribeResponse", + "type": "object" + }, + "ThreadCatalogUnsubscribeResponse": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadCatalogUnsubscribeResponse", + "type": "object" + }, "ThreadClosedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { @@ -19494,6 +19583,118 @@ "title": "ThreadStatusChangedNotification", "type": "object" }, + "ThreadSummary": { + "properties": { + "agentNickname": { + "type": [ + "string", + "null" + ] + }, + "agentRole": { + "type": [ + "string", + "null" + ] + }, + "archivedAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "cliVersion": { + "type": "string" + }, + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cwd": { + "$ref": "#/definitions/v2/AbsolutePathBuf" + }, + "ephemeral": { + "type": "boolean" + }, + "forkedFromId": { + "type": [ + "string", + "null" + ] + }, + "gitInfo": { + "anyOf": [ + { + "$ref": "#/definitions/v2/GitInfo" + }, + { + "type": "null" + } + ] + }, + "id": { + "type": "string" + }, + "modelProvider": { + "type": "string" + }, + "name": { + "type": [ + "string", + "null" + ] + }, + "parentThreadId": { + "type": [ + "string", + "null" + ] + }, + "path": { + "type": [ + "string", + "null" + ] + }, + "preview": { + "type": "string" + }, + "sessionId": { + "type": "string" + }, + "source": { + "$ref": "#/definitions/v2/SessionSource" + }, + "threadSource": { + "anyOf": [ + { + "$ref": "#/definitions/v2/ThreadSource" + }, + { + "type": "null" + } + ] + }, + "updatedAt": { + "format": "int64", + "type": "integer" + } + }, + "required": [ + "cliVersion", + "createdAt", + "cwd", + "ephemeral", + "id", + "modelProvider", + "preview", + "sessionId", + "source", + "updatedAt" + ], + "type": "object" + }, "ThreadTokenUsage": { "properties": { "last": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 9cae10233e..25d63874df 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -1648,6 +1648,52 @@ "title": "Thread/listRequest", "type": "object" }, + { + "properties": { + "id": { + "$ref": "#/definitions/RequestId" + }, + "method": { + "enum": [ + "thread/catalog/subscribe" + ], + "title": "Thread/catalog/subscribeRequestMethod", + "type": "string" + }, + "params": { + "type": "null" + } + }, + "required": [ + "id", + "method" + ], + "title": "Thread/catalog/subscribeRequest", + "type": "object" + }, + { + "properties": { + "id": { + "$ref": "#/definitions/RequestId" + }, + "method": { + "enum": [ + "thread/catalog/unsubscribe" + ], + "title": "Thread/catalog/unsubscribeRequestMethod", + "type": "string" + }, + "params": { + "type": "null" + } + }, + "required": [ + "id", + "method" + ], + "title": "Thread/catalog/unsubscribeRequest", + "type": "object" + }, { "properties": { "id": { @@ -12576,6 +12622,26 @@ "title": "Thread/status/changedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/catalog/changed" + ], + "title": "Thread/catalog/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadCatalogChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/catalog/changedNotification", + "type": "object" + }, { "properties": { "method": { @@ -14789,6 +14855,29 @@ "title": "ThreadArchivedNotification", "type": "object" }, + "ThreadCatalogChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "thread": { + "$ref": "#/definitions/ThreadSummary" + } + }, + "required": [ + "thread" + ], + "title": "ThreadCatalogChangedNotification", + "type": "object" + }, + "ThreadCatalogSubscribeResponse": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadCatalogSubscribeResponse", + "type": "object" + }, + "ThreadCatalogUnsubscribeResponse": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadCatalogUnsubscribeResponse", + "type": "object" + }, "ThreadClosedNotification": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { @@ -17273,6 +17362,118 @@ "title": "ThreadStatusChangedNotification", "type": "object" }, + "ThreadSummary": { + "properties": { + "agentNickname": { + "type": [ + "string", + "null" + ] + }, + "agentRole": { + "type": [ + "string", + "null" + ] + }, + "archivedAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "cliVersion": { + "type": "string" + }, + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cwd": { + "$ref": "#/definitions/AbsolutePathBuf" + }, + "ephemeral": { + "type": "boolean" + }, + "forkedFromId": { + "type": [ + "string", + "null" + ] + }, + "gitInfo": { + "anyOf": [ + { + "$ref": "#/definitions/GitInfo" + }, + { + "type": "null" + } + ] + }, + "id": { + "type": "string" + }, + "modelProvider": { + "type": "string" + }, + "name": { + "type": [ + "string", + "null" + ] + }, + "parentThreadId": { + "type": [ + "string", + "null" + ] + }, + "path": { + "type": [ + "string", + "null" + ] + }, + "preview": { + "type": "string" + }, + "sessionId": { + "type": "string" + }, + "source": { + "$ref": "#/definitions/SessionSource" + }, + "threadSource": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadSource" + }, + { + "type": "null" + } + ] + }, + "updatedAt": { + "format": "int64", + "type": "integer" + } + }, + "required": [ + "cliVersion", + "createdAt", + "cwd", + "ephemeral", + "id", + "modelProvider", + "preview", + "sessionId", + "source", + "updatedAt" + ], + "type": "object" + }, "ThreadTokenUsage": { "properties": { "last": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json new file mode 100644 index 0000000000..c9ba34a7b2 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json @@ -0,0 +1,284 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "AbsolutePathBuf": { + "description": "A path that is guaranteed to be absolute and normalized (though it is not guaranteed to be canonicalized or exist on the filesystem).\n\nIMPORTANT: When deserializing an `AbsolutePathBuf`, a base path must be set using [AbsolutePathBufGuard::new]. If no base path is set, the deserialization will fail unless the path being deserialized is already absolute.", + "type": "string" + }, + "AgentPath": { + "type": "string" + }, + "GitInfo": { + "properties": { + "branch": { + "type": [ + "string", + "null" + ] + }, + "originUrl": { + "type": [ + "string", + "null" + ] + }, + "sha": { + "type": [ + "string", + "null" + ] + } + }, + "type": "object" + }, + "SessionSource": { + "oneOf": [ + { + "enum": [ + "cli", + "vscode", + "exec", + "appServer", + "unknown" + ], + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "custom": { + "type": "string" + } + }, + "required": [ + "custom" + ], + "title": "CustomSessionSource", + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "subAgent": { + "$ref": "#/definitions/SubAgentSource" + } + }, + "required": [ + "subAgent" + ], + "title": "SubAgentSessionSource", + "type": "object" + } + ] + }, + "SubAgentSource": { + "oneOf": [ + { + "enum": [ + "review", + "compact", + "memory_consolidation" + ], + "type": "string" + }, + { + "additionalProperties": false, + "properties": { + "thread_spawn": { + "properties": { + "agent_nickname": { + "default": null, + "type": [ + "string", + "null" + ] + }, + "agent_path": { + "anyOf": [ + { + "$ref": "#/definitions/AgentPath" + }, + { + "type": "null" + } + ], + "default": null + }, + "agent_role": { + "default": null, + "type": [ + "string", + "null" + ] + }, + "depth": { + "format": "int32", + "type": "integer" + }, + "parent_thread_id": { + "$ref": "#/definitions/ThreadId" + } + }, + "required": [ + "depth", + "parent_thread_id" + ], + "type": "object" + } + }, + "required": [ + "thread_spawn" + ], + "title": "ThreadSpawnSubAgentSource", + "type": "object" + }, + { + "additionalProperties": false, + "properties": { + "other": { + "type": "string" + } + }, + "required": [ + "other" + ], + "title": "OtherSubAgentSource", + "type": "object" + } + ] + }, + "ThreadId": { + "type": "string" + }, + "ThreadSource": { + "enum": [ + "user", + "subagent", + "memory_consolidation" + ], + "type": "string" + }, + "ThreadSummary": { + "properties": { + "agentNickname": { + "type": [ + "string", + "null" + ] + }, + "agentRole": { + "type": [ + "string", + "null" + ] + }, + "archivedAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "cliVersion": { + "type": "string" + }, + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cwd": { + "$ref": "#/definitions/AbsolutePathBuf" + }, + "ephemeral": { + "type": "boolean" + }, + "forkedFromId": { + "type": [ + "string", + "null" + ] + }, + "gitInfo": { + "anyOf": [ + { + "$ref": "#/definitions/GitInfo" + }, + { + "type": "null" + } + ] + }, + "id": { + "type": "string" + }, + "modelProvider": { + "type": "string" + }, + "name": { + "type": [ + "string", + "null" + ] + }, + "parentThreadId": { + "type": [ + "string", + "null" + ] + }, + "path": { + "type": [ + "string", + "null" + ] + }, + "preview": { + "type": "string" + }, + "sessionId": { + "type": "string" + }, + "source": { + "$ref": "#/definitions/SessionSource" + }, + "threadSource": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadSource" + }, + { + "type": "null" + } + ] + }, + "updatedAt": { + "format": "int64", + "type": "integer" + } + }, + "required": [ + "cliVersion", + "createdAt", + "cwd", + "ephemeral", + "id", + "modelProvider", + "preview", + "sessionId", + "source", + "updatedAt" + ], + "type": "object" + } + }, + "properties": { + "thread": { + "$ref": "#/definitions/ThreadSummary" + } + }, + "required": [ + "thread" + ], + "title": "ThreadCatalogChangedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogSubscribeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogSubscribeResponse.json new file mode 100644 index 0000000000..35205001bd --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogSubscribeResponse.json @@ -0,0 +1,5 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadCatalogSubscribeResponse", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogUnsubscribeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogUnsubscribeResponse.json new file mode 100644 index 0000000000..76fbdbe930 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogUnsubscribeResponse.json @@ -0,0 +1,5 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadCatalogUnsubscribeResponse", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index f3f828b8ac..777da9c505 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -44,6 +44,7 @@ import type { ServerRequestResolvedNotification } from "./v2/ServerRequestResolv import type { SkillsChangedNotification } from "./v2/SkillsChangedNotification"; import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNotification"; import type { ThreadArchivedNotification } from "./v2/ThreadArchivedNotification"; +import type { ThreadCatalogChangedNotification } from "./v2/ThreadCatalogChangedNotification"; import type { ThreadClosedNotification } from "./v2/ThreadClosedNotification"; import type { ThreadDeletedNotification } from "./v2/ThreadDeletedNotification"; import type { ThreadGoalClearedNotification } from "./v2/ThreadGoalClearedNotification"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogChangedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogChangedNotification.ts new file mode 100644 index 0000000000..364f466549 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogChangedNotification.ts @@ -0,0 +1,6 @@ +// 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 { ThreadSummary } from "./ThreadSummary"; + +export type ThreadCatalogChangedNotification = { thread: ThreadSummary, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogSubscribeResponse.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogSubscribeResponse.ts new file mode 100644 index 0000000000..3050a2dad0 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogSubscribeResponse.ts @@ -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 ThreadCatalogSubscribeResponse = Record; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogUnsubscribeResponse.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogUnsubscribeResponse.ts new file mode 100644 index 0000000000..8f84d8bf67 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogUnsubscribeResponse.ts @@ -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 ThreadCatalogUnsubscribeResponse = Record; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts new file mode 100644 index 0000000000..f36064c464 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts @@ -0,0 +1,9 @@ +// 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 { AbsolutePathBuf } from "../AbsolutePathBuf"; +import type { GitInfo } from "./GitInfo"; +import type { SessionSource } from "./SessionSource"; +import type { ThreadSource } from "./ThreadSource"; + +export type ThreadSummary = { id: string, sessionId: string, forkedFromId: string | null, parentThreadId: string | null, preview: string, ephemeral: boolean, modelProvider: string, createdAt: number, updatedAt: number, archivedAt: number | null, path: string | null, cwd: AbsolutePathBuf, cliVersion: string, source: SessionSource, threadSource: ThreadSource | null, agentNickname: string | null, agentRole: string | null, gitInfo: GitInfo | null, name: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index ff7df9e9ec..5f5b5e935b 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -390,6 +390,9 @@ export type { ThreadApproveGuardianDeniedActionResponse } from "./ThreadApproveG export type { ThreadArchiveParams } from "./ThreadArchiveParams"; export type { ThreadArchiveResponse } from "./ThreadArchiveResponse"; export type { ThreadArchivedNotification } from "./ThreadArchivedNotification"; +export type { ThreadCatalogChangedNotification } from "./ThreadCatalogChangedNotification"; +export type { ThreadCatalogSubscribeResponse } from "./ThreadCatalogSubscribeResponse"; +export type { ThreadCatalogUnsubscribeResponse } from "./ThreadCatalogUnsubscribeResponse"; export type { ThreadClosedNotification } from "./ThreadClosedNotification"; export type { ThreadCompactStartParams } from "./ThreadCompactStartParams"; export type { ThreadCompactStartResponse } from "./ThreadCompactStartResponse"; @@ -453,6 +456,7 @@ export type { ThreadStartSource } from "./ThreadStartSource"; export type { ThreadStartedNotification } from "./ThreadStartedNotification"; export type { ThreadStatus } from "./ThreadStatus"; export type { ThreadStatusChangedNotification } from "./ThreadStatusChangedNotification"; +export type { ThreadSummary } from "./ThreadSummary"; export type { ThreadTokenUsage } from "./ThreadTokenUsage"; export type { ThreadTokenUsageUpdatedNotification } from "./ThreadTokenUsageUpdatedNotification"; export type { ThreadUnarchiveParams } from "./ThreadUnarchiveParams"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 51a37fbaa9..6e74763718 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -618,6 +618,16 @@ client_request_definitions! { serialization: None, response: v2::ThreadListResponse, }, + ThreadCatalogSubscribe => "thread/catalog/subscribe" { + params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>, + serialization: None, + response: v2::ThreadCatalogSubscribeResponse, + }, + ThreadCatalogUnsubscribe => "thread/catalog/unsubscribe" { + params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>, + serialization: None, + response: v2::ThreadCatalogUnsubscribeResponse, + }, #[experimental("thread/search")] ThreadSearch => "thread/search" { params: v2::ThreadSearchParams, @@ -1602,6 +1612,7 @@ server_notification_definitions! { Error => "error" (v2::ErrorNotification), ThreadStarted => "thread/started" (v2::ThreadStartedNotification), ThreadStatusChanged => "thread/status/changed" (v2::ThreadStatusChangedNotification), + ThreadCatalogChanged => "thread/catalog/changed" (v2::ThreadCatalogChangedNotification), ThreadArchived => "thread/archived" (v2::ThreadArchivedNotification), ThreadDeleted => "thread/deleted" (v2::ThreadDeletedNotification), ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index cba4f1b246..059552d58d 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -6,6 +6,7 @@ use super::SandboxPolicy; use super::Thread; use super::ThreadItem; use super::ThreadSource; +use super::ThreadSummary; use super::Turn; use super::TurnEnvironmentParams; use super::TurnItemsView; @@ -1094,6 +1095,16 @@ pub struct ThreadListParams { pub parent_thread_id: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadCatalogSubscribeResponse {} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadCatalogUnsubscribeResponse {} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -1419,6 +1430,13 @@ pub struct ThreadStatusChangedNotification { pub status: ThreadStatus, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadCatalogChangedNotification { + pub thread: ThreadSummary, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs index 3338513df7..356264e2e1 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs @@ -193,6 +193,34 @@ pub struct Thread { pub turns: Vec, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadSummary { + pub id: String, + pub session_id: String, + pub forked_from_id: Option, + pub parent_thread_id: Option, + pub preview: String, + pub ephemeral: bool, + pub model_provider: String, + #[ts(type = "number")] + pub created_at: i64, + #[ts(type = "number")] + pub updated_at: i64, + #[ts(type = "number | null")] + pub archived_at: Option, + pub path: Option, + pub cwd: AbsolutePathBuf, + pub cli_version: String, + pub source: SessionSource, + pub thread_source: Option, + pub agent_nickname: Option, + pub agent_role: Option, + pub git_info: Option, + pub name: Option, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index c7ec2c15cf..561659d5d7 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -421,6 +421,18 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li } } ``` +`thread/catalog/subscribe` watches all future thread metadata changes without draining existing history. Install the subscription before calling `thread/list` when you need a race-free sidebar bootstrap; after the subscribe response arrives, buffer catalog notifications until the list response arrives, then apply the complete summaries idempotently and filter them client-side. The summary contains sidebar metadata such as id, preview/name, cwd, timestamps, archive state, git info, and source; it does not contain turns, items, messages, deltas, tool state, status, or runtime history. + +```json +{ "method": "thread/catalog/subscribe", "id": 21 } +{ "id": 21, "result": {} } +{ "method": "thread/catalog/changed", "params": { + "thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "archivedAt": null } +} } +``` + +The subscription is connection-local. Repeated subscribe and unsubscribe calls are harmless. Reconnecting clients should subscribe again and refetch the current page with `thread/list`; catalog subscriptions do not replay changes that happened while the connection was offline. + ### Example: List loaded threads `thread/loaded/list` returns thread ids currently loaded in memory. This is useful when you want to check which sessions are active without scanning rollouts on disk. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 7c9f006d26..72a80c0ca8 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -395,6 +395,8 @@ impl MessageProcessor { let pending_thread_unloads = Arc::new(Mutex::new(HashSet::new())); let thread_watch_manager = crate::thread_status::ThreadWatchManager::new_with_outgoing(outgoing.clone()); + let thread_catalog_subscriptions = + crate::request_processors::ThreadCatalogSubscriptions::new(outgoing.clone()); let thread_list_state_permit = Arc::new(Semaphore::new(/*permits*/ 1)); let workspace_settings_cache = Arc::new(workspace_settings::WorkspaceSettingsCache::default()); @@ -490,6 +492,7 @@ impl MessageProcessor { Arc::clone(&pending_thread_unloads), thread_state_manager.clone(), thread_watch_manager.clone(), + thread_catalog_subscriptions.clone(), Arc::clone(&thread_list_state_permit), thread_goal_processor.clone(), state_db.clone(), @@ -505,8 +508,10 @@ impl MessageProcessor { Arc::clone(&config), config_manager.clone(), pending_thread_unloads, + Arc::clone(&thread_store), thread_state_manager, thread_watch_manager, + thread_catalog_subscriptions, thread_list_state_permit, Arc::clone(&skills_watcher), ); @@ -1211,6 +1216,16 @@ impl MessageProcessor { ClientRequest::ThreadList { params, .. } => { self.thread_processor.thread_list(params).await } + ClientRequest::ThreadCatalogSubscribe { .. } => { + self.thread_processor + .thread_catalog_subscribe(connection_id) + .await + } + ClientRequest::ThreadCatalogUnsubscribe { .. } => { + self.thread_processor + .thread_catalog_unsubscribe(connection_id) + .await + } ClientRequest::ThreadSearch { params, .. } => { self.thread_processor.thread_search(params).await } diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 500ecc4734..dc2dd62837 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -184,6 +184,9 @@ use codex_app_server_protocol::ThreadBackgroundTerminalsListParams; use codex_app_server_protocol::ThreadBackgroundTerminalsListResponse; use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateParams; use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateResponse; +use codex_app_server_protocol::ThreadCatalogChangedNotification; +use codex_app_server_protocol::ThreadCatalogSubscribeResponse; +use codex_app_server_protocol::ThreadCatalogUnsubscribeResponse; use codex_app_server_protocol::ThreadClosedNotification; use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadCompactStartResponse; @@ -257,6 +260,7 @@ 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::ThreadSummary; use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::ThreadUnarchiveParams; @@ -499,6 +503,7 @@ mod plugins; mod process_exec_processor; mod remote_control_processor; mod search; +mod thread_catalog_subscription; mod thread_processor; mod token_usage_replay; mod turn_processor; @@ -522,6 +527,7 @@ pub(crate) use plugins::PluginRequestProcessor; pub(crate) use process_exec_processor::ProcessExecRequestProcessor; pub(crate) use remote_control_processor::RemoteControlRequestProcessor; pub(crate) use search::SearchRequestProcessor; +pub(crate) use thread_catalog_subscription::ThreadCatalogSubscriptions; pub(crate) use thread_goal_processor::ThreadGoalRequestProcessor; pub(crate) use thread_processor::ThreadRequestProcessor; pub(crate) use turn_processor::TurnRequestProcessor; diff --git a/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs b/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs new file mode 100644 index 0000000000..a59d0865aa --- /dev/null +++ b/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs @@ -0,0 +1,84 @@ +use super::*; + +#[derive(Clone)] +pub(crate) struct ThreadCatalogSubscriptions { + outgoing: Arc, + connection_ids: Arc>>, +} + +impl ThreadCatalogSubscriptions { + pub(crate) fn new(outgoing: Arc) -> Self { + Self { + outgoing, + connection_ids: Arc::new(Mutex::new(HashSet::new())), + } + } + + pub(super) async fn subscribe( + &self, + connection_id: ConnectionId, + ) -> ThreadCatalogSubscribeResponse { + self.connection_ids.lock().await.insert(connection_id); + ThreadCatalogSubscribeResponse {} + } + + pub(super) async fn unsubscribe( + &self, + connection_id: ConnectionId, + ) -> ThreadCatalogUnsubscribeResponse { + self.connection_ids.lock().await.remove(&connection_id); + ThreadCatalogUnsubscribeResponse {} + } + + pub(super) async fn connection_closed(&self, connection_id: ConnectionId) { + self.connection_ids.lock().await.remove(&connection_id); + } + + pub(super) async fn publish_thread_summary(&self, thread: ThreadSummary) { + let connection_ids = self + .connection_ids + .lock() + .await + .iter() + .copied() + .collect::>(); + if connection_ids.is_empty() { + return; + } + self.outgoing + .send_server_notification_to_connections( + &connection_ids, + ServerNotification::ThreadCatalogChanged(ThreadCatalogChangedNotification { + thread, + }), + ) + .await; + } + + pub(super) async fn publish_thread_change( + &self, + thread_store: &Arc, + thread_id: ThreadId, + fallback_provider: &str, + fallback_cwd: &AbsolutePathBuf, + ) { + let stored_thread = match thread_store + .read_thread(StoreReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + { + Ok(stored_thread) => stored_thread, + Err(ThreadStoreError::ThreadNotFound { .. }) => return, + Err(err) => { + warn!("failed to read thread {thread_id} for catalog notification: {err}"); + return; + } + }; + let summary = + thread_summary_from_stored_thread(stored_thread, fallback_provider, fallback_cwd); + self.publish_thread_summary(summary).await; + } +} diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 21adf46134..33ea9c96d3 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -8,9 +8,12 @@ pub(super) struct ListenerTaskContext { pub(super) thread_state_manager: ThreadStateManager, pub(super) outgoing: Arc, pub(super) pending_thread_unloads: Arc>>, + pub(super) thread_store: Arc, pub(super) thread_watch_manager: ThreadWatchManager, + pub(super) thread_catalog_subscriptions: ThreadCatalogSubscriptions, pub(super) thread_list_state_permit: Arc, pub(super) fallback_model_provider: String, + pub(super) fallback_cwd: AbsolutePathBuf, pub(super) codex_home: PathBuf, pub(super) skills_watcher: Arc, } @@ -119,6 +122,17 @@ impl UnloadingState { } } +fn catalog_worthy_event(event: &EventMsg) -> bool { + matches!( + event, + EventMsg::TurnStarted(_) + | EventMsg::TurnComplete(_) + | EventMsg::TurnAborted(_) + | EventMsg::ThreadSettingsApplied(_) + | EventMsg::ThreadRolledBack(_) + ) +} + pub(super) enum ThreadShutdownResult { Complete, SubmitFailed, @@ -266,9 +280,12 @@ pub(super) async fn ensure_listener_task_running( thread_manager, thread_state_manager, pending_thread_unloads, + thread_store, thread_watch_manager, + thread_catalog_subscriptions, thread_list_state_permit, fallback_model_provider, + fallback_cwd, codex_home, .. } = listener_task_context; @@ -349,6 +366,16 @@ pub(super) async fn ensure_listener_task_running( fallback_model_provider.clone(), ) .await; + if catalog_worthy_event(&event.msg) { + thread_catalog_subscriptions + .publish_thread_change( + &thread_store, + conversation_id, + fallback_model_provider.as_str(), + &fallback_cwd, + ) + .await; + } } unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => { if !unloading_watchers_open { diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index a2966243c6..0f2bd6a206 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -354,6 +354,7 @@ pub(crate) struct ThreadRequestProcessor { pub(super) pending_thread_unloads: Arc>>, pub(super) thread_state_manager: ThreadStateManager, pub(super) thread_watch_manager: ThreadWatchManager, + pub(super) thread_catalog_subscriptions: ThreadCatalogSubscriptions, pub(super) thread_list_state_permit: Arc, pub(super) thread_goal_processor: ThreadGoalRequestProcessor, pub(super) state_db: Option, @@ -386,6 +387,7 @@ impl ThreadRequestProcessor { pending_thread_unloads: Arc>>, thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, + thread_catalog_subscriptions: ThreadCatalogSubscriptions, thread_list_state_permit: Arc, thread_goal_processor: ThreadGoalRequestProcessor, state_db: Option, @@ -403,6 +405,7 @@ impl ThreadRequestProcessor { pending_thread_unloads, thread_state_manager, thread_watch_manager, + thread_catalog_subscriptions, thread_list_state_permit, thread_goal_processor, state_db, @@ -492,9 +495,12 @@ impl ThreadRequestProcessor { .send_response(request_id.clone(), response) .await; for thread_id in archived_thread_ids { + self.publish_thread_catalog_change(thread_id).await; self.outgoing .send_server_notification(ServerNotification::ThreadArchived( - ThreadArchivedNotification { thread_id }, + ThreadArchivedNotification { + thread_id: thread_id.to_string(), + }, )) .await; } @@ -533,6 +539,9 @@ impl ThreadRequestProcessor { .send_response(request_id.clone(), response) .await; if let Some(notification) = notification { + if let Ok(thread_id) = ThreadId::from_string(¬ification.thread_id) { + self.publish_thread_catalog_change(thread_id).await; + } self.outgoing .send_server_notification(ServerNotification::ThreadNameUpdated( notification, @@ -549,9 +558,12 @@ impl ThreadRequestProcessor { &self, params: ThreadMetadataUpdateParams, ) -> Result, JSONRPCErrorError> { - self.thread_metadata_update_response_inner(params) - .await - .map(|response| Some(response.into())) + let thread_id = ThreadId::from_string(¶ms.thread_id).ok(); + let response = self.thread_metadata_update_response_inner(params).await?; + if let Some(thread_id) = thread_id { + self.publish_thread_catalog_change(thread_id).await; + } + Ok(Some(response.into())) } pub(crate) async fn thread_memory_mode_set( @@ -581,6 +593,9 @@ impl ThreadRequestProcessor { self.outgoing .send_response(request_id.clone(), response) .await; + if let Ok(thread_id) = ThreadId::from_string(¬ification.thread_id) { + self.publish_thread_catalog_change(thread_id).await; + } self.outgoing .send_server_notification(ServerNotification::ThreadUnarchived(notification)) .await; @@ -647,6 +662,30 @@ impl ThreadRequestProcessor { .map(|response| Some(response.into())) } + pub(crate) async fn thread_catalog_subscribe( + &self, + connection_id: ConnectionId, + ) -> Result, JSONRPCErrorError> { + Ok(Some( + self.thread_catalog_subscriptions + .subscribe(connection_id) + .await + .into(), + )) + } + + pub(crate) async fn thread_catalog_unsubscribe( + &self, + connection_id: ConnectionId, + ) -> Result, JSONRPCErrorError> { + Ok(Some( + self.thread_catalog_subscriptions + .unsubscribe(connection_id) + .await + .into(), + )) + } + pub(crate) async fn thread_search( &self, params: ThreadSearchParams, @@ -837,9 +876,12 @@ impl ThreadRequestProcessor { thread_state_manager: self.thread_state_manager.clone(), outgoing: Arc::clone(&self.outgoing), pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), + thread_store: Arc::clone(&self.thread_store), thread_watch_manager: self.thread_watch_manager.clone(), + thread_catalog_subscriptions: self.thread_catalog_subscriptions.clone(), thread_list_state_permit: self.thread_list_state_permit.clone(), fallback_model_provider: self.config.model_provider_id.clone(), + fallback_cwd: self.config.cwd.clone(), codex_home: self.config.codex_home.to_path_buf(), skills_watcher: Arc::clone(&self.skills_watcher), } @@ -937,9 +979,12 @@ impl ThreadRequestProcessor { thread_state_manager: self.thread_state_manager.clone(), outgoing: Arc::clone(&self.outgoing), pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), + thread_store: Arc::clone(&self.thread_store), thread_watch_manager: self.thread_watch_manager.clone(), + thread_catalog_subscriptions: self.thread_catalog_subscriptions.clone(), thread_list_state_permit: self.thread_list_state_permit.clone(), fallback_model_provider: self.config.model_provider_id.clone(), + fallback_cwd: self.config.cwd.clone(), codex_home: self.config.codex_home.to_path_buf(), skills_watcher: Arc::clone(&self.skills_watcher), }; @@ -1330,7 +1375,7 @@ impl ThreadRequestProcessor { async fn thread_archive_inner( &self, params: ThreadArchiveParams, - ) -> Result<(ThreadArchiveResponse, Vec), JSONRPCErrorError> { + ) -> Result<(ThreadArchiveResponse, Vec), JSONRPCErrorError> { let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; self.thread_archive_response(params).await } @@ -1338,7 +1383,7 @@ impl ThreadRequestProcessor { async fn thread_archive_response( &self, params: ThreadArchiveParams, - ) -> Result<(ThreadArchiveResponse, Vec), JSONRPCErrorError> { + ) -> Result<(ThreadArchiveResponse, Vec), JSONRPCErrorError> { let thread_id = ThreadId::from_string(¶ms.thread_id) .map_err(|err| invalid_request(format!("invalid session id: {err}")))?; @@ -1399,7 +1444,7 @@ impl ThreadRequestProcessor { .await { Ok(()) => { - archived_thread_ids.push(parent_thread_id.to_string()); + archived_thread_ids.push(*parent_thread_id); } Err(err) => return Err(thread_store_archive_error("archive", err)), } @@ -1414,7 +1459,7 @@ impl ThreadRequestProcessor { .await { Ok(()) => { - archived_thread_ids.push(descendant_thread_id.to_string()); + archived_thread_ids.push(descendant_thread_id); } Err(err) => { warn!( @@ -1977,6 +2022,17 @@ impl ThreadRequestProcessor { }) } + async fn publish_thread_catalog_change(&self, thread_id: ThreadId) { + self.thread_catalog_subscriptions + .publish_thread_change( + &self.thread_store, + thread_id, + self.config.model_provider_id.as_str(), + &self.config.cwd, + ) + .await; + } + async fn thread_search_response_inner( &self, params: ThreadSearchParams, @@ -2522,6 +2578,9 @@ impl ThreadRequestProcessor { } pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) { + self.thread_catalog_subscriptions + .connection_closed(connection_id) + .await; let thread_ids = self .thread_state_manager .remove_connection(connection_id) @@ -3602,6 +3661,8 @@ impl ThreadRequestProcessor { .await; } + self.publish_thread_catalog_change(thread_id).await; + self.outgoing .send_server_notification(ServerNotification::ThreadStarted(notif)) .await; diff --git a/codex-rs/app-server/src/request_processors/thread_summary.rs b/codex-rs/app-server/src/request_processors/thread_summary.rs index 3b668e0a1e..bf04c2098d 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -275,6 +275,36 @@ pub(super) fn thread_started_notification(mut thread: Thread) -> ThreadStartedNo ThreadStartedNotification { thread } } +pub(super) fn thread_summary_from_stored_thread( + thread: StoredThread, + fallback_provider: &str, + fallback_cwd: &AbsolutePathBuf, +) -> ThreadSummary { + let archived_at = thread.archived_at.map(|dt| dt.timestamp()); + let (thread, _) = thread_from_stored_thread(thread, fallback_provider, fallback_cwd); + ThreadSummary { + id: thread.id, + session_id: thread.session_id, + forked_from_id: thread.forked_from_id, + parent_thread_id: thread.parent_thread_id, + preview: thread.preview, + ephemeral: thread.ephemeral, + model_provider: thread.model_provider, + created_at: thread.created_at, + updated_at: thread.updated_at, + archived_at, + path: thread.path, + cwd: thread.cwd, + cli_version: thread.cli_version, + source: thread.source, + thread_source: thread.thread_source, + agent_nickname: thread.agent_nickname, + agent_role: thread.agent_role, + git_info: thread.git_info, + name: thread.name, + } +} + #[cfg(test)] pub(crate) fn summary_to_thread( summary: ConversationSummary, diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index e0d79a0c3b..d068453834 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -76,8 +76,10 @@ pub(crate) struct TurnRequestProcessor { config: Arc, config_manager: ConfigManager, pending_thread_unloads: Arc>>, + thread_store: Arc, thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, + thread_catalog_subscriptions: ThreadCatalogSubscriptions, thread_list_state_permit: Arc, skills_watcher: Arc, } @@ -133,8 +135,10 @@ impl TurnRequestProcessor { config: Arc, config_manager: ConfigManager, pending_thread_unloads: Arc>>, + thread_store: Arc, thread_state_manager: ThreadStateManager, thread_watch_manager: ThreadWatchManager, + thread_catalog_subscriptions: ThreadCatalogSubscriptions, thread_list_state_permit: Arc, skills_watcher: Arc, ) -> Self { @@ -147,8 +151,10 @@ impl TurnRequestProcessor { config, config_manager, pending_thread_unloads, + thread_store, thread_state_manager, thread_watch_manager, + thread_catalog_subscriptions, thread_list_state_permit, skills_watcher, } @@ -1419,9 +1425,12 @@ impl TurnRequestProcessor { thread_state_manager: self.thread_state_manager.clone(), outgoing: Arc::clone(&self.outgoing), pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), + thread_store: Arc::clone(&self.thread_store), thread_watch_manager: self.thread_watch_manager.clone(), + thread_catalog_subscriptions: self.thread_catalog_subscriptions.clone(), thread_list_state_permit: self.thread_list_state_permit.clone(), fallback_model_provider: self.config.model_provider_id.clone(), + fallback_cwd: self.config.cwd.clone(), codex_home: self.config.codex_home.to_path_buf(), skills_watcher: Arc::clone(&self.skills_watcher), } diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index bbd9952754..b74cb76cfd 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -59,6 +59,7 @@ mod safety_check_downgrade; mod skills_list; mod sleep; mod thread_archive; +mod thread_catalog_subscription; mod thread_delete; mod thread_fork; mod thread_inject_items; diff --git a/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs b/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs new file mode 100644 index 0000000000..232e4a360f --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs @@ -0,0 +1,125 @@ +use anyhow::Result; +use app_test_support::TestAppServer; +use app_test_support::create_fake_rollout; +use app_test_support::to_response; +use app_test_support::write_mock_responses_config_toml; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadCatalogChangedNotification; +use codex_app_server_protocol::ThreadListParams; +use codex_app_server_protocol::ThreadListResponse; +use codex_app_server_protocol::ThreadSetNameParams; +use codex_app_server_protocol::ThreadSetNameResponse; +use codex_app_server_protocol::ThreadSortKey; +use pretty_assertions::assert_eq; +use serde::de::DeserializeOwned; +use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); + +#[tokio::test] +async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> { + let codex_home = TempDir::new()?; + write_mock_responses_config_toml( + codex_home.path(), + "http://localhost:1", + &Default::default(), + i64::MAX, + /*requires_openai_auth*/ None, + "mock_provider", + "", + )?; + let older_thread_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Older thread", + Some("mock_provider"), + /*git_info*/ None, + )?; + let newer_thread_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-05-00", + "2025-01-05T12:05:00Z", + "Newer thread", + Some("mock_provider"), + /*git_info*/ None, + )?; + + let mut app = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, app.initialize()).await??; + + let subscribe_id = app + .send_raw_request("thread/catalog/subscribe", /*params*/ None) + .await?; + let _: codex_app_server_protocol::ThreadCatalogSubscribeResponse = + read_response(&mut app, subscribe_id).await?; + + let list_id = app + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: Some(1), + sort_key: Some(ThreadSortKey::CreatedAt), + sort_direction: None, + model_providers: Some(vec!["mock_provider".to_string()]), + source_kinds: None, + archived: None, + cwd: None, + use_state_db_only: false, + search_term: None, + }) + .await?; + let page = read_response::(&mut app, list_id).await?; + assert_eq!(page.data.len(), 1); + assert_eq!(page.data[0].id, newer_thread_id); + + rename_thread( + &mut app, + older_thread_id.clone(), + "Renamed outside first page", + ) + .await?; + + let notification: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + app.read_stream_until_notification_message("thread/catalog/changed"), + ) + .await??; + let raw_params = notification + .params + .expect("thread/catalog/changed should have params"); + assert_eq!(raw_params["thread"].get("turns"), None); + + let changed: ThreadCatalogChangedNotification = serde_json::from_value(raw_params)?; + assert_eq!(changed.thread.id, older_thread_id); + assert_eq!( + changed.thread.name.as_deref(), + Some("Renamed outside first page") + ); + assert_eq!(changed.thread.archived_at, None); + + Ok(()) +} + +async fn rename_thread(app: &mut TestAppServer, thread_id: String, name: &str) -> Result<()> { + let rename_id = app + .send_thread_set_name_request(ThreadSetNameParams { + thread_id, + name: name.to_string(), + }) + .await?; + let _: ThreadSetNameResponse = read_response(app, rename_id).await?; + Ok(()) +} + +async fn read_response(app: &mut TestAppServer, id: i64) -> Result { + let response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + app.read_stream_until_response_message(RequestId::Integer(id)), + ) + .await??; + to_response(response) +}