diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index e1a7cd3f3d..a1e4ca7caf 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -2985,6 +2985,81 @@ } ] }, + "ThreadJob": { + "properties": { + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cronExpression": { + "type": "string" + }, + "id": { + "type": "string" + }, + "lastRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "nextRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "prompt": { + "type": "string" + }, + "runOnce": { + "type": "boolean" + } + }, + "required": [ + "createdAt", + "cronExpression", + "id", + "prompt", + "runOnce" + ], + "type": "object" + }, + "ThreadJobFiredNotification": { + "properties": { + "job": { + "$ref": "#/definitions/ThreadJob" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "job", + "threadId" + ], + "type": "object" + }, + "ThreadJobUpdatedNotification": { + "properties": { + "jobs": { + "items": { + "$ref": "#/definitions/ThreadJob" + }, + "type": "array" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "jobs", + "threadId" + ], + "type": "object" + }, "ThreadNameUpdatedNotification": { "properties": { "threadId": { @@ -3940,6 +4015,46 @@ "title": "Thread/name/updatedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/job/updated" + ], + "title": "Thread/job/updatedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadJobUpdatedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/job/updatedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/job/fired" + ], + "title": "Thread/job/firedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadJobFiredNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/job/firedNotification", + "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 084da013e4..35357cdd0c 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 @@ -3557,6 +3557,46 @@ "title": "Thread/name/updatedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/job/updated" + ], + "title": "Thread/job/updatedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadJobUpdatedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/job/updatedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/job/fired" + ], + "title": "Thread/job/firedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadJobFiredNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/job/firedNotification", + "type": "object" + }, { "properties": { "method": { @@ -12960,6 +13000,85 @@ } ] }, + "ThreadJob": { + "properties": { + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cronExpression": { + "type": "string" + }, + "id": { + "type": "string" + }, + "lastRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "nextRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "prompt": { + "type": "string" + }, + "runOnce": { + "type": "boolean" + } + }, + "required": [ + "createdAt", + "cronExpression", + "id", + "prompt", + "runOnce" + ], + "type": "object" + }, + "ThreadJobFiredNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "job": { + "$ref": "#/definitions/v2/ThreadJob" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "job", + "threadId" + ], + "title": "ThreadJobFiredNotification", + "type": "object" + }, + "ThreadJobUpdatedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "jobs": { + "items": { + "$ref": "#/definitions/v2/ThreadJob" + }, + "type": "array" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "jobs", + "threadId" + ], + "title": "ThreadJobUpdatedNotification", + "type": "object" + }, "ThreadListParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { 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 59dab32b19..b4152bd1b6 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 @@ -8344,6 +8344,46 @@ "title": "Thread/name/updatedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/job/updated" + ], + "title": "Thread/job/updatedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadJobUpdatedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/job/updatedNotification", + "type": "object" + }, + { + "properties": { + "method": { + "enum": [ + "thread/job/fired" + ], + "title": "Thread/job/firedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadJobFiredNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/job/firedNotification", + "type": "object" + }, { "properties": { "method": { @@ -10806,6 +10846,85 @@ } ] }, + "ThreadJob": { + "properties": { + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cronExpression": { + "type": "string" + }, + "id": { + "type": "string" + }, + "lastRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "nextRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "prompt": { + "type": "string" + }, + "runOnce": { + "type": "boolean" + } + }, + "required": [ + "createdAt", + "cronExpression", + "id", + "prompt", + "runOnce" + ], + "type": "object" + }, + "ThreadJobFiredNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "job": { + "$ref": "#/definitions/ThreadJob" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "job", + "threadId" + ], + "title": "ThreadJobFiredNotification", + "type": "object" + }, + "ThreadJobUpdatedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "jobs": { + "items": { + "$ref": "#/definitions/ThreadJob" + }, + "type": "array" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "jobs", + "threadId" + ], + "title": "ThreadJobUpdatedNotification", + "type": "object" + }, "ThreadListParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadJobFiredNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadJobFiredNotification.json new file mode 100644 index 0000000000..db63c3320c --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadJobFiredNotification.json @@ -0,0 +1,61 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "ThreadJob": { + "properties": { + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cronExpression": { + "type": "string" + }, + "id": { + "type": "string" + }, + "lastRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "nextRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "prompt": { + "type": "string" + }, + "runOnce": { + "type": "boolean" + } + }, + "required": [ + "createdAt", + "cronExpression", + "id", + "prompt", + "runOnce" + ], + "type": "object" + } + }, + "properties": { + "job": { + "$ref": "#/definitions/ThreadJob" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "job", + "threadId" + ], + "title": "ThreadJobFiredNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadJobUpdatedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadJobUpdatedNotification.json new file mode 100644 index 0000000000..0a2c3523a7 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadJobUpdatedNotification.json @@ -0,0 +1,64 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "ThreadJob": { + "properties": { + "createdAt": { + "format": "int64", + "type": "integer" + }, + "cronExpression": { + "type": "string" + }, + "id": { + "type": "string" + }, + "lastRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "nextRunAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "prompt": { + "type": "string" + }, + "runOnce": { + "type": "boolean" + } + }, + "required": [ + "createdAt", + "cronExpression", + "id", + "prompt", + "runOnce" + ], + "type": "object" + } + }, + "properties": { + "jobs": { + "items": { + "$ref": "#/definitions/ThreadJob" + }, + "type": "array" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "jobs", + "threadId" + ], + "title": "ThreadJobUpdatedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 85ebe847f4..286b3169fa 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -36,6 +36,8 @@ 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 { ThreadJobFiredNotification } from "./v2/ThreadJobFiredNotification"; +import type { ThreadJobUpdatedNotification } from "./v2/ThreadJobUpdatedNotification"; import type { ThreadNameUpdatedNotification } from "./v2/ThreadNameUpdatedNotification"; import type { ThreadRealtimeClosedNotification } from "./v2/ThreadRealtimeClosedNotification"; import type { ThreadRealtimeErrorNotification } from "./v2/ThreadRealtimeErrorNotification"; @@ -57,4 +59,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/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": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "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": "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": "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/transcriptUpdated", "params": ThreadRealtimeTranscriptUpdatedNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "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/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/job/updated", "params": ThreadJobUpdatedNotification } | { "method": "thread/job/fired", "params": ThreadJobFiredNotification } | { "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": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "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": "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": "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/transcriptUpdated", "params": ThreadRealtimeTranscriptUpdatedNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "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/ThreadJob.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJob.ts new file mode 100644 index 0000000000..9503ca07c9 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJob.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 ThreadJob = { id: string, cronExpression: string, prompt: string, runOnce: boolean, createdAt: number, nextRunAt: number | null, lastRunAt: number | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJobFiredNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJobFiredNotification.ts new file mode 100644 index 0000000000..7ef9c494da --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJobFiredNotification.ts @@ -0,0 +1,6 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadJob } from "./ThreadJob"; + +export type ThreadJobFiredNotification = { threadId: string, job: ThreadJob, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJobUpdatedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJobUpdatedNotification.ts new file mode 100644 index 0000000000..df15b1a350 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadJobUpdatedNotification.ts @@ -0,0 +1,6 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ThreadJob } from "./ThreadJob"; + +export type ThreadJobUpdatedNotification = { threadId: string, jobs: Array, }; 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 d0687e5f1d..f5aa81b88d 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -274,6 +274,9 @@ export type { ThreadCompactStartResponse } from "./ThreadCompactStartResponse"; export type { ThreadForkParams } from "./ThreadForkParams"; export type { ThreadForkResponse } from "./ThreadForkResponse"; export type { ThreadItem } from "./ThreadItem"; +export type { ThreadJob } from "./ThreadJob"; +export type { ThreadJobFiredNotification } from "./ThreadJobFiredNotification"; +export type { ThreadJobUpdatedNotification } from "./ThreadJobUpdatedNotification"; export type { ThreadListParams } from "./ThreadListParams"; export type { ThreadListResponse } from "./ThreadListResponse"; export type { ThreadLoadedListParams } from "./ThreadLoadedListParams"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 30061c716e..5dd2ac52c6 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -317,6 +317,21 @@ client_request_definitions! { params: v2::ThreadReadParams, response: v2::ThreadReadResponse, }, + #[experimental("thread/job/create")] + ThreadJobCreate => "thread/job/create" { + params: v2::ThreadJobCreateParams, + response: v2::ThreadJobCreateResponse, + }, + #[experimental("thread/job/delete")] + ThreadJobDelete => "thread/job/delete" { + params: v2::ThreadJobDeleteParams, + response: v2::ThreadJobDeleteResponse, + }, + #[experimental("thread/job/list")] + ThreadJobList => "thread/job/list" { + params: v2::ThreadJobListParams, + response: v2::ThreadJobListResponse, + }, SkillsList => "skills/list" { params: v2::SkillsListParams, response: v2::SkillsListResponse, @@ -918,6 +933,10 @@ server_notification_definitions! { ThreadClosed => "thread/closed" (v2::ThreadClosedNotification), SkillsChanged => "skills/changed" (v2::SkillsChangedNotification), ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification), + #[experimental("thread/job/updated")] + ThreadJobUpdated => "thread/job/updated" (v2::ThreadJobUpdatedNotification), + #[experimental("thread/job/fired")] + ThreadJobFired => "thread/job/fired" (v2::ThreadJobFiredNotification), ThreadTokenUsageUpdated => "thread/tokenUsage/updated" (v2::ThreadTokenUsageUpdatedNotification), TurnStarted => "turn/started" (v2::TurnStartedNotification), HookStarted => "hook/started" (v2::HookStartedNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 9373c852d8..9d12bc6c16 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -3149,6 +3149,85 @@ pub struct ThreadReadResponse { pub thread: Thread, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJob { + pub id: String, + pub cron_expression: String, + pub prompt: String, + pub run_once: bool, + #[ts(type = "number")] + pub created_at: i64, + #[ts(type = "number | null")] + pub next_run_at: Option, + #[ts(type = "number | null")] + pub last_run_at: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobCreateParams { + pub thread_id: String, + pub cron_expression: String, + pub prompt: String, + #[ts(optional = nullable)] + pub run_once: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobCreateResponse { + pub job: ThreadJob, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobDeleteParams { + pub thread_id: String, + pub id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobDeleteResponse { + pub deleted: bool, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobListParams { + pub thread_id: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobListResponse { + pub data: Vec, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobUpdatedNotification { + pub thread_id: String, + pub jobs: Vec, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadJobFiredNotification { + pub thread_id: String, + pub job: ThreadJob, +} + #[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 4365cb8e29..fd356e3be2 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -144,6 +144,11 @@ Example with notification opt-out: - `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, 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`. +- `thread/job/create` — create a runtime-only job on a loaded thread; accepts `cronExpression`, `prompt`, and optional `runOnce`, returns the created job, and requires the experimental feature flag `job_scheduler` to be enabled for that thread. +- `thread/job/delete` — delete one runtime-only job from a loaded thread by id; returns `{}` and requires `job_scheduler`. +- `thread/job/list` — list runtime-only jobs for a loaded thread; returns `{ data }` and requires `job_scheduler`. +- `thread/job/updated` — notification emitted when a loaded thread’s job list changes, including run-once jobs deleting themselves after execution. +- `thread/job/fired` — notification emitted when a thread job fires and starts a synthetic follow-on turn. - `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications. - `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream. - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted. @@ -173,7 +178,7 @@ Example with notification opt-out: - `fs/changed` — notification emitted when watched paths change, including the `watchId` and `changedPaths`. - `model/list` — list available models (set `includeHidden: true` to include entries with `hidden: true`), with reasoning effort options, optional legacy `upgrade` model ids, optional `upgradeInfo` metadata (`model`, `upgradeCopy`, `modelLink`, `migrationMarkdown`), and optional `availabilityNux` metadata. - `experimentalFeature/list` — list feature flags with stage metadata (`beta`, `underDevelopment`, `stable`, etc.), enabled/default-enabled state, and cursor pagination. For non-beta flags, `displayName`/`description`/`announcement` are `null`. -- `experimentalFeature/enablement/set` — patch the in-memory process-wide runtime feature enablement for the currently supported feature keys (`apps`, `plugins`). For each feature, precedence is: cloud requirements > --enable > config.toml > experimentalFeature/enablement/set (new) > code default. +- `experimentalFeature/enablement/set` — patch the in-memory process-wide runtime feature enablement for the currently supported feature keys (`apps`, `plugins`, `job_scheduler`). For each feature, precedence is: cloud requirements > --enable > config.toml > experimentalFeature/enablement/set (new) > code default. - `collaborationMode/list` — list available collaboration mode presets (experimental, no pagination). This response omits built-in developer instructions; clients should either pass `settings.developer_instructions: null` when setting a mode to use Codex's built-in instructions, or provide their own instructions explicitly. - `skills/list` — list skills for one or more `cwd` values (optional `forceReload`). - `plugin/list` — list discovered plugin marketplaces and plugin state, including effective marketplace install/auth policy metadata, fail-open `marketplaceLoadErrors` entries for marketplace files that could not be parsed or loaded, and best-effort `featuredPluginIds` for the official curated marketplace. `interface.category` uses the marketplace category when present; otherwise it falls back to the plugin manifest category. Pass `forceRemoteSync: true` to refresh curated plugin state before listing (**under development; do not call from production clients yet**). diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 9a850c0b1e..ddeae12d1a 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -79,6 +79,9 @@ use codex_app_server_protocol::ServerRequestPayload; use codex_app_server_protocol::SkillsChangedNotification; use codex_app_server_protocol::TerminalInteractionNotification; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadJob; +use codex_app_server_protocol::ThreadJobFiredNotification; +use codex_app_server_protocol::ThreadJobUpdatedNotification; use codex_app_server_protocol::ThreadNameUpdatedNotification; use codex_app_server_protocol::ThreadRealtimeClosedNotification; use codex_app_server_protocol::ThreadRealtimeErrorNotification; @@ -107,6 +110,8 @@ use codex_app_server_protocol::convert_patch_changes; use codex_core::CodexThread; use codex_core::ThreadManager; use codex_core::find_thread_name_by_id; +use codex_core::jobs::JOB_FIRED_BACKGROUND_EVENT_PREFIX; +use codex_core::jobs::JOB_UPDATED_BACKGROUND_EVENT_PREFIX; use codex_core::review_format::format_review_findings_block; use codex_core::review_prompts; use codex_protocol::ThreadId; @@ -251,6 +256,18 @@ fn guardian_auto_approval_review_notification( } } +fn thread_job_from_core(value: codex_core::jobs::ThreadJob) -> ThreadJob { + ThreadJob { + id: value.id, + cron_expression: value.cron_expression, + prompt: value.prompt, + run_once: value.run_once, + created_at: value.created_at, + next_run_at: value.next_run_at, + last_run_at: value.last_run_at, + } +} + #[allow(clippy::too_many_arguments)] pub(crate) async fn apply_bespoke_event_handling( event: Event, @@ -1309,6 +1326,33 @@ pub(crate) async fn apply_bespoke_event_handling( .send_server_notification(ServerNotification::DeprecationNotice(notification)) .await; } + EventMsg::BackgroundEvent(event) => { + if let Some(payload) = event + .message + .strip_prefix(JOB_UPDATED_BACKGROUND_EVENT_PREFIX) + && let Ok(jobs) = serde_json::from_str::>(payload) + { + let notification = ThreadJobUpdatedNotification { + thread_id: conversation_id.to_string(), + jobs: jobs.into_iter().map(thread_job_from_core).collect(), + }; + outgoing + .send_server_notification(ServerNotification::ThreadJobUpdated(notification)) + .await; + } else if let Some(payload) = event + .message + .strip_prefix(JOB_FIRED_BACKGROUND_EVENT_PREFIX) + && let Ok(job) = serde_json::from_str::(payload) + { + let notification = ThreadJobFiredNotification { + thread_id: conversation_id.to_string(), + job: thread_job_from_core(job), + }; + outgoing + .send_server_notification(ServerNotification::ThreadJobFired(notification)) + .await; + } + } EventMsg::ReasoningContentDelta(event) => { let notification = ReasoningSummaryTextDeltaNotification { thread_id: conversation_id.to_string(), diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 2798dcb92e..477777372b 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -124,6 +124,13 @@ use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadIncrementElicitationParams; use codex_app_server_protocol::ThreadIncrementElicitationResponse; use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadJob as ApiThreadJob; +use codex_app_server_protocol::ThreadJobCreateParams; +use codex_app_server_protocol::ThreadJobCreateResponse; +use codex_app_server_protocol::ThreadJobDeleteParams; +use codex_app_server_protocol::ThreadJobDeleteResponse; +use codex_app_server_protocol::ThreadJobListParams; +use codex_app_server_protocol::ThreadJobListResponse; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadLoadedListParams; @@ -763,6 +770,18 @@ impl CodexMessageProcessor { self.thread_read(to_connection_request_id(request_id), params) .await; } + ClientRequest::ThreadJobCreate { request_id, params } => { + self.thread_job_create(to_connection_request_id(request_id), params) + .await; + } + ClientRequest::ThreadJobDelete { request_id, params } => { + self.thread_job_delete(to_connection_request_id(request_id), params) + .await; + } + ClientRequest::ThreadJobList { request_id, params } => { + self.thread_job_list(to_connection_request_id(request_id), params) + .await; + } ClientRequest::ThreadShellCommand { request_id, params } => { self.thread_shell_command(to_connection_request_id(request_id), params) .await; @@ -3553,6 +3572,96 @@ impl CodexMessageProcessor { self.outgoing.send_response(request_id, response).await; } + async fn thread_job_create( + &mut self, + request_id: ConnectionRequestId, + params: ThreadJobCreateParams, + ) { + let ThreadJobCreateParams { + thread_id, + cron_expression, + prompt, + run_once, + } = params; + let Some(thread) = self.load_job_thread(request_id.clone(), &thread_id).await else { + return; + }; + match thread + .create_job(cron_expression, prompt, run_once.unwrap_or(false)) + .await + { + Ok(job) => { + self.outgoing + .send_response( + request_id, + ThreadJobCreateResponse { + job: api_thread_job_from_core(job), + }, + ) + .await; + } + Err(err) => self.send_invalid_request_error(request_id, err).await, + } + } + + async fn thread_job_delete( + &mut self, + request_id: ConnectionRequestId, + params: ThreadJobDeleteParams, + ) { + let ThreadJobDeleteParams { thread_id, id } = params; + let Some(thread) = self.load_job_thread(request_id.clone(), &thread_id).await else { + return; + }; + let deleted = thread.delete_job(&id).await; + self.outgoing + .send_response(request_id, ThreadJobDeleteResponse { deleted }) + .await; + } + + async fn thread_job_list( + &mut self, + request_id: ConnectionRequestId, + params: ThreadJobListParams, + ) { + let ThreadJobListParams { thread_id } = params; + let Some(thread) = self.load_job_thread(request_id.clone(), &thread_id).await else { + return; + }; + let data = thread + .list_jobs() + .await + .into_iter() + .map(api_thread_job_from_core) + .collect(); + self.outgoing + .send_response(request_id, ThreadJobListResponse { data }) + .await; + } + + async fn load_job_thread( + &mut self, + request_id: ConnectionRequestId, + thread_id: &str, + ) -> Option> { + let (_, thread) = match self.load_thread(thread_id).await { + Ok(value) => value, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return None; + } + }; + if !thread.enabled(Feature::JobScheduler) { + self.send_invalid_request_error( + request_id, + format!("thread {thread_id} does not support job scheduling"), + ) + .await; + return None; + } + Some(thread) + } + pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver { self.thread_manager.subscribe_thread_created() } @@ -8620,6 +8729,18 @@ fn build_thread_from_snapshot( } } +fn api_thread_job_from_core(value: codex_core::jobs::ThreadJob) -> ApiThreadJob { + ApiThreadJob { + id: value.id, + cron_expression: value.cron_expression, + prompt: value.prompt, + run_once: value.run_once, + created_at: value.created_at, + next_run_at: value.next_run_at, + last_run_at: value.last_run_at, + } +} + pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread { let ConversationSummary { conversation_id, diff --git a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs index 8f944dca07..b18f004723 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs @@ -22,7 +22,7 @@ use codex_app_server_protocol::UserInput as V2UserInput; use tempfile::TempDir; use tokio::time::timeout; -const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20); #[tokio::test] async fn turn_interrupt_aborts_running_turn() -> Result<()> { @@ -124,14 +124,11 @@ async fn turn_interrupt_aborts_running_turn() -> Result<()> { #[tokio::test] async fn turn_interrupt_resolves_pending_command_approval_request() -> Result<()> { - #[cfg(target_os = "windows")] let shell_command = vec![ - "powershell".to_string(), - "-Command".to_string(), - "Start-Sleep -Seconds 10".to_string(), + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), ]; - #[cfg(not(target_os = "windows"))] - let shell_command = vec!["sleep".to_string(), "10".to_string()]; let tmp = TempDir::new()?; let codex_home = tmp.path().join("codex_home"); 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 3155a4be9c..9018d5d6cf 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -67,7 +67,7 @@ use tokio::time::timeout; #[cfg(windows)] const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25); #[cfg(not(windows))] -const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20); const TEST_ORIGINATOR: &str = "codex_vscode"; const LOCAL_PRAGMATIC_TEMPLATE: &str = "You are a deeply pragmatic, effective software engineer."; diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index e36f495a1b..573c69de8a 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -404,6 +404,9 @@ "include_apply_patch_tool": { "type": "boolean" }, + "job_scheduler": { + "type": "boolean" + }, "js_repl": { "type": "boolean" }, @@ -2046,6 +2049,9 @@ "include_apply_patch_tool": { "type": "boolean" }, + "job_scheduler": { + "type": "boolean" + }, "js_repl": { "type": "boolean" }, diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index e440f98385..a88c9c3170 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -495,7 +495,7 @@ async fn append_message_records_assistant_message() { .expect("append_message should succeed"); assert!(!submission_id.is_empty()); - timeout(Duration::from_secs(5), async { + timeout(Duration::from_secs(15), async { loop { let history_items = thread .codex @@ -1478,7 +1478,7 @@ async fn resume_thread_subagent_restores_stored_nickname_and_role() { let state_db = child_thread .state_db() .expect("sqlite state db should be available for nickname resume test"); - timeout(Duration::from_secs(5), async { + timeout(Duration::from_secs(15), async { loop { if let Ok(Some(metadata)) = state_db.get_thread(child_thread_id).await && metadata.agent_nickname.is_some() diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 9a8cc9cea7..7009469be0 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -259,6 +259,13 @@ use crate::injection::ToolMentionKind; use crate::injection::app_id_from_path; use crate::injection::tool_kind_for_path; use crate::instructions::UserInstructions; +use crate::jobs::ClaimedJob; +use crate::jobs::JOB_UPDATED_BACKGROUND_EVENT_PREFIX; +use crate::jobs::JobTurnContext; +use crate::jobs::JobsState; +use crate::jobs::ThreadJob; +use crate::jobs::job_prompt_input_item; +use crate::jobs::job_turn_developer_instructions; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::McpManager; use crate::mcp::auth::compute_auth_statuses; @@ -812,6 +819,7 @@ pub(crate) struct Session { mailbox: Mailbox, mailbox_rx: Mutex, idle_pending_input: Mutex>, // TODO (jif) merge with mailbox! + jobs: Mutex, pub(crate) guardian_review_session: GuardianReviewSessionManager, pub(crate) services: SessionServices, js_repl: Arc, @@ -856,6 +864,7 @@ pub(crate) struct TurnContext { pub(crate) timezone: Option, pub(crate) app_server_client_name: Option, pub(crate) developer_instructions: Option, + pub(crate) job_context: Option, pub(crate) compact_prompt: Option, pub(crate) user_instructions: Option, pub(crate) collaboration_mode: CollaborationMode, @@ -964,6 +973,7 @@ impl TurnContext { timezone: self.timezone.clone(), app_server_client_name: self.app_server_client_name.clone(), developer_instructions: self.developer_instructions.clone(), + job_context: self.job_context.clone(), compact_prompt: self.compact_prompt.clone(), user_instructions: self.user_instructions.clone(), collaboration_mode, @@ -1430,6 +1440,7 @@ impl Session { timezone: Some(timezone), app_server_client_name: session_configuration.app_server_client_name.clone(), developer_instructions: session_configuration.developer_instructions.clone(), + job_context: None, compact_prompt: session_configuration.compact_prompt.clone(), user_instructions: session_configuration.user_instructions.clone(), collaboration_mode: session_configuration.collaboration_mode.clone(), @@ -1926,6 +1937,7 @@ impl Session { mailbox, mailbox_rx: Mutex::new(mailbox_rx), idle_pending_input: Mutex::new(Vec::new()), + jobs: Mutex::new(JobsState::default()), guardian_review_session: GuardianReviewSessionManager::default(), services, js_repl, @@ -2413,6 +2425,7 @@ impl Session { session_configuration, updates.final_output_json_schema, sandbox_policy_changed, + /*job_context*/ None, ) .await) } @@ -2423,6 +2436,7 @@ impl Session { session_configuration: SessionConfiguration, final_output_json_schema: Option>, sandbox_policy_changed: bool, + job_context: Option, ) -> Arc { let per_turn_config = Self::build_per_turn_config(&session_configuration); self.services @@ -2495,6 +2509,7 @@ impl Session { if let Some(final_schema) = final_output_json_schema { turn_context.final_output_json_schema = final_schema; } + turn_context.job_context = job_context; let turn_context = Arc::new(turn_context); turn_context.turn_metadata_state.spawn_git_enrichment_task(); turn_context @@ -2600,10 +2615,150 @@ impl Session { session_configuration, /*final_output_json_schema*/ None, /*sandbox_policy_changed*/ false, + /*job_context*/ None, ) .await } + pub(crate) async fn new_job_turn_with_sub_id( + &self, + sub_id: String, + job_context: JobTurnContext, + ) -> Arc { + let session_configuration = { + let state = self.state.lock().await; + state.session_configuration.clone() + }; + self.new_turn_from_configuration( + sub_id, + session_configuration, + /*final_output_json_schema*/ None, + /*sandbox_policy_changed*/ false, + Some(job_context), + ) + .await + } + + pub(crate) async fn list_jobs(&self) -> Vec { + self.jobs.lock().await.list_jobs() + } + + pub(crate) async fn create_job( + self: &Arc, + cron_expression: String, + prompt: String, + run_once: bool, + ) -> Result { + let schedule = crate::jobs::JobSchedule::parse(&cron_expression)?; + let timer_cancel = matches!(schedule, crate::jobs::JobSchedule::EverySeconds(_)) + .then(CancellationToken::new); + let id = uuid::Uuid::new_v4().to_string(); + let job = { + let mut jobs = self.jobs.lock().await; + jobs.create_job( + id.clone(), + cron_expression, + prompt, + run_once, + Utc::now(), + timer_cancel.clone(), + )? + }; + if let (crate::jobs::JobSchedule::EverySeconds(seconds), Some(cancel)) = + (schedule, timer_cancel) + { + self.spawn_job_timer(id, seconds, cancel); + } + self.emit_job_updated_notification().await; + self.maybe_start_pending_job().await; + Ok(job) + } + + pub(crate) async fn delete_job(&self, id: &str) -> bool { + let deleted = self.jobs.lock().await.delete_job(id); + if deleted { + self.emit_job_updated_notification().await; + } + deleted + } + + pub(crate) async fn mark_after_turn_jobs_due(&self) { + self.jobs.lock().await.mark_after_turn_jobs_due(); + } + + pub(crate) async fn maybe_start_pending_job(self: &Arc) { + if self.active_turn.lock().await.is_some() { + return; + } + if self.has_queued_response_items_for_next_turn().await + || self.has_trigger_turn_mailbox_items().await + { + return; + } + let Some(ClaimedJob { + job: _, + context, + deleted_run_once_job, + }) = self.jobs.lock().await.claim_next_job(Utc::now()) + else { + return; + }; + if deleted_run_once_job { + self.emit_job_updated_notification().await; + } + self.queue_response_items_for_next_turn(vec![job_prompt_input_item(&context.prompt)]) + .await; + let turn_context = self + .new_job_turn_with_sub_id(uuid::Uuid::new_v4().to_string(), context) + .await; + self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) + .await; + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(self.start_task( + turn_context, + Vec::new(), + crate::tasks::RegularTask::new(), + )); + }); + } + + fn spawn_job_timer( + self: &Arc, + id: String, + seconds: u64, + cancellation_token: CancellationToken, + ) { + let weak = Arc::downgrade(self); + tokio::spawn(async move { + loop { + tokio::select! { + _ = cancellation_token.cancelled() => break, + _ = tokio::time::sleep(std::time::Duration::from_secs(seconds)) => {} + } + let Some(session) = weak.upgrade() else { + break; + }; + session.jobs.lock().await.mark_job_due(&id, Utc::now()); + session.maybe_start_pending_job().await; + } + }); + } + + async fn emit_job_updated_notification(&self) { + let jobs = self.list_jobs().await; + let Ok(payload) = serde_json::to_string(&jobs) else { + warn!("failed to serialize job update payload"); + return; + }; + self.send_event_raw(Event { + id: INITIAL_SUBMIT_ID.to_owned(), + msg: EventMsg::BackgroundEvent(BackgroundEventEvent { + message: format!("{JOB_UPDATED_BACKGROUND_EVENT_PREFIX}{payload}"), + }), + }) + .await; + } + async fn build_settings_update_items( &self, reference_context_item: Option<&TurnContextItem>, @@ -3519,6 +3674,9 @@ impl Session { { developer_sections.push(developer_instructions.to_string()); } + if let Some(job_context) = turn_context.job_context.as_ref() { + developer_sections.push(job_turn_developer_instructions(job_context)); + } // Add developer instructions for memories. if turn_context.features.enabled(Feature::MemoryTool) && turn_context.config.memories.use_memories @@ -4019,8 +4177,6 @@ impl Session { } } - /// Queue response items to be injected into the next active turn created for this session. - #[cfg(test)] pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec) { if items.is_empty() { return; @@ -5466,6 +5622,7 @@ async fn spawn_review_thread( timezone: parent_turn_context.timezone.clone(), app_server_client_name: parent_turn_context.app_server_client_name.clone(), developer_instructions: None, + job_context: None, user_instructions: None, compact_prompt: parent_turn_context.compact_prompt.clone(), collaboration_mode: parent_turn_context.collaboration_mode.clone(), diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index f5e5d4463e..1e8684edce 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -12,6 +12,7 @@ use crate::config_loader::Sourced; use crate::exec::ExecCapturePolicy; use crate::exec::ExecToolCallOutput; use crate::function_tool::FunctionCallError; +use crate::jobs::JobsState; use crate::mcp_connection_manager::ToolInfo; use crate::models_manager::model_info; use crate::shell::default_user_shell; @@ -2751,6 +2752,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { mailbox, mailbox_rx: Mutex::new(mailbox_rx), idle_pending_input: Mutex::new(Vec::new()), + jobs: Mutex::new(JobsState::default()), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, js_repl, @@ -3594,6 +3596,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( mailbox, mailbox_rx: Mutex::new(mailbox_rx), idle_pending_input: Mutex::new(Vec::new()), + jobs: Mutex::new(JobsState::default()), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, js_repl, diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 033f07aa15..34062db8b2 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -26,6 +26,7 @@ use std::path::PathBuf; use tokio::sync::Mutex; use tokio::sync::watch; +use crate::jobs::ThreadJob; use crate::state_db::StateDbHandle; #[derive(Clone, Debug)] @@ -236,6 +237,26 @@ impl CodexThread { Ok(*guard) } + + pub async fn create_job( + &self, + cron_expression: String, + prompt: String, + run_once: bool, + ) -> Result { + self.codex + .session + .create_job(cron_expression, prompt, run_once) + .await + } + + pub async fn delete_job(&self, id: &str) -> bool { + self.codex.session.delete_job(id).await + } + + pub async fn list_jobs(&self) -> Vec { + self.codex.session.list_jobs().await + } } fn pending_message_input_item(message: &ResponseItem) -> CodexResult { diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 00227fe2b3..868aef74af 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -34,6 +34,7 @@ use core_test_support::PathExt; use core_test_support::TempDirExt; use core_test_support::test_absolute_path; use pretty_assertions::assert_eq; +use serial_test::serial; use std::collections::BTreeMap; use std::collections::HashMap; @@ -5724,79 +5725,95 @@ shell_tool = true } #[tokio::test] +#[serial] async fn approvals_reviewer_defaults_to_manual_only_without_guardian_feature() -> std::io::Result<()> { let codex_home = TempDir::new()?; - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .build() - .await?; + let config = Config::load_from_base_config_with_overrides( + ConfigToml::default(), + ConfigOverrides { + cwd: Some(codex_home.path().to_path_buf()), + ..Default::default() + }, + codex_home.path().to_path_buf(), + )?; assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User); Ok(()) } #[tokio::test] +#[serial] async fn approvals_reviewer_stays_manual_only_when_guardian_feature_is_enabled() -> std::io::Result<()> { let codex_home = TempDir::new()?; - std::fs::write( - codex_home.path().join(CONFIG_TOML_FILE), - r#"[features] + let raw_config = r#"[features] guardian_approval = true -"#, - )?; +"#; + std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?; + let config_toml = toml::from_str::(raw_config) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .build() - .await?; + let config = Config::load_from_base_config_with_overrides( + config_toml, + ConfigOverrides { + cwd: Some(codex_home.path().to_path_buf()), + ..Default::default() + }, + codex_home.path().to_path_buf(), + )?; assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User); Ok(()) } #[tokio::test] +#[serial] async fn approvals_reviewer_can_be_set_in_config_without_guardian_approval() -> std::io::Result<()> { let codex_home = TempDir::new()?; - std::fs::write( - codex_home.path().join(CONFIG_TOML_FILE), - r#"approvals_reviewer = "user" -"#, - )?; + let raw_config = r#"approvals_reviewer = "user" +"#; + std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?; + let config_toml = toml::from_str::(raw_config) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .build() - .await?; + let config = Config::load_from_base_config_with_overrides( + config_toml, + ConfigOverrides { + cwd: Some(codex_home.path().to_path_buf()), + ..Default::default() + }, + codex_home.path().to_path_buf(), + )?; assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User); Ok(()) } #[tokio::test] +#[serial] async fn approvals_reviewer_can_be_set_in_profile_without_guardian_approval() -> std::io::Result<()> { let codex_home = TempDir::new()?; - std::fs::write( - codex_home.path().join(CONFIG_TOML_FILE), - r#"profile = "guardian" + let raw_config = r#"profile = "guardian" [profiles.guardian] approvals_reviewer = "guardian_subagent" -"#, - )?; +"#; + std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?; + let config_toml = toml::from_str::(raw_config) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .build() - .await?; + let config = Config::load_from_base_config_with_overrides( + config_toml, + ConfigOverrides { + cwd: Some(codex_home.path().to_path_buf()), + ..Default::default() + }, + codex_home.path().to_path_buf(), + )?; assert_eq!( config.approvals_reviewer, @@ -5806,20 +5823,24 @@ approvals_reviewer = "guardian_subagent" } #[tokio::test] +#[serial] async fn smart_approvals_alias_is_ignored() -> std::io::Result<()> { let codex_home = TempDir::new()?; - std::fs::write( - codex_home.path().join(CONFIG_TOML_FILE), - r#"[features] + let raw_config = r#"[features] smart_approvals = true -"#, - )?; +"#; + std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?; + let config_toml = toml::from_str::(raw_config) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .build() - .await?; + let config = Config::load_from_base_config_with_overrides( + config_toml, + ConfigOverrides { + cwd: Some(codex_home.path().to_path_buf()), + ..Default::default() + }, + codex_home.path().to_path_buf(), + )?; assert!(!config.features.enabled(Feature::GuardianApproval)); assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User); @@ -5833,22 +5854,26 @@ smart_approvals = true } #[tokio::test] +#[serial] async fn smart_approvals_alias_is_ignored_in_profiles() -> std::io::Result<()> { let codex_home = TempDir::new()?; - std::fs::write( - codex_home.path().join(CONFIG_TOML_FILE), - r#"profile = "guardian" + let raw_config = r#"profile = "guardian" [profiles.guardian.features] smart_approvals = true -"#, - )?; +"#; + std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?; + let config_toml = toml::from_str::(raw_config) + .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?; - let config = ConfigBuilder::default() - .codex_home(codex_home.path().to_path_buf()) - .fallback_cwd(Some(codex_home.path().to_path_buf())) - .build() - .await?; + let config = Config::load_from_base_config_with_overrides( + config_toml, + ConfigOverrides { + cwd: Some(codex_home.path().to_path_buf()), + ..Default::default() + }, + codex_home.path().to_path_buf(), + )?; assert!(!config.features.enabled(Feature::GuardianApproval)); assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User); diff --git a/codex-rs/core/src/jobs.rs b/codex-rs/core/src/jobs.rs new file mode 100644 index 0000000000..559d508d91 --- /dev/null +++ b/codex-rs/core/src/jobs.rs @@ -0,0 +1,323 @@ +//! Runtime-only thread-local job scheduling for follow-on turns. +//! +//! This module owns the in-memory job registry, limited schedule parsing, and +//! the hidden turn context injected when a job fires so the model can act on +//! the scheduled prompt and deschedule itself via `JobDelete(currentJobId)`. + +use chrono::DateTime; +use chrono::Duration as ChronoDuration; +use chrono::Utc; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseInputItem; +use serde::Deserialize; +use serde::Serialize; +use std::collections::HashMap; +use tokio_util::sync::CancellationToken; + +pub const AFTER_TURN_CRON_EXPRESSION: &str = "@after-turn"; +const EVERY_PREFIX: &str = "@every "; +const EVERY_SECONDS_PREFIX: &str = "@every:"; +pub const JOB_UPDATED_BACKGROUND_EVENT_PREFIX: &str = "job_updated:"; +pub const JOB_FIRED_BACKGROUND_EVENT_PREFIX: &str = "job_fired:"; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ThreadJob { + pub id: String, + pub cron_expression: String, + pub prompt: String, + pub run_once: bool, + pub created_at: i64, + pub next_run_at: Option, + pub last_run_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct JobTurnContext { + pub(crate) current_job_id: String, + pub(crate) prompt: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ClaimedJob { + pub(crate) job: ThreadJob, + pub(crate) context: JobTurnContext, + pub(crate) deleted_run_once_job: bool, +} + +#[derive(Debug, Default)] +pub(crate) struct JobsState { + jobs: HashMap, +} + +#[derive(Debug)] +struct JobRuntime { + job: ThreadJob, + schedule: JobSchedule, + pending_run: bool, + timer_cancel: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum JobSchedule { + AfterTurn, + EverySeconds(u64), +} + +impl JobSchedule { + pub(crate) fn parse(cron_expression: &str) -> Result { + if cron_expression == AFTER_TURN_CRON_EXPRESSION { + return Ok(Self::AfterTurn); + } + + if let Some(seconds) = cron_expression + .strip_prefix(EVERY_PREFIX) + .map(str::trim) + .and_then(parse_duration_literal) + { + return Ok(Self::EverySeconds(seconds)); + } + + if let Some(seconds) = cron_expression + .strip_prefix(EVERY_SECONDS_PREFIX) + .and_then(|raw| raw.parse::().ok()) + .filter(|seconds| *seconds > 0) + { + return Ok(Self::EverySeconds(seconds)); + } + + Err(format!( + "unsupported cron_expression `{cron_expression}`; supported values are `{AFTER_TURN_CRON_EXPRESSION}`, `@every 5m`, or `@every:300`" + )) + } + + fn next_run_at(self, now: DateTime) -> Option> { + match self { + Self::AfterTurn => None, + Self::EverySeconds(seconds) => Some(now + ChronoDuration::seconds(seconds as i64)), + } + } +} + +impl JobsState { + pub(crate) fn list_jobs(&self) -> Vec { + let mut jobs = self + .jobs + .values() + .map(|runtime| runtime.job.clone()) + .collect::>(); + jobs.sort_by(|left, right| { + left.created_at + .cmp(&right.created_at) + .then_with(|| left.id.cmp(&right.id)) + }); + jobs + } + + pub(crate) fn create_job( + &mut self, + id: String, + cron_expression: String, + prompt: String, + run_once: bool, + now: DateTime, + timer_cancel: Option, + ) -> Result { + let schedule = JobSchedule::parse(&cron_expression)?; + let job = ThreadJob { + id: id.clone(), + cron_expression, + prompt, + run_once, + created_at: now.timestamp(), + next_run_at: schedule.next_run_at(now).map(|value| value.timestamp()), + last_run_at: None, + }; + self.jobs.insert( + id, + JobRuntime { + job: job.clone(), + schedule, + pending_run: matches!(schedule, JobSchedule::AfterTurn), + timer_cancel, + }, + ); + Ok(job) + } + + pub(crate) fn delete_job(&mut self, id: &str) -> bool { + let Some(runtime) = self.jobs.remove(id) else { + return false; + }; + if let Some(cancel) = runtime.timer_cancel { + cancel.cancel(); + } + true + } + + pub(crate) fn mark_after_turn_jobs_due(&mut self) { + for runtime in self.jobs.values_mut() { + if matches!(runtime.schedule, JobSchedule::AfterTurn) { + runtime.pending_run = true; + } + } + } + + pub(crate) fn mark_job_due(&mut self, id: &str, now: DateTime) { + let Some(runtime) = self.jobs.get_mut(id) else { + return; + }; + runtime.pending_run = true; + runtime.job.next_run_at = runtime + .schedule + .next_run_at(now) + .map(|value| value.timestamp()); + } + + pub(crate) fn claim_next_job(&mut self, now: DateTime) -> Option { + let next_job_id = self + .jobs + .values() + .filter(|runtime| runtime.pending_run) + .min_by(|left, right| { + left.job + .created_at + .cmp(&right.job.created_at) + .then_with(|| left.job.id.cmp(&right.job.id)) + }) + .map(|runtime| runtime.job.id.clone())?; + + let runtime = self.jobs.remove(&next_job_id)?; + let JobRuntime { + mut job, + schedule, + pending_run: _, + timer_cancel, + } = runtime; + let deleted_run_once_job = job.run_once; + if deleted_run_once_job { + if let Some(cancel) = timer_cancel { + cancel.cancel(); + } + } else { + job.last_run_at = Some(now.timestamp()); + self.jobs.insert( + job.id.clone(), + JobRuntime { + job: job.clone(), + schedule, + pending_run: false, + timer_cancel, + }, + ); + } + Some(ClaimedJob { + job: job.clone(), + context: JobTurnContext { + current_job_id: job.id, + prompt: job.prompt, + }, + deleted_run_once_job, + }) + } +} + +pub(crate) fn job_turn_developer_instructions(job: &JobTurnContext) -> String { + format!( + "This turn was triggered by a thread job.\ncurrentJobId: {}\n\nThe job prompt has already been injected as hidden context for this turn.\nIf you determine the job should stop, call JobDelete with currentJobId. Do not expose scheduler internals unless they matter to the user.", + job.current_job_id + ) +} + +pub(crate) fn job_prompt_input_item(prompt: &str) -> ResponseInputItem { + ResponseInputItem::Message { + role: "developer".to_string(), + content: vec![ContentItem::InputText { + text: prompt.to_string(), + }], + } +} + +fn parse_duration_literal(raw: &str) -> Option { + let mut digits = String::new(); + let mut unit = String::new(); + for ch in raw.chars() { + if ch.is_ascii_digit() && unit.is_empty() { + digits.push(ch); + } else if !ch.is_whitespace() { + unit.push(ch); + } + } + let value = digits.parse::().ok().filter(|value| *value > 0)?; + match unit.as_str() { + "s" | "sec" | "secs" | "second" | "seconds" => Some(value), + "m" | "min" | "mins" | "minute" | "minutes" => value.checked_mul(60), + "h" | "hr" | "hrs" | "hour" | "hours" => value.checked_mul(60 * 60), + "d" | "day" | "days" => value.checked_mul(60 * 60 * 24), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::AFTER_TURN_CRON_EXPRESSION; + use super::JobSchedule; + use super::JobsState; + use super::job_prompt_input_item; + use chrono::TimeZone; + use chrono::Utc; + use codex_protocol::models::ContentItem; + use codex_protocol::models::ResponseInputItem; + use pretty_assertions::assert_eq; + + #[test] + fn parses_supported_job_schedules() { + assert_eq!( + JobSchedule::parse(AFTER_TURN_CRON_EXPRESSION), + Ok(JobSchedule::AfterTurn) + ); + assert_eq!( + JobSchedule::parse("@every 5m"), + Ok(JobSchedule::EverySeconds(300)) + ); + assert_eq!( + JobSchedule::parse("@every:3600"), + Ok(JobSchedule::EverySeconds(3600)) + ); + } + + #[test] + fn claim_run_once_job_removes_it() { + let now = Utc.timestamp_opt(100, 0).single().expect("valid timestamp"); + let mut jobs = JobsState::default(); + let job = jobs + .create_job( + "job-1".to_string(), + AFTER_TURN_CRON_EXPRESSION.to_string(), + "run tests".to_string(), + true, + now, + None, + ) + .expect("job should be created"); + assert_eq!(jobs.list_jobs(), vec![job]); + + let claimed = jobs.claim_next_job(now).expect("job should be claimed"); + assert_eq!(claimed.context.current_job_id, "job-1"); + assert!(claimed.deleted_run_once_job); + assert!(jobs.list_jobs().is_empty()); + } + + #[test] + fn job_prompt_input_is_hidden_developer_input() { + let item = job_prompt_input_item("run tests"); + assert_eq!( + item, + ResponseInputItem::Message { + role: "developer".to_string(), + content: vec![ContentItem::InputText { + text: "run tests".to_string(), + }], + } + ); + } +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 5276b09de3..6a1c835cfe 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -44,6 +44,7 @@ mod git_info_tests; mod guardian; mod hook_runtime; pub mod instructions; +pub mod jobs; pub mod landlock; pub mod mcp; mod mcp_connection_manager; diff --git a/codex-rs/core/src/shell_snapshot.rs b/codex-rs/core/src/shell_snapshot.rs index 29b50cb9e8..4169406335 100644 --- a/codex-rs/core/src/shell_snapshot.rs +++ b/codex-rs/core/src/shell_snapshot.rs @@ -29,7 +29,7 @@ pub struct ShellSnapshot { pub cwd: PathBuf, } -const SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(10); +const SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(20); const SNAPSHOT_RETENTION: Duration = Duration::from_secs(60 * 60 * 24 * 3); // 3 days retention. const SNAPSHOT_DIR: &str = "shell_snapshots"; const EXCLUDED_EXPORT_VARS: &[&str] = &["PWD", "OLDPWD"]; diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 07d040ca21..7b3153709f 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -48,7 +48,7 @@ pub(crate) enum TaskKind { pub(crate) struct RunningTask { pub(crate) done: Arc, pub(crate) kind: TaskKind, - pub(crate) task: Arc, + pub(crate) task: Arc, pub(crate) cancellation_token: CancellationToken, pub(crate) handle: Arc>, pub(crate) turn_context: Arc, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index e1db8b78c2..6f1127ce92 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -172,13 +172,13 @@ impl Session { self.start_task(turn_context, input, task).await; } - async fn start_task( + pub(crate) async fn start_task( self: &Arc, turn_context: Arc, input: Vec, task: T, ) { - let task: Arc = Arc::new(task); + let task: Arc = Arc::new(task); let task_kind = task.kind(); let span_name = task.span_name(); let started_at = Instant::now(); @@ -436,6 +436,8 @@ impl Session { last_agent_message, }); self.send_event(turn_context.as_ref(), event).await; + self.mark_after_turn_jobs_due().await; + self.maybe_start_pending_job().await; } async fn take_active_turn(&self) -> Option { diff --git a/codex-rs/core/src/tools/handlers/jobs.rs b/codex-rs/core/src/tools/handlers/jobs.rs new file mode 100644 index 0000000000..7eee972396 --- /dev/null +++ b/codex-rs/core/src/tools/handlers/jobs.rs @@ -0,0 +1,105 @@ +use async_trait::async_trait; +use serde::Deserialize; + +use crate::function_tool::FunctionCallError; +use crate::tools::context::FunctionToolOutput; +use crate::tools::context::ToolInvocation; +use crate::tools::context::ToolPayload; +use crate::tools::handlers::parse_arguments; +use crate::tools::registry::ToolHandler; +use crate::tools::registry::ToolKind; + +#[derive(Deserialize)] +struct JobCreateArgs { + cron_expression: String, + prompt: String, + run_once: Option, +} + +#[derive(Deserialize)] +struct JobDeleteArgs { + id: String, +} + +pub struct JobCreateHandler; + +#[async_trait] +impl ToolHandler for JobCreateHandler { + type Output = FunctionToolOutput; + + fn kind(&self) -> ToolKind { + ToolKind::Function + } + + async fn handle(&self, invocation: ToolInvocation) -> Result { + let ToolPayload::Function { arguments } = invocation.payload else { + return Err(FunctionCallError::RespondToModel( + "JobCreate received unsupported payload".to_string(), + )); + }; + let args: JobCreateArgs = parse_arguments(&arguments)?; + let job = invocation + .session + .create_job( + args.cron_expression, + args.prompt, + args.run_once.unwrap_or(false), + ) + .await + .map_err(FunctionCallError::RespondToModel)?; + let content = serde_json::to_string(&job).map_err(|err| { + FunctionCallError::Fatal(format!("failed to serialize JobCreate response: {err}")) + })?; + Ok(FunctionToolOutput::from_text(content, Some(true))) + } +} + +pub struct JobDeleteHandler; + +#[async_trait] +impl ToolHandler for JobDeleteHandler { + type Output = FunctionToolOutput; + + fn kind(&self) -> ToolKind { + ToolKind::Function + } + + async fn handle(&self, invocation: ToolInvocation) -> Result { + let ToolPayload::Function { arguments } = invocation.payload else { + return Err(FunctionCallError::RespondToModel( + "JobDelete received unsupported payload".to_string(), + )); + }; + let args: JobDeleteArgs = parse_arguments(&arguments)?; + let deleted = invocation.session.delete_job(&args.id).await; + let content = serde_json::json!({ "deleted": deleted }).to_string(); + Ok(FunctionToolOutput::from_text(content, Some(deleted))) + } +} + +pub struct JobListHandler; + +#[async_trait] +impl ToolHandler for JobListHandler { + type Output = FunctionToolOutput; + + fn kind(&self) -> ToolKind { + ToolKind::Function + } + + async fn handle(&self, invocation: ToolInvocation) -> Result { + match invocation.payload { + ToolPayload::Function { .. } => {} + _ => { + return Err(FunctionCallError::RespondToModel( + "JobList received unsupported payload".to_string(), + )); + } + } + let jobs = invocation.session.list_jobs().await; + let content = serde_json::to_string(&jobs).map_err(|err| { + FunctionCallError::Fatal(format!("failed to serialize JobList response: {err}")) + })?; + Ok(FunctionToolOutput::from_text(content, Some(true))) + } +} diff --git a/codex-rs/core/src/tools/handlers/mod.rs b/codex-rs/core/src/tools/handlers/mod.rs index 86f2175be6..06455cdac9 100644 --- a/codex-rs/core/src/tools/handlers/mod.rs +++ b/codex-rs/core/src/tools/handlers/mod.rs @@ -1,6 +1,7 @@ pub(crate) mod agent_jobs; pub mod apply_patch; mod dynamic; +mod jobs; mod js_repl; mod list_dir; mod mcp; @@ -37,6 +38,9 @@ pub use apply_patch::ApplyPatchHandler; use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::AskForApproval; pub use dynamic::DynamicToolHandler; +pub use jobs::JobCreateHandler; +pub use jobs::JobDeleteHandler; +pub use jobs::JobListHandler; pub use js_repl::JsReplHandler; pub use js_repl::JsReplResetHandler; pub use list_dir::ListDirHandler; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 091b421574..49c03bccf9 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1074,7 +1074,7 @@ async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message ) })); - timeout(Duration::from_secs(5), async { + timeout(Duration::from_secs(15), async { loop { let history_items = thread .codex diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index 6852671576..96690eac2d 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/codex-rs/core/src/tools/spec.rs @@ -8,6 +8,9 @@ use crate::shell::ShellType; use crate::tools::code_mode::PUBLIC_TOOL_NAME; use crate::tools::code_mode::WAIT_TOOL_NAME; use crate::tools::discoverable::DiscoverableTool; +use crate::tools::handlers::JobCreateHandler; +use crate::tools::handlers::JobDeleteHandler; +use crate::tools::handlers::JobListHandler; use crate::tools::handlers::PLAN_TOOL; use crate::tools::handlers::TOOL_SEARCH_DEFAULT_LIMIT; use crate::tools::handlers::TOOL_SEARCH_TOOL_NAME; @@ -51,6 +54,9 @@ use codex_tools::create_close_agent_tool_v1; use codex_tools::create_close_agent_tool_v2; use codex_tools::create_code_mode_tool; use codex_tools::create_exec_command_tool; +use codex_tools::create_job_create_tool; +use codex_tools::create_job_delete_tool; +use codex_tools::create_job_list_tool; use codex_tools::create_js_repl_reset_tool; use codex_tools::create_js_repl_tool; use codex_tools::create_list_agents_tool; @@ -167,6 +173,7 @@ pub(crate) struct ToolsConfig { pub collab_tools: bool, pub multi_agent_v2: bool, pub request_user_input: bool, + pub job_scheduler: bool, pub default_mode_request_user_input: bool, pub experimental_supported_tools: Vec, pub agent_jobs_tools: bool, @@ -217,6 +224,7 @@ impl ToolsConfig { let include_multi_agent_v2 = features.enabled(Feature::MultiAgentV2); let include_agent_jobs = features.enabled(Feature::SpawnCsv); let include_request_user_input = !matches!(session_source, SessionSource::SubAgent(_)); + let include_job_scheduler = features.enabled(Feature::JobScheduler); let include_default_mode_request_user_input = include_request_user_input && features.enabled(Feature::DefaultModeRequestUserInput); let include_search_tool = @@ -301,6 +309,7 @@ impl ToolsConfig { collab_tools: include_collab_tools, multi_agent_v2: include_multi_agent_v2, request_user_input: include_request_user_input, + job_scheduler: include_job_scheduler, default_mode_request_user_input: include_default_mode_request_user_input, experimental_supported_tools: model_info.experimental_supported_tools.clone(), agent_jobs_tools: include_agent_jobs, @@ -626,6 +635,30 @@ pub(crate) fn build_specs_with_discoverable_tools( builder.register_handler("request_permissions", request_permissions_handler); } + if config.job_scheduler { + push_tool_spec( + &mut builder, + create_job_create_tool(), + /*supports_parallel_tool_calls*/ false, + config.code_mode_enabled, + ); + push_tool_spec( + &mut builder, + create_job_delete_tool(), + /*supports_parallel_tool_calls*/ false, + config.code_mode_enabled, + ); + push_tool_spec( + &mut builder, + create_job_list_tool(), + /*supports_parallel_tool_calls*/ false, + config.code_mode_enabled, + ); + builder.register_handler("JobCreate", Arc::new(JobCreateHandler)); + builder.register_handler("JobDelete", Arc::new(JobDeleteHandler)); + builder.register_handler("JobList", Arc::new(JobListHandler)); + } + if config.search_tool && let Some(app_tools) = app_tools { diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 7f0d326a06..3ca287d619 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -170,6 +170,8 @@ pub enum Feature { Personality, /// Enable native artifact tools. Artifact, + /// Enable per-thread runtime job scheduling tools and APIs. + JobScheduler, /// Enable Fast mode selection in the TUI and request layer. FastMode, /// Enable experimental realtime voice conversation mode in the TUI. @@ -805,6 +807,16 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::JobScheduler, + key: "job_scheduler", + stage: Stage::Experimental { + name: "Job Scheduler", + menu_description: "Enable thread-local scheduled jobs and the /loop command.", + announcement: "", + }, + default_enabled: false, + }, FeatureSpec { id: Feature::FastMode, key: "fast_mode", diff --git a/codex-rs/tools/src/job_tool.rs b/codex-rs/tools/src/job_tool.rs new file mode 100644 index 0000000000..fc0d27da6e --- /dev/null +++ b/codex-rs/tools/src/job_tool.rs @@ -0,0 +1,89 @@ +use crate::JsonSchema; +use crate::ResponsesApiTool; +use crate::ToolSpec; +use std::collections::BTreeMap; + +pub fn create_job_create_tool() -> ToolSpec { + let properties = BTreeMap::from([ + ( + "cron_expression".to_string(), + JsonSchema::String { + description: Some( + "Scheduler expression for the job. Supported values are scheduler-specific." + .to_string(), + ), + }, + ), + ( + "prompt".to_string(), + JsonSchema::String { + description: Some("Prompt to execute when the job fires.".to_string()), + }, + ), + ( + "run_once".to_string(), + JsonSchema::Boolean { + description: Some( + "Optional. When true, delete the job after its next execution is claimed." + .to_string(), + ), + }, + ), + ]); + + ToolSpec::Function(ResponsesApiTool { + name: "JobCreate".to_string(), + description: + "Create a runtime-only thread job using a structured scheduler expression and prompt." + .to_string(), + strict: false, + defer_loading: None, + parameters: JsonSchema::Object { + properties, + required: Some(vec!["cron_expression".to_string(), "prompt".to_string()]), + additional_properties: Some(false.into()), + }, + output_schema: None, + }) +} + +pub fn create_job_delete_tool() -> ToolSpec { + let properties = BTreeMap::from([( + "id".to_string(), + JsonSchema::String { + description: Some("Identifier of the job to delete.".to_string()), + }, + )]); + + ToolSpec::Function(ResponsesApiTool { + name: "JobDelete".to_string(), + description: "Delete a runtime-only thread job by id.".to_string(), + strict: false, + defer_loading: None, + parameters: JsonSchema::Object { + properties, + required: Some(vec!["id".to_string()]), + additional_properties: Some(false.into()), + }, + output_schema: None, + }) +} + +pub fn create_job_list_tool() -> ToolSpec { + ToolSpec::Function(ResponsesApiTool { + name: "JobList".to_string(), + description: "List runtime-only thread jobs for the current thread.".to_string(), + strict: false, + defer_loading: None, + parameters: JsonSchema::Object { + properties: BTreeMap::new(), + required: None, + additional_properties: Some(false.into()), + }, + output_schema: None, + }) +} + +#[cfg(test)] +#[path = "job_tool_tests.rs"] +mod tests; diff --git a/codex-rs/tools/src/job_tool_tests.rs b/codex-rs/tools/src/job_tool_tests.rs new file mode 100644 index 0000000000..be74e0b25e --- /dev/null +++ b/codex-rs/tools/src/job_tool_tests.rs @@ -0,0 +1,31 @@ +use crate::ResponsesApiTool; +use crate::ToolSpec; +use pretty_assertions::assert_eq; + +use super::create_job_create_tool; +use super::create_job_delete_tool; +use super::create_job_list_tool; + +#[test] +fn job_create_tool_uses_expected_name() { + let ToolSpec::Function(ResponsesApiTool { name, .. }) = create_job_create_tool() else { + panic!("expected function tool"); + }; + assert_eq!(name, "JobCreate"); +} + +#[test] +fn job_delete_tool_uses_expected_name() { + let ToolSpec::Function(ResponsesApiTool { name, .. }) = create_job_delete_tool() else { + panic!("expected function tool"); + }; + assert_eq!(name, "JobDelete"); +} + +#[test] +fn job_list_tool_uses_expected_name() { + let ToolSpec::Function(ResponsesApiTool { name, .. }) = create_job_list_tool() else { + panic!("expected function tool"); + }; + assert_eq!(name, "JobList"); +} diff --git a/codex-rs/tools/src/lib.rs b/codex-rs/tools/src/lib.rs index 9dd629f916..7663faa167 100644 --- a/codex-rs/tools/src/lib.rs +++ b/codex-rs/tools/src/lib.rs @@ -5,6 +5,7 @@ mod agent_job_tool; mod agent_tool; mod code_mode; mod dynamic_tool; +mod job_tool; mod js_repl_tool; mod json_schema; mod local_tool; @@ -38,6 +39,9 @@ pub use code_mode::create_code_mode_tool; pub use code_mode::create_wait_tool; pub use code_mode::tool_spec_to_code_mode_tool_definition; pub use dynamic_tool::parse_dynamic_tool; +pub use job_tool::create_job_create_tool; +pub use job_tool::create_job_delete_tool; +pub use job_tool::create_job_list_tool; pub use js_repl_tool::create_js_repl_reset_tool; pub use js_repl_tool::create_js_repl_tool; pub use json_schema::AdditionalProperties; diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 5ea10e70ea..6dd2dc0628 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -30,6 +30,8 @@ use crate::history_cell; use crate::history_cell::HistoryCell; #[cfg(not(debug_assertions))] use crate::history_cell::UpdateAvailableHistoryCell; +use crate::job_scheduler::format_job_summary; +use crate::job_scheduler::parse_job_spec; use crate::model_catalog::ModelCatalog; use crate::model_migration::ModelMigrationOutcome; use crate::model_migration::migration_copy_for_models; @@ -1942,6 +1944,26 @@ impl App { }); } + fn parse_thread_job_spec(&mut self, thread_id: ThreadId, spec: String) { + let config = self.config.clone(); + let target = match self.remote_app_server_url.clone() { + Some(websocket_url) => crate::AppServerTarget::Remote { + websocket_url, + auth_token: self.remote_app_server_auth_token.clone(), + }, + None => crate::AppServerTarget::Embedded, + }; + let app_event_tx = self.app_event_tx.clone(); + tokio::spawn(async move { + let result = parse_job_spec(config, target, spec.clone()).await; + app_event_tx.send(AppEvent::ThreadJobSpecParsed { + thread_id, + spec, + result, + }); + }); + } + /// Process the completed MCP inventory fetch: clear the loading spinner, then /// render either the full tool/resource listing or an error into chat history. /// @@ -3123,11 +3145,20 @@ impl App { app_server: &mut AppServerSession, started: AppServerStartedThread, ) -> Result<()> { + let thread_id = started.session.thread_id; self.reset_thread_event_state(); let init = self.chatwidget_init_for_forked_or_resumed_thread(tui, self.config.clone()); self.replace_chat_widget(ChatWidget::new_with_app_event(init)); self.enqueue_primary_thread_session(started.session, started.turns) .await?; + if self.config.features.enabled(Feature::JobScheduler) { + match app_server.thread_job_list(thread_id).await { + Ok(jobs) => self.chat_widget.on_thread_jobs_updated(jobs), + Err(err) => { + tracing::warn!(%err, "failed to load thread jobs while attaching thread"); + } + } + } self.backfill_loaded_subagent_threads(app_server).await; Ok(()) } @@ -4009,6 +4040,69 @@ impl App { tui.frame_requester().schedule_frame(); } + AppEvent::OpenThreadJobs { thread_id } => { + match app_server.thread_job_list(thread_id).await { + Ok(jobs) => self.chat_widget.open_thread_jobs_popup(thread_id, jobs), + Err(err) => { + self.chat_widget + .add_error_message(format!("Failed to load thread jobs: {err}")); + } + } + } + AppEvent::CreateThreadJobFromSpec { thread_id, spec } => { + self.parse_thread_job_spec(thread_id, spec); + } + AppEvent::ThreadJobSpecParsed { + thread_id, + spec, + result, + } => match result { + Ok(parsed) => match app_server + .thread_job_create( + thread_id, + parsed.cron_expression.clone(), + parsed.prompt.clone(), + parsed.run_once, + ) + .await + { + Ok(job) => { + let summary = + format_job_summary(&job.cron_expression, job.run_once, &job.prompt); + self.chat_widget.add_info_message( + format!("Created thread job from `/loop {spec}`."), + Some(summary), + ); + } + Err(err) => { + self.chat_widget + .add_error_message(format!("Failed to create thread job: {err}")); + } + }, + Err(err) => { + self.chat_widget + .add_error_message(format!("Failed to parse `/loop {spec}`: {err}")); + } + }, + AppEvent::DeleteThreadJob { thread_id, id } => { + match app_server.thread_job_delete(thread_id, id.clone()).await { + Ok(true) => { + self.chat_widget.add_info_message( + format!("Deleted thread job `{id}`."), + /*hint*/ None, + ); + } + Ok(false) => { + self.chat_widget + .add_error_message(format!("No thread job matched `{id}`.")); + } + Err(err) => { + self.chat_widget.add_error_message(format!( + "Failed to delete thread job `{id}`: {err}" + )); + } + } + } AppEvent::InsertHistoryCell(cell) => { let cell: Arc = cell.into(); if let Some(Overlay::Transcript(t)) = &mut self.overlay { @@ -7431,6 +7525,14 @@ mod tests { let codex_home = tempdir()?; app.config.codex_home = codex_home.path().to_path_buf(); let guardian_approvals = guardian_approvals_mode(); + app.config + .features + .set_enabled(Feature::GuardianApproval, /*enabled*/ false)?; + app.chat_widget + .set_feature_enabled(Feature::GuardianApproval, /*enabled*/ false); + app.config.approvals_reviewer = ApprovalsReviewer::User; + app.chat_widget + .set_approvals_reviewer(ApprovalsReviewer::User); app.update_feature_flags(vec![(Feature::GuardianApproval, true)]) .await; @@ -7518,9 +7620,7 @@ mod tests { let config_toml = "approvals_reviewer = \"guardian_subagent\"\napproval_policy = \"on-request\"\nsandbox_mode = \"workspace-write\"\n\n[features]\nguardian_approval = true\n"; std::fs::write(config_toml_path.as_path(), config_toml)?; let user_config = toml::from_str::(config_toml)?; - app.config.config_layer_stack = app - .config - .config_layer_stack + app.config.config_layer_stack = codex_core::config_loader::ConfigLayerStack::default() .with_user_config(&config_toml_path, user_config); app.config .features diff --git a/codex-rs/tui/src/app/app_server_adapter.rs b/codex-rs/tui/src/app/app_server_adapter.rs index 12b1c669b6..06a33575af 100644 --- a/codex-rs/tui/src/app/app_server_adapter.rs +++ b/codex-rs/tui/src/app/app_server_adapter.rs @@ -329,6 +329,8 @@ fn server_notification_thread_target( ServerNotification::ThreadNameUpdated(notification) => { Some(notification.thread_id.as_str()) } + ServerNotification::ThreadJobFired(notification) => Some(notification.thread_id.as_str()), + ServerNotification::ThreadJobUpdated(notification) => Some(notification.thread_id.as_str()), ServerNotification::ThreadTokenUsageUpdated(notification) => { Some(notification.thread_id.as_str()) } diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 2ae175013b..379d262de0 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -30,6 +30,7 @@ use crate::bottom_pane::ApprovalRequest; use crate::bottom_pane::StatusLineItem; use crate::bottom_pane::TerminalTitleItem; use crate::history_cell::HistoryCell; +use crate::job_scheduler::ParsedJobSpec; use codex_core::config::types::ApprovalsReviewer; use codex_features::Feature; @@ -108,6 +109,30 @@ pub(crate) enum AppEvent { /// Fork the current session into a new thread. ForkCurrentSession, + /// List jobs for the specified thread and open the management UI. + OpenThreadJobs { + thread_id: ThreadId, + }, + + /// Parse a `/loop ` input into structured job params. + CreateThreadJobFromSpec { + thread_id: ThreadId, + spec: String, + }, + + /// Result of parsing a `/loop ` input. + ThreadJobSpecParsed { + thread_id: ThreadId, + spec: String, + result: Result, + }, + + /// Delete one job from the specified thread. + DeleteThreadJob { + thread_id: ThreadId, + id: String, + }, + /// Request to exit the application. /// /// Use `ShutdownFirst` for user-initiated quits so core cleanup runs and the diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index eb6c176076..aaf35bdd7d 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -30,6 +30,13 @@ use codex_app_server_protocol::ThreadCompactStartParams; use codex_app_server_protocol::ThreadCompactStartResponse; use codex_app_server_protocol::ThreadForkParams; use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadJob; +use codex_app_server_protocol::ThreadJobCreateParams; +use codex_app_server_protocol::ThreadJobCreateResponse; +use codex_app_server_protocol::ThreadJobDeleteParams; +use codex_app_server_protocol::ThreadJobDeleteResponse; +use codex_app_server_protocol::ThreadJobListParams; +use codex_app_server_protocol::ThreadJobListResponse; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadLoadedListParams; @@ -297,6 +304,18 @@ impl AppServerSession { started_thread_from_start_response(response, config).await } + pub(crate) async fn start_ephemeral_thread(&mut self, config: &Config) -> Result { + let request_id = self.next_request_id(); + let mut params = thread_start_params_from_config(config, self.thread_params_mode()); + params.ephemeral = Some(true); + let response: ThreadStartResponse = self + .client + .request_typed(ClientRequest::ThreadStart { request_id, params }) + .await + .wrap_err("thread/start failed for ephemeral TUI session")?; + Ok(response.thread) + } + pub(crate) async fn resume_thread( &mut self, config: Config, @@ -393,6 +412,65 @@ impl AppServerSession { Ok(response.thread) } + pub(crate) async fn thread_job_create( + &mut self, + thread_id: ThreadId, + cron_expression: String, + prompt: String, + run_once: Option, + ) -> Result { + let request_id = self.next_request_id(); + let response: ThreadJobCreateResponse = self + .client + .request_typed(ClientRequest::ThreadJobCreate { + request_id, + params: ThreadJobCreateParams { + thread_id: thread_id.to_string(), + cron_expression, + prompt, + run_once, + }, + }) + .await + .wrap_err("thread/job/create failed in TUI")?; + Ok(response.job) + } + + pub(crate) async fn thread_job_delete( + &mut self, + thread_id: ThreadId, + id: String, + ) -> Result { + let request_id = self.next_request_id(); + let response: ThreadJobDeleteResponse = self + .client + .request_typed(ClientRequest::ThreadJobDelete { + request_id, + params: ThreadJobDeleteParams { + thread_id: thread_id.to_string(), + id, + }, + }) + .await + .wrap_err("thread/job/delete failed in TUI")?; + Ok(response.deleted) + } + + pub(crate) async fn thread_job_list(&mut self, thread_id: ThreadId) -> Result> { + let request_id = self.next_request_id(); + let response: ThreadJobListResponse = self + .client + .request_typed(ClientRequest::ThreadJobList { + request_id, + params: ThreadJobListParams { + thread_id: thread_id.to_string(), + }, + }) + .await + .wrap_err("thread/job/list failed in TUI")?; + Ok(response.data) + } + #[allow(clippy::too_many_arguments)] pub(crate) async fn turn_start( &mut self, diff --git a/codex-rs/tui/src/bottom_pane/chat_composer.rs b/codex-rs/tui/src/bottom_pane/chat_composer.rs index 22da977e77..181950f986 100644 --- a/codex-rs/tui/src/bottom_pane/chat_composer.rs +++ b/codex-rs/tui/src/bottom_pane/chat_composer.rs @@ -328,6 +328,7 @@ pub(crate) struct ChatComposer { collaboration_mode_indicator: Option, connectors_enabled: bool, plugins_command_enabled: bool, + job_scheduler_command_enabled: bool, fast_command_enabled: bool, personality_command_enabled: bool, realtime_conversation_enabled: bool, @@ -367,6 +368,7 @@ impl ChatComposer { collaboration_modes_enabled: self.collaboration_modes_enabled, connectors_enabled: self.connectors_enabled, plugins_command_enabled: self.plugins_command_enabled, + job_scheduler_command_enabled: self.job_scheduler_command_enabled, fast_command_enabled: self.fast_command_enabled, personality_command_enabled: self.personality_command_enabled, realtime_conversation_enabled: self.realtime_conversation_enabled, @@ -449,6 +451,7 @@ impl ChatComposer { collaboration_mode_indicator: None, connectors_enabled: false, plugins_command_enabled: false, + job_scheduler_command_enabled: false, fast_command_enabled: false, personality_command_enabled: false, realtime_conversation_enabled: false, @@ -487,6 +490,10 @@ impl ChatComposer { self.plugins_command_enabled = enabled; } + pub fn set_job_scheduler_command_enabled(&mut self, enabled: bool) { + self.job_scheduler_command_enabled = enabled; + } + /// Toggle composer-side image paste handling. /// /// This only affects whether image-like paste content is converted into attachments; the @@ -3063,6 +3070,7 @@ impl ChatComposer { let collaboration_modes_enabled = self.collaboration_modes_enabled; let connectors_enabled = self.connectors_enabled; let plugins_command_enabled = self.plugins_command_enabled; + let job_scheduler_command_enabled = self.job_scheduler_command_enabled; let fast_command_enabled = self.fast_command_enabled; let personality_command_enabled = self.personality_command_enabled; let realtime_conversation_enabled = self.realtime_conversation_enabled; @@ -3071,6 +3079,7 @@ impl ChatComposer { collaboration_modes_enabled, connectors_enabled, plugins_command_enabled, + job_scheduler_command_enabled, fast_command_enabled, personality_command_enabled, realtime_conversation_enabled, diff --git a/codex-rs/tui/src/bottom_pane/command_popup.rs b/codex-rs/tui/src/bottom_pane/command_popup.rs index 8bef8ddbc8..05d803df83 100644 --- a/codex-rs/tui/src/bottom_pane/command_popup.rs +++ b/codex-rs/tui/src/bottom_pane/command_popup.rs @@ -33,6 +33,7 @@ pub(crate) struct CommandPopupFlags { pub(crate) collaboration_modes_enabled: bool, pub(crate) connectors_enabled: bool, pub(crate) plugins_command_enabled: bool, + pub(crate) job_scheduler_command_enabled: bool, pub(crate) fast_command_enabled: bool, pub(crate) personality_command_enabled: bool, pub(crate) realtime_conversation_enabled: bool, @@ -46,6 +47,7 @@ impl From for slash_commands::BuiltinCommandFlags { collaboration_modes_enabled: value.collaboration_modes_enabled, connectors_enabled: value.connectors_enabled, plugins_command_enabled: value.plugins_command_enabled, + job_scheduler_command_enabled: value.job_scheduler_command_enabled, fast_command_enabled: value.fast_command_enabled, personality_command_enabled: value.personality_command_enabled, realtime_conversation_enabled: value.realtime_conversation_enabled, @@ -355,6 +357,7 @@ mod tests { connectors_enabled: false, plugins_command_enabled: false, fast_command_enabled: false, + job_scheduler_command_enabled: false, personality_command_enabled: true, realtime_conversation_enabled: false, audio_device_selection_enabled: false, @@ -375,6 +378,7 @@ mod tests { connectors_enabled: false, plugins_command_enabled: false, fast_command_enabled: false, + job_scheduler_command_enabled: false, personality_command_enabled: true, realtime_conversation_enabled: false, audio_device_selection_enabled: false, @@ -395,6 +399,7 @@ mod tests { connectors_enabled: false, plugins_command_enabled: false, fast_command_enabled: false, + job_scheduler_command_enabled: false, personality_command_enabled: false, realtime_conversation_enabled: false, audio_device_selection_enabled: false, @@ -422,6 +427,7 @@ mod tests { connectors_enabled: false, plugins_command_enabled: false, fast_command_enabled: false, + job_scheduler_command_enabled: false, personality_command_enabled: true, realtime_conversation_enabled: false, audio_device_selection_enabled: false, @@ -442,6 +448,7 @@ mod tests { connectors_enabled: false, plugins_command_enabled: false, fast_command_enabled: false, + job_scheduler_command_enabled: false, personality_command_enabled: true, realtime_conversation_enabled: true, audio_device_selection_enabled: false, diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index 4016821d44..b4afad0a0e 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -270,6 +270,11 @@ impl BottomPane { self.request_redraw(); } + pub fn set_job_scheduler_command_enabled(&mut self, enabled: bool) { + self.composer.set_job_scheduler_command_enabled(enabled); + self.request_redraw(); + } + pub fn take_mention_bindings(&mut self) -> Vec { self.composer.take_mention_bindings() } diff --git a/codex-rs/tui/src/bottom_pane/slash_commands.rs b/codex-rs/tui/src/bottom_pane/slash_commands.rs index 54b1a8cf4e..77dfcf5eb1 100644 --- a/codex-rs/tui/src/bottom_pane/slash_commands.rs +++ b/codex-rs/tui/src/bottom_pane/slash_commands.rs @@ -15,6 +15,7 @@ pub(crate) struct BuiltinCommandFlags { pub(crate) collaboration_modes_enabled: bool, pub(crate) connectors_enabled: bool, pub(crate) plugins_command_enabled: bool, + pub(crate) job_scheduler_command_enabled: bool, pub(crate) fast_command_enabled: bool, pub(crate) personality_command_enabled: bool, pub(crate) realtime_conversation_enabled: bool, @@ -33,6 +34,7 @@ pub(crate) fn builtins_for_input(flags: BuiltinCommandFlags) -> Vec<(&'static st }) .filter(|(_, cmd)| flags.connectors_enabled || *cmd != SlashCommand::Apps) .filter(|(_, cmd)| flags.plugins_command_enabled || *cmd != SlashCommand::Plugins) + .filter(|(_, cmd)| flags.job_scheduler_command_enabled || *cmd != SlashCommand::Loop) .filter(|(_, cmd)| flags.fast_command_enabled || *cmd != SlashCommand::Fast) .filter(|(_, cmd)| flags.personality_command_enabled || *cmd != SlashCommand::Personality) .filter(|(_, cmd)| flags.realtime_conversation_enabled || *cmd != SlashCommand::Realtime) @@ -66,6 +68,7 @@ mod tests { collaboration_modes_enabled: true, connectors_enabled: true, plugins_command_enabled: true, + job_scheduler_command_enabled: true, fast_command_enabled: true, personality_command_enabled: true, realtime_conversation_enabled: true, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 1267e6b732..10e4069316 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -345,6 +345,7 @@ use crate::status_indicator_widget::STATUS_DETAILS_DEFAULT_MAX_LINES; use crate::status_indicator_widget::StatusDetailsCapitalization; use crate::text_formatting::truncate_text; use crate::tui::FrameRequester; +use codex_app_server_protocol::ThreadJob; mod interrupts; use self::interrupts::InterruptManager; mod session_header; @@ -839,6 +840,7 @@ pub(crate) struct ChatWidget { thread_id: Option, thread_name: Option, forked_from: Option, + thread_jobs: Vec, frame_requester: FrameRequester, // Whether to include the initial welcome banner on session configured show_welcome_banner: bool, @@ -1949,6 +1951,7 @@ impl ChatWidget { self.thread_id = Some(event.session_id); self.thread_name = event.thread_name.clone(); self.forked_from = event.forked_from_id; + self.thread_jobs.clear(); self.current_rollout_path = event.rollout_path.clone(); self.current_cwd = Some(event.cwd.clone()); match AbsolutePathBuf::try_from(event.cwd.clone()) { @@ -1996,6 +1999,7 @@ impl ChatWidget { self.refresh_model_display(); self.refresh_status_surfaces(); self.sync_fast_command_enabled(); + self.sync_job_scheduler_command_enabled(); self.sync_personality_command_enabled(); self.sync_plugins_command_enabled(); self.refresh_plugin_mentions(); @@ -2400,6 +2404,9 @@ impl ChatWidget { if self.has_queued_follow_up_messages() { return; } + if !self.thread_jobs.is_empty() { + return; + } if self.active_mode_kind() != ModeKind::Plan { return; } @@ -4763,6 +4770,7 @@ impl ChatWidget { thread_id: None, thread_name: None, forked_from: None, + thread_jobs: Vec::new(), queued_user_messages: VecDeque::new(), rejected_steers_queue: VecDeque::new(), pending_steers: VecDeque::new(), @@ -4821,6 +4829,7 @@ impl ChatWidget { .bottom_pane .set_collaboration_modes_enabled(/*enabled*/ true); widget.sync_fast_command_enabled(); + widget.sync_job_scheduler_command_enabled(); widget.sync_personality_command_enabled(); widget.sync_plugins_command_enabled(); widget @@ -5147,6 +5156,14 @@ impl ChatWidget { SlashCommand::Review => { self.open_review_popup(); } + SlashCommand::Loop => { + let Some(thread_id) = self.thread_id else { + self.add_error_message("No active thread is available.".to_string()); + return; + }; + self.app_event_tx + .send(AppEvent::OpenThreadJobs { thread_id }); + } SlashCommand::Rename => { self.session_telemetry .counter("codex.thread.rename", /*inc*/ 1, &[]); @@ -5537,6 +5554,27 @@ impl ChatWidget { })); self.bottom_pane.drain_pending_submission_state(); } + SlashCommand::Loop if !trimmed.is_empty() => { + let Some(thread_id) = self.thread_id else { + self.add_error_message("No active thread is available.".to_string()); + return; + }; + let Some((prepared_args, _prepared_elements)) = self + .bottom_pane + .prepare_inline_args_submission(/*record_history*/ false) + else { + return; + }; + self.add_info_message( + format!("Scheduling `/loop {prepared_args}`..."), + Some("Parsing the spec and creating the thread job.".to_string()), + ); + self.app_event_tx.send(AppEvent::CreateThreadJobFromSpec { + thread_id, + spec: prepared_args, + }); + self.bottom_pane.drain_pending_submission_state(); + } SlashCommand::SandboxReadRoot if !trimmed.is_empty() => { let Some((prepared_args, _prepared_elements)) = self .bottom_pane @@ -6397,6 +6435,10 @@ impl ChatWidget { } } } + ServerNotification::ThreadJobUpdated(notification) => { + self.on_thread_jobs_updated(notification.jobs); + } + ServerNotification::ThreadJobFired(_) => {} ServerNotification::TurnStarted(_) => { self.last_non_retry_error = None; if !matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages)) { @@ -9372,6 +9414,9 @@ impl ChatWidget { if feature == Feature::FastMode { self.sync_fast_command_enabled(); } + if feature == Feature::JobScheduler { + self.sync_job_scheduler_command_enabled(); + } if feature == Feature::Personality { self.sync_personality_command_enabled(); } @@ -9612,6 +9657,11 @@ impl ChatWidget { .set_fast_command_enabled(self.fast_mode_enabled()); } + fn sync_job_scheduler_command_enabled(&mut self) { + self.bottom_pane + .set_job_scheduler_command_enabled(self.config.features.enabled(Feature::JobScheduler)); + } + fn sync_personality_command_enabled(&mut self) { self.bottom_pane .set_personality_command_enabled(self.config.features.enabled(Feature::Personality)); @@ -9933,6 +9983,56 @@ impl ChatWidget { self.request_redraw(); } + pub(crate) fn on_thread_jobs_updated(&mut self, jobs: Vec) { + self.thread_jobs = jobs; + } + + pub(crate) fn open_thread_jobs_popup(&mut self, thread_id: ThreadId, jobs: Vec) { + self.thread_jobs = jobs.clone(); + if jobs.is_empty() { + self.add_info_message( + "No thread jobs are currently scheduled.".to_string(), + Some("Use `/loop ` to create one.".to_string()), + ); + return; + } + + let items = jobs + .into_iter() + .map(|job| { + let job_id = job.id.clone(); + let actions: Vec = vec![Box::new(move |tx| { + tx.send(AppEvent::DeleteThreadJob { + thread_id, + id: job_id.clone(), + }); + })]; + let name = if job.run_once { + format!("{} • one-shot", job.cron_expression) + } else { + job.cron_expression.clone() + }; + SelectionItem { + name, + description: Some(job.prompt), + selected_description: Some("Delete this job.".to_string()), + is_current: false, + actions, + dismiss_on_select: true, + ..Default::default() + } + }) + .collect(); + self.bottom_pane.show_selection_view(SelectionViewParams { + title: Some("Thread jobs".to_string()), + subtitle: Some("Select a job to delete it.".to_string()), + footer_hint: Some(standard_popup_hint_line()), + items, + ..Default::default() + }); + self.request_redraw(); + } + pub(crate) fn add_plain_history_lines(&mut self, lines: Vec>) { self.add_boxed_history(Box::new(PlainHistoryCell::new(lines))); self.request_redraw(); diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__approvals_selection_popup.snap.new b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__approvals_selection_popup.snap.new deleted file mode 100644 index 7ac97d97c5..0000000000 --- a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__approvals_selection_popup.snap.new +++ /dev/null @@ -1,18 +0,0 @@ ---- -source: tui/src/chatwidget/tests.rs -assertion_line: 9607 -expression: popup ---- - Update Model Permissions - -› 1. Default Codex can read and edit files in the current - workspace, and run commands. Approval is required to - access the internet or edit other files. - 2. Guardian Approvals Same workspace-write permissions as Default, but - eligible `on-request` approvals are routed through - the guardian reviewer subagent. - 3. Full Access Codex can edit files outside this workspace and - access the internet without asking for approval. - Exercise caution when using. - - Press enter to confirm or esc to go back diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__user_prompt_submit_app_server_hook_notifications_render_snapshot.snap.new b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__user_prompt_submit_app_server_hook_notifications_render_snapshot.snap.new deleted file mode 100644 index c7e2394230..0000000000 --- a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__user_prompt_submit_app_server_hook_notifications_render_snapshot.snap.new +++ /dev/null @@ -1,10 +0,0 @@ ---- -source: tui/src/chatwidget/tests.rs -assertion_line: 12789 -expression: combined ---- -• Running UserPromptSubmit hook: checking go-workflow input policy - -UserPromptSubmit hook (stopped) - warning: go-workflow must start from PlanMode - stop: prompt blocked diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index ebb92909e8..cfe81da627 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -200,14 +200,14 @@ use tokio::sync::mpsc::unbounded_channel; use toml::Value as TomlValue; async fn test_config() -> Config { - // Use base defaults to avoid depending on host state. - let codex_home = std::env::temp_dir(); - ConfigBuilder::default() - .codex_home(codex_home.clone()) - .fallback_cwd(Some(PathBuf::from(test_path_display("/tmp/project")))) - .build() - .await - .expect("config") + // Start from built-in defaults so test feature state does not inherit + // host config layers or managed requirements from the machine. + let mut config = Config::load_default_with_cli_overrides(Vec::new()).expect("config"); + config.codex_home = std::env::temp_dir(); + config.cwd = + AbsolutePathBuf::try_from(PathBuf::from(test_path_display("/tmp/project"))).expect("cwd"); + config.config_layer_stack = codex_core::config_loader::ConfigLayerStack::default(); + config } fn test_project_path() -> PathBuf { @@ -2123,6 +2123,7 @@ async fn make_chatwidget_manual( pending_turn_copyable_output: None, running_commands: HashMap::new(), collab_agent_metadata: HashMap::new(), + thread_jobs: Vec::new(), pending_collab_spawn_requests: HashMap::new(), suppressed_exec_calls: HashSet::new(), skills_all: Vec::new(), @@ -9741,6 +9742,7 @@ async fn approvals_selection_popup_snapshot() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; chat.config.notices.hide_full_access_warning = None; + chat.set_feature_enabled(Feature::GuardianApproval, /*enabled*/ false); chat.open_approvals_popup(); let popup = render_bottom_popup(&chat, /*width*/ 80); @@ -10409,6 +10411,7 @@ async fn permissions_selection_history_snapshot_after_mode_switch() { chat.set_windows_sandbox_mode(Some(WindowsSandboxModeToml::Unelevated)); } chat.config.notices.hide_full_access_warning = Some(true); + chat.set_feature_enabled(Feature::GuardianApproval, /*enabled*/ false); chat.open_permissions_popup(); chat.handle_key_event(KeyEvent::from(KeyCode::Down)); @@ -10509,6 +10512,7 @@ async fn permissions_selection_hides_guardian_approvals_when_feature_disabled() chat.set_windows_sandbox_mode(Some(WindowsSandboxModeToml::Unelevated)); } chat.config.notices.hide_full_access_warning = Some(true); + chat.set_feature_enabled(Feature::GuardianApproval, /*enabled*/ false); chat.open_permissions_popup(); let popup = render_bottom_popup(&chat, /*width*/ 120); @@ -10529,6 +10533,7 @@ async fn permissions_selection_hides_guardian_approvals_when_feature_disabled_ev chat.set_windows_sandbox_mode(Some(WindowsSandboxModeToml::Unelevated)); } chat.config.notices.hide_full_access_warning = Some(true); + chat.set_feature_enabled(Feature::GuardianApproval, /*enabled*/ false); chat.config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent; chat.config .permissions @@ -10765,6 +10770,7 @@ async fn permissions_full_access_history_cell_emitted_only_after_confirmation() chat.set_windows_sandbox_mode(Some(WindowsSandboxModeToml::Unelevated)); } chat.config.notices.hide_full_access_warning = None; + chat.set_feature_enabled(Feature::GuardianApproval, /*enabled*/ false); chat.open_permissions_popup(); chat.handle_key_event(KeyEvent::from(KeyCode::Down)); diff --git a/codex-rs/tui/src/job_scheduler.rs b/codex-rs/tui/src/job_scheduler.rs new file mode 100644 index 0000000000..40f05b0546 --- /dev/null +++ b/codex-rs/tui/src/job_scheduler.rs @@ -0,0 +1,158 @@ +use crate::AppServerTarget; +use crate::start_app_server_for_picker; +use codex_app_server_client::AppServerEvent; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ThreadItem; +use codex_core::config::Config; +use codex_protocol::user_input::UserInput; +use serde::Deserialize; +use serde_json::json; +use tracing::warn; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ParsedJobSpec { + pub(crate) cron_expression: String, + pub(crate) prompt: String, + pub(crate) run_once: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ParsedJobSpecResponse { + cron_expression: String, + prompt: String, + run_once: Option, +} + +impl ParsedJobSpecResponse { + fn into_parsed(self) -> std::result::Result { + let cron_expression = self.cron_expression.trim().to_string(); + let prompt = self.prompt.trim().to_string(); + if cron_expression.is_empty() { + return Err("Could not determine a supported schedule from /loop input.".to_string()); + } + if prompt.is_empty() { + return Err("Could not determine a prompt from /loop input.".to_string()); + } + Ok(ParsedJobSpec { + cron_expression, + prompt, + run_once: self.run_once, + }) + } +} + +pub(crate) async fn parse_job_spec( + config: Config, + target: AppServerTarget, + spec: String, +) -> std::result::Result { + let mut app_server = start_app_server_for_picker(&config, &target) + .await + .map_err(|err| format!("Failed to start parser session: {err}"))?; + let result = async { + let thread = app_server + .start_ephemeral_thread(&config) + .await + .map_err(|err| format!("Failed to start parser thread: {err}"))?; + + let output_schema = json!({ + "type": "object", + "properties": { + "cronExpression": { "type": "string" }, + "prompt": { "type": "string" }, + "runOnce": { "type": ["boolean", "null"] } + }, + "required": ["cronExpression", "prompt", "runOnce"], + "additionalProperties": false + }); + + let parser_prompt = format!( + "{}\n{}", + concat!( + "Convert the user's /loop spec into JSON with keys cronExpression, prompt, and runOnce.\n", + "Only emit valid JSON matching the schema.\n", + "Supported cronExpression values for this client are:\n", + "- @after-turn\n", + "- @every s\n", + "- @every m\n", + "- @every h\n", + "- @every d\n", + "Use @after-turn when no explicit interval is present.\n", + "Set runOnce to true only when the user clearly asked for a one-shot run.\n", + "Keep stop conditions and job intent inside prompt.\n", + "If the schedule request is unsupported by the allowed cronExpression formats, return an empty cronExpression.\n", + "User spec:" + ), + spec + ); + + app_server + .turn_start( + codex_protocol::ThreadId::from_string(&thread.id) + .map_err(|err| format!("Parser thread id was invalid: {err}"))?, + vec![UserInput::Text { + text: parser_prompt, + text_elements: Vec::new(), + }], + config.cwd.to_path_buf(), + config.permissions.approval_policy.value(), + config.approvals_reviewer, + config.permissions.sandbox_policy.get().clone(), + config.model.clone().unwrap_or_default(), + /*effort*/ None, + /*summary*/ None, + config.service_tier.map(Some), + /*collaboration_mode*/ None, + config.personality, + Some(output_schema), + ) + .await + .map_err(|err| format!("Failed to run /loop parser prompt: {err}"))?; + + let mut final_message = None; + while let Some(event) = app_server.next_event().await { + match event { + AppServerEvent::ServerNotification(ServerNotification::ItemCompleted( + notification, + )) => { + if let ThreadItem::AgentMessage { text, .. } = notification.item { + final_message = Some(text); + } + } + AppServerEvent::ServerNotification(ServerNotification::TurnCompleted( + notification, + )) => { + if notification.turn.error.is_some() { + return Err("The /loop parser turn failed.".to_string()); + } + break; + } + AppServerEvent::Disconnected { message } => { + return Err(format!("The /loop parser session disconnected: {message}")); + } + AppServerEvent::Lagged { .. } + | AppServerEvent::ServerNotification(_) + | AppServerEvent::ServerRequest(_) => {} + } + } + + let response = + final_message.ok_or_else(|| "The /loop parser returned no output.".to_string())?; + let parsed: ParsedJobSpecResponse = serde_json::from_str(&response) + .map_err(|err| format!("Invalid /loop parser JSON: {err}"))?; + parsed.into_parsed() + } + .await; + + if let Err(err) = app_server.shutdown().await { + warn!(%err, "failed to shut down temporary /loop parser session"); + } + + result +} + +pub(crate) fn format_job_summary(cron_expression: &str, run_once: bool, prompt: &str) -> String { + let mode = if run_once { "one-shot" } else { "recurring" }; + format!("{cron_expression} ({mode}) -> {prompt}") +} diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 02b743fa8a..adc4f100e9 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -113,6 +113,7 @@ mod frames; mod get_git_diff; mod history_cell; pub mod insert_history; +mod job_scheduler; mod key_hint; mod line_truncation; pub mod live_wrap; diff --git a/codex-rs/tui/src/slash_command.rs b/codex-rs/tui/src/slash_command.rs index ec624d3fb9..7dd81e9dcd 100644 --- a/codex-rs/tui/src/slash_command.rs +++ b/codex-rs/tui/src/slash_command.rs @@ -23,6 +23,7 @@ pub enum SlashCommand { Experimental, Skills, Review, + Loop, Rename, New, Resume, @@ -75,6 +76,7 @@ impl SlashCommand { SlashCommand::Init => "create an AGENTS.md file with instructions for Codex", SlashCommand::Compact => "summarize conversation to prevent hitting the context limit", SlashCommand::Review => "review my current changes and find issues", + SlashCommand::Loop => "create or list thread jobs", SlashCommand::Rename => "rename the current thread", SlashCommand::Resume => "resume a saved chat", SlashCommand::Clear => "clear the terminal and start a new chat", @@ -130,6 +132,7 @@ impl SlashCommand { self, SlashCommand::Review | SlashCommand::Rename + | SlashCommand::Loop | SlashCommand::Plan | SlashCommand::Fast | SlashCommand::SandboxReadRoot @@ -162,6 +165,7 @@ impl SlashCommand { SlashCommand::Diff | SlashCommand::Copy | SlashCommand::Rename + | SlashCommand::Loop | SlashCommand::Mention | SlashCommand::Skills | SlashCommand::Status diff --git a/job-scheduler-plan.md b/job-scheduler-plan.md new file mode 100644 index 0000000000..d640a100d6 --- /dev/null +++ b/job-scheduler-plan.md @@ -0,0 +1,127 @@ +# `/loop` Backed by Thread Jobs + +## Summary + +Implement `/loop` as a TUI command that creates and manages per-thread jobs. + +Defaults locked in: + +- `/loop ` creates a job. +- Bare `/loop` lists jobs for the current thread and offers delete actions. +- The entire feature is gated behind the experimental feature flag `job_scheduler`. +- Natural-language parsing happens in the caller, not in core or app-server. +- `JobCreate` accepts structured inputs: `cron_expression`, `prompt`, and optional `run_once`. +- `thread/job/create` also accepts the optional `run_once` field, exposed on the v2 wire as `runOnce`. +- Each thread has a hard limit of 256 active jobs to prevent resource exhaustion. +- Jobs are runtime-only and do not persist across restart, resume, or fork. +- Conditions like “until tests pass” stay inside `prompt`; the model stops the job by calling `JobDelete`. + +## Key Changes + +- Add a job scheduler module in core, preferably `codex-rs/core/src/jobs.rs`, to own: + - `Job { id, cron_expression, prompt, run_once, created_at, next_run_at, last_run_at, pending_run }` + - due-run tracking, idle-thread dispatch, and `next_run_at` computation +- Gate all core, app-server, and TUI job functionality behind `Feature::JobScheduler` / the experimental feature key `job_scheduler`. +- Add three built-in tools in core: + - `JobCreate { cron_expression: String, prompt: String, run_once: Option }` + - `JobDelete { id: String }` + - `JobList {}` +- Tool semantics: + - `JobCreate` validates the structured job and registers it on the current thread. + - `JobCreate` rejects creation when the thread already has 256 active jobs. + - `JobDelete` removes one job by id on the current thread. + - `JobList` returns all jobs on the current thread. + - `run_once=true` means the job is deleted immediately after its execution is claimed, removed from `JobList` results, and the thread emits `thread/job/updated` with the new list. +- Job execution behavior in core: + - due jobs run only when the owning thread is idle + - if a job becomes due while busy, keep at most one pending firing + - manual user turns and queued user prompts beat job runs + - job-triggered turns inherit the current thread context, including plan mode, sandbox, approvals, and model settings + - interrupts stop the current turn but do not implicitly delete jobs +- Job-triggered prompt execution stays off-transcript as a normal user-authored message. Clients should show explicit runtime/info events instead. +- Add app-server v2 RPCs in `codex-rs/app-server-protocol/src/protocol/v2.rs`: + - `thread/job/create { threadId, cronExpression, prompt, runOnce? }` + - `thread/job/delete { threadId, id }` + - `thread/job/list { threadId }` + - `thread/job/updated` notification with `threadId` and the full current list + - when `job_scheduler` is disabled, job RPCs fail with a clear unsupported-feature error instead of silently no-oping + - `thread/job/create` fails with a clear invalid-request or limit-exceeded error when the thread already has 256 active jobs + +## Natural-Language Parsing + +- The caller converts `/loop ` into structured job params by issuing a small model parse request before calling `thread/job/create`. +- Clients should use the existing app-server side-prompt path for that parse request rather than inventing a separate parser-specific backend surface. +- Define a shared parse contract for clients: + - input: raw `spec: String` + - output: `{ cronExpression: String, prompt: String, runOnce?: bool }` +- Parsing prompt should be minimal and deterministic: + - extract only schedule timing into `cronExpression` + - keep all stop conditions and task intent inside `prompt` + - set `runOnce=true` only when the spec clearly describes a one-shot run + - if no explicit wall-clock schedule is present, return the special immediate-follow-up expression used by the scheduler + - return structured JSON only, no prose +- TUI uses this parse step internally for `/loop `. +- Other clients can reuse the same prompt/contract, but core and app-server only validate the structured fields they receive. +- Parse failures or invalid structured output fail closed: do not create a job. + +## Job Identity And Subagents + +- Every job-triggered turn carries `currentJobId` in hidden turn context. The model uses that id when it decides the job should stop and calls `JobDelete(currentJobId)`. +- `currentJobId` is present only for turns fired by a job on that same thread. It must not be written into visible prompt text. +- Jobs are strictly thread-local: + - parent and child threads have separate job lists + - spawning, resuming, or forking a child does not copy the parent’s jobs + - a child spawned from a parent job turn does not inherit the parent’s `currentJobId` +- Built-in job tools are available on any thread, including subagents, but always operate on that thread’s own jobs only. No cross-thread job mutation is allowed in v1. +- If a parent job spawns a child and the child concludes the parent’s stop condition is met, the child must report that back to the parent thread; it cannot directly delete the parent’s job. +- Closing a child thread implicitly drops that child’s runtime jobs with the thread. +- Multi-agent clients must treat `thread/job/updated` and job-fire events as thread-scoped so background child-thread jobs are not mistaken for parent-thread activity. + +## TUI Behavior + +- Add `SlashCommand::Loop` in `codex-rs/tui/src/slash_command.rs`. +- `/loop `: + - runs the parse prompt + - maps the structured result into `thread/job/create` + - shows a confirmation/info event with the normalized job +- Bare `/loop` lists current jobs for the active thread and supports deletion. +- While a turn is running, `/loop ` still works so the next job can be armed immediately. +- In multi-agent views, job events and lists should remain attached to the owning thread so users can tell which agent armed or fired a job. +- TUI only shows `/loop` when `job_scheduler` is enabled for the current session/config. + +## Test Plan + +- Core: + - create, list, and delete multiple jobs on one thread + - reject the 257th active job on a thread + - validate `cron_expression` and `run_once` + - ensure due jobs fire once when idle and do not backlog endlessly while busy + - ensure a `run_once` job is deleted after execution and disappears from subsequent list results + - ensure job-triggered turns include hidden `currentJobId` + - ensure non-job turns do not include `currentJobId` + - ensure parent `currentJobId` is not inherited by spawned child threads + - ensure child-thread jobs are isolated from parent-thread jobs + - ensure closing a child thread drops its runtime jobs + - ensure restart, resume, and fork clear runtime jobs +- App-server: + - create/list/delete RPC round-trips, including omitted and explicit `runOnce` + - `thread/job/create` returns the expected error once the thread reaches 256 active jobs + - `thread/job/updated` emits the full current job set after creates, deletes, and run-once deletion after execution + - thread-scoped notifications stay associated with the correct parent or child thread + - disabled-feature behavior rejects job RPCs cleanly +- TUI: + - `/loop ` issues the parse request and converts the structured result into the expected create RPC + - parser snapshots cover recurring, immediate-follow-up, and one-shot examples + - bare `/loop` lists jobs for the active thread only + - delete actions call the delete RPC on the active thread + - job-fire info events render with the owning thread identity in multi-agent scenarios + - `/loop` is hidden or rejected when `job_scheduler` is disabled + +## Assumptions + +- `job` is the public product and API term; the underlying module/file can still be named `scheduled.rs` if that keeps implementation boundaries cleaner. +- Core and app-server do not parse freeform schedule text. +- The parser uses a model call rather than handwritten NLP rules. +- The scheduler supports a special non-cron expression for immediate post-turn execution, because plain cron is a bad fit for “run again right after this turn.” +- Cron expressions are evaluated in the local timezone of the process hosting the thread unless the existing scheduler infrastructure already defines a better thread-level timezone source. +- The experimental gate is named exactly `job_scheduler`.