From c6465c1ec203d36e697227b8afa63d0283a6613d Mon Sep 17 00:00:00 2001 From: Ruslan Nigmatullin Date: Tue, 28 Apr 2026 16:52:14 -0700 Subject: [PATCH] app-server: notify clients of remote-control status changes (#19919) ## Why Remote-control app-server enrollments have both an internal server id and the environment id exposed to remote-control clients. App-server clients need one current status snapshot that says whether remote control is usable and which environment id, if any, is exposed. A temporary websocket disconnect is not itself an identity change. Account changes, stale enrollment invalidation, successful re-enrollment, and missing ChatGPT auth are meaningful status changes. Disabled remote control remains `disabled` regardless of auth or SQLite state. SQLite startup failure disablement and enrollment persistence failures are handled in #20068; this PR reports the resulting effective status to clients. ## What changed - Adds v2 `remoteControl/status/changed` carrying `state` and `environmentId`. - Adds `RemoteControlConnectionState` values: `disabled`, `connecting`, `connected`, and `errored`. - Exposes remote-control status updates through `RemoteControlHandle` using a Tokio watch channel. - Always sends the current remote-control status snapshot to newly initialized app-server clients. - Broadcasts status changes to initialized app-server clients when state or environment id changes. - Treats missing ChatGPT auth as an `errored` status while leaving it retryable because auth can change at runtime. - Clears `environmentId` when enrollment is cleared for account changes, auth loss, stale backend invalidation, or disabled remote control. - Updates app-server protocol schema fixtures, generated TypeScript, app-server README, remote-control tests, and TUI exhaustive notification matches. ## Stack - Builds on #20068. ## Verification - `just write-app-server-schema` - `cargo test -p codex-app-server-protocol` - `cargo test -p codex-app-server transport::remote_control --lib` - `cargo check -p codex-tui` - `just fix -p codex-app-server-protocol` - `just fix -p codex-app-server` - `just fix -p codex-tui` --- .../schema/json/ServerNotification.json | 47 +++ .../codex_app_server_protocol.schemas.json | 49 +++ .../codex_app_server_protocol.v2.schemas.json | 49 +++ ...emoteControlStatusChangedNotification.json | 31 ++ .../schema/typescript/ServerNotification.ts | 3 +- .../v2/RemoteControlConnectionStatus.ts | 5 + .../RemoteControlStatusChangedNotification.ts | 9 + .../schema/typescript/v2/index.ts | 2 + .../src/protocol/common.rs | 1 + .../app-server-protocol/src/protocol/v2.rs | 19 + codex-rs/app-server/README.md | 1 + codex-rs/app-server/src/lib.rs | 33 +- .../src/transport/remote_control/mod.rs | 27 +- .../src/transport/remote_control/tests.rs | 148 +++++++- .../src/transport/remote_control/websocket.rs | 352 ++++++++++++++++-- codex-rs/tui/src/app/app_server_adapter.rs | 1 + codex-rs/tui/src/chatwidget.rs | 1 + 17 files changed, 743 insertions(+), 35 deletions(-) create mode 100644 codex-rs/app-server-protocol/schema/json/v2/RemoteControlStatusChangedNotification.json create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlConnectionStatus.ts create mode 100644 codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlStatusChangedNotification.ts diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index c94559c11..5bbd51bb9 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -2603,6 +2603,33 @@ ], "type": "object" }, + "RemoteControlConnectionStatus": { + "enum": [ + "disabled", + "connecting", + "connected", + "errored" + ], + "type": "string" + }, + "RemoteControlStatusChangedNotification": { + "description": "Current remote-control connection status and environment id exposed to clients.", + "properties": { + "environmentId": { + "type": [ + "string", + "null" + ] + }, + "status": { + "$ref": "#/definitions/RemoteControlConnectionStatus" + } + }, + "required": [ + "status" + ], + "type": "object" + }, "RequestId": { "anyOf": [ { @@ -5342,6 +5369,26 @@ "title": "App/list/updatedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "remoteControl/status/changed" + ], + "title": "RemoteControl/status/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/RemoteControlStatusChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "RemoteControl/status/changedNotification", + "type": "object" + }, { "properties": { "method": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 2f94b072c..4b6d42920 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 @@ -4348,6 +4348,26 @@ "title": "App/list/updatedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "remoteControl/status/changed" + ], + "title": "RemoteControl/status/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/RemoteControlStatusChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "RemoteControl/status/changedNotification", + "type": "object" + }, { "properties": { "method": { @@ -12499,6 +12519,35 @@ ], "type": "string" }, + "RemoteControlConnectionStatus": { + "enum": [ + "disabled", + "connecting", + "connected", + "errored" + ], + "type": "string" + }, + "RemoteControlStatusChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "Current remote-control connection status and environment id exposed to clients.", + "properties": { + "environmentId": { + "type": [ + "string", + "null" + ] + }, + "status": { + "$ref": "#/definitions/v2/RemoteControlConnectionStatus" + } + }, + "required": [ + "status" + ], + "title": "RemoteControlStatusChangedNotification", + "type": "object" + }, "RequestId": { "anyOf": [ { 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 88f81ad46..06a228f76 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 @@ -9173,6 +9173,35 @@ ], "type": "string" }, + "RemoteControlConnectionStatus": { + "enum": [ + "disabled", + "connecting", + "connected", + "errored" + ], + "type": "string" + }, + "RemoteControlStatusChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "description": "Current remote-control connection status and environment id exposed to clients.", + "properties": { + "environmentId": { + "type": [ + "string", + "null" + ] + }, + "status": { + "$ref": "#/definitions/RemoteControlConnectionStatus" + } + }, + "required": [ + "status" + ], + "title": "RemoteControlStatusChangedNotification", + "type": "object" + }, "RequestId": { "anyOf": [ { @@ -10909,6 +10938,26 @@ "title": "App/list/updatedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "remoteControl/status/changed" + ], + "title": "RemoteControl/status/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/RemoteControlStatusChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "RemoteControl/status/changedNotification", + "type": "object" + }, { "properties": { "method": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/RemoteControlStatusChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/RemoteControlStatusChangedNotification.json new file mode 100644 index 000000000..8286815ff --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/RemoteControlStatusChangedNotification.json @@ -0,0 +1,31 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "RemoteControlConnectionStatus": { + "enum": [ + "disabled", + "connecting", + "connected", + "errored" + ], + "type": "string" + } + }, + "description": "Current remote-control connection status and environment id exposed to clients.", + "properties": { + "environmentId": { + "type": [ + "string", + "null" + ] + }, + "status": { + "$ref": "#/definitions/RemoteControlConnectionStatus" + } + }, + "required": [ + "status" + ], + "title": "RemoteControlStatusChangedNotification", + "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 41d4754bc..c32618c09 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -35,6 +35,7 @@ import type { RawResponseItemCompletedNotification } from "./v2/RawResponseItemC import type { ReasoningSummaryPartAddedNotification } from "./v2/ReasoningSummaryPartAddedNotification"; import type { ReasoningSummaryTextDeltaNotification } from "./v2/ReasoningSummaryTextDeltaNotification"; import type { ReasoningTextDeltaNotification } from "./v2/ReasoningTextDeltaNotification"; +import type { RemoteControlStatusChangedNotification } from "./v2/RemoteControlStatusChangedNotification"; import type { ServerRequestResolvedNotification } from "./v2/ServerRequestResolvedNotification"; import type { SkillsChangedNotification } from "./v2/SkillsChangedNotification"; import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNotification"; @@ -66,4 +67,4 @@ import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldW /** * Notification sent from the server to the client. */ -export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/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": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; +export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/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": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlConnectionStatus.ts b/codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlConnectionStatus.ts new file mode 100644 index 000000000..3e6197f5b --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlConnectionStatus.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 RemoteControlConnectionStatus = "disabled" | "connecting" | "connected" | "errored"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlStatusChangedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlStatusChangedNotification.ts new file mode 100644 index 000000000..16a913855 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/RemoteControlStatusChangedNotification.ts @@ -0,0 +1,9 @@ +// GENERATED CODE! DO NOT MODIFY BY HAND! + +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { RemoteControlConnectionStatus } from "./RemoteControlConnectionStatus"; + +/** + * Current remote-control connection status and environment id exposed to clients. + */ +export type RemoteControlStatusChangedNotification = { status: RemoteControlConnectionStatus, environmentId: string | null, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 3c0178c29..5753272da 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -282,6 +282,8 @@ export type { ReasoningSummaryTextDeltaNotification } from "./ReasoningSummaryTe export type { ReasoningTextDeltaNotification } from "./ReasoningTextDeltaNotification"; export type { RemoteControlClientConnectionAudience } from "./RemoteControlClientConnectionAudience"; export type { RemoteControlClientEnrollmentAudience } from "./RemoteControlClientEnrollmentAudience"; +export type { RemoteControlConnectionStatus } from "./RemoteControlConnectionStatus"; +export type { RemoteControlStatusChangedNotification } from "./RemoteControlStatusChangedNotification"; export type { RequestPermissionProfile } from "./RequestPermissionProfile"; export type { ResidencyRequirement } from "./ResidencyRequirement"; export type { ReviewDelivery } from "./ReviewDelivery"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 99f4d7d3e..5d506d05b 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1258,6 +1258,7 @@ server_notification_definitions! { AccountUpdated => "account/updated" (v2::AccountUpdatedNotification), AccountRateLimitsUpdated => "account/rateLimits/updated" (v2::AccountRateLimitsUpdatedNotification), AppListUpdated => "app/list/updated" (v2::AppListUpdatedNotification), + RemoteControlStatusChanged => "remoteControl/status/changed" (v2::RemoteControlStatusChangedNotification), ExternalAgentConfigImportCompleted => "externalAgentConfig/import/completed" (v2::ExternalAgentConfigImportCompletedNotification), FsChanged => "fs/changed" (v2::FsChangedNotification), ReasoningSummaryTextDelta => "item/reasoning/summaryTextDelta" (v2::ReasoningSummaryTextDeltaNotification), diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index ccefe15a3..c5a300943 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -2878,6 +2878,25 @@ pub struct DeviceKeyPublicResponse { pub protection_class: DeviceKeyProtectionClass, } +/// Current remote-control connection status and environment id exposed to clients. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct RemoteControlStatusChangedNotification { + pub status: RemoteControlConnectionStatus, + pub environment_id: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(rename_all = "camelCase", export_to = "v2/")] +pub enum RemoteControlConnectionStatus { + Disabled, + Connecting, + Connected, + Errored, +} + /// Audience for a remote-control client connection device-key proof. #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "snake_case")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 248f5d415..ac1574b16 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -205,6 +205,7 @@ Example with notification opt-out: - `device/key/create` — create or load a controller-local device signing key for an account/client binding. This local-key API is available only over local transports such as stdio and in-process; remote transports reject it. Hardware-backed providers are the target protection class; an OS-protected non-extractable fallback is allowed only with `protectionPolicy: "allow_os_protected_nonextractable"` and returns the reported `protectionClass`. - `device/key/public` — return a device key's SPKI DER public key as base64 plus its `algorithm` and `protectionClass`. - `device/key/sign` — sign one of the accepted structured payload variants with a controller-local device key. The only accepted payload today is `remoteControlClientConnection`, which binds a server-issued `/client` websocket challenge to the enrolled controller device without signing the bearer token itself; this is intentionally not an arbitrary-byte signing API. +- `remoteControl/status/changed` — notification emitted when the remote-control status or client-visible environment id changes. `status` is one of `disabled`, `connecting`, `connected`, or `errored`; `environmentId` is a string when the app-server has a current enrollment and `null` when that enrollment is cleared, invalidated, or remote control is disabled. Newly initialized app-server clients always receive the current status snapshot. - `skills/config/write` — write user-level skill config by name or absolute path. - `plugin/install` — install a plugin from a discovered marketplace entry, rejecting marketplace entries marked unavailable for install, install MCPs if any, and return the effective plugin auth policy plus any apps that still need auth (**under development; do not call from production clients yet**). - `plugin/uninstall` — uninstall a local plugin by `pluginId` in `@` form by removing its cached files and clearing its user-level config entry, or uninstall a remote ChatGPT plugin by backend `pluginId` by forwarding the uninstall to the ChatGPT plugin backend and removing any downloaded remote-plugin cache (**under development; do not call from production clients yet**). diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index beacb3251..b80510b0f 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -40,6 +40,8 @@ use codex_analytics::AppServerRpcTransport; use codex_app_server_protocol::ConfigLayerSource; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::RemoteControlStatusChangedNotification; +use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::TextPosition as AppTextPosition; use codex_app_server_protocol::TextRange as AppTextRange; use codex_config::ConfigLoadError; @@ -724,6 +726,7 @@ pub async fn run_main_with_transport_options( let processor_handle = tokio::spawn({ let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(outgoing_tx)); + let initialize_notification_sender = outgoing_message_sender.clone(); let outbound_control_tx = outbound_control_tx; let auth_manager = AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await; @@ -739,12 +742,14 @@ pub async fn run_main_with_transport_options( session_source, auth_manager, rpc_transport: analytics_rpc_transport(&transport), - remote_control_handle: Some(remote_control_handle), + remote_control_handle: Some(remote_control_handle.clone()), plugin_startup_tasks: runtime_options.plugin_startup_tasks, })); let mut thread_created_rx = processor.thread_created_receiver(); let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count(); let mut connections = HashMap::::new(); + let mut remote_control_status_rx = remote_control_handle.status_receiver(); + let mut remote_control_status = remote_control_status_rx.borrow().clone(); let transport_shutdown_token = transport_shutdown_token.clone(); async move { let mut listen_for_threads = true; @@ -884,6 +889,14 @@ pub async fn run_main_with_transport_options( connection_id, ) .await; + initialize_notification_sender + .send_server_notification_to_connections( + &[connection_id], + ServerNotification::RemoteControlStatusChanged( + remote_control_status.clone(), + ), + ) + .await; processor.connection_initialized(connection_id).await; connection_state .outbound_initialized @@ -915,6 +928,24 @@ pub async fn run_main_with_transport_options( } } } + changed = remote_control_status_rx.changed() => { + if changed.is_err() { + continue; + } + let status = remote_control_status_rx.borrow().clone(); + if remote_control_status == status { + continue; + } + remote_control_status = status.clone(); + initialize_notification_sender + .send_server_notification(ServerNotification::RemoteControlStatusChanged( + RemoteControlStatusChangedNotification { + status: status.status, + environment_id: status.environment_id, + }, + )) + .await; + } created = thread_created_rx.recv(), if listen_for_threads => { match created { Ok(thread_id) => { diff --git a/codex-rs/app-server/src/transport/remote_control/mod.rs b/codex-rs/app-server/src/transport/remote_control/mod.rs index eca6047ef..ef517e1ae 100644 --- a/codex-rs/app-server/src/transport/remote_control/mod.rs +++ b/codex-rs/app-server/src/transport/remote_control/mod.rs @@ -3,6 +3,8 @@ mod enroll; mod protocol; mod websocket; +use crate::transport::remote_control::websocket::RemoteControlChannels; +use crate::transport::remote_control::websocket::RemoteControlStatusPublisher; use crate::transport::remote_control::websocket::RemoteControlWebsocket; pub use self::protocol::ClientId; @@ -12,6 +14,8 @@ use self::protocol::normalize_remote_control_url; use super::CHANNEL_CAPACITY; use super::TransportEvent; use super::next_connection_id; +use codex_app_server_protocol::RemoteControlConnectionStatus; +use codex_app_server_protocol::RemoteControlStatusChangedNotification; use codex_login::AuthManager; use codex_state::StateRuntime; use std::io; @@ -33,6 +37,7 @@ pub(super) struct QueuedServerEnvelope { #[derive(Clone)] pub(crate) struct RemoteControlHandle { enabled_tx: Arc>, + status_tx: Arc>, state_db_available: bool, } @@ -49,6 +54,12 @@ impl RemoteControlHandle { changed }); } + + pub(crate) fn status_receiver( + &self, + ) -> watch::Receiver { + self.status_tx.subscribe() + } } pub(crate) async fn start_remote_control( @@ -73,13 +84,26 @@ pub(crate) async fn start_remote_control( }; let (enabled_tx, enabled_rx) = watch::channel(initial_enabled); + let initial_status = RemoteControlStatusChangedNotification { + status: if initial_enabled { + RemoteControlConnectionStatus::Connecting + } else { + RemoteControlConnectionStatus::Disabled + }, + environment_id: None, + }; + let (status_tx, _status_rx) = watch::channel(initial_status); + let status_publisher = RemoteControlStatusPublisher::new(status_tx.clone()); let join_handle = tokio::spawn(async move { RemoteControlWebsocket::new( remote_control_url, remote_control_target, state_db, auth_manager, - transport_event_tx, + RemoteControlChannels { + transport_event_tx, + status_publisher, + }, shutdown_token, enabled_rx, ) @@ -91,6 +115,7 @@ pub(crate) async fn start_remote_control( join_handle, RemoteControlHandle { enabled_tx: Arc::new(enabled_tx), + status_tx: Arc::new(status_tx), state_db_available, }, )) diff --git a/codex-rs/app-server/src/transport/remote_control/tests.rs b/codex-rs/app-server/src/transport/remote_control/tests.rs index 9b0896184..92ac9e431 100644 --- a/codex-rs/app-server/src/transport/remote_control/tests.rs +++ b/codex-rs/app-server/src/transport/remote_control/tests.rs @@ -18,6 +18,8 @@ use base64::Engine; use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::ConfigWarningNotification; use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::RemoteControlConnectionStatus; +use codex_app_server_protocol::RemoteControlStatusChangedNotification; use codex_app_server_protocol::ServerNotification; use codex_config::types::AuthCredentialsStoreMode; use codex_core::test_support::auth_manager_from_auth; @@ -45,6 +47,7 @@ use tokio::net::TcpListener; use tokio::net::TcpStream; use tokio::sync::mpsc; use tokio::sync::oneshot; +use tokio::sync::watch; use tokio::time::Duration; use tokio::time::timeout; use tokio_tungstenite::WebSocketStream; @@ -115,6 +118,50 @@ fn remote_control_url_for_listener(listener: &TcpListener) -> String { format!("http://{addr}/backend-api/") } +async fn expect_remote_control_status( + status_rx: &mut watch::Receiver, + expected_status: Option, + expected_environment_id: Option<&str>, +) { + timeout(Duration::from_secs(5), status_rx.changed()) + .await + .expect("remote control status event should arrive in time") + .expect("remote control status watch should remain open"); + let status = status_rx.borrow(); + if let Some(expected_status) = expected_status { + assert_eq!(status.status, expected_status); + } + assert_eq!(status.environment_id.as_deref(), expected_environment_id); +} + +async fn expect_remote_control_status_snapshot( + status_rx: &mut watch::Receiver, + expected_status: RemoteControlStatusChangedNotification, +) { + if *status_rx.borrow() == expected_status { + return; + } + + let expected_status_for_wait = expected_status.clone(); + let result = timeout(Duration::from_secs(5), async { + loop { + status_rx + .changed() + .await + .expect("remote control status watch should remain open"); + if *status_rx.borrow() == expected_status_for_wait { + return; + } + } + }) + .await; + assert!( + result.is_ok(), + "remote control status snapshot should arrive in time; expected {expected_status:?}, latest {:?}", + status_rx.borrow().clone() + ); +} + #[tokio::test] async fn remote_control_transport_manages_virtual_clients_and_routes_messages() { let listener = TcpListener::bind("127.0.0.1:0") @@ -125,7 +172,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages() let (transport_event_tx, mut transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let shutdown_token = CancellationToken::new(); - let (remote_task, _remote_handle) = start_remote_control( + let (remote_task, remote_handle) = start_remote_control( remote_control_url, Some(remote_control_state_runtime(&codex_home).await), remote_control_auth_manager(), @@ -136,6 +183,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages() ) .await .expect("remote control should start"); + let mut status_rx = remote_handle.status_receiver(); let enroll_request = accept_http_request(&listener).await; assert_eq!( enroll_request.request_line, @@ -147,6 +195,12 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages() ) .await; let mut websocket = accept_remote_control_connection(&listener).await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_test"), + ) + .await; let client_id = ClientId("client-1".to_string()); send_client_event( @@ -394,7 +448,7 @@ async fn remote_control_transport_reconnects_after_disconnect() { let (transport_event_tx, mut transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let shutdown_token = CancellationToken::new(); - let (remote_task, _remote_handle) = start_remote_control( + let (remote_task, remote_handle) = start_remote_control( remote_control_url, Some(remote_control_state_runtime(&codex_home).await), remote_control_auth_manager(), @@ -405,6 +459,7 @@ async fn remote_control_transport_reconnects_after_disconnect() { ) .await .expect("remote control should start"); + let mut status_rx = remote_handle.status_receiver(); let enroll_request = accept_http_request(&listener).await; assert_eq!( @@ -424,6 +479,12 @@ async fn remote_control_transport_reconnects_after_disconnect() { drop(first_websocket); let mut second_websocket = accept_remote_control_connection(&listener).await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_test"), + ) + .await; send_client_event( &mut second_websocket, ClientEnvelope { @@ -526,7 +587,7 @@ async fn remote_control_start_allows_missing_auth_when_enabled() { } #[tokio::test] -async fn remote_control_start_disables_remote_control_without_state_db() { +async fn remote_control_start_reports_missing_state_db_as_disabled_when_enabled() { let listener = TcpListener::bind("127.0.0.1:0") .await .expect("listener should bind"); @@ -545,6 +606,14 @@ async fn remote_control_start_disables_remote_control_without_state_db() { ) .await .expect("remote control should start disabled without sqlite state db"); + let mut status_rx = remote_handle.status_receiver(); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Disabled, + environment_id: None, + } + ); timeout(Duration::from_millis(100), listener.accept()) .await @@ -554,6 +623,9 @@ async fn remote_control_start_disables_remote_control_without_state_db() { timeout(Duration::from_millis(100), listener.accept()) .await .expect_err("remote control should remain disabled without sqlite state db"); + timeout(Duration::from_millis(20), status_rx.changed()) + .await + .expect_err("status should remain disabled without sqlite state db"); shutdown_token.cancel(); timeout(Duration::from_secs(1), remote_task) @@ -583,6 +655,7 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() { ) .await .expect("remote control should start"); + let mut status_rx = remote_handle.status_receiver(); let enroll_request = accept_http_request(&listener).await; assert_eq!( @@ -595,8 +668,24 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() { ) .await; let mut first_websocket = accept_remote_control_connection(&listener).await; + expect_remote_control_status_snapshot( + &mut status_rx, + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connected, + environment_id: Some("env_test".to_string()), + }, + ) + .await; remote_handle.set_enabled(/*enabled*/ false); + expect_remote_control_status_snapshot( + &mut status_rx, + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Disabled, + environment_id: None, + }, + ) + .await; timeout(Duration::from_secs(1), first_websocket.next()) .await .expect("disabling remote control should close the websocket"); @@ -605,7 +694,21 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() { .expect_err("disabled remote control should not reconnect"); remote_handle.set_enabled(/*enabled*/ true); + expect_remote_control_status_snapshot( + &mut status_rx, + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + environment_id: Some("env_test".to_string()), + }, + ) + .await; let mut second_websocket = accept_remote_control_connection(&listener).await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_test"), + ) + .await; second_websocket .close(None) .await @@ -625,7 +728,7 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() { let (transport_event_tx, mut transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let shutdown_token = CancellationToken::new(); - let (remote_task, _remote_handle) = start_remote_control( + let (remote_task, remote_handle) = start_remote_control( remote_control_url, Some(remote_control_state_runtime(&codex_home).await), remote_control_auth_manager(), @@ -636,6 +739,7 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() { ) .await .expect("remote control should start"); + let mut status_rx = remote_handle.status_receiver(); let enroll_request = accept_http_request(&listener).await; respond_with_json( @@ -644,6 +748,12 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() { ) .await; let mut first_websocket = accept_remote_control_connection(&listener).await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_test"), + ) + .await; let client_id = ClientId("client-1".to_string()); let initialize_message = JSONRPCMessage::Request(codex_app_server_protocol::JSONRPCRequest { @@ -793,7 +903,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() { mpsc::channel::(CHANNEL_CAPACITY); let expected_server_name = gethostname().to_string_lossy().trim().to_string(); let shutdown_token = CancellationToken::new(); - let (remote_task, _remote_handle) = start_remote_control( + let (remote_task, remote_handle) = start_remote_control( remote_control_url, Some(remote_control_state_runtime(&codex_home).await), remote_control_auth_manager(), @@ -804,6 +914,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() { ) .await .expect("remote control should start"); + let mut status_rx = remote_handle.status_receiver(); let enroll_request = accept_http_request(&listener).await; assert_eq!( @@ -836,6 +947,12 @@ async fn remote_control_http_mode_enrolls_before_connecting() { let (handshake_request, mut websocket) = accept_remote_control_backend_connection(&listener).await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_test"), + ) + .await; assert_eq!( handshake_request.path, "/backend-api/wham/remote/control/server" @@ -1220,7 +1337,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() let (transport_event_tx, _transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let shutdown_token = CancellationToken::new(); - let (remote_task, _remote_handle) = start_remote_control( + let (remote_task, remote_handle) = start_remote_control( remote_control_url, Some(state_db.clone()), remote_control_auth_manager_with_home(&codex_home), @@ -1231,6 +1348,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() ) .await .expect("remote control should start"); + let mut status_rx = remote_handle.status_receiver(); let websocket_request = accept_http_request(&listener).await; assert_eq!( @@ -1241,7 +1359,19 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() websocket_request.headers.get("x-codex-server-id"), Some(&stale_enrollment.server_id) ); + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_stale"), + ) + .await; respond_with_status(websocket_request.stream, "404 Not Found", "").await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + /*expected_environment_id*/ None, + ) + .await; let enroll_request = accept_http_request(&listener).await; assert_eq!( @@ -1258,6 +1388,12 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() .await; let (handshake_request, _websocket) = accept_remote_control_backend_connection(&listener).await; + expect_remote_control_status( + &mut status_rx, + /*expected_status*/ None, + Some("env_refreshed"), + ) + .await; assert_eq!( handshake_request.headers.get("x-codex-server-id"), Some(&refreshed_enrollment.server_id) diff --git a/codex-rs/app-server/src/transport/remote_control/websocket.rs b/codex-rs/app-server/src/transport/remote_control/websocket.rs index 5e0af24fb..df673079c 100644 --- a/codex-rs/app-server/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server/src/transport/remote_control/websocket.rs @@ -17,6 +17,8 @@ use super::protocol::ServerEnvelope; use super::protocol::StreamId; use axum::http::HeaderValue; use base64::Engine; +use codex_app_server_protocol::RemoteControlConnectionStatus; +use codex_app_server_protocol::RemoteControlStatusChangedNotification; use codex_core::util::backoff; use codex_login::AuthManager; use codex_login::UnauthorizedRecovery; @@ -117,6 +119,7 @@ pub(crate) struct RemoteControlWebsocket { remote_control_target: Option, state_db: Option>, auth_manager: Arc, + status_publisher: RemoteControlStatusPublisher, shutdown_token: CancellationToken, reconnect_attempt: u64, enrollment: Option, @@ -134,20 +137,82 @@ enum ConnectOutcome { Shutdown, } +pub(super) struct RemoteControlChannels { + pub(super) transport_event_tx: mpsc::Sender, + pub(super) status_publisher: RemoteControlStatusPublisher, +} + +#[derive(Clone)] +pub(super) struct RemoteControlStatusPublisher { + tx: watch::Sender, +} + +impl RemoteControlStatusPublisher { + pub(super) fn new(tx: watch::Sender) -> Self { + Self { tx } + } + + fn publish_status(&self, connection_status: RemoteControlConnectionStatus) { + self.tx.send_if_modified(|status| { + let next_status = RemoteControlStatusChangedNotification { + status: connection_status, + environment_id: if connection_status == RemoteControlConnectionStatus::Disabled { + None + } else { + status.environment_id.clone() + }, + }; + if *status == next_status { + return false; + } + + *status = next_status; + true + }); + } + + fn publish_environment_id(&self, environment_id: Option) { + self.tx.send_if_modified(|status| { + if status.status == RemoteControlConnectionStatus::Disabled { + return false; + } + let next_status = RemoteControlStatusChangedNotification { + status: status.status, + environment_id, + }; + if *status == next_status { + return false; + } + + *status = next_status; + true + }); + } +} + +#[derive(Clone, Copy)] +pub(super) struct RemoteControlConnectOptions<'a> { + subscribe_cursor: Option<&'a str>, + app_server_client_name: Option<&'a str>, +} + impl RemoteControlWebsocket { pub(crate) fn new( remote_control_url: String, remote_control_target: Option, state_db: Option>, auth_manager: Arc, - transport_event_tx: mpsc::Sender, + channels: RemoteControlChannels, shutdown_token: CancellationToken, enabled_rx: watch::Receiver, ) -> Self { let shutdown_token = shutdown_token.child_token(); let (server_event_tx, server_event_rx) = mpsc::channel(super::CHANNEL_CAPACITY); - let client_tracker = - ClientTracker::new(server_event_tx, transport_event_tx, &shutdown_token); + let client_tracker = ClientTracker::new( + server_event_tx, + channels.transport_event_tx, + &shutdown_token, + ); let (outbound_buffer, used_rx) = BoundedOutboundBuffer::new(); let auth_recovery = auth_manager.unauthorized_recovery(); @@ -156,6 +221,7 @@ impl RemoteControlWebsocket { remote_control_target, state_db, auth_manager, + status_publisher: channels.status_publisher, shutdown_token, reconnect_attempt: 0, enrollment: None, @@ -202,7 +268,11 @@ impl RemoteControlWebsocket { .await { ConnectOutcome::Connected(websocket_connection) => *websocket_connection, - ConnectOutcome::Disabled => continue, + ConnectOutcome::Disabled => { + self.status_publisher + .publish_status(RemoteControlConnectionStatus::Disabled); + continue; + } ConnectOutcome::Shutdown => break, }; @@ -243,6 +313,8 @@ impl RemoteControlWebsocket { shutdown_token: &CancellationToken, app_server_client_name: Option<&str>, ) -> ConnectOutcome { + self.status_publisher + .publish_status(RemoteControlConnectionStatus::Connecting); let remote_control_target = match self.remote_control_target.as_ref() { Some(remote_control_target) => remote_control_target.clone(), None => match super::protocol::normalize_remote_control_url(&self.remote_control_url) { @@ -251,6 +323,8 @@ impl RemoteControlWebsocket { remote_control_target } Err(err) => { + self.status_publisher + .publish_status(RemoteControlConnectionStatus::Errored); warn!("remote control is enabled but the URL is invalid: {err}"); tokio::select! { _ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown, @@ -267,6 +341,10 @@ impl RemoteControlWebsocket { loop { let subscribe_cursor = self.state.lock().await.subscribe_cursor.clone(); + let connect_options = RemoteControlConnectOptions { + subscribe_cursor: subscribe_cursor.as_deref(), + app_server_client_name, + }; let connect_result = tokio::select! { _ = shutdown_token.cancelled() => return ConnectOutcome::Shutdown, changed = self.enabled_rx.wait_for(|enabled| !*enabled) => { @@ -281,15 +359,20 @@ impl RemoteControlWebsocket { &self.auth_manager, &mut self.auth_recovery, &mut self.enrollment, - subscribe_cursor.as_deref(), - app_server_client_name, + connect_options, + &self.status_publisher, ) => connect_result, }; match connect_result { Ok((websocket_connection, response)) => { + if !*self.enabled_rx.borrow() { + return ConnectOutcome::Disabled; + } self.reconnect_attempt = 0; self.auth_recovery = self.auth_manager.unauthorized_recovery(); + self.status_publisher + .publish_status(RemoteControlConnectionStatus::Connected); info!( "connected to app-server remote control websocket: {}, {}", remote_control_target.websocket_url, @@ -298,9 +381,14 @@ impl RemoteControlWebsocket { return ConnectOutcome::Connected(Box::new(websocket_connection)); } Err(err) => { + if !*self.enabled_rx.borrow() { + return ConnectOutcome::Disabled; + } let reconnect_delay = if err.kind() == ErrorKind::WouldBlock { REMOTE_CONTROL_ACCOUNT_ID_RETRY_INTERVAL } else { + self.status_publisher + .publish_status(RemoteControlConnectionStatus::Errored); warn!( "failed to connect to app-server remote control websocket: {}, err: {}", remote_control_target.websocket_url, err @@ -351,9 +439,15 @@ impl RemoteControlWebsocket { let mut enabled_rx = self.enabled_rx.clone(); tokio::select! { _ = shutdown_token.cancelled() => {} - _ = enabled_rx.wait_for(|enabled| !*enabled) => shutdown_token.cancel(), - _ = join_set.join_next() => shutdown_token.cancel(), - } + changed = enabled_rx.wait_for(|enabled| !*enabled) => { + if changed.is_ok() { + self.status_publisher + .publish_status(RemoteControlConnectionStatus::Disabled); + } + } + _ = join_set.join_next() => {} + }; + shutdown_token.cancel(); join_set.join_all().await; } @@ -745,8 +839,8 @@ pub(super) async fn connect_remote_control_websocket( auth_manager: &Arc, auth_recovery: &mut UnauthorizedRecovery, enrollment: &mut Option, - subscribe_cursor: Option<&str>, - app_server_client_name: Option<&str>, + connect_options: RemoteControlConnectOptions<'_>, + status_publisher: &RemoteControlStatusPublisher, ) -> io::Result<( WebSocketStream>, tungstenite::http::Response<()>, @@ -761,7 +855,16 @@ pub(super) async fn connect_remote_control_websocket( )); }; - let auth = load_remote_control_auth(auth_manager).await?; + let auth = match load_remote_control_auth(auth_manager).await { + Ok(auth) => auth, + Err(err) => { + if err.kind() == ErrorKind::PermissionDenied { + *enrollment = None; + status_publisher.publish_environment_id(/*environment_id*/ None); + } + return Err(err); + } + }; let enrollment_account_id = enrollment.as_ref().map(|enrollment| &enrollment.account_id); if enrollment_account_id.is_some_and(|account_id| account_id != &auth.account_id) { info!( @@ -773,16 +876,25 @@ pub(super) async fn connect_remote_control_websocket( auth.account_id ); *enrollment = None; + status_publisher.publish_environment_id(/*environment_id*/ None); + } + + if let Some(enrollment) = enrollment.as_ref() { + status_publisher.publish_environment_id(Some(enrollment.environment_id.clone())); } if enrollment.is_none() { - *enrollment = load_persisted_remote_control_enrollment( + let loaded_enrollment = load_persisted_remote_control_enrollment( Some(state_db), remote_control_target, &auth.account_id, - app_server_client_name, + connect_options.app_server_client_name, ) .await?; + if let Some(loaded_enrollment) = loaded_enrollment.as_ref() { + status_publisher.publish_environment_id(Some(loaded_enrollment.environment_id.clone())); + } + *enrollment = loaded_enrollment; } if enrollment.is_none() { @@ -807,7 +919,7 @@ pub(super) async fn connect_remote_control_websocket( Some(state_db), remote_control_target, &auth.account_id, - app_server_client_name, + connect_options.app_server_client_name, Some(&new_enrollment), ) .await @@ -823,6 +935,7 @@ pub(super) async fn connect_remote_control_websocket( new_enrollment.server_id, new_enrollment.environment_id ); + status_publisher.publish_environment_id(Some(new_enrollment.environment_id.clone())); *enrollment = Some(new_enrollment); } @@ -833,7 +946,7 @@ pub(super) async fn connect_remote_control_websocket( &remote_control_target.websocket_url, enrollment_ref, &auth, - subscribe_cursor, + connect_options.subscribe_cursor, )?; match connect_async(request).await { @@ -852,7 +965,7 @@ pub(super) async fn connect_remote_control_websocket( Some(state_db), remote_control_target, &auth.account_id, - app_server_client_name, + connect_options.app_server_client_name, /*enrollment*/ None, ) .await @@ -862,6 +975,7 @@ pub(super) async fn connect_remote_control_websocket( ); } *enrollment = None; + status_publisher.publish_environment_id(/*environment_id*/ None); } tungstenite::Error::Http(response) if matches!(response.status().as_u16(), 401 | 403) => @@ -968,6 +1082,17 @@ mod tests { #[cfg(not(windows))] const TEST_HTTP_ACCEPT_TIMEOUT: Duration = Duration::from_secs(5); + fn remote_control_status_channel() -> ( + RemoteControlStatusPublisher, + watch::Receiver, + ) { + let (status_tx, status_rx) = watch::channel(RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + environment_id: None, + }); + (RemoteControlStatusPublisher::new(status_tx), status_rx) + } + async fn remote_control_state_runtime(codex_home: &TempDir) -> Arc { StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()) .await @@ -1059,6 +1184,7 @@ mod tests { server_id: "srv_e_test".to_string(), server_name: "test-server".to_string(), }); + let (status_publisher, status_rx) = remote_control_status_channel(); let err = match connect_remote_control_websocket( &remote_control_target, @@ -1066,8 +1192,11 @@ mod tests { &auth_manager, &mut auth_recovery, &mut enrollment, - /*subscribe_cursor*/ None, - /*app_server_client_name*/ None, + RemoteControlConnectOptions { + subscribe_cursor: None, + app_server_client_name: None, + }, + &status_publisher, ) .await { @@ -1077,6 +1206,13 @@ mod tests { server_task.await.expect("server task should succeed"); assert_eq!(err.to_string(), expected_error); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + environment_id: Some("env_test".to_string()), + } + ); } #[tokio::test] @@ -1109,6 +1245,7 @@ mod tests { server_id: "srv_e_test".to_string(), server_name: "test-server".to_string(), }); + let (status_publisher, status_rx) = remote_control_status_channel(); save_auth( codex_home.path(), &remote_control_auth_dot_json("fresh-token"), @@ -1131,13 +1268,23 @@ mod tests { &auth_manager, &mut auth_recovery, &mut enrollment, - /*subscribe_cursor*/ None, - /*app_server_client_name*/ None, + RemoteControlConnectOptions { + subscribe_cursor: None, + app_server_client_name: None, + }, + &status_publisher, ) .await .expect_err("unauthorized response should fail the websocket connect"); server_task.await.expect("server task should succeed"); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + environment_id: Some("env_test".to_string()), + } + ); assert_eq!( err.to_string(), "remote control websocket auth failed with HTTP 401 Unauthorized; retrying after auth recovery" @@ -1187,6 +1334,7 @@ mod tests { .await; let mut auth_recovery = auth_manager.unauthorized_recovery(); let mut enrollment = None; + let (status_publisher, status_rx) = remote_control_status_channel(); save_auth( codex_home.path(), &remote_control_auth_dot_json("fresh-token"), @@ -1200,13 +1348,21 @@ mod tests { &auth_manager, &mut auth_recovery, &mut enrollment, - /*subscribe_cursor*/ None, - /*app_server_client_name*/ None, + RemoteControlConnectOptions { + subscribe_cursor: None, + app_server_client_name: None, + }, + &status_publisher, ) .await .expect_err("unauthorized enrollment should fail the websocket connect"); server_task.await.expect("server task should succeed"); + assert!( + !status_rx + .has_changed() + .expect("remote control status watch should remain open") + ); assert_eq!( err.to_string(), format!( @@ -1236,6 +1392,7 @@ mod tests { server_id: "srv_e_test".to_string(), server_name: "test-server".to_string(), }); + let (status_publisher, _status_rx) = remote_control_status_channel(); let err = connect_remote_control_websocket( &remote_control_target, @@ -1243,8 +1400,11 @@ mod tests { &auth_manager, &mut auth_recovery, &mut enrollment, - /*subscribe_cursor*/ None, - /*app_server_client_name*/ None, + RemoteControlConnectOptions { + subscribe_cursor: None, + app_server_client_name: None, + }, + &status_publisher, ) .await .expect_err("missing sqlite state db should fail remote control"); @@ -1254,6 +1414,63 @@ mod tests { assert_eq!(enrollment, None); } + #[tokio::test] + async fn connect_remote_control_websocket_requires_chatgpt_auth() { + let remote_control_target = normalize_remote_control_url("http://127.0.0.1:9/backend-api/") + .expect("target should parse"); + let codex_home = TempDir::new().expect("temp dir should create"); + let state_db = remote_control_state_runtime(&codex_home).await; + let auth_manager = AuthManager::shared( + codex_home.path().to_path_buf(), + /*enable_codex_api_key_env*/ false, + AuthCredentialsStoreMode::File, + /*chatgpt_base_url*/ None, + ) + .await; + let mut auth_recovery = auth_manager.unauthorized_recovery(); + let mut enrollment = Some(RemoteControlEnrollment { + account_id: "account_id".to_string(), + environment_id: "env_test".to_string(), + server_id: "srv_e_test".to_string(), + server_name: "test-server".to_string(), + }); + let (status_publisher, mut status_rx) = remote_control_status_channel(); + status_publisher.publish_environment_id(Some("env_test".to_string())); + status_rx + .changed() + .await + .expect("remote control status watch should remain open"); + + let err = connect_remote_control_websocket( + &remote_control_target, + Some(state_db.as_ref()), + &auth_manager, + &mut auth_recovery, + &mut enrollment, + RemoteControlConnectOptions { + subscribe_cursor: None, + app_server_client_name: None, + }, + &status_publisher, + ) + .await + .expect_err("missing auth should fail remote control"); + + assert_eq!(err.kind(), ErrorKind::PermissionDenied); + assert_eq!( + err.to_string(), + "remote control requires ChatGPT authentication" + ); + assert_eq!(enrollment, None); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + environment_id: None, + } + ); + } + #[tokio::test] async fn run_remote_control_websocket_loop_shutdown_cancels_reconnect_backoff() { let listener = TcpListener::bind("127.0.0.1:0") @@ -1266,6 +1483,7 @@ mod tests { normalize_remote_control_url(&remote_control_url).expect("target should parse"); let (transport_event_tx, transport_event_rx) = mpsc::channel(1); drop(transport_event_rx); + let (status_publisher, _status_rx) = remote_control_status_channel(); let shutdown_token = CancellationToken::new(); let (_enabled_tx, enabled_rx) = watch::channel(true); let websocket_task = tokio::spawn({ @@ -1276,7 +1494,10 @@ mod tests { Some(remote_control_target), /*state_db*/ None, remote_control_auth_manager(), - transport_event_tx, + RemoteControlChannels { + transport_event_tx, + status_publisher, + }, shutdown_token, enabled_rx, ) @@ -1294,6 +1515,85 @@ mod tests { .expect("websocket task should join"); } + #[tokio::test] + async fn publish_status_if_changed_sends_only_status_changes() { + let (status_publisher, mut status_rx) = remote_control_status_channel(); + + status_publisher.publish_environment_id(/*environment_id*/ None); + assert!( + timeout(Duration::from_millis(20), status_rx.changed()) + .await + .is_err() + ); + + status_publisher.publish_environment_id(Some("env_first".to_string())); + status_rx + .changed() + .await + .expect("remote control status watch should remain open"); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connecting, + environment_id: Some("env_first".to_string()), + } + ); + + status_publisher.publish_environment_id(Some("env_first".to_string())); + assert!( + timeout(Duration::from_millis(20), status_rx.changed()) + .await + .is_err() + ); + + status_publisher.publish_status(RemoteControlConnectionStatus::Connected); + status_rx + .changed() + .await + .expect("remote control status watch should remain open"); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connected, + environment_id: Some("env_first".to_string()), + } + ); + + status_publisher.publish_environment_id(/*environment_id*/ None); + status_rx + .changed() + .await + .expect("remote control status watch should remain open"); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Connected, + environment_id: None, + } + ); + + status_publisher.publish_environment_id(Some("env_disabled".to_string())); + status_publisher.publish_status(RemoteControlConnectionStatus::Disabled); + status_rx + .changed() + .await + .expect("remote control status watch should remain open"); + assert_eq!( + status_rx.borrow().clone(), + RemoteControlStatusChangedNotification { + status: RemoteControlConnectionStatus::Disabled, + environment_id: None, + } + ); + + status_publisher.publish_environment_id(Some("env_disabled".to_string())); + assert!( + timeout(Duration::from_millis(20), status_rx.changed()) + .await + .is_err() + ); + } + #[tokio::test] async fn run_server_writer_inner_sends_periodic_ping_frames() { let (client_stream, mut server_stream) = connected_websocket_pair().await; diff --git a/codex-rs/tui/src/app/app_server_adapter.rs b/codex-rs/tui/src/app/app_server_adapter.rs index 0a90c2c9b..0a1d01aec 100644 --- a/codex-rs/tui/src/app/app_server_adapter.rs +++ b/codex-rs/tui/src/app/app_server_adapter.rs @@ -441,6 +441,7 @@ fn server_notification_thread_target( | ServerNotification::AccountUpdated(_) | ServerNotification::AccountRateLimitsUpdated(_) | ServerNotification::AppListUpdated(_) + | ServerNotification::RemoteControlStatusChanged(_) | ServerNotification::ExternalAgentConfigImportCompleted(_) | ServerNotification::DeprecationNotice(_) | ServerNotification::ConfigWarning(_) diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 5756bf5d1..756de19b3 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -7261,6 +7261,7 @@ impl ChatWidget { | ServerNotification::McpToolCallProgress(_) | ServerNotification::McpServerOauthLoginCompleted(_) | ServerNotification::AppListUpdated(_) + | ServerNotification::RemoteControlStatusChanged(_) | ServerNotification::ExternalAgentConfigImportCompleted(_) | ServerNotification::FsChanged(_) | ServerNotification::FuzzyFileSearchSessionUpdated(_)