From a19d43a40aee5f3308ce57eda604fa085fd9f356 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Wed, 10 Jun 2026 11:22:12 -0700 Subject: [PATCH] Add app-server `thread/delete` API (#25018) ## Why Clients can archive and unarchive threads today, but there is no app-server API for permanently removing a thread. Deletion also needs to cover the full session tree: deleting a main thread should remove spawned subagent threads and the related local metadata instead of leaving orphaned rollout files, goals, or subagent state behind. ## What - Adds the v2 `thread/delete` request and `thread/deleted` notification, with the response shape kept consistent with `thread/archive`. - Implements local hard delete for active and archived rollout files. - Deletes the requested thread's state DB row as the commit point, then best-effort cleans associated state including spawned descendants, goals, spawn edges, logs, dynamic tools, and agent job assignments. - Updates app-server API docs and generated protocol schema/TypeScript fixtures. --- .../schema/json/ClientRequest.json | 35 +++ .../schema/json/ServerNotification.json | 31 ++ .../codex_app_server_protocol.schemas.json | 75 +++++ .../codex_app_server_protocol.v2.schemas.json | 75 +++++ .../schema/json/v2/ThreadDeleteParams.json | 13 + .../schema/json/v2/ThreadDeleteResponse.json | 5 + .../json/v2/ThreadDeletedNotification.json | 13 + .../schema/typescript/ClientRequest.ts | 3 +- .../schema/typescript/ServerNotification.ts | 3 +- .../typescript/v2/ThreadDeleteParams.ts | 5 + .../typescript/v2/ThreadDeleteResponse.ts | 5 + .../v2/ThreadDeletedNotification.ts | 5 + .../schema/typescript/v2/index.ts | 3 + .../src/protocol/common.rs | 6 + .../src/protocol/v2/thread.rs | 19 ++ codex-rs/app-server/README.md | 11 + codex-rs/app-server/src/message_processor.rs | 8 +- codex-rs/app-server/src/request_processors.rs | 5 + .../src/request_processors/thread_delete.rs | 188 +++++++++++ .../request_processors/thread_processor.rs | 60 ++-- .../tests/common/test_app_server.rs | 10 + codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../tests/suite/v2/remote_thread_store.rs | 108 ++++--- .../tests/suite/v2/thread_delete.rs | 200 ++++++++++++ .../app-server/tests/suite/v2/turn_start.rs | 48 +++ codex-rs/core/src/agent/control_tests.rs | 4 +- codex-rs/core/src/thread_manager.rs | 11 +- codex-rs/rollout/src/lib.rs | 1 + codex-rs/rollout/src/session_index.rs | 56 +++- codex-rs/state/src/runtime/threads.rs | 294 +++++++++++++++++- codex-rs/thread-store/src/in_memory.rs | 21 ++ codex-rs/thread-store/src/lib.rs | 1 + .../thread-store/src/local/delete_thread.rs | 210 +++++++++++++ codex-rs/thread-store/src/local/mod.rs | 6 + codex-rs/thread-store/src/store.rs | 4 + codex-rs/thread-store/src/types.rs | 7 + .../tui/src/app/app_server_event_targets.rs | 1 + codex-rs/tui/src/chatwidget/protocol.rs | 1 + 38 files changed, 1464 insertions(+), 88 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteParams.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteResponse.json create mode 100644 codex-rs/app-server-protocol/schema/json/v2/ThreadDeletedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteParams.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteResponse.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeletedNotification.ts create mode 100644 codex-rs/app-server/src/request_processors/thread_delete.rs create mode 100644 codex-rs/app-server/tests/suite/v2/thread_delete.rs create mode 100644 codex-rs/thread-store/src/local/delete_thread.rs diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 2701dfb34..6bdf19536 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -3203,6 +3203,17 @@ ], "type": "object" }, + "ThreadDeleteParams": { + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "type": "object" + }, "ThreadForkParams": { "description": "There are two ways to fork a thread: 1. By thread_id: load the thread from disk by thread_id and fork it into a new thread. 2. By path: load the thread from disk by path and fork it into a new thread.\n\nIf using a non-empty path, the thread_id param will be ignored. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", "properties": { @@ -4514,6 +4525,30 @@ "title": "Thread/archiveRequest", "type": "object" }, + { + "properties": { + "id": { + "$ref": "#/definitions/RequestId" + }, + "method": { + "enum": [ + "thread/delete" + ], + "title": "Thread/deleteRequestMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadDeleteParams" + } + }, + "required": [ + "id", + "method", + "params" + ], + "title": "Thread/deleteRequest", + "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 e0651d172..89831b3fe 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -3523,6 +3523,17 @@ ], "type": "object" }, + "ThreadDeletedNotification": { + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "type": "object" + }, "ThreadGoal": { "properties": { "createdAt": { @@ -5447,6 +5458,26 @@ "title": "Thread/archivedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/deleted" + ], + "title": "Thread/deletedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadDeletedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/deletedNotification", + "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 d65c043a7..fcf014d26 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 @@ -324,6 +324,30 @@ "title": "Thread/archiveRequest", "type": "object" }, + { + "properties": { + "id": { + "$ref": "#/definitions/v2/RequestId" + }, + "method": { + "enum": [ + "thread/delete" + ], + "title": "Thread/deleteRequestMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadDeleteParams" + } + }, + "required": [ + "id", + "method", + "params" + ], + "title": "Thread/deleteRequest", + "type": "object" + }, { "properties": { "id": { @@ -4061,6 +4085,26 @@ "title": "Thread/archivedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/deleted" + ], + "title": "Thread/deletedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadDeletedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/deletedNotification", + "type": "object" + }, { "properties": { "method": { @@ -16053,6 +16097,37 @@ "title": "ThreadCompactStartResponse", "type": "object" }, + "ThreadDeleteParams": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadDeleteParams", + "type": "object" + }, + "ThreadDeleteResponse": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadDeleteResponse", + "type": "object" + }, + "ThreadDeletedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadDeletedNotification", + "type": "object" + }, "ThreadForkParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are two ways to fork a thread: 1. By thread_id: load the thread from disk by thread_id and fork it into a new thread. 2. By path: load the thread from disk by path and fork it into a new thread.\n\nIf using a non-empty path, the thread_id param will be ignored. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", 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 8271dc485..ee0415e13 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 @@ -1263,6 +1263,30 @@ "title": "Thread/archiveRequest", "type": "object" }, + { + "properties": { + "id": { + "$ref": "#/definitions/RequestId" + }, + "method": { + "enum": [ + "thread/delete" + ], + "title": "Thread/deleteRequestMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadDeleteParams" + } + }, + "required": [ + "id", + "method", + "params" + ], + "title": "Thread/deleteRequest", + "type": "object" + }, { "properties": { "id": { @@ -11726,6 +11750,26 @@ "title": "Thread/archivedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/deleted" + ], + "title": "Thread/deletedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadDeletedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/deletedNotification", + "type": "object" + }, { "properties": { "method": { @@ -13870,6 +13914,37 @@ "title": "ThreadCompactStartResponse", "type": "object" }, + "ThreadDeleteParams": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadDeleteParams", + "type": "object" + }, + "ThreadDeleteResponse": { + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadDeleteResponse", + "type": "object" + }, + "ThreadDeletedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadDeletedNotification", + "type": "object" + }, "ThreadForkParams": { "$schema": "http://json-schema.org/draft-07/schema#", "description": "There are two ways to fork a thread: 1. By thread_id: load the thread from disk by thread_id and fork it into a new thread. 2. By path: load the thread from disk by path and fork it into a new thread.\n\nIf using a non-empty path, the thread_id param will be ignored. Empty string path values are treated as absent.\n\nPrefer using thread_id whenever possible.", diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteParams.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteParams.json new file mode 100644 index 000000000..1711e11a2 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteParams.json @@ -0,0 +1,13 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadDeleteParams", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteResponse.json new file mode 100644 index 000000000..ff9f48533 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadDeleteResponse.json @@ -0,0 +1,5 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "ThreadDeleteResponse", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadDeletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadDeletedNotification.json new file mode 100644 index 000000000..53011ea01 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadDeletedNotification.json @@ -0,0 +1,13 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "threadId": { + "type": "string" + } + }, + "required": [ + "threadId" + ], + "title": "ThreadDeletedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts b/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts index c91792c72..b6a13fe30 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts @@ -62,6 +62,7 @@ import type { SkillsListParams } from "./v2/SkillsListParams"; import type { ThreadApproveGuardianDeniedActionParams } from "./v2/ThreadApproveGuardianDeniedActionParams"; import type { ThreadArchiveParams } from "./v2/ThreadArchiveParams"; import type { ThreadCompactStartParams } from "./v2/ThreadCompactStartParams"; +import type { ThreadDeleteParams } from "./v2/ThreadDeleteParams"; import type { ThreadForkParams } from "./v2/ThreadForkParams"; import type { ThreadGoalClearParams } from "./v2/ThreadGoalClearParams"; import type { ThreadGoalGetParams } from "./v2/ThreadGoalGetParams"; @@ -86,4 +87,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/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/usage/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": "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": "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/usage/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": "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 2520c71c3..7cae63a55 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -43,6 +43,7 @@ import type { SkillsChangedNotification } from "./v2/SkillsChangedNotification"; import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNotification"; import type { ThreadArchivedNotification } from "./v2/ThreadArchivedNotification"; import type { ThreadClosedNotification } from "./v2/ThreadClosedNotification"; +import type { ThreadDeletedNotification } from "./v2/ThreadDeletedNotification"; import type { ThreadGoalClearedNotification } from "./v2/ThreadGoalClearedNotification"; import type { ThreadGoalUpdatedNotification } from "./v2/ThreadGoalUpdatedNotification"; import type { ThreadNameUpdatedNotification } from "./v2/ThreadNameUpdatedNotification"; @@ -71,4 +72,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/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/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": "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": "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/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": "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/ThreadDeleteParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteParams.ts new file mode 100644 index 000000000..909ccda71 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteParams.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 ThreadDeleteParams = { threadId: string, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteResponse.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteResponse.ts new file mode 100644 index 000000000..1af1c3077 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeleteResponse.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 ThreadDeleteResponse = Record; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeletedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeletedNotification.ts new file mode 100644 index 000000000..5122a2229 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadDeletedNotification.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 ThreadDeletedNotification = { threadId: string, }; 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 71f4c67cd..b4cd88aad 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -374,6 +374,9 @@ export type { ThreadArchivedNotification } from "./ThreadArchivedNotification"; export type { ThreadClosedNotification } from "./ThreadClosedNotification"; export type { ThreadCompactStartParams } from "./ThreadCompactStartParams"; export type { ThreadCompactStartResponse } from "./ThreadCompactStartResponse"; +export type { ThreadDeleteParams } from "./ThreadDeleteParams"; +export type { ThreadDeleteResponse } from "./ThreadDeleteResponse"; +export type { ThreadDeletedNotification } from "./ThreadDeletedNotification"; export type { ThreadForkParams } from "./ThreadForkParams"; export type { ThreadForkResponse } from "./ThreadForkResponse"; export type { ThreadGoal } from "./ThreadGoal"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 8bd9a2c17..5526f5c54 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -480,6 +480,11 @@ client_request_definitions! { serialization: thread_id(params.thread_id), response: v2::ThreadArchiveResponse, }, + ThreadDelete => "thread/delete" { + params: v2::ThreadDeleteParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadDeleteResponse, + }, ThreadUnsubscribe => "thread/unsubscribe" { params: v2::ThreadUnsubscribeParams, serialization: thread_id(params.thread_id), @@ -1534,6 +1539,7 @@ server_notification_definitions! { ThreadStarted => "thread/started" (v2::ThreadStartedNotification), ThreadStatusChanged => "thread/status/changed" (v2::ThreadStatusChangedNotification), ThreadArchived => "thread/archived" (v2::ThreadArchivedNotification), + ThreadDeleted => "thread/deleted" (v2::ThreadDeletedNotification), ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification), ThreadClosed => "thread/closed" (v2::ThreadClosedNotification), SkillsChanged => "skills/changed" (v2::SkillsChangedNotification), 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 9e1770fde..5b412c2ed 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -594,6 +594,18 @@ pub struct ThreadArchiveParams { #[ts(export_to = "v2/")] pub struct ThreadArchiveResponse {} +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadDeleteParams { + pub thread_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadDeleteResponse {} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -1381,6 +1393,13 @@ pub struct ThreadArchivedNotification { pub thread_id: String, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadDeletedNotification { + pub thread_id: String, +} + #[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 766a5b047..0532ffa70 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -151,6 +151,7 @@ Example with notification opt-out: - `thread/settings/updated` — experimental notification emitted to subscribed clients when a loaded thread’s effective next-turn settings change; includes `threadId` and the full `threadSettings`. - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`). - `thread/archive` — move a thread’s rollout file into the archived directory and attempt to move any spawned descendant thread rollout files; returns `{}` on success and emits `thread/archived` for each archived thread. +- `thread/delete` — hard-delete an active or archived thread and any spawned descendant threads; returns `{}` on success and emits `thread/deleted` for each deleted thread. - `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server keeps the thread loaded and unloads it only after it has had no subscribers and no thread activity for 30 minutes, then emits `thread/closed`. - `thread/name/set` — set or update a thread’s user-facing name for either a loaded thread or a persisted rollout; returns `{}` on success and emits `thread/name/updated` to initialized, opted-in clients. Thread names are not required to be unique; name lookups resolve to the most recently updated thread. - `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`. @@ -612,6 +613,16 @@ Use `thread/archive` to move the persisted rollout (stored as a JSONL file on di An archived thread will not appear in `thread/list` unless `archived` is set to `true`. +### Example: Delete a thread + +Use `thread/delete` to hard-delete a thread and its spawned descendant threads. Existing rollout files and associated metadata must be removed before the request succeeds; missing rollout files are treated as already deleted. + +```json +{ "method": "thread/delete", "id": 23, "params": { "threadId": "thr_b" } } +{ "id": 23, "result": {} } +{ "method": "thread/deleted", "params": { "threadId": "thr_b" } } +``` + ### Example: Unarchive a thread Use `thread/unarchive` to move an archived rollout back into the sessions directory. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index ce239a271..30e57208e 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -400,7 +400,7 @@ impl MessageProcessor { Arc::clone(&thread_manager), Arc::clone(&config), feedback, - log_db, + log_db.clone(), state_db.clone(), ); let git_processor = GitRequestProcessor::new(); @@ -454,6 +454,7 @@ impl MessageProcessor { Arc::clone(&thread_list_state_permit), thread_goal_processor.clone(), state_db, + log_db, Arc::clone(&skills_watcher), ); let turn_processor = TurnRequestProcessor::new( @@ -1071,6 +1072,11 @@ impl MessageProcessor { .thread_archive(request_id.clone(), params) .await } + ClientRequest::ThreadDelete { params, .. } => { + self.thread_processor + .thread_delete(request_id.clone(), params) + .await + } ClientRequest::ThreadIncrementElicitation { params, .. } => { self.thread_processor .thread_increment_elicitation(params) diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 2bce5b8dd..83e704cdf 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -183,6 +183,9 @@ use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadCompactStartResponse; use codex_app_server_protocol::ThreadDecrementElicitationParams; use codex_app_server_protocol::ThreadDecrementElicitationResponse; +use codex_app_server_protocol::ThreadDeleteParams; +use codex_app_server_protocol::ThreadDeleteResponse; +use codex_app_server_protocol::ThreadDeletedNotification; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadGoal; @@ -419,6 +422,7 @@ use codex_rollout::state_db::reconcile_rollout; use codex_state::ThreadMetadata; use codex_state::log_db::LogDbLayer; use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams; +use codex_thread_store::DeleteThreadParams as StoreDeleteThreadParams; use codex_thread_store::GitInfoPatch as StoreGitInfoPatch; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; use codex_thread_store::LocalThreadStore; @@ -539,6 +543,7 @@ fn resolve_runtime_workspace_roots(workspace_roots: Vec) -> Vec mod config_errors; mod request_errors; +mod thread_delete; mod thread_goal_processor; mod thread_lifecycle; mod thread_resume_redaction; diff --git a/codex-rs/app-server/src/request_processors/thread_delete.rs b/codex-rs/app-server/src/request_processors/thread_delete.rs new file mode 100644 index 000000000..fc7e58573 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/thread_delete.rs @@ -0,0 +1,188 @@ +//! `thread/delete` request handling. + +use super::thread_processor::core_thread_write_error; +use super::thread_processor::unsupported_thread_store_operation; +use super::*; + +impl ThreadRequestProcessor { + pub(crate) async fn thread_delete( + &self, + request_id: ConnectionRequestId, + params: ThreadDeleteParams, + ) -> Result, JSONRPCErrorError> { + let mut deleted_thread_ids = Vec::new(); + let result = { + let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; + self.thread_delete_response(params, &mut deleted_thread_ids) + .await + }; + match result { + Ok(response) => { + self.outgoing + .send_response(request_id.clone(), response) + .await; + self.send_thread_deleted_notifications(deleted_thread_ids) + .await; + Ok(None) + } + Err(error) => Err(error), + } + } + + async fn thread_delete_response( + &self, + params: ThreadDeleteParams, + deleted_thread_ids: &mut Vec, + ) -> Result { + let thread_id = ThreadId::from_string(¶ms.thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; + + let mut thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?; + let mut seen = thread_ids.iter().copied().collect::>(); + + match self + .thread_manager + .list_agent_subtree_thread_ids(thread_id) + .await + { + Ok(live_thread_ids) => { + for live_thread_id in live_thread_ids { + if seen.insert(live_thread_id) { + thread_ids.push(live_thread_id); + } + } + } + Err(err) => return Err(core_thread_write_error("delete thread", err)), + } + + self.validate_root_thread_delete(thread_id, thread_ids.len() > 1) + .await?; + for thread_id_to_delete in thread_ids.iter().copied() { + self.prepare_thread_for_delete(thread_id_to_delete).await; + } + + let mut delete_order: Vec<_> = thread_ids.iter().skip(1).rev().copied().collect(); + delete_order.push(thread_id); + + for thread_id_to_delete in delete_order.iter().copied() { + match self + .thread_store + .delete_thread(StoreDeleteThreadParams { + thread_id: thread_id_to_delete, + }) + .await + { + Ok(()) => {} + Err(ThreadStoreError::ThreadNotFound { .. }) => { + warn!( + "thread {thread_id_to_delete} was already missing while deleting {thread_id}" + ); + } + Err(err) => { + return Err(thread_store_delete_error(err)); + } + } + } + + if let Some(state_db) = self.state_db.as_ref() { + state_db + .delete_threads_strict(thread_ids.as_slice()) + .await + .map_err(|err| { + internal_error(format!( + "failed to delete app-server state for {thread_id}: {err}" + )) + })?; + } + + deleted_thread_ids.extend( + delete_order + .into_iter() + .map(|thread_id| thread_id.to_string()), + ); + Ok(ThreadDeleteResponse {}) + } + + async fn send_thread_deleted_notifications(&self, deleted_thread_ids: Vec) { + for thread_id in deleted_thread_ids { + self.outgoing + .send_server_notification(ServerNotification::ThreadDeleted( + ThreadDeletedNotification { thread_id }, + )) + .await; + } + } + + async fn validate_root_thread_delete( + &self, + thread_id: ThreadId, + has_descendants: bool, + ) -> Result<(), JSONRPCErrorError> { + if let Ok(thread) = self.thread_manager.get_thread(thread_id).await { + if !thread.config_snapshot().await.ephemeral { + return Ok(()); + } + return Err(invalid_request(format!( + "thread is not persisted and cannot be deleted: {thread_id}" + ))); + } + match self + .thread_store + .read_thread(StoreReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + { + Ok(_) => Ok(()), + Err(ThreadStoreError::ThreadNotFound { .. }) => { + if has_descendants { + return Ok(()); + } + let Some(state_db) = self.state_db.as_ref() else { + return Err(thread_store_delete_error( + ThreadStoreError::ThreadNotFound { thread_id }, + )); + }; + if state_db + .get_thread(thread_id) + .await + .map_err(|err| { + internal_error(format!( + "failed to read app-server state for {thread_id}: {err}" + )) + })? + .is_some() + { + Ok(()) + } else { + Err(thread_store_delete_error( + ThreadStoreError::ThreadNotFound { thread_id }, + )) + } + } + Err(err) => Err(thread_store_delete_error(err)), + } + } + + async fn prepare_thread_for_delete(&self, thread_id: ThreadId) { + self.prepare_thread_for_removal(thread_id, "delete").await; + if let Some(log_db) = self.log_db.as_ref() { + log_db.flush().await; + } + } +} + +fn thread_store_delete_error(err: ThreadStoreError) -> JSONRPCErrorError { + match err { + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("thread not found: {thread_id}")) + } + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::Unsupported { operation } => { + unsupported_thread_store_operation(operation) + } + err => internal_error(format!("failed to delete thread: {err}")), + } +} 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 b10358d8d..58719d55b 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -325,6 +325,7 @@ pub(crate) struct ThreadRequestProcessor { pub(super) thread_list_state_permit: Arc, pub(super) thread_goal_processor: ThreadGoalRequestProcessor, pub(super) state_db: Option, + pub(super) log_db: Option, pub(super) background_tasks: TaskTracker, pub(super) skills_watcher: Arc, } @@ -356,6 +357,7 @@ impl ThreadRequestProcessor { thread_list_state_permit: Arc, thread_goal_processor: ThreadGoalRequestProcessor, state_db: Option, + log_db: Option, skills_watcher: Arc, ) -> Self { Self { @@ -372,6 +374,7 @@ impl ThreadRequestProcessor { thread_list_state_permit, thread_goal_processor, state_db, + log_db, background_tasks: TaskTracker::new(), skills_watcher, } @@ -696,7 +699,7 @@ impl ThreadRequestProcessor { Ok((thread_id, thread)) } - async fn acquire_thread_list_state_permit( + pub(super) async fn acquire_thread_list_state_permit( &self, ) -> Result, JSONRPCErrorError> { self.thread_list_state_permit @@ -768,6 +771,10 @@ impl ThreadRequestProcessor { } async fn prepare_thread_for_archive(&self, thread_id: ThreadId) { + self.prepare_thread_for_removal(thread_id, "archive").await; + } + + pub(super) async fn prepare_thread_for_removal(&self, thread_id: ThreadId, operation: &str) { let removed_conversation = self.thread_manager.remove_thread(&thread_id).await; if let Some(conversation) = removed_conversation { info!("thread {thread_id} was active; shutting down"); @@ -775,11 +782,11 @@ impl ThreadRequestProcessor { ThreadShutdownResult::Complete => {} ThreadShutdownResult::SubmitFailed => { error!( - "failed to submit Shutdown to thread {thread_id}; proceeding with archive" + "failed to submit Shutdown to thread {thread_id}; proceeding with {operation}" ); } ThreadShutdownResult::TimedOut => { - warn!("thread {thread_id} shutdown timed out; proceeding with archive"); + warn!("thread {thread_id} shutdown timed out; proceeding with {operation}"); } } } @@ -1312,23 +1319,7 @@ impl ThreadRequestProcessor { let thread_id = ThreadId::from_string(¶ms.thread_id) .map_err(|err| invalid_request(format!("invalid session id: {err}")))?; - let mut thread_ids = vec![thread_id]; - if let Some(state_db_ctx) = self.state_db.as_ref() { - let descendants = state_db_ctx - .list_thread_spawn_descendants(thread_id) - .await - .map_err(|err| { - internal_error(format!( - "failed to list spawned descendants for session {thread_id}: {err}" - )) - })?; - let mut seen = HashSet::from([thread_id]); - for descendant_id in descendants { - if seen.insert(descendant_id) { - thread_ids.push(descendant_id); - } - } - } + let thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?; let mut archive_thread_ids = Vec::new(); match self @@ -1413,6 +1404,31 @@ impl ThreadRequestProcessor { Ok((ThreadArchiveResponse {}, archived_thread_ids)) } + pub(super) async fn state_db_spawn_subtree_thread_ids( + &self, + thread_id: ThreadId, + ) -> Result, JSONRPCErrorError> { + let mut thread_ids = vec![thread_id]; + let Some(state_db_ctx) = self.state_db.as_ref() else { + return Ok(thread_ids); + }; + let mut seen = HashSet::from([thread_id]); + let descendants = state_db_ctx + .list_thread_spawn_descendants(thread_id) + .await + .map_err(|err| { + internal_error(format!( + "failed to list spawned descendants for thread id {thread_id}: {err}" + )) + })?; + for descendant_id in descendants { + if seen.insert(descendant_id) { + thread_ids.push(descendant_id); + } + } + Ok(thread_ids) + } + async fn thread_increment_elicitation_inner( &self, params: ThreadIncrementElicitationParams, @@ -3909,7 +3925,7 @@ fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { } } -fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError { +pub(super) fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError { method_not_found(format!("{operation} is not supported yet")) } @@ -4030,7 +4046,7 @@ fn conversation_summary_rollout_path_read_error( } } -fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPCErrorError { +pub(super) fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPCErrorError { match err { CodexErr::ThreadNotFound(thread_id) => { invalid_request(format!("thread not found: {thread_id}")) diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index 646de6ad2..91c42e2fe 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -79,6 +79,7 @@ use codex_app_server_protocol::SkillsExtraRootsSetParams; use codex_app_server_protocol::SkillsListParams; use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadCompactStartParams; +use codex_app_server_protocol::ThreadDeleteParams; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadInjectItemsParams; use codex_app_server_protocol::ThreadListParams; @@ -456,6 +457,15 @@ impl TestAppServer { self.send_request("thread/archive", params).await } + /// Send a `thread/delete` JSON-RPC request. + pub async fn send_thread_delete_request( + &mut self, + params: ThreadDeleteParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/delete", params).await + } + /// Send a `thread/name/set` JSON-RPC request. pub async fn send_thread_set_name_request( &mut self, diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 77d4b3758..e038a131e 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -50,6 +50,7 @@ mod review; mod safety_check_downgrade; mod skills_list; mod thread_archive; +mod thread_delete; mod thread_fork; mod thread_inject_items; mod thread_list; diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index ca2121dd4..18d1dcd43 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -20,6 +20,7 @@ use std::sync::Arc; use anyhow::Result; use app_test_support::create_mock_responses_server_repeating_assistant; use codex_app_server::in_process; +use codex_app_server::in_process::InProcessClientHandle; use codex_app_server::in_process::InProcessServerEvent; use codex_app_server::in_process::InProcessStartArgs; use codex_app_server_protocol::ClientInfo; @@ -27,6 +28,8 @@ use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ThreadDeleteParams; +use codex_app_server_protocol::ThreadDeleteResponse; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadResumeParams; @@ -42,8 +45,14 @@ use codex_core::config::Config; use codex_core::config::ConfigBuilder; use codex_exec_server::EnvironmentManager; use codex_feedback::CodexFeedback; +use codex_protocol::ThreadId; +use codex_protocol::models::BaseInstructions; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadMemoryMode; +use codex_thread_store::CreateThreadParams as StoreCreateThreadParams; use codex_thread_store::InMemoryThreadStore; +use codex_thread_store::ThreadPersistenceMetadata; +use codex_thread_store::ThreadStore; use pretty_assertions::assert_eq; use tempfile::TempDir; use tokio::time::timeout; @@ -52,7 +61,7 @@ use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] -async fn thread_start_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()> +async fn thread_delete_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; @@ -61,43 +70,10 @@ async fn thread_start_with_non_local_thread_store_does_not_create_local_persiste // here so this regression stays focused on thread persistence artifacts. create_config_toml_with_thread_store(codex_home.path(), &server.uri(), &store_id)?; - let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .loader_overrides(loader_overrides.clone()) - .build() - .await?; - let thread_store = InMemoryThreadStore::for_id(store_id.clone()); let _in_memory_store = InMemoryThreadStoreId { store_id }; - let mut client = in_process::start(InProcessStartArgs { - arg0_paths: Arg0DispatchPaths::default(), - config: Arc::new(config), - cli_overrides: Vec::new(), - loader_overrides, - strict_config: false, - cloud_config_bundle: CloudConfigBundleLoader::default(), - thread_config_loader: Arc::new(NoopThreadConfigLoader), - feedback: CodexFeedback::new(), - log_db: None, - state_db: None, - environment_manager: Arc::new(EnvironmentManager::default_for_tests()), - config_warnings: Vec::new(), - session_source: SessionSource::Cli, - enable_codex_api_key_env: false, - initialize: InitializeParams { - client_info: ClientInfo { - name: "codex-app-server-tests".to_string(), - title: None, - version: "0.1.0".to_string(), - }, - capabilities: None, - }, - channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, - }) - .await?; + let mut client = start_in_process_server(codex_home.path()).await?; let response = client .request(ClientRequest::ThreadStart { @@ -166,11 +142,39 @@ async fn thread_start_with_non_local_thread_store_does_not_create_local_persiste assert_eq!(data[0].id, thread.id); assert_eq!(data[0].path, None); + delete_thread(&client, /*request_id*/ 4, thread.id.clone()).await?; + let unloaded_thread_id = ThreadId::from_string(&Uuid::new_v4().to_string())?; + thread_store + .create_thread(StoreCreateThreadParams { + thread_id: unloaded_thread_id, + extra_config: None, + forked_from_id: None, + parent_thread_id: None, + source: SessionSource::Cli, + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + multi_agent_version: None, + metadata: ThreadPersistenceMetadata { + cwd: Some(codex_home.path().to_path_buf()), + model_provider: "mock_provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await?; + delete_thread( + &client, + /*request_id*/ 5, + unloaded_thread_id.to_string(), + ) + .await?; + client.shutdown().await?; let calls = thread_store.calls().await; - assert_eq!(calls.create_thread, 1); + assert_eq!(calls.create_thread, 2); assert_eq!(calls.list_threads, 1); + assert_eq!(calls.delete_thread, 2); assert!( calls.append_items > 0, "turn/start should append rollout items through the injected store" @@ -269,10 +273,24 @@ async fn cold_thread_resume_reuses_non_local_history_probe() -> Result<()> { Ok(()) } +async fn start_in_process_server(codex_home: &Path) -> Result { + let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); + let config = Arc::new( + ConfigBuilder::default() + .codex_home(codex_home.to_path_buf()) + .fallback_cwd(Some(codex_home.to_path_buf())) + .loader_overrides(loader_overrides.clone()) + .build() + .await?, + ); + + Ok(start_in_process_client(config, loader_overrides).await?) +} + async fn start_in_process_client( config: Arc, loader_overrides: LoaderOverrides, -) -> std::io::Result { +) -> std::io::Result { in_process::start(InProcessStartArgs { arg0_paths: Arg0DispatchPaths::default(), config, @@ -301,6 +319,22 @@ async fn start_in_process_client( .await } +async fn delete_thread( + client: &InProcessClientHandle, + request_id: i64, + thread_id: String, +) -> Result<()> { + let response = client + .request(ClientRequest::ThreadDelete { + request_id: RequestId::Integer(request_id), + params: ThreadDeleteParams { thread_id }, + }) + .await? + .map_err(|error| anyhow::anyhow!("thread/delete failed: {}", error.message))?; + let _: ThreadDeleteResponse = serde_json::from_value(response)?; + Ok(()) +} + fn assert_no_local_persistence_artifacts(codex_home: &Path) -> Result<()> { // These are the observable tripwires for accidental local persistence. If a // future code path constructs a local rollout/session store or opens the diff --git a/codex-rs/app-server/tests/suite/v2/thread_delete.rs b/codex-rs/app-server/tests/suite/v2/thread_delete.rs new file mode 100644 index 000000000..bef155f9a --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_delete.rs @@ -0,0 +1,200 @@ +use anyhow::Result; +use app_test_support::TestAppServer; +use app_test_support::create_fake_rollout; +use app_test_support::to_response; +use codex_app_server_protocol::JSONRPCError; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadDeleteParams; +use codex_app_server_protocol::ThreadDeleteResponse; +use codex_app_server_protocol::ThreadDeletedNotification; +use codex_app_server_protocol::ThreadLoadedListParams; +use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_core::find_thread_path_by_id_str; +use codex_protocol::ThreadId; +use codex_state::DirectionalThreadSpawnEdgeStatus; +use codex_state::StateRuntime; +use pretty_assertions::assert_eq; +use std::path::Path; +use tempfile::TempDir; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[tokio::test] +async fn thread_delete_deletes_spawned_descendants() -> Result<()> { + let codex_home = TempDir::new()?; + + let parent_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 0, "parent")?; + let child_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 1, "child")?; + let grandchild_id = + create_delete_test_rollout(codex_home.path(), /*minute*/ 2, "grandchild")?; + + let state_db = + StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()).await?; + let parent_thread_id = ThreadId::from_string(&parent_id)?; + let child_thread_id = ThreadId::from_string(&child_id)?; + let grandchild_thread_id = ThreadId::from_string(&grandchild_id)?; + + for (parent, child, status) in [ + ( + parent_thread_id, + child_thread_id, + DirectionalThreadSpawnEdgeStatus::Closed, + ), + ( + child_thread_id, + grandchild_thread_id, + DirectionalThreadSpawnEdgeStatus::Open, + ), + ] { + state_db + .upsert_thread_spawn_edge(parent, child, status) + .await?; + } + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let delete_id = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: parent_id.clone(), + }) + .await?; + let delete_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(delete_id)), + ) + .await??; + let _: ThreadDeleteResponse = to_response::(delete_resp)?; + + let mut deleted_ids = Vec::new(); + for _ in 0..3 { + let notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/deleted"), + ) + .await??; + let deleted_notification: ThreadDeletedNotification = serde_json::from_value( + notification + .params + .expect("thread/deleted notification params"), + )?; + deleted_ids.push(deleted_notification.thread_id); + } + assert_eq!(deleted_ids, vec![grandchild_id, child_id, parent_id]); + + for thread_id in [parent_thread_id, child_thread_id, grandchild_thread_id] { + let rollout_path = find_thread_path_by_id_str( + codex_home.path(), + &thread_id.to_string(), + /*state_db_ctx*/ None, + ) + .await?; + assert!( + rollout_path.is_none(), + "expected active rollout for {thread_id} to be deleted" + ); + } + assert_eq!( + state_db + .list_thread_spawn_descendants(parent_thread_id) + .await?, + Vec::::new() + ); + Ok(()) +} + +fn create_delete_test_rollout(codex_home: &Path, minute: u8, preview: &str) -> Result { + create_fake_rollout( + codex_home, + &format!("2025-01-01T00-{minute:02}-00"), + &format!("2025-01-01T00:{minute:02}:00Z"), + preview, + Some("mock_provider"), + /*git_info*/ None, + ) +} + +#[tokio::test] +async fn thread_delete_handles_live_threads_before_rollout_exists() -> Result<()> { + let codex_home = TempDir::new()?; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let persisted_thread = to_response::(start_resp)?.thread; + let rollout_path = find_thread_path_by_id_str( + codex_home.path(), + &persisted_thread.id, + /*state_db_ctx*/ None, + ) + .await?; + assert_eq!(rollout_path, None); + + let delete_id = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: persisted_thread.id, + }) + .await?; + let delete_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(delete_id)), + ) + .await??; + let _: ThreadDeleteResponse = to_response::(delete_resp)?; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + ephemeral: Some(true), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + + let delete_id = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: thread.id.clone(), + }) + .await?; + let delete_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(delete_id)), + ) + .await??; + let expected_message = format!( + "thread is not persisted and cannot be deleted: {}", + thread.id + ); + assert_eq!(delete_err.error.message, expected_message); + + let list_id = mcp + .send_thread_loaded_list_request(ThreadLoadedListParams::default()) + .await?; + let list_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(list_id)), + ) + .await??; + let ThreadLoadedListResponse { mut data, .. } = + to_response::(list_resp)?; + data.sort(); + assert_eq!(data, vec![thread.id]); + + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 823851b71..7fc0d8c71 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -43,7 +43,12 @@ use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerRequestResolvedNotification; use codex_app_server_protocol::SubAgentActivityKind; use codex_app_server_protocol::TextElement; +use codex_app_server_protocol::ThreadDeleteParams; +use codex_app_server_protocol::ThreadDeleteResponse; +use codex_app_server_protocol::ThreadDeletedNotification; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadLoadedListParams; +use codex_app_server_protocol::ThreadLoadedListResponse; use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; @@ -3370,6 +3375,49 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<() assert_eq!(turn_completed.thread_id, thread.id); assert_eq!(turn_completed.turn.id, turn.turn.id); + // Reuse this live spawn setup to cover thread/delete's ThreadManager descendant path. + let delete_req = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: thread.id.clone(), + }) + .await?; + let delete_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(delete_req)), + ) + .await??; + let _: ThreadDeleteResponse = to_response::(delete_resp)?; + + let mut deleted_thread_ids = Vec::new(); + for _ in 0..2 { + let deleted_notif = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("thread/deleted"), + ) + .await??; + let deleted: ThreadDeletedNotification = serde_json::from_value( + deleted_notif + .params + .expect("thread/deleted notification params"), + )?; + deleted_thread_ids.push(deleted.thread_id); + } + assert_eq!( + deleted_thread_ids, + vec![receiver_thread_id, thread.id.clone()] + ); + + let list_req = mcp + .send_thread_loaded_list_request(ThreadLoadedListParams::default()) + .await?; + let list_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(list_req)), + ) + .await??; + let ThreadLoadedListResponse { data, .. } = to_response::(list_resp)?; + assert_eq!(data, Vec::::new()); + Ok(()) } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 772da10c3..ac42984b5 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -2491,7 +2491,7 @@ async fn list_agent_subtree_thread_ids_includes_anonymous_and_closed_descendants } #[tokio::test] -async fn list_agent_subtree_thread_ids_includes_live_descendants_without_state_db() { +async fn list_agent_subtree_thread_ids_finds_live_descendants_of_unloaded_root() { let (_home, config) = test_config().await; let manager = ThreadManager::with_models_provider_home_and_state_for_tests( CodexAuth::from_api_key("dummy"), @@ -2536,6 +2536,8 @@ async fn list_agent_subtree_thread_ids_includes_live_descendants_without_state_d .await .expect("grandchild spawn should succeed"); + manager.remove_thread(&parent_thread_id).await; + let mut subtree_thread_ids = manager .list_agent_subtree_thread_ids(parent_thread_id) .await diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index d5f54d555..5758c0cad 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -515,14 +515,12 @@ impl ThreadManager { &self, thread_id: ThreadId, ) -> CodexResult> { - let thread = self.state.get_thread(thread_id).await?; - let mut subtree_thread_ids = Vec::new(); let mut seen_thread_ids = HashSet::new(); subtree_thread_ids.push(thread_id); seen_thread_ids.insert(thread_id); - if let Some(state_db_ctx) = thread.state_db() { + if let Some(state_db_ctx) = self.state.state_db() { for status in [ DirectionalThreadSpawnEdgeStatus::Open, DirectionalThreadSpawnEdgeStatus::Closed, @@ -541,11 +539,8 @@ impl ThreadManager { } } - for descendant_id in thread - .codex - .session - .services - .agent_control + for descendant_id in self + .agent_control() .list_live_agent_subtree_thread_ids(thread_id) .await? { diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 9e6db0fb8..23be07996 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -73,6 +73,7 @@ pub use session_index::append_thread_name; pub use session_index::find_thread_meta_by_name_str; pub use session_index::find_thread_name_by_id; pub use session_index::find_thread_names_by_ids; +pub use session_index::remove_thread_name_entries; pub use state_db::StateDbHandle; pub use state_db::sqlite_telemetry_recorder; diff --git a/codex-rs/rollout/src/session_index.rs b/codex-rs/rollout/src/session_index.rs index e22751535..96ff9d825 100644 --- a/codex-rs/rollout/src/session_index.rs +++ b/codex-rs/rollout/src/session_index.rs @@ -1,21 +1,25 @@ use std::collections::HashMap; use std::collections::HashSet; use std::fs::File; +use std::io::ErrorKind; use std::io::Read; use std::io::Seek; use std::io::SeekFrom; +use std::io::Write; use std::path::Path; use std::path::PathBuf; +use std::sync::LazyLock; +use std::sync::Mutex; use codex_protocol::ThreadId; use codex_protocol::protocol::SessionMetaLine; use serde::Deserialize; use serde::Serialize; use tokio::io::AsyncBufReadExt; -use tokio::io::AsyncWriteExt; const SESSION_INDEX_FILE: &str = "session_index.jsonl"; const READ_CHUNK_SIZE: usize = 8192; +static SESSION_INDEX_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct SessionIndexEntry { @@ -25,7 +29,7 @@ pub struct SessionIndexEntry { } /// Append a thread name update to the session index. -/// The index is append-only; the most recent entry wins when resolving names or ids. +/// Name updates are append-only; the most recent entry wins when resolving names or ids. pub async fn append_thread_name( codex_home: &Path, thread_id: ThreadId, @@ -46,24 +50,60 @@ pub async fn append_thread_name( } /// Append a raw session index entry to `session_index.jsonl`. -/// The file is append-only; consumers scan from the end to find the newest match. +/// Consumers scan from the end to find the newest match. pub async fn append_session_index_entry( codex_home: &Path, entry: &SessionIndexEntry, ) -> std::io::Result<()> { + let _guard = SESSION_INDEX_LOCK + .lock() + .map_err(|err| std::io::Error::other(err.to_string()))?; let path = session_index_path(codex_home); - let mut file = tokio::fs::OpenOptions::new() + let mut file = std::fs::OpenOptions::new() .create(true) .append(true) - .open(&path) - .await?; + .open(&path)?; let mut line = serde_json::to_string(entry).map_err(std::io::Error::other)?; line.push('\n'); - file.write_all(line.as_bytes()).await?; - file.flush().await?; + file.write_all(line.as_bytes())?; + file.flush()?; Ok(()) } +/// Remove all recorded names for a thread from the session index. +pub async fn remove_thread_name_entries( + codex_home: &Path, + thread_id: ThreadId, +) -> std::io::Result<()> { + let _guard = SESSION_INDEX_LOCK + .lock() + .map_err(|err| std::io::Error::other(err.to_string()))?; + let path = session_index_path(codex_home); + let contents = match std::fs::read_to_string(&path) { + Ok(contents) => contents, + Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()), + Err(err) => return Err(err), + }; + let mut removed = false; + let mut remaining = String::with_capacity(contents.len()); + for line in contents.lines() { + let should_remove = serde_json::from_str::(line.trim()) + .is_ok_and(|entry| entry.id == thread_id); + if should_remove { + removed = true; + } else { + remaining.push_str(line); + remaining.push('\n'); + } + } + if !removed { + return Ok(()); + } + let temp_path = path.with_extension("jsonl.tmp"); + std::fs::write(&temp_path, remaining)?; + std::fs::rename(temp_path, path) +} + /// Find the latest thread name for a thread id, if any. pub async fn find_thread_name_by_id( codex_home: &Path, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 1ccfab5e1..98eff16c5 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -884,17 +884,117 @@ ON CONFLICT(id) DO UPDATE SET self.upsert_thread(&metadata).await } - /// Delete a thread metadata row by id. + /// Delete a thread and all associated state by id. pub async fn delete_thread(&self, thread_id: ThreadId) -> anyhow::Result { - let result = sqlx::query("DELETE FROM threads WHERE id = ?") - .bind(thread_id.to_string()) - .execute(self.pool.as_ref()) - .await?; - let rows_affected = result.rows_affected(); - self.memories.delete_thread_memory(thread_id).await?; - if rows_affected > 0 { - let _ = self.thread_goals.delete_thread_goal(thread_id).await?; + self.delete_threads_strict(&[thread_id]).await + } + + /// Delete a set of threads and all associated state. + /// + /// Spawn edges and thread rows are deleted last so a failed delete can be retried with enough + /// state left to rediscover the same spawned subtree. + pub async fn delete_threads_strict(&self, thread_ids: &[ThreadId]) -> anyhow::Result { + if thread_ids.is_empty() { + return Ok(0); } + + let thread_id_strings = thread_ids + .iter() + .map(ThreadId::to_string) + .collect::>(); + for (thread_id, thread_id_string) in thread_ids.iter().zip(&thread_id_strings) { + sqlx::query("DELETE FROM logs WHERE thread_id = ?") + .bind(thread_id_string) + .execute(self.logs_pool.as_ref()) + .await?; + self.memories.delete_thread_memory(*thread_id).await?; + self.thread_goals.delete_thread_goal(*thread_id).await?; + } + + let now = Utc::now().timestamp(); + let mut tx = self.pool.begin().await?; + for thread_id_string in &thread_id_strings { + for parent_thread_id_string in &thread_id_strings { + // If both the job runner and worker are being deleted, requeueing + // the worker item would leave a running job with no loop to consume it. + sqlx::query( + r#" +UPDATE agent_jobs +SET status = ?, updated_at = ?, completed_at = ?, last_error = ? +WHERE status IN (?, ?) + AND id IN ( + SELECT item.job_id + FROM agent_job_items AS item + JOIN thread_spawn_edges AS edge ON edge.child_thread_id = item.assigned_thread_id + WHERE item.status = ? AND item.assigned_thread_id = ? AND edge.parent_thread_id = ? + ) + "#, + ) + .bind(AgentJobStatus::Cancelled.as_str()) + .bind(now) + .bind(now) + .bind("agent job runner thread was deleted") + .bind(AgentJobStatus::Pending.as_str()) + .bind(AgentJobStatus::Running.as_str()) + .bind(AgentJobItemStatus::Running.as_str()) + .bind(thread_id_string) + .bind(parent_thread_id_string) + .execute(&mut *tx) + .await?; + } + sqlx::query("DELETE FROM thread_dynamic_tools WHERE thread_id = ?") + .bind(thread_id_string) + .execute(&mut *tx) + .await?; + sqlx::query( + r#" +UPDATE agent_job_items +SET + status = ?, + assigned_thread_id = NULL, + updated_at = ?, + last_error = ? +WHERE assigned_thread_id = ? AND status = ? + "#, + ) + .bind(AgentJobItemStatus::Pending.as_str()) + .bind(now) + .bind("assigned thread was deleted") + .bind(thread_id_string) + .bind(AgentJobItemStatus::Running.as_str()) + .execute(&mut *tx) + .await?; + sqlx::query( + r#" +UPDATE agent_job_items +SET assigned_thread_id = NULL, updated_at = ? +WHERE assigned_thread_id = ? + "#, + ) + .bind(now) + .bind(thread_id_string) + .execute(&mut *tx) + .await?; + } + for thread_id_string in &thread_id_strings { + sqlx::query( + "DELETE FROM thread_spawn_edges WHERE parent_thread_id = ? OR child_thread_id = ?", + ) + .bind(thread_id_string) + .bind(thread_id_string) + .execute(&mut *tx) + .await?; + } + let mut rows_affected = 0; + for thread_id_string in &thread_id_strings { + rows_affected += sqlx::query("DELETE FROM threads WHERE id = ?") + .bind(thread_id_string) + .execute(&mut *tx) + .await? + .rows_affected(); + } + tx.commit().await?; + Ok(rows_affected) } } @@ -1136,12 +1236,14 @@ mod tests { use crate::DirectionalThreadSpawnEdgeStatus; use crate::runtime::test_support::test_thread_metadata; use crate::runtime::test_support::unique_temp_dir; + use anyhow::Result; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use pretty_assertions::assert_eq; + use serde_json::json; use std::path::PathBuf; #[tokio::test] @@ -1182,6 +1284,180 @@ mod tests { assert_eq!(memory_mode, "disabled"); } + #[tokio::test] + async fn delete_thread_cleans_associated_state() -> Result<()> { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()).await?; + let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000401")?; + let child_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000402")?; + runtime + .upsert_thread(&test_thread_metadata( + &codex_home, + thread_id, + codex_home.clone(), + )) + .await?; + seed_thread_cleanup_state(&runtime, thread_id, child_thread_id).await?; + sqlx::query("INSERT INTO thread_dynamic_tools (thread_id, position, name, description, input_schema) VALUES (?, ?, ?, ?, ?)") + .bind(thread_id.to_string()) + .bind(0_i64) + .bind("test_tool") + .bind("test dynamic tool") + .bind("{}") + .execute(runtime.pool.as_ref()) + .await?; + runtime + .create_agent_job( + &AgentJobCreateParams { + id: "job-1".to_string(), + name: "test-job".to_string(), + instruction: "Return a result".to_string(), + auto_export: true, + max_runtime_seconds: None, + output_schema_json: None, + input_headers: vec!["path".to_string()], + input_csv_path: "/tmp/in.csv".to_string(), + output_csv_path: "/tmp/out.csv".to_string(), + }, + &[AgentJobItemCreateParams { + item_id: "item-1".to_string(), + row_index: 0, + source_id: None, + row_json: json!({"path": "file-1"}), + }], + ) + .await?; + runtime.mark_agent_job_running("job-1").await?; + runtime + .mark_agent_job_item_running_with_thread( + "job-1", + "item-1", + &child_thread_id.to_string(), + ) + .await?; + + let rows = runtime + .delete_threads_strict(&[thread_id, child_thread_id]) + .await?; + + assert_eq!(rows, 1); + assert!(runtime.get_thread(thread_id).await?.is_none()); + let dynamic_tool_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM thread_dynamic_tools WHERE thread_id = ?") + .bind(thread_id.to_string()) + .fetch_one(runtime.pool.as_ref()) + .await?; + assert_eq!(dynamic_tool_count, 0); + assert_thread_cleanup_state(&runtime, thread_id).await?; + let job_item = runtime + .get_agent_job_item("job-1", "item-1") + .await? + .expect("job item should exist"); + assert_eq!(job_item.status, AgentJobItemStatus::Pending); + assert_eq!(job_item.assigned_thread_id, None); + assert_eq!( + job_item.last_error, + Some("assigned thread was deleted".to_string()) + ); + let job = runtime + .get_agent_job("job-1") + .await? + .expect("job should exist"); + assert_eq!(job.status, AgentJobStatus::Cancelled); + + let missing_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000403")?; + let missing_child_thread_id = + ThreadId::from_string("00000000-0000-0000-0000-000000000404")?; + seed_thread_cleanup_state(&runtime, missing_thread_id, missing_child_thread_id).await?; + + assert_eq!(runtime.delete_thread(missing_thread_id).await?, 0); + assert_thread_cleanup_state(&runtime, missing_thread_id).await?; + Ok(()) + } + + #[tokio::test] + async fn delete_thread_keeps_retry_graph_on_cleanup_failure() -> Result<()> { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()).await?; + let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000405")?; + let child_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000406")?; + runtime + .upsert_thread(&test_thread_metadata( + &codex_home, + thread_id, + codex_home.clone(), + )) + .await?; + seed_thread_cleanup_state(&runtime, thread_id, child_thread_id).await?; + + runtime.logs_pool.close().await; + runtime + .delete_thread(thread_id) + .await + .expect_err("closed log db should fail deletion"); + + assert!(runtime.get_thread(thread_id).await?.is_some()); + assert_eq!( + runtime.list_thread_spawn_descendants(thread_id).await?, + vec![child_thread_id] + ); + Ok(()) + } + + async fn seed_thread_cleanup_state( + runtime: &StateRuntime, + thread_id: ThreadId, + child_thread_id: ThreadId, + ) -> Result<()> { + runtime + .upsert_thread_spawn_edge( + thread_id, + child_thread_id, + DirectionalThreadSpawnEdgeStatus::Closed, + ) + .await?; + runtime + .thread_goals() + .replace_thread_goal( + thread_id, + "test goal", + crate::ThreadGoalStatus::Active, + /*token_budget*/ None, + ) + .await?; + sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (1, 0, 'INFO', 'test', 'feedback log', ?)") + .bind(thread_id.to_string()) + .execute(runtime.logs_pool.as_ref()) + .await?; + Ok(()) + } + + async fn assert_thread_cleanup_state( + runtime: &StateRuntime, + thread_id: ThreadId, + ) -> Result<()> { + let spawn_edge_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM thread_spawn_edges WHERE parent_thread_id = ? OR child_thread_id = ?", + ) + .bind(thread_id.to_string()) + .bind(thread_id.to_string()) + .fetch_one(runtime.pool.as_ref()) + .await?; + assert_eq!(spawn_edge_count, 0); + assert_eq!( + runtime.thread_goals().get_thread_goal(thread_id).await?, + None + ); + let logs = runtime + .query_logs(&LogQuery { + thread_ids: vec![thread_id.to_string()], + ..Default::default() + }) + .await?; + assert!(logs.is_empty()); + Ok(()) + } + #[tokio::test] async fn list_threads_updated_after_returns_oldest_changes_first() { let codex_home = unique_temp_dir(); diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 2ab510366..07156fe35 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -18,6 +18,7 @@ use codex_protocol::protocol::ThreadMemoryMode; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; +use crate::DeleteThreadParams; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; @@ -115,6 +116,7 @@ pub struct InMemoryThreadStoreCalls { pub update_thread_metadata: usize, pub archive_thread: usize, pub unarchive_thread: usize, + pub delete_thread: usize, } /// In-memory [`ThreadStore`] implementation for tests and debug configs. @@ -333,6 +335,25 @@ impl ThreadStore for InMemoryThreadStore { state.calls.unarchive_thread += 1; stored_thread_from_state(&state, params.thread_id, /*include_history*/ false) } + + async fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreResult<()> { + let mut state = self.state.lock().await; + state.calls.delete_thread += 1; + let existed = state.histories.remove(¶ms.thread_id).is_some(); + state.created_threads.remove(¶ms.thread_id); + state.names.remove(¶ms.thread_id); + state.metadata_updates.remove(¶ms.thread_id); + state + .rollout_paths + .retain(|_, thread_id| *thread_id != params.thread_id); + if existed { + Ok(()) + } else { + Err(ThreadStoreError::ThreadNotFound { + thread_id: params.thread_id, + }) + } + } } fn stored_thread_from_state( diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 84b72f23f..284614bf1 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -25,6 +25,7 @@ pub use types::AppendThreadItemsParams; pub use types::ArchiveThreadParams; pub use types::ClearableField; pub use types::CreateThreadParams; +pub use types::DeleteThreadParams; pub use types::ExtraConfig; pub use types::GitInfoPatch; pub use types::ItemPage; diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs new file mode 100644 index 000000000..ecd962f64 --- /dev/null +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -0,0 +1,210 @@ +//! Local hard-delete support for persisted threads. +//! +//! Existing rollout files are deleted before this operation reports success. A rollout file that +//! vanishes after discovery counts as already deleted. SQLite cleanup happens at the app-server +//! layer after every associated rollout has been removed so failed deletes can be retried. + +use std::io::ErrorKind; +use std::path::Path; + +use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; +use codex_rollout::SESSIONS_SUBDIR; +use codex_rollout::find_archived_thread_path_by_id_str; +use codex_rollout::find_thread_path_by_id_str; +use codex_rollout::remove_thread_name_entries; + +use super::LocalThreadStore; +use super::helpers::matching_rollout_file_name; +use super::helpers::scoped_rollout_path; +use crate::DeleteThreadParams; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) async fn delete_thread( + store: &LocalThreadStore, + params: DeleteThreadParams, +) -> ThreadStoreResult<()> { + let thread_id = params.thread_id; + let thread_id_str = thread_id.to_string(); + let state_db_ctx = store.state_db().await; + let mut rollout_paths = Vec::new(); + + match find_thread_path_by_id_str( + store.config.codex_home.as_path(), + thread_id_str.as_str(), + state_db_ctx.as_deref(), + ) + .await + { + Ok(Some(path)) => rollout_paths.push(path), + Ok(None) => {} + Err(err) => { + return Err(ThreadStoreError::InvalidRequest { + message: format!("failed to locate thread id {thread_id}: {err}"), + }); + } + } + + match find_archived_thread_path_by_id_str( + store.config.codex_home.as_path(), + thread_id_str.as_str(), + state_db_ctx.as_deref(), + ) + .await + { + Ok(Some(path)) => { + if !rollout_paths.contains(&path) { + rollout_paths.push(path); + } + } + Ok(None) => {} + Err(err) => { + return Err(ThreadStoreError::InvalidRequest { + message: format!("failed to locate archived thread id {thread_id}: {err}"), + }); + } + } + + let found_rollout_path = !rollout_paths.is_empty(); + for rollout_path in rollout_paths { + delete_rollout_file(store, rollout_path.as_path(), thread_id)?; + } + remove_thread_name_entries(store.config.codex_home.as_path(), thread_id) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to delete thread name index entries for {thread_id}: {err}"), + })?; + + if !found_rollout_path { + return Err(ThreadStoreError::ThreadNotFound { thread_id }); + } + + store.live_recorders.lock().await.remove(&thread_id); + + Ok(()) +} + +fn delete_rollout_file( + store: &LocalThreadStore, + rollout_path: &Path, + thread_id: codex_protocol::ThreadId, +) -> ThreadStoreResult { + let plain_path = codex_rollout::plain_rollout_path(rollout_path); + let compressed_path = plain_path.with_extension("jsonl.zst"); + let deleted_plain = delete_rollout_path(store, plain_path.as_path(), thread_id)?; + let deleted_compressed = delete_rollout_path(store, compressed_path.as_path(), thread_id)?; + Ok(deleted_plain || deleted_compressed) +} + +fn delete_rollout_path( + store: &LocalThreadStore, + rollout_path: &Path, + thread_id: codex_protocol::ThreadId, +) -> ThreadStoreResult { + let canonical_rollout_path = scoped_rollout_path( + store.config.codex_home.join(SESSIONS_SUBDIR), + rollout_path, + "sessions", + ) + .or_else(|_| { + scoped_rollout_path( + store.config.codex_home.join(ARCHIVED_SESSIONS_SUBDIR), + rollout_path, + "archived sessions", + ) + }) + .or_else(|err| match rollout_path.try_exists() { + Ok(false) => Ok(rollout_path.to_path_buf()), + Ok(true) | Err(_) => Err(err), + })?; + matching_rollout_file_name(&canonical_rollout_path, thread_id, rollout_path)?; + match std::fs::remove_file(&canonical_rollout_path) { + Ok(()) => Ok(true), + Err(err) if err.kind() == ErrorKind::NotFound => Ok(false), + Err(err) => Err(ThreadStoreError::Internal { + message: format!( + "failed to delete rollout file `{}`: {err}", + canonical_rollout_path.display() + ), + }), + } +} + +#[cfg(test)] +mod tests { + use codex_protocol::ThreadId; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + use uuid::Uuid; + + use super::*; + use crate::ThreadStore; + use crate::local::LocalThreadStore; + use crate::local::test_support::test_config; + use crate::local::test_support::write_archived_session_file; + use crate::local::test_support::write_session_file; + + #[tokio::test] + async fn delete_thread_removes_active_and_archived_rollouts() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let active_path = + write_session_file(home.path(), "2025-01-03T12-00-00", Uuid::from_u128(301)) + .expect("session file"); + let compressed_path = active_path.with_extension("jsonl.zst"); + std::fs::write(&compressed_path, b"compressed sibling").expect("compressed sibling"); + let cases = [ + (Uuid::from_u128(301), active_path), + ( + Uuid::from_u128(302), + write_archived_session_file( + home.path(), + "2025-01-03T12-00-00", + Uuid::from_u128(302), + ) + .expect("archived session file"), + ), + ]; + + for (uuid, path) in cases { + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + store + .delete_thread(DeleteThreadParams { thread_id }) + .await + .expect("delete thread"); + + assert!(!path.exists()); + } + assert!(!compressed_path.exists()); + } + + #[tokio::test] + async fn delete_rollout_file_treats_vanished_path_as_already_deleted() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let uuid = Uuid::from_u128(305); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + std::fs::remove_file(&path).expect("remove session file"); + + assert!(!delete_rollout_file(&store, path.as_path(), thread_id).expect("delete rollout")); + } + + #[tokio::test] + async fn delete_thread_reports_missing_thread() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let thread_id = + ThreadId::from_string("00000000-0000-0000-0000-000000000304").expect("valid thread id"); + + let err = store + .delete_thread(DeleteThreadParams { thread_id }) + .await + .expect_err("missing thread should fail"); + assert_eq!( + err.to_string(), + "thread 00000000-0000-0000-0000-000000000304 not found" + ); + } +} diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index cef4496a0..3000fee9d 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1,5 +1,6 @@ mod archive_thread; mod create_thread; +mod delete_thread; mod helpers; mod list_threads; mod live_writer; @@ -24,6 +25,7 @@ use tokio::sync::Mutex; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; +use crate::DeleteThreadParams; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; @@ -288,6 +290,10 @@ impl ThreadStore for LocalThreadStore { ) -> ThreadStoreResult { unarchive_thread::unarchive_thread(self, params).await } + + async fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreResult<()> { + delete_thread::delete_thread(self, params).await + } } #[cfg(test)] diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 7276a980d..6373dd9b1 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -5,6 +5,7 @@ use std::any::Any; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; +use crate::DeleteThreadParams; use crate::ItemPage; use crate::ListItemsParams; use crate::ListThreadsParams; @@ -119,4 +120,7 @@ pub trait ThreadStore: Any + Send + Sync { &self, params: ArchiveThreadParams, ) -> ThreadStoreResult; + + /// Deletes a thread's persisted rollout data and associated metadata. + async fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreResult<()>; } diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 1f18adf08..853d866bc 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -653,6 +653,13 @@ pub struct ArchiveThreadParams { pub thread_id: ThreadId, } +/// Parameters for deleting a thread. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DeleteThreadParams { + /// Thread id to delete. + pub thread_id: ThreadId, +} + #[cfg(test)] mod tests { use pretty_assertions::assert_eq; diff --git a/codex-rs/tui/src/app/app_server_event_targets.rs b/codex-rs/tui/src/app/app_server_event_targets.rs index 800560d52..fa2bab011 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -49,6 +49,7 @@ pub(super) fn server_notification_thread_target( Some(notification.thread_id.as_str()) } ServerNotification::ThreadArchived(notification) => Some(notification.thread_id.as_str()), + ServerNotification::ThreadDeleted(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadUnarchived(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadClosed(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadNameUpdated(notification) => { diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 3682f245f..51e4cedf8 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -224,6 +224,7 @@ impl ChatWidget { | ServerNotification::ThreadStarted(_) | ServerNotification::ThreadStatusChanged(_) | ServerNotification::ThreadArchived(_) + | ServerNotification::ThreadDeleted(_) | ServerNotification::ThreadUnarchived(_) | ServerNotification::RawResponseItemCompleted(_) | ServerNotification::CommandExecOutputDelta(_)