diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 4f414c1b37..01bd7d0a4d 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -5204,9 +5204,9 @@ }, "method": { "enum": [ - "thread/catalog/subscribe" + "threadCatalog/subscribe" ], - "title": "Thread/catalog/subscribeRequestMethod", + "title": "ThreadCatalog/subscribeRequestMethod", "type": "string" }, "params": { @@ -5217,7 +5217,7 @@ "id", "method" ], - "title": "Thread/catalog/subscribeRequest", + "title": "ThreadCatalog/subscribeRequest", "type": "object" }, { @@ -5227,9 +5227,9 @@ }, "method": { "enum": [ - "thread/catalog/unsubscribe" + "threadCatalog/unsubscribe" ], - "title": "Thread/catalog/unsubscribeRequestMethod", + "title": "ThreadCatalog/unsubscribeRequestMethod", "type": "string" }, "params": { @@ -5240,7 +5240,7 @@ "id", "method" ], - "title": "Thread/catalog/unsubscribeRequest", + "title": "ThreadCatalog/unsubscribeRequest", "type": "object" }, { diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index c504ce13b3..eacd4ced53 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -5041,6 +5041,10 @@ "format": "int64", "type": "integer" }, + "createdAtMs": { + "format": "int64", + "type": "integer" + }, "cwd": { "$ref": "#/definitions/AbsolutePathBuf" }, @@ -5109,11 +5113,16 @@ "updatedAt": { "format": "int64", "type": "integer" + }, + "updatedAtMs": { + "format": "int64", + "type": "integer" } }, "required": [ "cliVersion", "createdAt", + "createdAtMs", "cwd", "ephemeral", "id", @@ -5121,7 +5130,8 @@ "preview", "sessionId", "source", - "updatedAt" + "updatedAt", + "updatedAtMs" ], "type": "object" }, @@ -5839,9 +5849,9 @@ "properties": { "method": { "enum": [ - "thread/catalog/changed" + "threadCatalog/changed" ], - "title": "Thread/catalog/changedNotificationMethod", + "title": "ThreadCatalog/changedNotificationMethod", "type": "string" }, "params": { @@ -5852,7 +5862,7 @@ "method", "params" ], - "title": "Thread/catalog/changedNotification", + "title": "ThreadCatalog/changedNotification", "type": "object" }, { 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 918d6117dd..3e6dd79cb4 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 @@ -643,9 +643,9 @@ }, "method": { "enum": [ - "thread/catalog/subscribe" + "threadCatalog/subscribe" ], - "title": "Thread/catalog/subscribeRequestMethod", + "title": "ThreadCatalog/subscribeRequestMethod", "type": "string" }, "params": { @@ -656,7 +656,7 @@ "id", "method" ], - "title": "Thread/catalog/subscribeRequest", + "title": "ThreadCatalog/subscribeRequest", "type": "object" }, { @@ -666,9 +666,9 @@ }, "method": { "enum": [ - "thread/catalog/unsubscribe" + "threadCatalog/unsubscribe" ], - "title": "Thread/catalog/unsubscribeRequestMethod", + "title": "ThreadCatalog/unsubscribeRequestMethod", "type": "string" }, "params": { @@ -679,7 +679,7 @@ "id", "method" ], - "title": "Thread/catalog/unsubscribeRequest", + "title": "ThreadCatalog/unsubscribeRequest", "type": "object" }, { @@ -4218,9 +4218,9 @@ "properties": { "method": { "enum": [ - "thread/catalog/changed" + "threadCatalog/changed" ], - "title": "Thread/catalog/changedNotificationMethod", + "title": "ThreadCatalog/changedNotificationMethod", "type": "string" }, "params": { @@ -4231,7 +4231,7 @@ "method", "params" ], - "title": "Thread/catalog/changedNotification", + "title": "ThreadCatalog/changedNotification", "type": "object" }, { @@ -19611,6 +19611,10 @@ "format": "int64", "type": "integer" }, + "createdAtMs": { + "format": "int64", + "type": "integer" + }, "cwd": { "$ref": "#/definitions/v2/AbsolutePathBuf" }, @@ -19679,11 +19683,16 @@ "updatedAt": { "format": "int64", "type": "integer" + }, + "updatedAtMs": { + "format": "int64", + "type": "integer" } }, "required": [ "cliVersion", "createdAt", + "createdAtMs", "cwd", "ephemeral", "id", @@ -19691,7 +19700,8 @@ "preview", "sessionId", "source", - "updatedAt" + "updatedAt", + "updatedAtMs" ], "type": "object" }, 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 25d63874df..2b13aa04f5 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 @@ -1655,9 +1655,9 @@ }, "method": { "enum": [ - "thread/catalog/subscribe" + "threadCatalog/subscribe" ], - "title": "Thread/catalog/subscribeRequestMethod", + "title": "ThreadCatalog/subscribeRequestMethod", "type": "string" }, "params": { @@ -1668,7 +1668,7 @@ "id", "method" ], - "title": "Thread/catalog/subscribeRequest", + "title": "ThreadCatalog/subscribeRequest", "type": "object" }, { @@ -1678,9 +1678,9 @@ }, "method": { "enum": [ - "thread/catalog/unsubscribe" + "threadCatalog/unsubscribe" ], - "title": "Thread/catalog/unsubscribeRequestMethod", + "title": "ThreadCatalog/unsubscribeRequestMethod", "type": "string" }, "params": { @@ -1691,7 +1691,7 @@ "id", "method" ], - "title": "Thread/catalog/unsubscribeRequest", + "title": "ThreadCatalog/unsubscribeRequest", "type": "object" }, { @@ -12626,9 +12626,9 @@ "properties": { "method": { "enum": [ - "thread/catalog/changed" + "threadCatalog/changed" ], - "title": "Thread/catalog/changedNotificationMethod", + "title": "ThreadCatalog/changedNotificationMethod", "type": "string" }, "params": { @@ -12639,7 +12639,7 @@ "method", "params" ], - "title": "Thread/catalog/changedNotification", + "title": "ThreadCatalog/changedNotification", "type": "object" }, { @@ -17390,6 +17390,10 @@ "format": "int64", "type": "integer" }, + "createdAtMs": { + "format": "int64", + "type": "integer" + }, "cwd": { "$ref": "#/definitions/AbsolutePathBuf" }, @@ -17458,11 +17462,16 @@ "updatedAt": { "format": "int64", "type": "integer" + }, + "updatedAtMs": { + "format": "int64", + "type": "integer" } }, "required": [ "cliVersion", "createdAt", + "createdAtMs", "cwd", "ephemeral", "id", @@ -17470,7 +17479,8 @@ "preview", "sessionId", "source", - "updatedAt" + "updatedAt", + "updatedAtMs" ], "type": "object" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json index 7a42120171..56d0e12f49 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json @@ -181,6 +181,10 @@ "format": "int64", "type": "integer" }, + "createdAtMs": { + "format": "int64", + "type": "integer" + }, "cwd": { "$ref": "#/definitions/AbsolutePathBuf" }, @@ -249,11 +253,16 @@ "updatedAt": { "format": "int64", "type": "integer" + }, + "updatedAtMs": { + "format": "int64", + "type": "integer" } }, "required": [ "cliVersion", "createdAt", + "createdAtMs", "cwd", "ephemeral", "id", @@ -261,7 +270,8 @@ "preview", "sessionId", "source", - "updatedAt" + "updatedAt", + "updatedAtMs" ], "type": "object" } diff --git a/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts b/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts index 6af04ef401..e149fbeef8 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts @@ -88,4 +88,4 @@ import type { WindowsSandboxSetupStartParams } from "./v2/WindowsSandboxSetupSta /** * Request from the client to the server. */ -export type ClientRequest ={ "method": "initialize", id: RequestId, params: InitializeParams, } | { "method": "thread/start", id: RequestId, params: ThreadStartParams, } | { "method": "thread/resume", id: RequestId, params: ThreadResumeParams, } | { "method": "thread/fork", id: RequestId, params: ThreadForkParams, } | { "method": "thread/archive", id: RequestId, params: ThreadArchiveParams, } | { "method": "thread/delete", id: RequestId, params: ThreadDeleteParams, } | { "method": "thread/unsubscribe", id: RequestId, params: ThreadUnsubscribeParams, } | { "method": "thread/name/set", id: RequestId, params: ThreadSetNameParams, } | { "method": "thread/goal/set", id: RequestId, params: ThreadGoalSetParams, } | { "method": "thread/goal/get", id: RequestId, params: ThreadGoalGetParams, } | { "method": "thread/goal/clear", id: RequestId, params: ThreadGoalClearParams, } | { "method": "thread/metadata/update", id: RequestId, params: ThreadMetadataUpdateParams, } | { "method": "thread/unarchive", id: RequestId, params: ThreadUnarchiveParams, } | { "method": "thread/compact/start", id: RequestId, params: ThreadCompactStartParams, } | { "method": "thread/shellCommand", id: RequestId, params: ThreadShellCommandParams, } | { "method": "thread/approveGuardianDeniedAction", id: RequestId, params: ThreadApproveGuardianDeniedActionParams, } | { "method": "thread/rollback", id: RequestId, params: ThreadRollbackParams, } | { "method": "thread/list", id: RequestId, params: ThreadListParams, } | { "method": "thread/loaded/list", id: RequestId, params: ThreadLoadedListParams, } | { "method": "thread/read", id: RequestId, params: ThreadReadParams, } | { "method": "thread/inject_items", id: RequestId, params: ThreadInjectItemsParams, } | { "method": "skills/list", id: RequestId, params: SkillsListParams, } | { "method": "skills/extraRoots/set", id: RequestId, params: SkillsExtraRootsSetParams, } | { "method": "hooks/list", id: RequestId, params: HooksListParams, } | { "method": "marketplace/add", id: RequestId, params: MarketplaceAddParams, } | { "method": "marketplace/remove", id: RequestId, params: MarketplaceRemoveParams, } | { "method": "marketplace/upgrade", id: RequestId, params: MarketplaceUpgradeParams, } | { "method": "plugin/list", id: RequestId, params: PluginListParams, } | { "method": "plugin/installed", id: RequestId, params: PluginInstalledParams, } | { "method": "plugin/read", id: RequestId, params: PluginReadParams, } | { "method": "plugin/skill/read", id: RequestId, params: PluginSkillReadParams, } | { "method": "plugin/share/save", id: RequestId, params: PluginShareSaveParams, } | { "method": "plugin/share/updateTargets", id: RequestId, params: PluginShareUpdateTargetsParams, } | { "method": "plugin/share/list", id: RequestId, params: PluginShareListParams, } | { "method": "plugin/share/checkout", id: RequestId, params: PluginShareCheckoutParams, } | { "method": "plugin/share/delete", id: RequestId, params: PluginShareDeleteParams, } | { "method": "app/list", id: RequestId, params: AppsListParams, } | { "method": "fs/readFile", id: RequestId, params: FsReadFileParams, } | { "method": "fs/writeFile", id: RequestId, params: FsWriteFileParams, } | { "method": "fs/createDirectory", id: RequestId, params: FsCreateDirectoryParams, } | { "method": "fs/getMetadata", id: RequestId, params: FsGetMetadataParams, } | { "method": "fs/readDirectory", id: RequestId, params: FsReadDirectoryParams, } | { "method": "fs/remove", id: RequestId, params: FsRemoveParams, } | { "method": "fs/copy", id: RequestId, params: FsCopyParams, } | { "method": "fs/watch", id: RequestId, params: FsWatchParams, } | { "method": "fs/unwatch", id: RequestId, params: FsUnwatchParams, } | { "method": "skills/config/write", id: RequestId, params: SkillsConfigWriteParams, } | { "method": "plugin/install", id: RequestId, params: PluginInstallParams, } | { "method": "plugin/uninstall", id: RequestId, params: PluginUninstallParams, } | { "method": "turn/start", id: RequestId, params: TurnStartParams, } | { "method": "turn/steer", id: RequestId, params: TurnSteerParams, } | { "method": "turn/interrupt", id: RequestId, params: TurnInterruptParams, } | { "method": "review/start", id: RequestId, params: ReviewStartParams, } | { "method": "model/list", id: RequestId, params: ModelListParams, } | { "method": "modelProvider/capabilities/read", id: RequestId, params: ModelProviderCapabilitiesReadParams, } | { "method": "experimentalFeature/list", id: RequestId, params: ExperimentalFeatureListParams, } | { "method": "permissionProfile/list", id: RequestId, params: PermissionProfileListParams, } | { "method": "experimentalFeature/enablement/set", id: RequestId, params: ExperimentalFeatureEnablementSetParams, } | { "method": "mcpServer/oauth/login", id: RequestId, params: McpServerOauthLoginParams, } | { "method": "config/mcpServer/reload", id: RequestId, params: undefined, } | { "method": "mcpServerStatus/list", id: RequestId, params: ListMcpServerStatusParams, } | { "method": "mcpServer/resource/read", id: RequestId, params: McpResourceReadParams, } | { "method": "mcpServer/tool/call", id: RequestId, params: McpServerToolCallParams, } | { "method": "windowsSandbox/setupStart", id: RequestId, params: WindowsSandboxSetupStartParams, } | { "method": "windowsSandbox/readiness", id: RequestId, params: undefined, } | { "method": "account/login/start", id: RequestId, params: LoginAccountParams, } | { "method": "account/login/cancel", id: RequestId, params: CancelLoginAccountParams, } | { "method": "account/logout", id: RequestId, params: undefined, } | { "method": "account/rateLimits/read", id: RequestId, params: undefined, } | { "method": "account/rateLimitResetCredit/consume", id: RequestId, params: ConsumeAccountRateLimitResetCreditParams, } | { "method": "account/usage/read", id: RequestId, params: undefined, } | { "method": "account/workspaceMessages/read", id: RequestId, params: undefined, } | { "method": "account/sendAddCreditsNudgeEmail", id: RequestId, params: SendAddCreditsNudgeEmailParams, } | { "method": "feedback/upload", id: RequestId, params: FeedbackUploadParams, } | { "method": "command/exec", id: RequestId, params: CommandExecParams, } | { "method": "command/exec/write", id: RequestId, params: CommandExecWriteParams, } | { "method": "command/exec/terminate", id: RequestId, params: CommandExecTerminateParams, } | { "method": "command/exec/resize", id: RequestId, params: CommandExecResizeParams, } | { "method": "config/read", id: RequestId, params: ConfigReadParams, } | { "method": "externalAgentConfig/detect", id: RequestId, params: ExternalAgentConfigDetectParams, } | { "method": "externalAgentConfig/import", id: RequestId, params: ExternalAgentConfigImportParams, } | { "method": "externalAgentConfig/import/readHistories", id: RequestId, params: undefined, } | { "method": "config/value/write", id: RequestId, params: ConfigValueWriteParams, } | { "method": "config/batchWrite", id: RequestId, params: ConfigBatchWriteParams, } | { "method": "configRequirements/read", id: RequestId, params: undefined, } | { "method": "account/read", id: RequestId, params: GetAccountParams, } | { "method": "getConversationSummary", id: RequestId, params: GetConversationSummaryParams, } | { "method": "gitDiffToRemote", id: RequestId, params: GitDiffToRemoteParams, } | { "method": "getAuthStatus", id: RequestId, params: GetAuthStatusParams, } | { "method": "fuzzyFileSearch", id: RequestId, params: FuzzyFileSearchParams, }; +export type ClientRequest ={ "method": "initialize", id: RequestId, params: InitializeParams, } | { "method": "thread/start", id: RequestId, params: ThreadStartParams, } | { "method": "thread/resume", id: RequestId, params: ThreadResumeParams, } | { "method": "thread/fork", id: RequestId, params: ThreadForkParams, } | { "method": "thread/archive", id: RequestId, params: ThreadArchiveParams, } | { "method": "thread/delete", id: RequestId, params: ThreadDeleteParams, } | { "method": "thread/unsubscribe", id: RequestId, params: ThreadUnsubscribeParams, } | { "method": "thread/name/set", id: RequestId, params: ThreadSetNameParams, } | { "method": "thread/goal/set", id: RequestId, params: ThreadGoalSetParams, } | { "method": "thread/goal/get", id: RequestId, params: ThreadGoalGetParams, } | { "method": "thread/goal/clear", id: RequestId, params: ThreadGoalClearParams, } | { "method": "thread/metadata/update", id: RequestId, params: ThreadMetadataUpdateParams, } | { "method": "thread/unarchive", id: RequestId, params: ThreadUnarchiveParams, } | { "method": "thread/compact/start", id: RequestId, params: ThreadCompactStartParams, } | { "method": "thread/shellCommand", id: RequestId, params: ThreadShellCommandParams, } | { "method": "thread/approveGuardianDeniedAction", id: RequestId, params: ThreadApproveGuardianDeniedActionParams, } | { "method": "thread/rollback", id: RequestId, params: ThreadRollbackParams, } | { "method": "thread/list", id: RequestId, params: ThreadListParams, } | { "method": "threadCatalog/subscribe", id: RequestId, params: undefined, } | { "method": "threadCatalog/unsubscribe", id: RequestId, params: undefined, } | { "method": "thread/loaded/list", id: RequestId, params: ThreadLoadedListParams, } | { "method": "thread/read", id: RequestId, params: ThreadReadParams, } | { "method": "thread/inject_items", id: RequestId, params: ThreadInjectItemsParams, } | { "method": "skills/list", id: RequestId, params: SkillsListParams, } | { "method": "skills/extraRoots/set", id: RequestId, params: SkillsExtraRootsSetParams, } | { "method": "hooks/list", id: RequestId, params: HooksListParams, } | { "method": "marketplace/add", id: RequestId, params: MarketplaceAddParams, } | { "method": "marketplace/remove", id: RequestId, params: MarketplaceRemoveParams, } | { "method": "marketplace/upgrade", id: RequestId, params: MarketplaceUpgradeParams, } | { "method": "plugin/list", id: RequestId, params: PluginListParams, } | { "method": "plugin/installed", id: RequestId, params: PluginInstalledParams, } | { "method": "plugin/read", id: RequestId, params: PluginReadParams, } | { "method": "plugin/skill/read", id: RequestId, params: PluginSkillReadParams, } | { "method": "plugin/share/save", id: RequestId, params: PluginShareSaveParams, } | { "method": "plugin/share/updateTargets", id: RequestId, params: PluginShareUpdateTargetsParams, } | { "method": "plugin/share/list", id: RequestId, params: PluginShareListParams, } | { "method": "plugin/share/checkout", id: RequestId, params: PluginShareCheckoutParams, } | { "method": "plugin/share/delete", id: RequestId, params: PluginShareDeleteParams, } | { "method": "app/list", id: RequestId, params: AppsListParams, } | { "method": "fs/readFile", id: RequestId, params: FsReadFileParams, } | { "method": "fs/writeFile", id: RequestId, params: FsWriteFileParams, } | { "method": "fs/createDirectory", id: RequestId, params: FsCreateDirectoryParams, } | { "method": "fs/getMetadata", id: RequestId, params: FsGetMetadataParams, } | { "method": "fs/readDirectory", id: RequestId, params: FsReadDirectoryParams, } | { "method": "fs/remove", id: RequestId, params: FsRemoveParams, } | { "method": "fs/copy", id: RequestId, params: FsCopyParams, } | { "method": "fs/watch", id: RequestId, params: FsWatchParams, } | { "method": "fs/unwatch", id: RequestId, params: FsUnwatchParams, } | { "method": "skills/config/write", id: RequestId, params: SkillsConfigWriteParams, } | { "method": "plugin/install", id: RequestId, params: PluginInstallParams, } | { "method": "plugin/uninstall", id: RequestId, params: PluginUninstallParams, } | { "method": "turn/start", id: RequestId, params: TurnStartParams, } | { "method": "turn/steer", id: RequestId, params: TurnSteerParams, } | { "method": "turn/interrupt", id: RequestId, params: TurnInterruptParams, } | { "method": "review/start", id: RequestId, params: ReviewStartParams, } | { "method": "model/list", id: RequestId, params: ModelListParams, } | { "method": "modelProvider/capabilities/read", id: RequestId, params: ModelProviderCapabilitiesReadParams, } | { "method": "experimentalFeature/list", id: RequestId, params: ExperimentalFeatureListParams, } | { "method": "permissionProfile/list", id: RequestId, params: PermissionProfileListParams, } | { "method": "experimentalFeature/enablement/set", id: RequestId, params: ExperimentalFeatureEnablementSetParams, } | { "method": "mcpServer/oauth/login", id: RequestId, params: McpServerOauthLoginParams, } | { "method": "config/mcpServer/reload", id: RequestId, params: undefined, } | { "method": "mcpServerStatus/list", id: RequestId, params: ListMcpServerStatusParams, } | { "method": "mcpServer/resource/read", id: RequestId, params: McpResourceReadParams, } | { "method": "mcpServer/tool/call", id: RequestId, params: McpServerToolCallParams, } | { "method": "windowsSandbox/setupStart", id: RequestId, params: WindowsSandboxSetupStartParams, } | { "method": "windowsSandbox/readiness", id: RequestId, params: undefined, } | { "method": "account/login/start", id: RequestId, params: LoginAccountParams, } | { "method": "account/login/cancel", id: RequestId, params: CancelLoginAccountParams, } | { "method": "account/logout", id: RequestId, params: undefined, } | { "method": "account/rateLimits/read", id: RequestId, params: undefined, } | { "method": "account/rateLimitResetCredit/consume", id: RequestId, params: ConsumeAccountRateLimitResetCreditParams, } | { "method": "account/usage/read", id: RequestId, params: undefined, } | { "method": "account/workspaceMessages/read", id: RequestId, params: undefined, } | { "method": "account/sendAddCreditsNudgeEmail", id: RequestId, params: SendAddCreditsNudgeEmailParams, } | { "method": "feedback/upload", id: RequestId, params: FeedbackUploadParams, } | { "method": "command/exec", id: RequestId, params: CommandExecParams, } | { "method": "command/exec/write", id: RequestId, params: CommandExecWriteParams, } | { "method": "command/exec/terminate", id: RequestId, params: CommandExecTerminateParams, } | { "method": "command/exec/resize", id: RequestId, params: CommandExecResizeParams, } | { "method": "config/read", id: RequestId, params: ConfigReadParams, } | { "method": "externalAgentConfig/detect", id: RequestId, params: ExternalAgentConfigDetectParams, } | { "method": "externalAgentConfig/import", id: RequestId, params: ExternalAgentConfigImportParams, } | { "method": "externalAgentConfig/import/readHistories", id: RequestId, params: undefined, } | { "method": "config/value/write", id: RequestId, params: ConfigValueWriteParams, } | { "method": "config/batchWrite", id: RequestId, params: ConfigBatchWriteParams, } | { "method": "configRequirements/read", id: RequestId, params: undefined, } | { "method": "account/read", id: RequestId, params: GetAccountParams, } | { "method": "getConversationSummary", id: RequestId, params: GetConversationSummaryParams, } | { "method": "gitDiffToRemote", id: RequestId, params: GitDiffToRemoteParams, } | { "method": "getAuthStatus", id: RequestId, params: GetAuthStatusParams, } | { "method": "fuzzyFileSearch", id: RequestId, params: FuzzyFileSearchParams, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 777da9c505..1f657a125b 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -75,4 +75,4 @@ import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldW /** * Notification sent from the server to the client. */ -export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; +export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "threadCatalog/changed", "params": ThreadCatalogChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts index f36064c464..ee4ed5bf17 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts @@ -6,4 +6,4 @@ 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, }; +export type ThreadSummary = { id: string, sessionId: string, forkedFromId: string | null, parentThreadId: string | null, preview: string, ephemeral: boolean, modelProvider: string, createdAt: number, createdAtMs: number, updatedAt: number, updatedAtMs: 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/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 6e74763718..f4e38efbfb 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -618,12 +618,12 @@ client_request_definitions! { serialization: None, response: v2::ThreadListResponse, }, - ThreadCatalogSubscribe => "thread/catalog/subscribe" { + ThreadCatalogSubscribe => "threadCatalog/subscribe" { params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>, serialization: None, response: v2::ThreadCatalogSubscribeResponse, }, - ThreadCatalogUnsubscribe => "thread/catalog/unsubscribe" { + ThreadCatalogUnsubscribe => "threadCatalog/unsubscribe" { params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>, serialization: None, response: v2::ThreadCatalogUnsubscribeResponse, @@ -1612,7 +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), + ThreadCatalogChanged => "threadCatalog/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_data.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs index 356264e2e1..5446821056 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 @@ -207,7 +207,11 @@ pub struct ThreadSummary { #[ts(type = "number")] pub created_at: i64, #[ts(type = "number")] + pub created_at_ms: i64, + #[ts(type = "number")] pub updated_at: i64, + #[ts(type = "number")] + pub updated_at_ms: i64, #[ts(type = "number | null")] pub archived_at: Option, pub path: Option, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 0a1db57621..31cb4b13fb 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -421,13 +421,13 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li } } ``` -`thread/catalog/subscribe` watches all future thread metadata changes without draining existing history. For a race-free sidebar bootstrap, start buffering `thread/catalog/changed` notifications before sending the subscribe request. After the subscribe response arrives, call `thread/list`, then apply the buffered 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. +`threadCatalog/subscribe` watches all future thread metadata changes without draining existing history. For a race-free sidebar bootstrap, start buffering `threadCatalog/changed` notifications before sending the subscribe request. After the subscribe response arrives, call `thread/list`, then apply the buffered complete summaries idempotently and filter them client-side. The summary contains sidebar metadata such as id, preview/name, cwd, second and millisecond 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 } +{ "method": "threadCatalog/subscribe", "id": 21 } { "id": 21, "result": {} } -{ "method": "thread/catalog/changed", "params": { - "thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "archivedAt": null } +{ "method": "threadCatalog/changed", "params": { + "thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "updatedAtMs": 1730832222000, "archivedAt": null } } } ``` diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 72a80c0ca8..385552914f 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -477,7 +477,9 @@ impl MessageProcessor { Arc::clone(&thread_manager), outgoing.clone(), Arc::clone(&config), + Arc::clone(&thread_store), thread_state_manager.clone(), + thread_catalog_subscriptions.clone(), state_db.clone(), Arc::clone(&goal_service), ); @@ -508,7 +510,6 @@ impl MessageProcessor { Arc::clone(&config), config_manager.clone(), pending_thread_unloads, - Arc::clone(&thread_store), thread_state_manager, thread_watch_manager, thread_catalog_subscriptions, 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 index 1192e9c6c6..99037eceef 100644 --- a/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs +++ b/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs @@ -4,6 +4,7 @@ use super::*; pub(crate) struct ThreadCatalogSubscriptions { outgoing: Arc, connection_ids: Arc>>, + ephemeral_threads: Arc>>, } impl ThreadCatalogSubscriptions { @@ -11,6 +12,7 @@ impl ThreadCatalogSubscriptions { Self { outgoing, connection_ids: Arc::new(Mutex::new(HashSet::new())), + ephemeral_threads: Arc::new(Mutex::new(HashMap::new())), } } @@ -35,6 +37,12 @@ impl ThreadCatalogSubscriptions { } pub(super) async fn publish_thread_summary(&self, thread: ThreadSummary) { + if thread.ephemeral { + self.ephemeral_threads + .lock() + .await + .insert(thread.id.clone(), thread.clone()); + } let connection_ids = self .connection_ids .lock() @@ -84,4 +92,30 @@ impl ThreadCatalogSubscriptions { thread_summary_from_stored_thread(stored_thread, fallback_provider, fallback_cwd); self.publish_thread_summary(summary).await; } + + pub(super) async fn update_ephemeral_thread_summary( + &self, + thread_id: ThreadId, + event: &EventMsg, + ) -> Option { + let mut threads = self.ephemeral_threads.lock().await; + let thread = threads.get_mut(&thread_id.to_string())?; + match event { + EventMsg::UserMessage(user) if thread.preview.is_empty() => { + thread.preview = user.message.clone(); + } + EventMsg::UserMessage(_) + | EventMsg::ThreadSettingsApplied(_) + | EventMsg::ThreadRolledBack(_) + | EventMsg::TurnComplete(_) + | EventMsg::TurnAborted(_) => {} + _ => return None, + } + let updated_at_ms = + i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000) + .ok()?; + thread.updated_at = updated_at_ms / 1_000; + thread.updated_at_ms = updated_at_ms; + Some(thread.clone()) + } } diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index 5878047237..2d2663eab2 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -10,17 +10,25 @@ pub(crate) struct ThreadGoalRequestProcessor { thread_manager: Arc, outgoing: Arc, config: Arc, + thread_store: Arc, thread_state_manager: ThreadStateManager, + thread_catalog_subscriptions: ThreadCatalogSubscriptions, state_db: Option, goal_service: Arc, } impl ThreadGoalRequestProcessor { + #[expect( + clippy::too_many_arguments, + reason = "request processor dependencies stay explicit" + )] pub(crate) fn new( thread_manager: Arc, outgoing: Arc, config: Arc, + thread_store: Arc, thread_state_manager: ThreadStateManager, + thread_catalog_subscriptions: ThreadCatalogSubscriptions, state_db: Option, goal_service: Arc, ) -> Self { @@ -28,7 +36,9 @@ impl ThreadGoalRequestProcessor { thread_manager, outgoing, config, + thread_store, thread_state_manager, + thread_catalog_subscriptions, state_db, goal_service, } @@ -158,6 +168,7 @@ impl ThreadGoalRequestProcessor { .await; self.emit_thread_goal_updated_ordered(thread_id, goal, listener_command_tx) .await; + self.publish_thread_catalog_change(thread_id).await; outcome.apply_runtime_effects(&self.goal_service).await; Ok(()) } @@ -212,10 +223,22 @@ impl ThreadGoalRequestProcessor { if cleared { self.emit_thread_goal_cleared_ordered(thread_id, listener_command_tx) .await; + self.publish_thread_catalog_change(thread_id).await; } Ok(()) } + 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 state_db_for_materialized_thread( &self, thread_id: ThreadId, 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 33ea9c96d3..3233a00092 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -8,7 +8,6 @@ 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, @@ -122,17 +121,6 @@ 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, @@ -280,7 +268,6 @@ 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, @@ -290,6 +277,8 @@ pub(super) async fn ensure_listener_task_running( .. } = listener_task_context; let outgoing_for_task = Arc::clone(&outgoing); + let mut thread_metadata_updated_rx = conversation.subscribe_thread_metadata_updated(); + let is_ephemeral = conversation.config_snapshot().await.ephemeral; tokio::spawn(async move { loop { tokio::select! { @@ -315,6 +304,22 @@ pub(super) async fn ensure_listener_task_running( ) .await; } + stored_thread = async { + match &mut thread_metadata_updated_rx { + Some(rx) => rx.recv().await.ok(), + None => futures::future::pending().await, + } + } => { + if let Some(stored_thread) = stored_thread { + thread_catalog_subscriptions + .publish_thread_summary(thread_summary_from_stored_thread( + stored_thread, + fallback_model_provider.as_str(), + &fallback_cwd, + )) + .await; + } + } event = conversation.next_event() => { let event = match event { Ok(event) => event, @@ -366,14 +371,13 @@ pub(super) async fn ensure_listener_task_running( fallback_model_provider.clone(), ) .await; - if catalog_worthy_event(&event.msg) { + if is_ephemeral + && let Some(summary) = thread_catalog_subscriptions + .update_ephemeral_thread_summary(conversation_id, &event.msg) + .await + { thread_catalog_subscriptions - .publish_thread_change( - &thread_store, - conversation_id, - fallback_model_provider.as_str(), - &fallback_cwd, - ) + .publish_thread_summary(summary) .await; } } 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 9025156cd2..1ac02251cd 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -876,7 +876,6 @@ 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(), @@ -979,7 +978,6 @@ 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(), @@ -1239,6 +1237,12 @@ impl ThreadRequestProcessor { otel.name = "app_server.thread_start.config_snapshot", )) .await; + let stored_thread = thread + .read_thread( + /*include_archived*/ true, /*include_history*/ false, + ) + .await + .ok(); let mut thread = build_thread_from_snapshot( thread_id, session_configured.session_id.to_string(), @@ -1309,7 +1313,16 @@ impl ThreadRequestProcessor { reasoning_effort: config_snapshot.reasoning_effort, multi_agent_mode: config_snapshot.multi_agent_mode, }; - let catalog_summary = thread_summary_from_thread(thread.clone(), /*archived_at*/ None); + let catalog_summary = stored_thread.map_or_else( + || thread_summary_from_thread(thread.clone(), /*archived_at*/ None), + |stored_thread| { + thread_summary_from_stored_thread( + stored_thread, + thread.model_provider.as_str(), + &thread.cwd, + ) + }, + ); let notif = thread_started_notification(thread); listener_task_context .outgoing 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 d67da42d8f..574ba7140b 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -280,9 +280,14 @@ pub(super) fn thread_summary_from_stored_thread( fallback_provider: &str, fallback_cwd: &AbsolutePathBuf, ) -> ThreadSummary { - let archived_at = thread.archived_at.map(|dt| dt.timestamp()); + let created_at_ms = thread.created_at.timestamp_millis(); + let updated_at_ms = thread.updated_at.timestamp_millis(); + let archived_at = thread.archived_at.as_ref().map(chrono::DateTime::timestamp); let (thread, _) = thread_from_stored_thread(thread, fallback_provider, fallback_cwd); - thread_summary_from_thread(thread, archived_at) + let mut summary = thread_summary_from_thread(thread, archived_at); + summary.created_at_ms = created_at_ms; + summary.updated_at_ms = updated_at_ms; + summary } pub(super) fn thread_summary_from_thread( @@ -298,7 +303,9 @@ pub(super) fn thread_summary_from_thread( ephemeral: thread.ephemeral, model_provider: thread.model_provider, created_at: thread.created_at, + created_at_ms: thread.created_at.saturating_mul(1_000), updated_at: thread.updated_at, + updated_at_ms: thread.updated_at.saturating_mul(1_000), archived_at, path: thread.path, cwd: thread.cwd, 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 d068453834..deb1db3992 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -76,7 +76,6 @@ 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, @@ -135,7 +134,6 @@ 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, @@ -151,7 +149,6 @@ impl TurnRequestProcessor { config, config_manager, pending_thread_unloads, - thread_store, thread_state_manager, thread_watch_manager, thread_catalog_subscriptions, @@ -1425,7 +1422,6 @@ 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(), 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 index ff5a006866..a8ea2e5a92 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs @@ -7,6 +7,8 @@ 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::ThreadInjectItemsParams; +use codex_app_server_protocol::ThreadInjectItemsResponse; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadSetNameParams; @@ -23,41 +25,37 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); #[tokio::test] -async fn catalog_subscription_reports_new_empty_thread() -> Result<()> { +async fn catalog_subscription_reports_new_and_injected_thread() -> Result<()> { let codex_home = TempDir::new()?; let mut app = start_app(&codex_home).await?; + subscribe(&mut app).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 start_id = app - .send_thread_start_request(ThreadStartParams::default()) - .await?; - let started = read_response::(&mut app, start_id).await?; - let notification: JSONRPCNotification = timeout( - DEFAULT_READ_TIMEOUT, - app.read_stream_until_notification_message("thread/catalog/changed"), - ) - .await??; - let changed: ThreadCatalogChangedNotification = serde_json::from_value( - notification - .params - .expect("thread/catalog/changed should have params"), - )?; - - assert_eq!(changed.thread.id, started.thread.id); + let started = start_thread(&mut app, ThreadStartParams::default()).await?; + let changed = read_catalog_change(&mut app).await?; + assert_eq!(changed.thread.id, started.id); assert_eq!(changed.thread.preview, ""); assert!( !started - .thread .path .expect("thread path should be present") .exists() ); + let inject_id = app + .send_thread_inject_items_request(ThreadInjectItemsParams { + thread_id: started.id.clone(), + items: vec![serde_json::json!({ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "injected"}], + })], + }) + .await?; + let _: ThreadInjectItemsResponse = read_response(&mut app, inject_id).await?; + let injected = read_catalog_change(&mut app).await?; + assert_eq!(injected.thread.id, started.id); + assert!(injected.thread.updated_at_ms >= changed.thread.updated_at_ms); + Ok(()) } @@ -82,11 +80,7 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> )?; let mut app = start_app(&codex_home).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?; + subscribe(&mut app).await?; let list_id = app .send_thread_list_request(ThreadListParams { @@ -98,6 +92,7 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> source_kinds: None, archived: None, cwd: None, + parent_thread_id: None, use_state_db_only: false, search_term: None, }) @@ -115,12 +110,12 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> let notification: JSONRPCNotification = timeout( DEFAULT_READ_TIMEOUT, - app.read_stream_until_notification_message("thread/catalog/changed"), + app.read_stream_until_notification_message("threadCatalog/changed"), ) .await??; let raw_params = notification .params - .expect("thread/catalog/changed should have params"); + .expect("threadCatalog/changed should have params"); assert_eq!(raw_params["thread"].get("turns"), None); let changed: ThreadCatalogChangedNotification = serde_json::from_value(raw_params)?; @@ -134,6 +129,41 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> Ok(()) } +#[tokio::test] +async fn catalog_subscription_reports_ephemeral_thread_updates() -> Result<()> { + let codex_home = TempDir::new()?; + let mut app = start_app(&codex_home).await?; + subscribe(&mut app).await?; + + let thread = start_thread( + &mut app, + ThreadStartParams { + ephemeral: Some(true), + ..Default::default() + }, + ) + .await?; + let created = read_catalog_change(&mut app).await?; + + let turn_id = app + .send_turn_start_request(codex_app_server_protocol::TurnStartParams { + thread_id: thread.id.clone(), + input: vec![codex_app_server_protocol::UserInput::Text { + text: "Ephemeral catalog preview".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let _: codex_app_server_protocol::TurnStartResponse = read_response(&mut app, turn_id).await?; + let changed = read_catalog_change(&mut app).await?; + assert_eq!(changed.thread.id, thread.id); + assert_eq!(changed.thread.preview, "Ephemeral catalog preview"); + assert!(changed.thread.updated_at_ms >= created.thread.updated_at_ms); + + Ok(()) +} + async fn start_app(codex_home: &TempDir) -> Result { write_mock_responses_config_toml( codex_home.path(), @@ -149,6 +179,38 @@ async fn start_app(codex_home: &TempDir) -> Result { Ok(app) } +async fn subscribe(app: &mut TestAppServer) -> Result<()> { + let subscribe_id = app + .send_raw_request("threadCatalog/subscribe", /*params*/ None) + .await?; + let _: codex_app_server_protocol::ThreadCatalogSubscribeResponse = + read_response(app, subscribe_id).await?; + Ok(()) +} + +async fn start_thread( + app: &mut TestAppServer, + params: ThreadStartParams, +) -> Result { + let start_id = app.send_thread_start_request(params).await?; + let started = read_response::(app, start_id).await?; + Ok(started.thread) +} + +async fn read_catalog_change(app: &mut TestAppServer) -> Result { + let notification: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + app.read_stream_until_notification_message("threadCatalog/changed"), + ) + .await??; + serde_json::from_value( + notification + .params + .ok_or_else(|| anyhow::anyhow!("threadCatalog/changed should have params"))?, + ) + .map_err(Into::into) +} + async fn rename_thread(app: &mut TestAppServer, thread_id: String, name: &str) -> Result<()> { let rename_id = app .send_thread_set_name_request(ThreadSetNameParams { diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index ce95f4810c..42e37be018 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -1080,6 +1080,14 @@ async fn thread_resume_keeps_paused_goal_paused() -> Result<()> { mcp.read_stream_until_notification_message("turn/completed"), ) .await??; + let subscribe_id = mcp + .send_raw_request("threadCatalog/subscribe", /*params*/ None) + .await?; + let _: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(subscribe_id)), + ) + .await??; let goal_id = mcp .send_raw_request( @@ -1102,6 +1110,11 @@ async fn thread_resume_keeps_paused_goal_paused() -> Result<()> { mcp.read_stream_until_notification_message("thread/goal/updated"), ) .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("threadCatalog/changed"), + ) + .await??; mcp.clear_message_buffer(); let resume_id = mcp diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index dccfd4a763..4b41805a9e 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -416,6 +416,12 @@ impl CodexThread { self.codex.next_event().await } + pub fn subscribe_thread_metadata_updated( + &self, + ) -> Option> { + self.codex.session.subscribe_thread_metadata_updated() + } + pub async fn agent_status(&self) -> AgentStatus { self.codex.agent_status().await } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 00d2fdca39..1990570359 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -153,6 +153,7 @@ use codex_thread_store::LiveThreadInitGuard; use codex_thread_store::LocalThreadStore; use codex_thread_store::ReadThreadParams; use codex_thread_store::ResumeThreadParams; +use codex_thread_store::StoredThread; use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadStore; use codex_utils_path_uri::PathUri; @@ -1154,6 +1155,13 @@ impl Session { self.services.live_thread.as_ref() } + pub(crate) fn subscribe_thread_metadata_updated( + &self, + ) -> Option> { + self.live_thread() + .map(LiveThread::subscribe_metadata_updated) + } + /// Flush rollout writes and return the final durability-barrier result. #[instrument(name = "session.flush_rollout", level = "trace", skip_all)] pub(crate) async fn flush_rollout(&self) -> std::io::Result<()> { diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index 793bbf9965..e942892402 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -8,6 +8,7 @@ use codex_rollout::RolloutPersistenceTelemetry; use codex_rollout::measure_and_filter_rollout_items; use codex_rollout::persisted_rollout_items; use tokio::sync::Mutex; +use tokio::sync::broadcast; use tracing::warn; use crate::AppendThreadItemsParams; @@ -35,6 +36,7 @@ pub struct LiveThread { thread_store: Arc, metadata_sync: Arc>, persistence_telemetry: RolloutPersistenceTelemetry, + metadata_updated_tx: broadcast::Sender, } /// Owns a live thread while session initialization is still fallible. @@ -87,6 +89,8 @@ impl Drop for LiveThreadInitGuard { } impl LiveThread { + const METADATA_UPDATED_CHANNEL_CAPACITY: usize = 16; + pub async fn create( thread_store: Arc, params: CreateThreadParams, @@ -94,11 +98,13 @@ impl LiveThread { let thread_id = params.thread_id; let metadata_sync = ThreadMetadataSync::for_create(¶ms).await; thread_store.create_thread(params).await?; + let (metadata_updated_tx, _) = broadcast::channel(Self::METADATA_UPDATED_CHANNEL_CAPACITY); Ok(Self { thread_id, thread_store, metadata_sync: Arc::new(Mutex::new(metadata_sync)), persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id), + metadata_updated_tx, }) } @@ -130,14 +136,20 @@ impl LiveThread { } } } + let (metadata_updated_tx, _) = broadcast::channel(Self::METADATA_UPDATED_CHANNEL_CAPACITY); Ok(Self { thread_id, thread_store, metadata_sync: Arc::new(Mutex::new(metadata_sync)), persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id), + metadata_updated_tx, }) } + pub fn subscribe_metadata_updated(&self) -> broadcast::Receiver { + self.metadata_updated_tx.subscribe() + } + #[tracing::instrument( level = "trace", skip_all, @@ -172,13 +184,15 @@ impl LiveThread { .await .observe_appended_items(canonical_items.as_slice()); if let Some(update) = update { - self.thread_store + let stored_thread = self + .thread_store .update_thread_metadata(UpdateThreadMetadataParams { thread_id: self.thread_id, patch: update.patch.clone(), include_archived: true, }) .await?; + self.publish_metadata_updated(stored_thread); self.metadata_sync .lock() .await @@ -240,7 +254,8 @@ impl LiveThread { include_archived: bool, ) -> ThreadStoreResult<()> { self.flush_pending_metadata_update().await?; - self.thread_store + let stored_thread = self + .thread_store .update_thread_metadata(UpdateThreadMetadataParams { thread_id: self.thread_id, patch: ThreadMetadataPatch { @@ -250,6 +265,7 @@ impl LiveThread { include_archived, }) .await?; + self.publish_metadata_updated(stored_thread); Ok(()) } @@ -259,13 +275,16 @@ impl LiveThread { include_archived: bool, ) -> ThreadStoreResult { self.flush_pending_metadata_update().await?; - self.thread_store + let stored_thread = self + .thread_store .update_thread_metadata(UpdateThreadMetadataParams { thread_id: self.thread_id, patch, include_archived, }) - .await + .await?; + self.publish_metadata_updated(stored_thread.clone()); + Ok(stored_thread) } /// Returns the live local rollout path for legacy local-only callers. @@ -306,17 +325,23 @@ impl LiveThread { let Some(update) = update else { return Ok(()); }; - self.thread_store + let stored_thread = self + .thread_store .update_thread_metadata(UpdateThreadMetadataParams { thread_id: self.thread_id, patch: update.patch.clone(), include_archived: true, }) .await?; + self.publish_metadata_updated(stored_thread); self.metadata_sync .lock() .await .mark_pending_update_applied(&update); Ok(()) } + + fn publish_metadata_updated(&self, stored_thread: StoredThread) { + let _ = self.metadata_updated_tx.send(stored_thread); + } }