diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 24e3f072c..5bd76ffee 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -6527,6 +6527,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -6548,11 +6556,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadArchivedNotification": { "properties": { "threadId": { @@ -7078,6 +7094,96 @@ ], "type": "object" }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, + "ThreadStatusChangedNotification": { + "properties": { + "status": { + "$ref": "#/definitions/ThreadStatus" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "status", + "threadId" + ], + "type": "object" + }, "ThreadTokenUsage": { "properties": { "last": { @@ -8151,6 +8257,26 @@ "title": "Thread/startedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/status/changed" + ], + "title": "Thread/status/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/ThreadStatusChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/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 ffcbdc842..741f50da1 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 @@ -8108,6 +8108,26 @@ "title": "Thread/startedNotification", "type": "object" }, + { + "properties": { + "method": { + "enum": [ + "thread/status/changed" + ], + "title": "Thread/status/changedNotificationMethod", + "type": "string" + }, + "params": { + "$ref": "#/definitions/v2/ThreadStatusChangedNotification" + } + }, + "required": [ + "method", + "params" + ], + "title": "Thread/status/changedNotification", + "type": "object" + }, { "properties": { "method": { @@ -15168,6 +15188,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/v2/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -15189,11 +15217,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadArchiveParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { @@ -16375,6 +16411,98 @@ "title": "ThreadStartedNotification", "type": "object" }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/v2/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, + "ThreadStatusChangedNotification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "properties": { + "status": { + "$ref": "#/definitions/v2/ThreadStatus" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "status", + "threadId" + ], + "title": "ThreadStatusChangedNotification", + "type": "object" + }, "ThreadTokenUsage": { "properties": { "last": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json index 214d54ae3..f7db8aa66 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json @@ -794,6 +794,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -815,11 +823,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1306,6 +1322,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json index ddda3ca35..2ab535625 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json @@ -600,6 +600,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -621,11 +629,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1112,6 +1128,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json index 0a478b9f8..8aa8b0c4f 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json @@ -600,6 +600,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -621,11 +629,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1112,6 +1128,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json index d73833f10..198b91f05 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json @@ -794,6 +794,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -815,11 +823,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1306,6 +1322,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json index 63c9f794d..482102928 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json @@ -600,6 +600,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -621,11 +629,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1112,6 +1128,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json index 4f98fd138..00a619927 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json @@ -794,6 +794,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -815,11 +823,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1306,6 +1322,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json index 5d3011500..6f016b0c7 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json @@ -600,6 +600,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -621,11 +629,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1112,6 +1128,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStatusChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStatusChangedNotification.json new file mode 100644 index 000000000..bd6585042 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStatusChangedNotification.json @@ -0,0 +1,101 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "definitions": { + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + } + }, + "properties": { + "status": { + "$ref": "#/definitions/ThreadStatus" + }, + "threadId": { + "type": "string" + } + }, + "required": [ + "status", + "threadId" + ], + "title": "ThreadStatusChangedNotification", + "type": "object" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json index 64487a412..a49a3a8f9 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json @@ -600,6 +600,14 @@ ], "description": "Origin of the thread (CLI, VSCode, codex exec, codex app-server, etc.)." }, + "status": { + "allOf": [ + { + "$ref": "#/definitions/ThreadStatus" + } + ], + "description": "Current runtime status for the thread." + }, "turns": { "description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.", "items": { @@ -621,11 +629,19 @@ "modelProvider", "preview", "source", + "status", "turns", "updatedAt" ], "type": "object" }, + "ThreadActiveFlag": { + "enum": [ + "waitingOnApproval", + "waitingOnUserInput" + ], + "type": "string" + }, "ThreadId": { "type": "string" }, @@ -1112,6 +1128,81 @@ } ] }, + "ThreadStatus": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "notLoaded" + ], + "title": "NotLoadedThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "NotLoadedThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "idle" + ], + "title": "IdleThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "IdleThreadStatus", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "systemError" + ], + "title": "SystemErrorThreadStatusType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "SystemErrorThreadStatus", + "type": "object" + }, + { + "properties": { + "activeFlags": { + "items": { + "$ref": "#/definitions/ThreadActiveFlag" + }, + "type": "array" + }, + "type": { + "enum": [ + "active" + ], + "title": "ActiveThreadStatusType", + "type": "string" + } + }, + "required": [ + "activeFlags", + "type" + ], + "title": "ActiveThreadStatus", + "type": "object" + } + ] + }, "Turn": { "properties": { "error": { diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 82572bd02..20dfa846f 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -31,6 +31,7 @@ import type { TerminalInteractionNotification } from "./v2/TerminalInteractionNo import type { ThreadArchivedNotification } from "./v2/ThreadArchivedNotification"; import type { ThreadNameUpdatedNotification } from "./v2/ThreadNameUpdatedNotification"; import type { ThreadStartedNotification } from "./v2/ThreadStartedNotification"; +import type { ThreadStatusChangedNotification } from "./v2/ThreadStatusChangedNotification"; import type { ThreadTokenUsageUpdatedNotification } from "./v2/ThreadTokenUsageUpdatedNotification"; import type { ThreadUnarchivedNotification } from "./v2/ThreadUnarchivedNotification"; import type { TurnCompletedNotification } from "./v2/TurnCompletedNotification"; @@ -43,4 +44,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/archived", "params": ThreadArchivedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "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": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification } | { "method": "authStatusChange", "params": AuthStatusChangeNotification } | { "method": "loginChatGptComplete", "params": LoginChatGptCompleteNotification } | { "method": "sessionConfigured", "params": SessionConfiguredNotification }; +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/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "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": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification } | { "method": "authStatusChange", "params": AuthStatusChangeNotification } | { "method": "loginChatGptComplete", "params": LoginChatGptCompleteNotification } | { "method": "sessionConfigured", "params": SessionConfiguredNotification }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts b/codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts index 5ef567bd2..523a02858 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts @@ -3,6 +3,7 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { GitInfo } from "./GitInfo"; import type { SessionSource } from "./SessionSource"; +import type { ThreadStatus } from "./ThreadStatus"; import type { Turn } from "./Turn"; export type Thread = { id: string, @@ -22,6 +23,10 @@ createdAt: number, * Unix timestamp (in seconds) when the thread was last updated. */ updatedAt: number, +/** + * Current runtime status for the thread. + */ +status: ThreadStatus, /** * [UNSTABLE] Path to the thread on disk. */ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadActiveFlag.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadActiveFlag.ts new file mode 100644 index 000000000..73c875a00 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadActiveFlag.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 ThreadActiveFlag = "waitingOnApproval" | "waitingOnUserInput"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStatus.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStatus.ts new file mode 100644 index 000000000..7cc6c8a6a --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStatus.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 { ThreadActiveFlag } from "./ThreadActiveFlag"; + +export type ThreadStatus = { "type": "notLoaded" } | { "type": "idle" } | { "type": "systemError" } | { "type": "active", activeFlags: Array, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStatusChangedNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStatusChangedNotification.ts new file mode 100644 index 000000000..3242c892c --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadStatusChangedNotification.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 { ThreadStatus } from "./ThreadStatus"; + +export type ThreadStatusChangedNotification = { threadId: string, status: ThreadStatus, }; 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 7fb9681cc..e595d0ce0 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -146,6 +146,7 @@ export type { TextElement } from "./TextElement"; export type { TextPosition } from "./TextPosition"; export type { TextRange } from "./TextRange"; export type { Thread } from "./Thread"; +export type { ThreadActiveFlag } from "./ThreadActiveFlag"; export type { ThreadArchiveParams } from "./ThreadArchiveParams"; export type { ThreadArchiveResponse } from "./ThreadArchiveResponse"; export type { ThreadArchivedNotification } from "./ThreadArchivedNotification"; @@ -172,6 +173,8 @@ export type { ThreadSourceKind } from "./ThreadSourceKind"; export type { ThreadStartParams } from "./ThreadStartParams"; export type { ThreadStartResponse } from "./ThreadStartResponse"; export type { ThreadStartedNotification } from "./ThreadStartedNotification"; +export type { ThreadStatus } from "./ThreadStatus"; +export type { ThreadStatusChangedNotification } from "./ThreadStatusChangedNotification"; export type { ThreadTokenUsage } from "./ThreadTokenUsage"; export type { ThreadTokenUsageUpdatedNotification } from "./ThreadTokenUsageUpdatedNotification"; export type { ThreadUnarchiveParams } from "./ThreadUnarchiveParams"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 9f580de00..a9ce27ed0 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -774,6 +774,7 @@ server_notification_definitions! { /// NEW NOTIFICATIONS Error => "error" (v2::ErrorNotification), ThreadStarted => "thread/started" (v2::ThreadStartedNotification), + ThreadStatusChanged => "thread/status/changed" (v2::ThreadStatusChangedNotification), ThreadArchived => "thread/archived" (v2::ThreadArchivedNotification), ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification), ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification), @@ -1334,6 +1335,28 @@ mod tests { Ok(()) } + #[test] + fn serialize_thread_status_changed_notification() -> Result<()> { + let notification = + ServerNotification::ThreadStatusChanged(v2::ThreadStatusChangedNotification { + thread_id: "thr_123".to_string(), + status: v2::ThreadStatus::Idle, + }); + assert_eq!( + json!({ + "method": "thread/status/changed", + "params": { + "threadId": "thr_123", + "status": { + "type": "idle" + }, + } + }), + serde_json::to_value(¬ification)?, + ); + Ok(()) + } + #[test] fn mock_experimental_method_is_marked_experimental() { let request = ClientRequest::MockExperimentalMethod { diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 4c3d95378..ecc3ba259 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1837,6 +1837,29 @@ pub struct ThreadLoadedListResponse { pub next_cursor: Option, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type")] +#[ts(export_to = "v2/")] +pub enum ThreadStatus { + NotLoaded, + Idle, + SystemError, + #[serde(rename_all = "camelCase")] + #[ts(rename_all = "camelCase")] + Active { + active_flags: Vec, + }, +} + +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub enum ThreadActiveFlag { + WaitingOnApproval, + WaitingOnUserInput, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -2153,6 +2176,8 @@ pub struct Thread { /// Unix timestamp (in seconds) when the thread was last updated. #[ts(type = "number")] pub updated_at: i64, + /// Current runtime status for the thread. + pub status: ThreadStatus, /// [UNSTABLE] Path to the thread on disk. pub path: Option, /// Working directory captured for the thread. @@ -2941,6 +2966,14 @@ pub struct ThreadStartedNotification { pub thread: Thread, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadStatusChangedNotification { + pub thread_id: String, + pub status: ThreadStatus, +} + #[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 0b3b98c61..fcae5b306 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -117,9 +117,10 @@ Example with notification opt-out: - `thread/start` — create a new thread; emits `thread/started` and auto-subscribes you to turn/item events for that thread. - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. - `thread/fork` — fork an existing thread into a new thread id by copying the stored history; emits `thread/started` and auto-subscribes you to turn/item events for the new thread. -- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, and `cwd` filters. +- `thread/list` — page through stored rollouts; supports cursor-based pagination and optional `modelProviders`, `sourceKinds`, `archived`, and `cwd` filters. Each returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. - `thread/loaded/list` — list the thread ids currently loaded in memory. -- `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. +- `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. +- `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`). - `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`. - `thread/name/set` — set or update a thread’s user-facing name; returns `{}` on success. 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`. @@ -234,8 +235,8 @@ Example: } } { "id": 20, "result": { "data": [ - { "id": "thr_a", "preview": "Create a TUI", "modelProvider": "openai", "createdAt": 1730831111, "updatedAt": 1730831111 }, - { "id": "thr_b", "preview": "Fix tests", "modelProvider": "openai", "createdAt": 1730750000, "updatedAt": 1730750000 } + { "id": "thr_a", "preview": "Create a TUI", "modelProvider": "openai", "createdAt": 1730831111, "updatedAt": 1730831111, "status": { "type": "notLoaded" } }, + { "id": "thr_b", "preview": "Fix tests", "modelProvider": "openai", "createdAt": 1730750000, "updatedAt": 1730750000, "status": { "type": "notLoaded" } } ], "nextCursor": "opaque-token-or-null" } } @@ -254,18 +255,36 @@ When `nextCursor` is `null`, you’ve reached the final page. } } ``` +### Example: Track thread status changes + +`thread/status/changed` is emitted whenever a loaded thread's status changes: + +- Includes `threadId` and the new `status`. +- Status can be `notLoaded`, `idle`, `systemError`, or `active` (with `activeFlags`; `active` implies running). + +```json +{ "method": "thread/status/changed", "params": { + "threadId": "thr_123", + "status": { "type": "active", "activeFlags": [] } +} } +``` + ### Example: Read a thread Use `thread/read` to fetch a stored thread by id without resuming it. Pass `includeTurns` when you want the rollout history loaded into `thread.turns`. ```json { "method": "thread/read", "id": 22, "params": { "threadId": "thr_123" } } -{ "id": 22, "result": { "thread": { "id": "thr_123", "turns": [] } } } +{ "id": 22, "result": { + "thread": { "id": "thr_123", "status": { "type": "notLoaded" }, "turns": [] } +} } ``` ```json { "method": "thread/read", "id": 23, "params": { "threadId": "thr_123", "includeTurns": true } } -{ "id": 23, "result": { "thread": { "id": "thr_123", "turns": [ ... ] } } } +{ "id": 23, "result": { + "thread": { "id": "thr_123", "status": { "type": "notLoaded" }, "turns": [ ... ] } +} } ``` ### Example: Archive a thread diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 33bbb9c34..b206eac35 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -8,6 +8,8 @@ use crate::outgoing_message::ClientRequestResult; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; use crate::thread_state::ThreadState; use crate::thread_state::TurnSummary; +use crate::thread_status::ThreadWatchActiveGuard; +use crate::thread_status::ThreadWatchManager; use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; use codex_app_server_protocol::AgentMessageDeltaNotification; use codex_app_server_protocol::ApplyPatchApprovalParams; @@ -71,6 +73,7 @@ use codex_app_server_protocol::TurnPlanUpdatedNotification; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::build_turns_from_rollout_items; use codex_core::CodexThread; +use codex_core::ThreadManager; use codex_core::parse_command::shlex_join; use codex_core::protocol::ApplyPatchApprovalRequestEvent; use codex_core::protocol::CodexErrorInfo as CoreCodexErrorInfo; @@ -109,8 +112,10 @@ pub(crate) async fn apply_bespoke_event_handling( event: Event, conversation_id: ThreadId, conversation: Arc, + thread_manager: Arc, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc>, + thread_watch_manager: ThreadWatchManager, api_version: ApiVersion, fallback_model_provider: String, ) { @@ -119,8 +124,16 @@ pub(crate) async fn apply_bespoke_event_handling( msg, } = event; match msg { - EventMsg::TurnStarted(_) => {} + EventMsg::TurnStarted(_) => { + thread_watch_manager + .note_turn_started(&conversation_id.to_string()) + .await; + } EventMsg::TurnComplete(_ev) => { + let turn_failed = thread_state.lock().await.turn_summary.last_error.is_some(); + thread_watch_manager + .note_turn_completed(&conversation_id.to_string(), turn_failed) + .await; handle_turn_complete(conversation_id, event_turn_id, &outgoing, &thread_state).await; } EventMsg::Warning(_warning_event) => {} @@ -144,77 +157,87 @@ pub(crate) async fn apply_bespoke_event_handling( changes, reason, grant_root, - }) => match api_version { - ApiVersion::V1 => { - let params = ApplyPatchApprovalParams { - conversation_id, - call_id: call_id.clone(), - file_changes: changes.clone(), - reason, - grant_root, - }; - let rx = outgoing - .send_request(ServerRequestPayload::ApplyPatchApproval(params)) - .await; - tokio::spawn(async move { - on_patch_approval_response(call_id, rx, conversation).await; - }); - } - ApiVersion::V2 => { - // Until we migrate the core to be aware of a first class FileChangeItem - // and emit the corresponding EventMsg, we repurpose the call_id as the item_id. - let item_id = call_id.clone(); - let patch_changes = convert_patch_changes(&changes); - - let first_start = { - let mut state = thread_state.lock().await; - state - .turn_summary - .file_change_started - .insert(item_id.clone()) - }; - if first_start { - let item = ThreadItem::FileChange { - id: item_id.clone(), - changes: patch_changes.clone(), - status: PatchApplyStatus::InProgress, - }; - let notification = ItemStartedNotification { - thread_id: conversation_id.to_string(), - turn_id: event_turn_id.clone(), - item, - }; - outgoing - .send_server_notification(ServerNotification::ItemStarted(notification)) - .await; - } - - let params = FileChangeRequestApprovalParams { - thread_id: conversation_id.to_string(), - turn_id: turn_id.clone(), - item_id: item_id.clone(), - reason, - grant_root, - }; - let rx = outgoing - .send_request(ServerRequestPayload::FileChangeRequestApproval(params)) - .await; - tokio::spawn(async move { - on_file_change_request_approval_response( - event_turn_id, + }) => { + let permission_guard = thread_watch_manager + .note_permission_requested(&conversation_id.to_string()) + .await; + match api_version { + ApiVersion::V1 => { + let params = ApplyPatchApprovalParams { conversation_id, - item_id, - patch_changes, - rx, - conversation, - outgoing, - thread_state.clone(), - ) - .await; - }); + call_id: call_id.clone(), + file_changes: changes.clone(), + reason, + grant_root, + }; + let rx = outgoing + .send_request(ServerRequestPayload::ApplyPatchApproval(params)) + .await; + tokio::spawn(async move { + let _permission_guard = permission_guard; + on_patch_approval_response(call_id, rx, conversation).await; + }); + } + ApiVersion::V2 => { + // Until we migrate the core to be aware of a first class FileChangeItem + // and emit the corresponding EventMsg, we repurpose the call_id as the item_id. + let item_id = call_id.clone(); + let patch_changes = convert_patch_changes(&changes); + + let first_start = { + let mut state = thread_state.lock().await; + state + .turn_summary + .file_change_started + .insert(item_id.clone()) + }; + if first_start { + let item = ThreadItem::FileChange { + id: item_id.clone(), + changes: patch_changes.clone(), + status: PatchApplyStatus::InProgress, + }; + let notification = ItemStartedNotification { + thread_id: conversation_id.to_string(), + turn_id: event_turn_id.clone(), + item, + }; + outgoing + .send_server_notification(ServerNotification::ItemStarted(notification)) + .await; + } + + let params = FileChangeRequestApprovalParams { + thread_id: conversation_id.to_string(), + turn_id: turn_id.clone(), + item_id: item_id.clone(), + reason, + grant_root, + }; + let rx = outgoing + .send_request(ServerRequestPayload::FileChangeRequestApproval(params)) + .await; + tokio::spawn(async move { + on_file_change_request_approval_response( + event_turn_id, + conversation_id, + item_id, + patch_changes, + rx, + conversation, + outgoing, + thread_state.clone(), + permission_guard, + ) + .await; + }); + } } - }, + } EventMsg::ExecApprovalRequest(ev) => { + let permission_guard = thread_watch_manager + .note_permission_requested(&conversation_id.to_string()) + .await; let approval_id_for_op = ev.effective_approval_id(); let ExecApprovalRequestEvent { call_id, @@ -242,6 +265,7 @@ pub(crate) async fn apply_bespoke_event_handling( .send_request(ServerRequestPayload::ExecCommandApproval(params)) .await; tokio::spawn(async move { + let _permission_guard = permission_guard; on_exec_approval_response( approval_id_for_op, event_turn_id, @@ -290,6 +314,7 @@ pub(crate) async fn apply_bespoke_event_handling( conversation, outgoing, thread_state.clone(), + permission_guard, ) .await; }); @@ -298,6 +323,9 @@ pub(crate) async fn apply_bespoke_event_handling( } EventMsg::RequestUserInput(request) => { if matches!(api_version, ApiVersion::V2) { + let user_input_guard = thread_watch_manager + .note_user_input_requested(&conversation_id.to_string()) + .await; let questions = request .questions .into_iter() @@ -328,7 +356,13 @@ pub(crate) async fn apply_bespoke_event_handling( .send_request(ServerRequestPayload::ToolRequestUserInput(params)) .await; tokio::spawn(async move { - on_request_user_input_response(event_turn_id, rx, conversation).await; + on_request_user_input_response( + event_turn_id, + rx, + conversation, + user_input_guard, + ) + .await; }); } else { error!( @@ -589,6 +623,15 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } EventMsg::CollabCloseEnd(end_event) => { + if thread_manager + .get_thread(end_event.receiver_thread_id) + .await + .is_err() + { + thread_watch_manager + .remove_thread(&end_event.receiver_thread_id.to_string()) + .await; + } let status = match &end_event.status { codex_protocol::protocol::AgentStatus::Errored(_) | codex_protocol::protocol::AgentStatus::NotFound => V2CollabToolCallStatus::Failed, @@ -727,6 +770,10 @@ pub(crate) async fn apply_bespoke_event_handling( .await; } EventMsg::Error(ev) => { + thread_watch_manager + .note_system_error(&conversation_id.to_string()) + .await; + let message = ev.message.clone(); let codex_error_info = ev.codex_error_info.clone(); @@ -1106,6 +1153,9 @@ pub(crate) async fn apply_bespoke_event_handling( } } + thread_watch_manager + .note_turn_interrupted(&conversation_id.to_string()) + .await; handle_turn_interrupted(conversation_id, event_turn_id, &outgoing, &thread_state).await; } EventMsg::ThreadRolledBack(_rollback_event) => { @@ -1135,6 +1185,9 @@ pub(crate) async fn apply_bespoke_event_handling( match read_rollout_items_from_rollout(rollout_path.as_path()).await { Ok(items) => { thread.turns = build_turns_from_rollout_items(&items); + thread.status = thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; ThreadRollbackResponse { thread } } Err(err) => { @@ -1199,6 +1252,11 @@ pub(crate) async fn apply_bespoke_event_handling( ) .await; } + EventMsg::ShutdownComplete => { + thread_watch_manager + .note_thread_shutdown(&conversation_id.to_string()) + .await; + } _ => {} } @@ -1553,8 +1611,10 @@ async fn on_request_user_input_response( event_turn_id: String, receiver: oneshot::Receiver, conversation: Arc, + user_input_guard: ThreadWatchActiveGuard, ) { let response = receiver.await; + drop(user_input_guard); let value = match response { Ok(Ok(value)) => value, Ok(Err(err)) => { @@ -1711,8 +1771,10 @@ async fn on_file_change_request_approval_response( codex: Arc, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc>, + permission_guard: ThreadWatchActiveGuard, ) { let response = receiver.await; + drop(permission_guard); let (decision, completion_status) = match response { Ok(Ok(value)) => { let response = serde_json::from_value::(value) @@ -1776,8 +1838,10 @@ async fn on_command_execution_request_approval_response( conversation: Arc, outgoing: ThreadScopedOutgoingMessageSender, thread_state: Arc>, + permission_guard: ThreadWatchActiveGuard, ) { let response = receiver.await; + drop(permission_guard); let (decision, completion_status) = match response { Ok(Ok(value)) => { let response = serde_json::from_value::(value) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index b4663ea16..f4cdf6626 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -10,6 +10,7 @@ use crate::outgoing_message::ConnectionRequestId; use crate::outgoing_message::OutgoingMessageSender; use crate::outgoing_message::OutgoingNotification; use crate::outgoing_message::ThreadScopedOutgoingMessageSender; +use crate::thread_status::ThreadWatchManager; use chrono::DateTime; use chrono::SecondsFormat; use chrono::Utc; @@ -146,6 +147,7 @@ use codex_app_server_protocol::ThreadSourceKind; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadUnarchiveParams; use codex_app_server_protocol::ThreadUnarchiveResponse; use codex_app_server_protocol::ThreadUnarchivedNotification; @@ -338,12 +340,13 @@ pub(crate) struct CodexMessageProcessor { cloud_requirements: Arc>, active_login: Arc>>, thread_state_manager: ThreadStateManager, + thread_watch_manager: ThreadWatchManager, pending_fuzzy_searches: Arc>>>, fuzzy_search_sessions: Arc>>, feedback: CodexFeedback, } -#[derive(Clone, Copy, Debug, Default)] +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub(crate) enum ApiVersion { V1, #[default] @@ -399,13 +402,14 @@ impl CodexMessageProcessor { Self { auth_manager, thread_manager, - outgoing, + outgoing: outgoing.clone(), codex_linux_sandbox_exe, config, cli_overrides, cloud_requirements, active_login: Arc::new(Mutex::new(None)), thread_state_manager: ThreadStateManager::new(), + thread_watch_manager: ThreadWatchManager::new_with_outgoing(outgoing), pending_fuzzy_searches: Arc::new(Mutex::new(HashMap::new())), fuzzy_search_sessions: Arc::new(Mutex::new(HashMap::new())), feedback, @@ -2016,22 +2020,12 @@ impl CodexMessageProcessor { .. } = new_conv; let config_snapshot = thread.config_snapshot().await; - let thread = build_thread_from_snapshot( + let mut thread = build_thread_from_snapshot( thread_id, &config_snapshot, session_configured.rollout_path.clone(), ); - let response = ThreadStartResponse { - thread: thread.clone(), - model: config_snapshot.model, - model_provider: config_snapshot.model_provider_id, - cwd: config_snapshot.cwd, - approval_policy: config_snapshot.approval_policy.into(), - sandbox: config_snapshot.sandbox_policy.into(), - reasoning_effort: config_snapshot.reasoning_effort, - }; - // Auto-attach a thread listener when starting a thread. // Use the same behavior as the v1 API, with opt-in support for raw item events. if let Err(err) = self @@ -2050,6 +2044,25 @@ impl CodexMessageProcessor { ); } + self.thread_watch_manager + .upsert_thread(thread.clone()) + .await; + + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; + + let response = ThreadStartResponse { + thread: thread.clone(), + model: config_snapshot.model, + model_provider: config_snapshot.model_provider_id, + cwd: config_snapshot.cwd, + approval_policy: config_snapshot.approval_policy.into(), + sandbox: config_snapshot.sandbox_policy.into(), + reasoning_effort: config_snapshot.reasoning_effort, + }; + self.outgoing.send_response(request_id, response).await; let notif = ThreadStartedNotification { thread }; @@ -2353,7 +2366,11 @@ impl CodexMessageProcessor { .await; match result { - Ok(thread) => { + Ok(mut thread) => { + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; let thread_id = thread.id.clone(); let response = ThreadUnarchiveResponse { thread }; self.outgoing.send_response(request_id, response).await; @@ -2525,7 +2542,23 @@ impl CodexMessageProcessor { } }; - let data = summaries.into_iter().map(summary_to_thread).collect(); + let data = summaries + .into_iter() + .map(summary_to_thread) + .collect::>(); + let statuses = self + .thread_watch_manager + .loaded_statuses_for_threads(data.iter().map(|thread| thread.id.clone()).collect()) + .await; + let data = data + .into_iter() + .map(|mut thread| { + if let Some(status) = statuses.get(&thread.id) { + thread.status = status.clone(); + } + thread + }) + .collect(); let response = ThreadListResponse { data, next_cursor }; self.outgoing.send_response(request_id, response).await; } @@ -2711,6 +2744,10 @@ impl CodexMessageProcessor { } } + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; let response = ThreadReadResponse { thread }; self.outgoing.send_response(request_id, response).await; } @@ -2731,6 +2768,13 @@ impl CodexMessageProcessor { thread_id: ThreadId, connection_ids: Vec, ) { + if let Ok(thread) = self.thread_manager.get_thread(thread_id).await { + let config_snapshot = thread.config_snapshot().await; + let loaded_thread = + build_thread_from_snapshot(thread_id, &config_snapshot, thread.rollout_path()); + self.thread_watch_manager.upsert_thread(loaded_thread).await; + } + for connection_id in connection_ids { if let Err(err) = self .ensure_conversation_listener(thread_id, connection_id, false, ApiVersion::V2) @@ -2864,7 +2908,7 @@ impl CodexMessageProcessor { ); } - let Some(thread) = self + let Some(mut thread) = self .load_thread_from_rollout_or_send_internal( request_id.clone(), thread_id, @@ -2876,6 +2920,15 @@ impl CodexMessageProcessor { return; }; + self.thread_watch_manager + .upsert_thread(thread.clone()) + .await; + + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; + let response = ThreadResumeResponse { thread, model: session_configured.model, @@ -3015,7 +3068,7 @@ impl CodexMessageProcessor { ); } - let Some(thread) = self + let Some(mut thread) = self .load_thread_from_rollout_or_send_internal( request_id.clone(), existing_thread_id, @@ -3036,6 +3089,10 @@ impl CodexMessageProcessor { reasoning_effort, .. } = config_snapshot; + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; let response = ThreadResumeResponse { thread, model, @@ -3385,6 +3442,15 @@ impl CodexMessageProcessor { } } + self.thread_watch_manager + .upsert_thread(thread.clone()) + .await; + + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; + let response = ThreadForkResponse { thread: thread.clone(), model: session_configured.model, @@ -4658,6 +4724,10 @@ impl CodexMessageProcessor { .await; } + self.thread_watch_manager + .remove_thread(&thread_id.to_string()) + .await; + if state_db_ctx.is_none() { state_db_ctx = get_state_db(&self.config, None).await; } @@ -5512,7 +5582,14 @@ impl CodexMessageProcessor { if let Some(rollout_path) = review_thread.rollout_path() { match read_summary_from_rollout(rollout_path.as_path(), fallback_provider).await { Ok(summary) => { - let thread = summary_to_thread(summary); + let mut thread = summary_to_thread(summary); + self.thread_watch_manager + .upsert_thread(thread.clone()) + .await; + thread.status = self + .thread_watch_manager + .loaded_status_for_thread(&thread.id) + .await; let notif = ThreadStartedNotification { thread }; self.outgoing .send_server_notification(ServerNotification::ThreadStarted(notif)) @@ -5747,6 +5824,8 @@ impl CodexMessageProcessor { thread_state.set_listener(cancel_tx, &conversation); } let outgoing_for_task = self.outgoing.clone(); + let thread_manager = self.thread_manager.clone(); + let thread_watch_manager = self.thread_watch_manager.clone(); let fallback_model_provider = self.config.model_provider_id.clone(); tokio::spawn(async move { loop { @@ -5819,8 +5898,10 @@ impl CodexMessageProcessor { event.clone(), conversation_id, conversation.clone(), + thread_manager.clone(), thread_outgoing, thread_state.clone(), + thread_watch_manager.clone(), api_version, fallback_model_provider.clone(), ) @@ -6733,6 +6814,7 @@ fn build_thread_from_snapshot( model_provider: config_snapshot.model_provider_id.clone(), created_at: now, updated_at: now, + status: ThreadStatus::NotLoaded, path, cwd: config_snapshot.cwd.clone(), cli_version: env!("CARGO_PKG_VERSION").to_string(), @@ -6770,6 +6852,7 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread { model_provider, created_at: created_at.map(|dt| dt.timestamp()).unwrap_or(0), updated_at: updated_at.map(|dt| dt.timestamp()).unwrap_or(0), + status: ThreadStatus::NotLoaded, path: Some(path), cwd, cli_version, @@ -6786,6 +6869,7 @@ mod tests { use codex_protocol::protocol::SessionSource; use pretty_assertions::assert_eq; use serde_json::json; + use std::path::PathBuf; use tempfile::TempDir; #[test] diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 31bc831ca..7752ee92c 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -62,6 +62,7 @@ mod message_processor; mod models; mod outgoing_message; mod thread_state; +mod thread_status; mod transport; pub use crate::transport::AppServerTransport; @@ -550,14 +551,12 @@ pub async fn run_main_with_transport( connection_state.session.initialized.then_some(*connection_id) }) .collect(); - if !initialized_connection_ids.is_empty() { - processor - .try_attach_thread_listener( - thread_id, - initialized_connection_ids, - ) - .await; - } + processor + .try_attach_thread_listener( + thread_id, + initialized_connection_ids, + ) + .await; } Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { // TODO(jif) handle lag. diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index c7fd963c3..fd0f58abd 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -125,16 +125,13 @@ impl ThreadStateManager { self.thread_ids_by_connection .remove(&subscription_state.connection_id); } - if let Some(thread_state) = self.thread_states.get(&thread_id) { - thread_state - .lock() - .await - .remove_connection(subscription_state.connection_id); - } } if let Some(thread_state) = self.thread_states.get(&thread_id) { let mut thread_state = thread_state.lock().await; + if !connection_still_subscribed_to_thread { + thread_state.remove_connection(subscription_state.connection_id); + } if thread_state.subscribed_connection_ids().is_empty() { thread_state.clear_listener(); } diff --git a/codex-rs/app-server/src/thread_status.rs b/codex-rs/app-server/src/thread_status.rs new file mode 100644 index 000000000..7576791d3 --- /dev/null +++ b/codex-rs/app-server/src/thread_status.rs @@ -0,0 +1,630 @@ +#[cfg(test)] +use crate::outgoing_message::OutgoingEnvelope; +#[cfg(test)] +use crate::outgoing_message::OutgoingMessage; +use crate::outgoing_message::OutgoingMessageSender; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadActiveFlag; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::ThreadStatusChangedNotification; +use std::collections::HashMap; +#[cfg(test)] +use std::path::PathBuf; +use std::sync::Arc; +use tokio::sync::Mutex; +#[cfg(test)] +use tokio::sync::mpsc; + +#[derive(Clone)] +pub(crate) struct ThreadWatchManager { + state: Arc>, + outgoing: Option>, +} + +pub(crate) struct ThreadWatchActiveGuard { + manager: ThreadWatchManager, + thread_id: String, + guard_type: ThreadWatchActiveGuardType, + handle: tokio::runtime::Handle, +} + +impl ThreadWatchActiveGuard { + fn new( + manager: ThreadWatchManager, + thread_id: String, + guard_type: ThreadWatchActiveGuardType, + ) -> Self { + Self { + manager, + thread_id, + guard_type, + handle: tokio::runtime::Handle::current(), + } + } +} + +impl Drop for ThreadWatchActiveGuard { + fn drop(&mut self) { + let manager = self.manager.clone(); + let thread_id = self.thread_id.clone(); + let guard_type = self.guard_type; + self.handle.spawn(async move { + manager + .note_active_guard_released(thread_id, guard_type) + .await; + }); + } +} + +#[derive(Clone, Copy)] +enum ThreadWatchActiveGuardType { + Permission, + UserInput, +} + +impl Default for ThreadWatchManager { + fn default() -> Self { + Self::new() + } +} + +impl ThreadWatchManager { + pub(crate) fn new() -> Self { + Self { + state: Arc::new(Mutex::new(ThreadWatchState::default())), + outgoing: None, + } + } + + pub(crate) fn new_with_outgoing(outgoing: Arc) -> Self { + Self { + state: Arc::new(Mutex::new(ThreadWatchState::default())), + outgoing: Some(outgoing), + } + } + + pub(crate) async fn upsert_thread(&self, thread: Thread) { + self.mutate_and_publish(move |state| state.upsert_thread(thread.id)) + .await; + } + + pub(crate) async fn remove_thread(&self, thread_id: &str) { + let thread_id = thread_id.to_string(); + self.mutate_and_publish(move |state| state.remove_thread(&thread_id)) + .await; + } + + pub(crate) async fn loaded_status_for_thread(&self, thread_id: &str) -> ThreadStatus { + self.state.lock().await.loaded_status_for_thread(thread_id) + } + + pub(crate) async fn loaded_statuses_for_threads( + &self, + thread_ids: Vec, + ) -> HashMap { + let state = self.state.lock().await; + thread_ids + .into_iter() + .map(|thread_id| { + let status = state.loaded_status_for_thread(&thread_id); + (thread_id, status) + }) + .collect() + } + + pub(crate) async fn note_turn_started(&self, thread_id: &str) { + self.update_runtime_for_thread(thread_id, |runtime| { + runtime.is_loaded = true; + runtime.running = true; + runtime.has_system_error = false; + }) + .await; + } + + pub(crate) async fn note_turn_completed(&self, thread_id: &str, _failed: bool) { + self.clear_active_state(thread_id).await; + } + + pub(crate) async fn note_turn_interrupted(&self, thread_id: &str) { + self.clear_active_state(thread_id).await; + } + + pub(crate) async fn note_thread_shutdown(&self, thread_id: &str) { + self.update_runtime_for_thread(thread_id, |runtime| { + runtime.running = false; + runtime.pending_permission_requests = 0; + runtime.pending_user_input_requests = 0; + runtime.is_loaded = false; + }) + .await; + } + + pub(crate) async fn note_system_error(&self, thread_id: &str) { + self.update_runtime_for_thread(thread_id, |runtime| { + runtime.running = false; + runtime.pending_permission_requests = 0; + runtime.pending_user_input_requests = 0; + runtime.has_system_error = true; + }) + .await; + } + + async fn clear_active_state(&self, thread_id: &str) { + self.update_runtime_for_thread(thread_id, move |runtime| { + runtime.running = false; + runtime.pending_permission_requests = 0; + runtime.pending_user_input_requests = 0; + }) + .await; + } + + pub(crate) async fn note_permission_requested( + &self, + thread_id: &str, + ) -> ThreadWatchActiveGuard { + self.note_pending_request(thread_id, ThreadWatchActiveGuardType::Permission) + .await + } + + pub(crate) async fn note_user_input_requested( + &self, + thread_id: &str, + ) -> ThreadWatchActiveGuard { + self.note_pending_request(thread_id, ThreadWatchActiveGuardType::UserInput) + .await + } + + async fn note_pending_request( + &self, + thread_id: &str, + guard_type: ThreadWatchActiveGuardType, + ) -> ThreadWatchActiveGuard { + self.update_runtime_for_thread(thread_id, move |runtime| { + runtime.is_loaded = true; + let counter = Self::pending_counter(runtime, guard_type); + *counter = counter.saturating_add(1); + }) + .await; + ThreadWatchActiveGuard::new(self.clone(), thread_id.to_string(), guard_type) + } + + async fn mutate_and_publish(&self, mutate: F) + where + F: FnOnce(&mut ThreadWatchState) -> Option, + { + let notification = { + let mut state = self.state.lock().await; + mutate(&mut state) + }; + + if let Some(notification) = notification + && let Some(outgoing) = &self.outgoing + { + outgoing + .send_server_notification(ServerNotification::ThreadStatusChanged(notification)) + .await; + } + } + + async fn note_active_guard_released( + &self, + thread_id: String, + guard_type: ThreadWatchActiveGuardType, + ) { + self.update_runtime_for_thread(&thread_id, move |runtime| { + let counter = Self::pending_counter(runtime, guard_type); + *counter = counter.saturating_sub(1); + }) + .await; + } + + async fn update_runtime_for_thread(&self, thread_id: &str, update: F) + where + F: FnOnce(&mut RuntimeFacts), + { + let thread_id = thread_id.to_string(); + self.mutate_and_publish(move |state| state.update_runtime(&thread_id, update)) + .await; + } + + fn pending_counter( + runtime: &mut RuntimeFacts, + guard_type: ThreadWatchActiveGuardType, + ) -> &mut u32 { + match guard_type { + ThreadWatchActiveGuardType::Permission => &mut runtime.pending_permission_requests, + ThreadWatchActiveGuardType::UserInput => &mut runtime.pending_user_input_requests, + } + } +} + +#[derive(Default)] +struct ThreadWatchState { + runtime_by_thread_id: HashMap, +} + +impl ThreadWatchState { + fn upsert_thread(&mut self, thread_id: String) -> Option { + let previous_status = self.status_for(&thread_id); + let runtime = self + .runtime_by_thread_id + .entry(thread_id.clone()) + .or_default(); + runtime.is_loaded = true; + self.status_changed_notification(thread_id, previous_status) + } + + fn remove_thread(&mut self, thread_id: &str) -> Option { + self.runtime_by_thread_id.remove(thread_id); + None + } + + fn update_runtime( + &mut self, + thread_id: &str, + mutate: F, + ) -> Option + where + F: FnOnce(&mut RuntimeFacts), + { + let previous_status = self.status_for(thread_id); + let runtime = self + .runtime_by_thread_id + .entry(thread_id.to_string()) + .or_default(); + runtime.is_loaded = true; + mutate(runtime); + self.status_changed_notification(thread_id.to_string(), previous_status) + } + + fn status_for(&self, thread_id: &str) -> Option { + self.runtime_by_thread_id + .get(thread_id) + .map(loaded_thread_status) + } + + fn loaded_status_for_thread(&self, thread_id: &str) -> ThreadStatus { + self.status_for(thread_id) + .unwrap_or(ThreadStatus::NotLoaded) + } + + fn status_changed_notification( + &self, + thread_id: String, + previous_status: Option, + ) -> Option { + let status = self.status_for(&thread_id)?; + + if previous_status.as_ref() == Some(&status) { + return None; + } + + Some(ThreadStatusChangedNotification { thread_id, status }) + } +} + +#[derive(Clone, Default)] +struct RuntimeFacts { + is_loaded: bool, + running: bool, + pending_permission_requests: u32, + pending_user_input_requests: u32, + has_system_error: bool, +} + +fn loaded_thread_status(runtime: &RuntimeFacts) -> ThreadStatus { + if !runtime.is_loaded { + return ThreadStatus::NotLoaded; + } + + let mut active_flags = Vec::new(); + if runtime.pending_permission_requests > 0 { + active_flags.push(ThreadActiveFlag::WaitingOnApproval); + } + if runtime.pending_user_input_requests > 0 { + active_flags.push(ThreadActiveFlag::WaitingOnUserInput); + } + + if runtime.running || !active_flags.is_empty() { + return ThreadStatus::Active { active_flags }; + } + + if runtime.has_system_error { + return ThreadStatus::SystemError; + } + + ThreadStatus::Idle +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use tokio::time::Duration; + use tokio::time::timeout; + + const INTERACTIVE_THREAD_ID: &str = "00000000-0000-0000-0000-000000000001"; + const NON_INTERACTIVE_THREAD_ID: &str = "00000000-0000-0000-0000-000000000002"; + + #[tokio::test] + async fn loaded_status_defaults_to_not_loaded_for_untracked_threads() { + let manager = ThreadWatchManager::new(); + + assert_eq!( + manager + .loaded_status_for_thread("00000000-0000-0000-0000-000000000003") + .await, + ThreadStatus::NotLoaded, + ); + } + + #[tokio::test] + async fn tracks_non_interactive_thread_status() { + let manager = ThreadWatchManager::new(); + manager + .upsert_thread(test_thread( + NON_INTERACTIVE_THREAD_ID, + codex_app_server_protocol::SessionSource::AppServer, + )) + .await; + + manager.note_turn_started(NON_INTERACTIVE_THREAD_ID).await; + + assert_eq!( + manager + .loaded_status_for_thread(NON_INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::Active { + active_flags: vec![], + }, + ); + } + + #[tokio::test] + async fn status_updates_track_single_thread() { + let manager = ThreadWatchManager::new(); + manager + .upsert_thread(test_thread( + INTERACTIVE_THREAD_ID, + codex_app_server_protocol::SessionSource::Cli, + )) + .await; + + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::Active { + active_flags: vec![], + }, + ); + + let permission_guard = manager + .note_permission_requested(INTERACTIVE_THREAD_ID) + .await; + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::Active { + active_flags: vec![ThreadActiveFlag::WaitingOnApproval], + }, + ); + + let user_input_guard = manager + .note_user_input_requested(INTERACTIVE_THREAD_ID) + .await; + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::Active { + active_flags: vec![ + ThreadActiveFlag::WaitingOnApproval, + ThreadActiveFlag::WaitingOnUserInput, + ], + }, + ); + + drop(permission_guard); + wait_for_status( + &manager, + INTERACTIVE_THREAD_ID, + ThreadStatus::Active { + active_flags: vec![ThreadActiveFlag::WaitingOnUserInput], + }, + ) + .await; + + drop(user_input_guard); + wait_for_status( + &manager, + INTERACTIVE_THREAD_ID, + ThreadStatus::Active { + active_flags: vec![], + }, + ) + .await; + + manager + .note_turn_completed(INTERACTIVE_THREAD_ID, false) + .await; + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::Idle, + ); + } + + #[tokio::test] + async fn system_error_sets_idle_flag_until_next_turn() { + let manager = ThreadWatchManager::new(); + manager + .upsert_thread(test_thread( + INTERACTIVE_THREAD_ID, + codex_app_server_protocol::SessionSource::Cli, + )) + .await; + + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + manager.note_system_error(INTERACTIVE_THREAD_ID).await; + + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::SystemError, + ); + + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::Active { + active_flags: vec![], + }, + ); + } + + #[tokio::test] + async fn shutdown_marks_thread_not_loaded() { + let manager = ThreadWatchManager::new(); + manager + .upsert_thread(test_thread( + INTERACTIVE_THREAD_ID, + codex_app_server_protocol::SessionSource::Cli, + )) + .await; + + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + manager.note_thread_shutdown(INTERACTIVE_THREAD_ID).await; + + assert_eq!( + manager + .loaded_status_for_thread(INTERACTIVE_THREAD_ID) + .await, + ThreadStatus::NotLoaded, + ); + } + + #[tokio::test] + async fn loaded_statuses_default_to_not_loaded_for_untracked_threads() { + let manager = ThreadWatchManager::new(); + manager + .upsert_thread(test_thread( + INTERACTIVE_THREAD_ID, + codex_app_server_protocol::SessionSource::Cli, + )) + .await; + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + + let statuses = manager + .loaded_statuses_for_threads(vec![ + INTERACTIVE_THREAD_ID.to_string(), + NON_INTERACTIVE_THREAD_ID.to_string(), + ]) + .await; + + assert_eq!( + statuses.get(INTERACTIVE_THREAD_ID), + Some(&ThreadStatus::Active { + active_flags: vec![], + }), + ); + assert_eq!( + statuses.get(NON_INTERACTIVE_THREAD_ID), + Some(&ThreadStatus::NotLoaded), + ); + } + + #[tokio::test] + async fn status_change_emits_notification() { + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(8); + let manager = ThreadWatchManager::new_with_outgoing(Arc::new(OutgoingMessageSender::new( + outgoing_tx, + ))); + + manager + .upsert_thread(test_thread( + INTERACTIVE_THREAD_ID, + codex_app_server_protocol::SessionSource::Cli, + )) + .await; + assert_eq!( + recv_status_changed_notification(&mut outgoing_rx).await, + ThreadStatusChangedNotification { + thread_id: INTERACTIVE_THREAD_ID.to_string(), + status: ThreadStatus::Idle, + }, + ); + + manager.note_turn_started(INTERACTIVE_THREAD_ID).await; + assert_eq!( + recv_status_changed_notification(&mut outgoing_rx).await, + ThreadStatusChangedNotification { + thread_id: INTERACTIVE_THREAD_ID.to_string(), + status: ThreadStatus::Active { + active_flags: vec![], + }, + }, + ); + } + + async fn wait_for_status( + manager: &ThreadWatchManager, + thread_id: &str, + expected_status: ThreadStatus, + ) { + timeout(Duration::from_secs(1), async { + loop { + let status = manager.loaded_status_for_thread(thread_id).await; + if status == expected_status { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("timed out waiting for status"); + } + + async fn recv_status_changed_notification( + outgoing_rx: &mut mpsc::Receiver, + ) -> ThreadStatusChangedNotification { + let envelope = timeout(Duration::from_secs(1), outgoing_rx.recv()) + .await + .expect("timed out waiting for outgoing notification") + .expect("outgoing channel closed unexpectedly"); + let OutgoingEnvelope::Broadcast { message } = envelope else { + panic!("expected broadcast notification"); + }; + let OutgoingMessage::AppServerNotification(ServerNotification::ThreadStatusChanged( + notification, + )) = message + else { + panic!("expected thread/status/changed notification"); + }; + notification + } + + fn test_thread(thread_id: &str, source: codex_app_server_protocol::SessionSource) -> Thread { + Thread { + id: thread_id.to_string(), + preview: String::new(), + model_provider: "mock-provider".to_string(), + created_at: 0, + updated_at: 0, + status: ThreadStatus::NotLoaded, + path: None, + cwd: PathBuf::from("/tmp"), + cli_version: "test".to_string(), + source, + git_info: None, + turns: Vec::new(), + } + } +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 5cfe7de46..ac8f9c988 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -25,6 +25,7 @@ mod thread_read; mod thread_resume; mod thread_rollback; mod thread_start; +mod thread_status; mod thread_unarchive; mod turn_interrupt; mod turn_start; diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 1e2e172cf..7047622a4 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -14,6 +14,7 @@ use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; use pretty_assertions::assert_eq; @@ -80,6 +81,7 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { assert_ne!(thread.id, conversation_id); assert_eq!(thread.preview, preview); assert_eq!(thread.model_provider, "mock_provider"); + assert_eq!(thread.status, ThreadStatus::Idle); let thread_path = thread.path.clone().expect("thread path"); assert!(thread_path.is_absolute()); assert_ne!(thread_path, original_path); diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index cee88a18d..46e1d170f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -2,6 +2,8 @@ use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_fake_rollout; use app_test_support::create_fake_rollout_with_source; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_sequence; use app_test_support::rollout_path; use app_test_support::to_response; use chrono::DateTime; @@ -14,6 +16,12 @@ use codex_app_server_protocol::SessionSource; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadSortKey; use codex_app_server_protocol::ThreadSourceKind; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_protocol::ThreadId; use codex_protocol::protocol::GitInfo as CoreGitInfo; @@ -21,6 +29,7 @@ use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionSource as CoreSessionSource; use codex_protocol::protocol::SubAgentSource; +use core_test_support::responses; use pretty_assertions::assert_eq; use std::cmp::Reverse; use std::fs; @@ -157,7 +166,9 @@ async fn thread_list_basic_empty() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(10), @@ -172,6 +183,97 @@ async fn thread_list_basic_empty() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_list_reports_system_error_idle_flag_after_failed_turn() -> Result<()> { + let responses = vec![ + create_final_assistant_message_sse_response("seeded")?, + responses::sse_failed("resp-2", "server_error", "simulated failure"), + ]; + let server = create_mock_responses_server_sequence(responses).await; + + let codex_home = TempDir::new()?; + create_runtime_config(codex_home.path(), &server.uri())?; + let mut mcp = init_mcp(codex_home.path()).await?; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + + let seed_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "seed history".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let seed_turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(seed_turn_id)), + ) + .await??; + let _: TurnStartResponse = to_response::(seed_turn_resp)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let failed_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "fail turn".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let failed_turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(failed_turn_id)), + ) + .await??; + let _: TurnStartResponse = to_response::(failed_turn_resp)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("error"), + ) + .await??; + + let ThreadListResponse { data, .. } = list_threads( + &mut mcp, + None, + Some(10), + Some(vec!["mock_provider".to_string()]), + Some(vec![ + ThreadSourceKind::AppServer, + ThreadSourceKind::Cli, + ThreadSourceKind::VsCode, + ]), + None, + ) + .await?; + let listed = data + .iter() + .find(|candidate| candidate.id == thread.id) + .expect("expected started thread to be listed"); + assert_eq!(listed.status, ThreadStatus::SystemError,); + + Ok(()) +} + // Minimal config.toml for listing. fn create_minimal_config(codex_home: &std::path::Path) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); @@ -184,6 +286,29 @@ approval_policy = "never" ) } +fn create_runtime_config(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "never" +sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} + #[tokio::test] async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> { let codex_home = TempDir::new()?; @@ -240,6 +365,7 @@ async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> { assert_eq!(thread.cli_version, "0.0.0"); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); + assert_eq!(thread.status, ThreadStatus::NotLoaded); } let cursor1 = cursor1.expect("expected nextCursor on first page"); @@ -266,6 +392,7 @@ async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> { assert_eq!(thread.cli_version, "0.0.0"); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); + assert_eq!(thread.status, ThreadStatus::NotLoaded); } assert_eq!(cursor2, None, "expected nextCursor to be null on last page"); @@ -298,7 +425,9 @@ async fn thread_list_respects_provider_filter() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; // Filter to only other_provider; expect 1 item, nextCursor None. - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(10), @@ -369,7 +498,9 @@ async fn thread_list_respects_cwd_filter() -> Result<()> { mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; - let ThreadListResponse { data, next_cursor } = to_response::(resp)?; + let ThreadListResponse { + data, next_cursor, .. + } = to_response::(resp)?; assert_eq!(next_cursor, None); assert_eq!(data.len(), 1); @@ -405,7 +536,9 @@ async fn thread_list_empty_source_kinds_defaults_to_interactive_only() -> Result let mut mcp = init_mcp(codex_home.path()).await?; - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(10), @@ -454,7 +587,9 @@ async fn thread_list_filters_by_source_kind_subagent_thread_spawn() -> Result<() let mut mcp = init_mcp(codex_home.path()).await?; - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(10), @@ -607,7 +742,9 @@ async fn thread_list_fetches_until_limit_or_exhausted() -> Result<()> { // Request 8 threads for the target provider; the matches only start on the // third page so we rely on pagination to reach the limit. - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(8), @@ -653,7 +790,9 @@ async fn thread_list_enforces_max_limit() -> Result<()> { let mut mcp = init_mcp(codex_home.path()).await?; - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(200), @@ -700,7 +839,9 @@ async fn thread_list_stops_when_not_enough_filtered_results_exist() -> Result<() // Request more threads than exist after filtering; expect all matches to be // returned with nextCursor None. - let ThreadListResponse { data, next_cursor } = list_threads( + let ThreadListResponse { + data, next_cursor, .. + } = list_threads( &mut mcp, None, Some(10), @@ -934,6 +1075,7 @@ async fn thread_list_updated_at_paginates_with_cursor() -> Result<()> { let ThreadListResponse { data: page1, next_cursor: cursor1, + .. } = list_threads_with_sort( &mut mcp, None, @@ -951,6 +1093,7 @@ async fn thread_list_updated_at_paginates_with_cursor() -> Result<()> { let ThreadListResponse { data: page2, next_cursor: cursor2, + .. } = list_threads_with_sort( &mut mcp, Some(cursor1), diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 338728a12..5a6925bcb 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -12,10 +12,14 @@ use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; +use core_test_support::responses; use pretty_assertions::assert_eq; use std::path::Path; use std::path::PathBuf; @@ -73,6 +77,7 @@ async fn thread_read_returns_summary_without_turns() -> Result<()> { assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); assert_eq!(thread.turns.len(), 0); + assert_eq!(thread.status, ThreadStatus::NotLoaded); Ok(()) } @@ -133,6 +138,7 @@ async fn thread_read_can_include_turns() -> Result<()> { } other => panic!("expected user message item, got {other:?}"), } + assert_eq!(thread.status, ThreadStatus::NotLoaded); Ok(()) } @@ -181,6 +187,7 @@ async fn thread_read_loaded_thread_returns_precomputed_path_before_materializati assert_eq!(read.path, Some(thread_path)); assert!(read.preview.is_empty()); assert_eq!(read.turns.len(), 0); + assert_eq!(read.status, ThreadStatus::Idle); Ok(()) } @@ -236,6 +243,73 @@ async fn thread_read_include_turns_rejects_unmaterialized_loaded_thread() -> Res Ok(()) } +#[tokio::test] +async fn thread_read_reports_system_error_idle_flag_after_failed_turn() -> Result<()> { + let server = responses::start_mock_server().await; + let _response_mock = responses::mount_sse_once( + &server, + responses::sse_failed("resp-1", "server_error", "simulated failure"), + ) + .await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + + let turn_start_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "fail this turn".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_start_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + let _: TurnStartResponse = to_response::(turn_start_response)?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("error"), + ) + .await??; + + let read_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id, + include_turns: false, + }) + .await?; + let read_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(read_id)), + ) + .await??; + let ThreadReadResponse { thread } = to_response::(read_resp)?; + + assert_eq!(thread.status, ThreadStatus::SystemError,); + + Ok(()) +} + // Helper to create a config.toml pointing at the mock model server. fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 3976d3eea..4ec348564 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -14,6 +14,7 @@ use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; @@ -129,6 +130,7 @@ async fn thread_resume_returns_rollout_history() -> Result<()> { assert_eq!(thread.cli_version, "0.0.0"); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); + assert_eq!(thread.status, ThreadStatus::Idle); assert_eq!( thread.turns.len(), @@ -178,6 +180,7 @@ async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() - let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; assert_eq!(thread.updated_at, rollout.expected_updated_at); + assert_eq!(thread.status, ThreadStatus::Idle); let after_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?; assert_eq!(after_modified, rollout.before_modified); @@ -283,11 +286,16 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { ..Default::default() }) .await?; - timeout( + let resume_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, secondary.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; + let ThreadResumeResponse { + thread: resumed_thread, + .. + } = to_response::(resume_resp)?; + assert_ne!(resumed_thread.status, ThreadStatus::NotLoaded); timeout( DEFAULT_READ_TIMEOUT, @@ -582,8 +590,15 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R primary.read_stream_until_response_message(RequestId::Integer(resume_id)), ) .await??; - let ThreadResumeResponse { model, .. } = to_response::(resume_resp)?; + let ThreadResumeResponse { thread, model, .. } = + to_response::(resume_resp)?; assert_eq!(model, "gpt-5.1-codex-max"); + assert_eq!( + thread.status, + ThreadStatus::Active { + active_flags: vec![], + } + ); timeout( DEFAULT_READ_TIMEOUT, @@ -630,6 +645,7 @@ async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Re } = to_response::(resume_resp)?; assert_eq!(resumed_thread.updated_at, expected_updated_at); + assert_eq!(resumed_thread.status, ThreadStatus::Idle); let after_resume_modified = std::fs::metadata(&rollout_file_path)?.modified()?; assert_eq!(after_resume_modified, before_modified); @@ -761,6 +777,7 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { } = to_response::(resume_resp)?; assert_eq!(resumed.id, thread.id); assert_eq!(resumed.path, thread.path); + assert_eq!(resumed.status, ThreadStatus::Idle); Ok(()) } @@ -809,6 +826,7 @@ async fn thread_resume_supports_history_and_overrides() -> Result<()> { assert!(!resumed.id.is_empty()); assert_eq!(model_provider, "mock_provider"); assert_eq!(resumed.preview, history_text); + assert_eq!(resumed.status, ThreadStatus::Idle); Ok(()) } @@ -951,6 +969,7 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { ) .await??; let resume: ThreadResumeResponse = to_response::(resume_resp)?; + assert_eq!(resume.thread.status, ThreadStatus::Idle); let turn_id = secondary .send_turn_start_request(TurnStartParams { diff --git a/codex-rs/app-server/tests/suite/v2/thread_rollback.rs b/codex-rs/app-server/tests/suite/v2/thread_rollback.rs index 6e3767db9..47a358728 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_rollback.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_rollback.rs @@ -12,6 +12,7 @@ use codex_app_server_protocol::ThreadRollbackParams; use codex_app_server_protocol::ThreadRollbackResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::UserInput as V2UserInput; use pretty_assertions::assert_eq; @@ -111,6 +112,7 @@ async fn thread_rollback_drops_last_turns_and_persists_to_rollout() -> Result<() } = to_response::(rollback_resp)?; assert_eq!(rolled_back_thread.turns.len(), 1); + assert_eq!(rolled_back_thread.status, ThreadStatus::Idle); assert_eq!(rolled_back_thread.turns[0].items.len(), 2); match &rolled_back_thread.turns[0].items[0] { ThreadItem::UserMessage { content, .. } => { @@ -140,6 +142,7 @@ async fn thread_rollback_drops_last_turns_and_persists_to_rollout() -> Result<() let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; assert_eq!(thread.turns.len(), 1); + assert_eq!(thread.status, ThreadStatus::Idle); assert_eq!(thread.turns[0].items.len(), 2); match &thread.turns[0].items[0] { ThreadItem::UserMessage { content, .. } => { diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index b7ee9a13e..bcee63df9 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -9,6 +9,7 @@ use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; +use codex_app_server_protocol::ThreadStatus; use codex_core::config::set_project_trust_level; use codex_protocol::config_types::TrustLevel; use codex_protocol::openai_models::ReasoningEffort; @@ -59,6 +60,7 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> { thread.created_at > 0, "created_at should be a positive UNIX timestamp" ); + assert_eq!(thread.status, ThreadStatus::Idle); let thread_path = thread.path.clone().expect("thread path should be present"); assert!(thread_path.is_absolute(), "thread path should be absolute"); assert!( diff --git a/codex-rs/app-server/tests/suite/v2/thread_status.rs b/codex-rs/app-server/tests/suite/v2/thread_status.rs new file mode 100644 index 000000000..ad90e4900 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_status.rs @@ -0,0 +1,240 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_sequence; +use app_test_support::to_response; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::ThreadStatusChangedNotification; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +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); + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn thread_status_changed_emits_runtime_updates() -> Result<()> { + let codex_home = TempDir::new()?; + let responses = vec![create_final_assistant_message_sse_response("done")?]; + let server = create_mock_responses_server_sequence(responses).await; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = + McpProcess::new_with_env(codex_home.path(), &[("RUST_LOG", Some("info"))]).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_start_resp)?; + + let turn_start_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "collect status updates".to_string(), + text_elements: Vec::new(), + }], + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let turn_start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + let _: TurnStartResponse = to_response(turn_start_resp)?; + + let mut saw_active_running = false; + let mut saw_idle_after_turn = false; + let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT; + while tokio::time::Instant::now() < deadline { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + let message = match timeout(remaining, mcp.read_next_message()).await { + Ok(Ok(message)) => message, + _ => break, + }; + match message { + JSONRPCMessage::Notification(JSONRPCNotification { + method, + params: Some(params), + }) if method == "thread/status/changed" => { + let notification: ThreadStatusChangedNotification = serde_json::from_value(params)?; + if notification.thread_id != thread.id { + continue; + } + match notification.status { + ThreadStatus::Active { .. } => { + saw_active_running = true; + } + ThreadStatus::Idle => { + if saw_active_running { + saw_idle_after_turn = true; + } + } + ThreadStatus::SystemError => { + if saw_active_running { + saw_idle_after_turn = true; + } + } + ThreadStatus::NotLoaded => { + if saw_active_running { + saw_idle_after_turn = true; + } + } + } + } + _ => {} + } + + if saw_active_running && saw_idle_after_turn { + break; + } + } + + assert!( + saw_active_running, + "expected running active flag in thread/status/changed notifications" + ); + assert!( + saw_idle_after_turn, + "expected idle status after turn completion in thread/status/changed notifications" + ); + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + Ok(()) +} + +#[tokio::test] +async fn thread_status_changed_can_be_opted_out() -> Result<()> { + let codex_home = TempDir::new()?; + let responses = vec![create_final_assistant_message_sse_response("done")?]; + let server = create_mock_responses_server_sequence(responses).await; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + let message = timeout( + DEFAULT_READ_TIMEOUT, + mcp.initialize_with_capabilities( + ClientInfo { + name: "codex_vscode".to_string(), + title: Some("Codex VS Code Extension".to_string()), + version: "0.1.0".to_string(), + }, + Some(InitializeCapabilities { + experimental_api: true, + opt_out_notification_methods: Some(vec!["thread/status/changed".to_string()]), + }), + ), + ) + .await??; + let JSONRPCMessage::Response(_) = message else { + anyhow::bail!("expected initialize response, got {message:?}"); + }; + + let thread_start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_start_resp)?; + + let turn_start_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id, + input: vec![V2UserInput::Text { + text: "run once".to_string(), + text_elements: Vec::new(), + }], + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let turn_start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + let _: TurnStartResponse = to_response(turn_start_resp)?; + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let status_update = timeout( + std::time::Duration::from_millis(500), + mcp.read_stream_until_notification_message("thread/status/changed"), + ) + .await; + match status_update { + Err(_) => {} + Ok(Ok(notification)) => { + anyhow::bail!( + "thread/status/changed should be filtered by optOutNotificationMethods; got: {notification:?}" + ); + } + Ok(Err(err)) => { + anyhow::bail!( + "expected timeout waiting for filtered thread/status/changed, got: {err}" + ); + } + } + + Ok(()) +} + +fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { + let config_toml = codex_home.join("config.toml"); + std::fs::write( + config_toml, + format!( + r#" +model = "mock-model" +approval_policy = "untrusted" +sandbox_mode = "read-only" + +model_provider = "mock_provider" + +[features] +collaboration_modes = true + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + ) +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index ba3b740e2..35de7eee9 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -8,6 +8,7 @@ use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus; use codex_app_server_protocol::ThreadUnarchiveParams; use codex_app_server_protocol::ThreadUnarchiveResponse; use codex_app_server_protocol::ThreadUnarchivedNotification; @@ -137,6 +138,7 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result unarchived_thread.updated_at > old_timestamp, "expected updated_at to be bumped on unarchive" ); + assert_eq!(unarchived_thread.status, ThreadStatus::NotLoaded); let rollout_path_display = rollout_path.display(); assert!(