From 0d27f0e495fa2160fd014053e57a07482de03a69 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Wed, 2 Sep 2026 22:56:07 +0400 Subject: [PATCH 1/6] feat: expose background terminals as async tasks Publish Codex-owned background terminals through the AIR async task lifecycle. Use the app-server process handle only for targeted stop requests. --- README.md | 7 + docs/async-tasks.md | 32 +++ src/AcpExtensions.ts | 8 +- src/AirExtension.ts | 1 + src/CodexAcpServer.ts | 57 +++++ src/CodexAppServerClient.ts | 17 +- .../CodexACPAgent/async-tasks.test.ts | 241 ++++++++++++++++++ .../CodexACPAgent/initialize.test.ts | 2 +- src/__tests__/acp-test-utils.ts | 7 + src/async-tasks/AcpAsyncTasks.ts | 41 +++ src/async-tasks/AsyncTaskExtension.ts | 17 ++ src/async-tasks/BackgroundTerminalApi.ts | 34 +++ .../CodexBackgroundTerminalTasks.ts | 149 +++++++++++ src/index.ts | 7 + src/subagents/AcpSubagents.ts | 4 +- 15 files changed, 620 insertions(+), 4 deletions(-) create mode 100644 docs/async-tasks.md create mode 100644 src/__tests__/CodexACPAgent/async-tasks.test.ts create mode 100644 src/async-tasks/AcpAsyncTasks.ts create mode 100644 src/async-tasks/AsyncTaskExtension.ts create mode 100644 src/async-tasks/BackgroundTerminalApi.ts create mode 100644 src/async-tasks/CodexBackgroundTerminalTasks.ts diff --git a/README.md b/README.md index 092f61ba..88a2ad51 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol] - Text prompts, embedded context, images, resource links, and additional workspace directories. - Shell command, file change, [permission request](docs/permission-extension.md), MCP tool call, terminal output, reasoning, plan, web search, image generation, image view, token usage, and review events. - [Native ACP subagent sessions](docs/subagent-sessions.md) (after capability negotiation) with separate child histories and root-routed permissions; a legacy tool-call fallback otherwise. +- [Background terminal tasks](docs/async-tasks.md) in AIR, with task status and targeted stop support after capability negotiation. - Session-scoped long-running goals through the provider-neutral [goal extension](docs/goal-extension.md). - A per-turn [agent file-change report](docs/agent-file-change-report.md) after capability negotiation. - Client-provided MCP servers over command-based stdio config and HTTP transport. @@ -82,6 +83,12 @@ Subagent sessions follow the draft [ACP subagent RFD](https://github.com/agentcl See [docs/subagent-sessions.md](docs/subagent-sessions.md) for the negotiation, lifecycle events, `session/load` reconstruction, and legacy fallback details. +### Background terminal tasks + +Codex can keep a shell command running after a turn continues. AIR clients can show this work in the Async Tasks panel and stop one command. + +See [docs/async-tasks.md](docs/async-tasks.md) for the capability, lifecycle events, and stop request. + ## License By contributing, you agree that your contributions will be licensed under the Apache 2.0 License. diff --git a/docs/async-tasks.md b/docs/async-tasks.md new file mode 100644 index 00000000..79f444d2 --- /dev/null +++ b/docs/async-tasks.md @@ -0,0 +1,32 @@ +# Background terminal tasks + +Codex app-server owns shell commands that continue after their initial tool call. The adapter exposes these commands through the AIR async task extension. + +## Negotiation + +The client adds `asyncTasks` to `_meta.jetbrains.air.capabilities`. The adapter advertises the same capability in its `initialize` response. + +The adapter emits no async task updates when the client does not advertise this capability. + +## Lifecycle + +The adapter uses `thread/backgroundTerminals/list` as the source of active processes. It maps each active process to `async_task_spawned`. + +The command item ID is both the async task ID and the related tool call ID. The app-server process ID remains an internal control handle. + +The existing command card owns the command output. Therefore, a background terminal task sets `showInTranscript` to `false`. + +When the command ends, the adapter emits `async_task_state_update` with `completed` or `failed`. + +## Stop request + +The client sends `_session/async_task/stop` with the ACP session ID and async task ID: + +```json +{ + "sessionId": "thread-id", + "asyncTaskId": "command-item-id" +} +``` + +The adapter resolves the app-server process ID and calls `thread/backgroundTerminals/terminate`. It returns `{ "stopped": true }` after app-server accepts the termination. diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index a469d515..d3299b7f 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -11,6 +11,10 @@ import { LEGACY_GOAL_CONTROL_METHOD, type GoalControlRequest, } from "./GoalExtension"; +import { + ASYNC_TASK_STOP_METHOD, + type AsyncTaskStopExtRequest, +} from "./async-tasks/AsyncTaskExtension"; export { GOAL_CONTROL_ACTIONS, @@ -63,6 +67,7 @@ export type ExtMethodRequest = | LegacySetSessionModelExtRequest | SessionSteeringExtRequest | GoalControlExtRequest + | AsyncTaskStopExtRequest export function isExtMethodRequest(request: { method: string, params: Record }): request is ExtMethodRequest { return request.method === "authentication/status" @@ -70,7 +75,8 @@ export function isExtMethodRequest(request: { method: string, params: Record sessionState.asyncTasks.stop(methodRequest.params.asyncTaskId), + ), + }; + } case GOAL_CONTROL_METHOD: case LEGACY_GOAL_CONTROL_METHOD: { const sessionState = this.sessions.get(methodRequest.params.sessionId); @@ -646,6 +660,7 @@ export class CodexAcpServer { clientSupportsSubagents(this.clientCapabilities), new ACPSessionConnection(this.connection, sessionId), ), + asyncTasks: this.createAsyncTasks(sessionId), }; sessionState.titleGen = new TitleGenerator( this.codexAcpClient.appServerClient, @@ -670,6 +685,7 @@ export class CodexAcpServer { } if (operation === "resume") { this.publishCurrentGoalAsync(sessionState, sessionGeneration); + this.publishAsyncTasksAsync(sessionState, sessionGeneration); } const sessionModelState: LegacySessionModelState = this.createModelState(models, currentModelId); const sessionModeState: SessionModeState = sessionState.agentMode.toSessionModeState(); @@ -702,6 +718,15 @@ export class CodexAcpServer { return a === b; } + private createAsyncTasks(sessionId: string): CodexBackgroundTerminalTasks { + return new CodexBackgroundTerminalTasks( + clientSupportsAirCapability(this.clientCapabilities, AIR_ASYNC_TASKS_KEY), + sessionId, + this.codexAcpClient.appServerClient, + new ACPSessionConnection(this.connection, sessionId), + ); + } + private getAuthProviderForAuthenticateRequest(request: acp.AuthenticateRequest): string | null { if (isCodexAuthRequest(request) && request.methodId === "gateway") { return "custom-gateway"; @@ -802,6 +827,7 @@ export class CodexAcpServer { try { if (sessionState) { await this.interruptSessionTurn(sessionState, "Close", true); + sessionState.asyncTasks.clear(); } else { logger.log("Close request received for unknown local session", {sessionId: params.sessionId}); } @@ -1535,6 +1561,15 @@ export class CodexAcpServer { void this.publishCurrentGoalBestEffort(sessionState, sessionGeneration, true); } + private publishAsyncTasksAsync(sessionState: SessionState, sessionGeneration: number): void { + if (!this.sessionPublishIsCurrent(sessionState, sessionGeneration)) return; + void sessionState.asyncTasks.sync().catch((error) => { + if (this.sessionPublishIsCurrent(sessionState, sessionGeneration)) { + logger.error(`Failed to list background terminals for ${sessionState.sessionId}`, error); + } + }); + } + private async publishCurrentGoalBestEffort( sessionState: SessionState, sessionGeneration: number, @@ -1697,6 +1732,7 @@ export class CodexAcpServer { clientSupportsSubagents(this.clientCapabilities), new ACPSessionConnection(this.connection, sessionId), ), + asyncTasks: this.createAsyncTasks(sessionId), }; sessionState.titleGen = new TitleGenerator( this.codexAcpClient.appServerClient, @@ -1706,6 +1742,7 @@ export class CodexAcpServer { ); this.sessions.set(sessionId, sessionState); subscribed = false; + this.publishAsyncTasksAsync(sessionState, requestedSessionGeneration); if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { this.pendingMcpStartupSessions.set(sessionId, { @@ -2554,6 +2591,26 @@ export class CodexAcpServer { await this.codexAcpClient.subscribeToSessionEvents(params.sessionId, async (event) => { await observeInteraction(event); + const startedCommand = event.method === "item/started" + && event.params.threadId === sessionState.sessionId + && event.params.item.type === "commandExecution" + ? event.params.item + : null; + if (startedCommand !== null) { + sessionState.asyncTasks.observeCommandStarted(startedCommand); + } + if (event.method === "item/completed" + && event.params.threadId === sessionState.sessionId + && event.params.item.type === "commandExecution") { + await sessionState.asyncTasks.observeCommandCompleted(event.params.item); + } + if ((event.method === "item/started" && event.params.threadId === sessionState.sessionId && startedCommand === null) + || (event.method === "turn/completed" && event.params.threadId === sessionState.sessionId)) { + this.publishAsyncTasksAsync( + sessionState, + this.getSessionGeneration(sessionState.sessionId), + ); + } if (!promptNotificationsActive) { await promptEventHandler.handleSessionScopedNotification(event); return; diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index 51521928..b17fc455 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -81,6 +81,13 @@ import type { PermissionsRequestApprovalResponse, ItemCompletedNotification, } from "./app-server/v2"; +import type { + ThreadBackgroundTerminalsRequest, + ThreadBackgroundTerminalsTerminateParams, + ThreadBackgroundTerminalsTerminateResponse, + ThreadBackgroundTerminalsListParams, + ThreadBackgroundTerminalsListResponse, +} from "./async-tasks/BackgroundTerminalApi"; export interface ApprovalHandler { handleCommandExecution(params: CommandExecutionRequestApprovalParams): Promise; @@ -580,6 +587,14 @@ export class CodexAppServerClient { return await this.sendRequest({ method: "thread/compact/start", params: params }); } + async threadBackgroundTerminalsList(params: ThreadBackgroundTerminalsListParams): Promise { + return await this.sendRequest({method: "thread/backgroundTerminals/list", params}); + } + + async threadBackgroundTerminalsTerminate(params: ThreadBackgroundTerminalsTerminateParams): Promise { + return await this.sendRequest({method: "thread/backgroundTerminals/terminate", params}); + } + async threadGoalSet(params: ThreadGoalSetParams): Promise { return await this.sendRequest({ method: "thread/goal/set", params: params }); } @@ -1016,7 +1031,7 @@ export type CompactionCompletedNotification = | { method: "thread/compacted", params: Extract["params"] } | { method: "item/completed", params: ItemCompletedNotification & { item: Extract } }; -type CodexRequest = DistributiveOmit +type CodexRequest = DistributiveOmit | ThreadBackgroundTerminalsRequest type DistributiveOmit = T extends any ? Omit diff --git a/src/__tests__/CodexACPAgent/async-tasks.test.ts b/src/__tests__/CodexACPAgent/async-tasks.test.ts new file mode 100644 index 00000000..d9b96dd4 --- /dev/null +++ b/src/__tests__/CodexACPAgent/async-tasks.test.ts @@ -0,0 +1,241 @@ +import {describe, expect, it, vi} from "vitest"; +import type {ThreadItem} from "../../app-server/v2"; +import {ACPSessionConnection} from "../../ACPSessionConnection"; +import type {CodexAppServerClient} from "../../CodexAppServerClient"; +import {CodexBackgroundTerminalTasks} from "../../async-tasks/CodexBackgroundTerminalTasks"; +import {ASYNC_TASK_STOP_METHOD} from "../../async-tasks/AsyncTaskExtension"; +import type { + ThreadBackgroundTerminal, + ThreadBackgroundTerminalsListResponse, +} from "../../async-tasks/BackgroundTerminalApi"; +import { + createCodexMockTestFixture, + createTestSessionState, + setupPromptAndSendNotifications, +} from "../acp-test-utils"; + +type CommandExecutionItem = Extract; + +describe("Codex background terminal tasks", () => { + it("discovers background work from the root session event stream", async () => { + const fixture = createCodexMockTestFixture(); + await fixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: { + _meta: {jetbrains: {air: {version: 1, capabilities: ["asyncTasks"]}}}, + }, + }); + const sessionState = createTestSessionState({sessionId: "thread-1"}); + sessionState.asyncTasks = new CodexBackgroundTerminalTasks( + true, + sessionState.sessionId, + fixture.getCodexAppServerClient(), + new ACPSessionConnection(fixture.getAcpConnection(), sessionState.sessionId), + ); + // @ts-expect-error - register the local session for session-generation checks + fixture.getCodexAcpAgent().sessions.set(sessionState.sessionId, sessionState); + vi.spyOn(fixture.getCodexAppServerClient(), "threadBackgroundTerminalsList") + .mockResolvedValue(page([terminal()])); + + await setupPromptAndSendNotifications(fixture, sessionState.sessionId, sessionState, [ + started(command()), + started({type: "reasoning", id: "reasoning-1", summary: [], content: []}), + ]); + + await vi.waitFor(() => { + const taskUpdates = fixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update) + .filter(update => update.sessionUpdate === "async_task_spawned"); + expect(taskUpdates).toEqual([expect.objectContaining({ + asyncTaskId: "command-1", + toolCallId: "command-1", + })]); + }); + }); + + it("publishes a background terminal as a task linked to its command", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + + fixture.tasks.observeCommandStarted(command()); + await fixture.tasks.sync(); + + expect(fixture.updates).toEqual([{ + sessionUpdate: "async_task_spawned", + asyncTaskId: "command-1", + name: "python -m http.server", + taskType: "shell", + description: "python -m http.server", + showInTranscript: false, + canStop: true, + toolCallId: "command-1", + }]); + }); + + it("does not publish a command that completes before it becomes background work", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([])); + const item = command(); + + fixture.tasks.observeCommandStarted(item); + await fixture.tasks.observeCommandCompleted({...item, status: "completed", exitCode: 0}); + await fixture.tasks.sync(); + + expect(fixture.updates).toEqual([]); + }); + + it("publishes the terminal state after a background command exits", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + const item = command(); + + fixture.tasks.observeCommandStarted(item); + await fixture.tasks.sync(); + await fixture.tasks.observeCommandCompleted({...item, status: "failed", exitCode: 1}); + + expect(fixture.updates.at(-1)).toEqual({ + sessionUpdate: "async_task_state_update", + asyncTaskId: "command-1", + state: "failed", + toolCallId: "command-1", + }); + }); + + it("stops one task through the app-server process id", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + fixture.terminate.mockResolvedValue({terminated: true}); + await fixture.tasks.sync(); + + await expect(fixture.tasks.stop("command-1")).resolves.toBe(true); + + expect(fixture.terminate).toHaveBeenCalledWith({ + threadId: "thread-1", + processId: "42", + }); + expect(fixture.updates.at(-1)).toEqual({ + sessionUpdate: "async_task_state_update", + asyncTaskId: "command-1", + state: "stopped", + toolCallId: "command-1", + }); + await expect(fixture.tasks.stop("command-1")).resolves.toBe(false); + }); + + it("routes the AIR stop request to the session task runtime", async () => { + const fixture = createCodexMockTestFixture(); + const sessionState = createTestSessionState({sessionId: "thread-1"}); + const stop = vi.spyOn(sessionState.asyncTasks, "stop").mockResolvedValue(true); + // @ts-expect-error - register the local session for the extension request path + fixture.getCodexAcpAgent().sessions.set(sessionState.sessionId, sessionState); + + await expect(fixture.getCodexAcpAgent().extMethod(ASYNC_TASK_STOP_METHOD, { + sessionId: sessionState.sessionId, + asyncTaskId: "command-1", + })).resolves.toEqual({stopped: true}); + expect(stop).toHaveBeenCalledWith("command-1"); + }); + + it("reads every background terminal page", async () => { + const fixture = createFixture(); + fixture.list + .mockResolvedValueOnce(page([terminal()], "42")) + .mockResolvedValueOnce(page([terminal({itemId: "command-2", processId: "84"})])); + + await fixture.tasks.sync(); + + expect(fixture.list).toHaveBeenNthCalledWith(1, { + threadId: "thread-1", + cursor: null, + limit: 64, + }); + expect(fixture.list).toHaveBeenNthCalledWith(2, { + threadId: "thread-1", + cursor: "42", + limit: 64, + }); + expect(fixture.updates).toHaveLength(2); + }); + + it("does nothing when the client did not negotiate async tasks", async () => { + const fixture = createFixture(false); + + fixture.tasks.observeCommandStarted(command()); + await fixture.tasks.sync(); + await expect(fixture.tasks.stop("command-1")).resolves.toBe(false); + + expect(fixture.list).not.toHaveBeenCalled(); + expect(fixture.terminate).not.toHaveBeenCalled(); + expect(fixture.updates).toEqual([]); + }); +}); + +function createFixture(enabled = true) { + const updates: unknown[] = []; + const list = vi.fn<() => Promise>(); + const terminate = vi.fn(); + const appServer = { + threadBackgroundTerminalsList: list, + threadBackgroundTerminalsTerminate: terminate, + } as unknown as CodexAppServerClient; + const session = new ACPSessionConnection({ + notify: vi.fn(async (_method, params) => { + updates.push((params as {update: unknown}).update); + }), + request: vi.fn(), + }, "thread-1"); + return { + updates, + list, + terminate, + tasks: new CodexBackgroundTerminalTasks(enabled, "thread-1", appServer, session), + }; +} + +function terminal(overrides: Partial = {}): ThreadBackgroundTerminal { + return { + itemId: "command-1", + processId: "42", + command: "python -m http.server", + cwd: "/workspace", + osPid: null, + cpuPercent: null, + rssKb: null, + ...overrides, + }; +} + +function page(data: ThreadBackgroundTerminal[], nextCursor: string | null = null): ThreadBackgroundTerminalsListResponse { + return {data, nextCursor}; +} + +function command(): CommandExecutionItem { + return { + type: "commandExecution", + id: "command-1", + pluginId: null, + scriptPath: null, + command: "python -m http.server", + cwd: "/workspace", + processId: "42", + source: "unifiedExecStartup", + status: "inProgress", + commandActions: [], + aggregatedOutput: null, + exitCode: null, + durationMs: null, + }; +} + +function started(item: ThreadItem) { + return { + method: "item/started" as const, + params: { + threadId: "thread-1", + turnId: "turn-id", + startedAtMs: 0, + item, + }, + }; +} diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 9f8fe458..44a2596f 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -75,7 +75,7 @@ describe('CodexACPAgent - initialize', () => { jetbrains: { air: { version: 1, - capabilities: ["sessionFailure", "agentFileChangeReport", "nativeSubagentSessions"], + capabilities: ["sessionFailure", "agentFileChangeReport", "nativeSubagentSessions", "asyncTasks"], }, }, }, diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts index 5cf73d5d..b74ec042 100644 --- a/src/__tests__/acp-test-utils.ts +++ b/src/__tests__/acp-test-utils.ts @@ -15,6 +15,7 @@ import {DEFAULT_COLLABORATION_MODE} from "../CollaborationModeConfig"; import {expect, vi} from "vitest"; import type {Model, ReasoningEffortOption} from "../app-server/v2"; import {CodexSubagentEventRouter} from "../subagents/CodexSubagentEventRouter"; +import {CodexBackgroundTerminalTasks} from "../async-tasks/CodexBackgroundTerminalTasks"; export type MethodCallEvent = { method: string; args: any[] }; @@ -414,6 +415,12 @@ export function createTestSessionState(overrides?: Partial): Sessi false, new ACPSessionConnection({notify: vi.fn(), request: vi.fn()} as AcpClientConnection, sessionId), ), + asyncTasks: new CodexBackgroundTerminalTasks( + false, + sessionId, + {} as CodexAppServerClient, + new ACPSessionConnection({notify: vi.fn(), request: vi.fn()} as AcpClientConnection, sessionId), + ), ...overrides, }; } diff --git a/src/async-tasks/AcpAsyncTasks.ts b/src/async-tasks/AcpAsyncTasks.ts new file mode 100644 index 00000000..9262b14d --- /dev/null +++ b/src/async-tasks/AcpAsyncTasks.ts @@ -0,0 +1,41 @@ +export type AsyncTaskState = "running" | "paused" | "completed" | "failed" | "stopped"; + +export type AsyncTaskSpawnedUpdate = { + sessionUpdate: "async_task_spawned"; + asyncTaskId: string; + name: string; + taskType: string; + description: string; + showInTranscript: boolean; + canStop: boolean; + outputFilePath?: string; + toolCallId?: string; + _meta?: Record | null; +}; + +export type AsyncTaskProgressUpdate = { + sessionUpdate: "async_task_progress"; + asyncTaskId: string; + description?: string; + summary?: string; + lastToolName?: string; + usage?: { totalTokens: number; toolUses: number; durationMs: number }; + outputFilePath?: string; + toolCallId?: string; + _meta?: Record | null; +}; + +export type AsyncTaskStateUpdate = { + sessionUpdate: "async_task_state_update"; + asyncTaskId: string; + state: AsyncTaskState; + summary?: string; + outputFilePath?: string; + toolCallId?: string; + _meta?: Record | null; +}; + +export type AsyncTaskUpdate = + | AsyncTaskSpawnedUpdate + | AsyncTaskProgressUpdate + | AsyncTaskStateUpdate; diff --git a/src/async-tasks/AsyncTaskExtension.ts b/src/async-tasks/AsyncTaskExtension.ts new file mode 100644 index 00000000..4ed8d7ab --- /dev/null +++ b/src/async-tasks/AsyncTaskExtension.ts @@ -0,0 +1,17 @@ +import type {SessionId} from "@agentclientprotocol/sdk"; + +export const ASYNC_TASK_STOP_METHOD = "_session/async_task/stop"; + +export type AsyncTaskStopRequest = { + sessionId: SessionId; + asyncTaskId: string; +}; + +export type AsyncTaskStopResponse = { + stopped: boolean; +}; + +export type AsyncTaskStopExtRequest = { + method: typeof ASYNC_TASK_STOP_METHOD; + params: AsyncTaskStopRequest; +}; diff --git a/src/async-tasks/BackgroundTerminalApi.ts b/src/async-tasks/BackgroundTerminalApi.ts new file mode 100644 index 00000000..849936a1 --- /dev/null +++ b/src/async-tasks/BackgroundTerminalApi.ts @@ -0,0 +1,34 @@ +/** Experimental app-server API types that `generate-ts` does not export yet. */ +export type ThreadBackgroundTerminal = { + itemId: string; + processId: string; + command: string; + cwd: string; + osPid: number | null; + cpuPercent: number | null; + rssKb: number | null; +}; + +export type ThreadBackgroundTerminalsListParams = { + threadId: string; + cursor?: string | null; + limit?: number | null; +}; + +export type ThreadBackgroundTerminalsListResponse = { + data: ThreadBackgroundTerminal[]; + nextCursor: string | null; +}; + +export type ThreadBackgroundTerminalsTerminateParams = { + threadId: string; + processId: string; +}; + +export type ThreadBackgroundTerminalsTerminateResponse = { + terminated: boolean; +}; + +export type ThreadBackgroundTerminalsRequest = + | { method: "thread/backgroundTerminals/list"; params: ThreadBackgroundTerminalsListParams } + | { method: "thread/backgroundTerminals/terminate"; params: ThreadBackgroundTerminalsTerminateParams }; diff --git a/src/async-tasks/CodexBackgroundTerminalTasks.ts b/src/async-tasks/CodexBackgroundTerminalTasks.ts new file mode 100644 index 00000000..ffbf1729 --- /dev/null +++ b/src/async-tasks/CodexBackgroundTerminalTasks.ts @@ -0,0 +1,149 @@ +import type {ThreadItem} from "../app-server/v2"; +import type {ACPSessionConnection} from "../ACPSessionConnection"; +import type {CodexAppServerClient} from "../CodexAppServerClient"; +import type {ThreadBackgroundTerminal} from "./BackgroundTerminalApi"; + +type CommandExecutionItem = Extract; +type TerminalState = "completed" | "failed" | "stopped"; + +type Task = { + processId: string; + itemId: string; + command: string; + announced: boolean; + state: "running" | "stopping" | TerminalState; +}; + +/** Maps Codex-owned background terminals to the AIR async task extension. */ +export class CodexBackgroundTerminalTasks { + private readonly tasksById = new Map(); + private disposed = false; + + constructor( + readonly enabled: boolean, + private readonly threadId: string, + private readonly appServer: CodexAppServerClient, + private readonly session: ACPSessionConnection, + ) {} + + observeCommandStarted(item: CommandExecutionItem): void { + if (!this.isActive() || item.processId === null) return; + this.remember({ + itemId: item.id, + processId: item.processId, + command: item.command, + }); + } + + async observeCommandCompleted(item: CommandExecutionItem): Promise { + if (!this.isActive()) return; + await this.finish(item.id, item.status === "completed" ? "completed" : "failed"); + } + + async sync(): Promise { + if (!this.isActive()) return; + for (const terminal of await this.listAll()) { + if (!this.isActive()) return; + const task = this.remember(terminal); + if (!task.announced && task.state === "running") { + task.announced = true; + try { + await this.session.update({ + sessionUpdate: "async_task_spawned", + asyncTaskId: task.itemId, + name: task.command, + taskType: "shell", + description: task.command, + showInTranscript: false, + canStop: true, + toolCallId: task.itemId, + }); + } catch (error) { + task.announced = false; + throw error; + } + } + } + } + + async stop(taskId: string): Promise { + if (!this.isActive()) return false; + const task = this.tasksById.get(taskId); + if (!task || task.state !== "running") return false; + task.state = "stopping"; + try { + const response = await this.appServer.threadBackgroundTerminalsTerminate({ + threadId: this.threadId, + processId: task.processId, + }); + if (!response.terminated) { + if (task.state === "stopping") task.state = "running"; + return false; + } + await this.finish(taskId, "stopped"); + return true; + } catch (error) { + if (task.state === "stopping") task.state = "running"; + throw error; + } + } + + clear(): void { + this.disposed = true; + this.tasksById.clear(); + } + + private remember(terminal: Pick): Task { + const existing = this.tasksById.get(terminal.itemId); + if (existing) { + existing.processId = terminal.processId; + return existing; + } + const task: Task = { + processId: terminal.processId, + itemId: terminal.itemId, + command: terminal.command, + announced: false, + state: "running", + }; + this.tasksById.set(task.itemId, task); + return task; + } + + private async finish(taskId: string, state: TerminalState): Promise { + const task = this.tasksById.get(taskId); + if (!task || (task.state !== "running" && task.state !== "stopping")) return; + task.state = state; + if (task.announced) { + await this.session.update({ + sessionUpdate: "async_task_state_update", + asyncTaskId: task.itemId, + state, + toolCallId: task.itemId, + }); + } + } + + private async listAll(): Promise { + const terminals: ThreadBackgroundTerminal[] = []; + const seenCursors = new Set(); + let cursor: string | null = null; + do { + const response = await this.appServer.threadBackgroundTerminalsList({ + threadId: this.threadId, + cursor, + limit: 64, + }); + terminals.push(...response.data); + cursor = response.nextCursor; + if (cursor !== null && !seenCursors.add(cursor)) { + throw new Error("Codex returned a repeated background terminal cursor"); + } + } while (cursor !== null); + return terminals; + } + + private isActive(): boolean { + return this.enabled && !this.disposed; + } +} diff --git a/src/index.ts b/src/index.ts index 68df2ccd..19759300 100644 --- a/src/index.ts +++ b/src/index.ts @@ -16,6 +16,7 @@ import { GOAL_CONTROL_METHOD, LEGACY_SET_SESSION_MODEL_METHOD, SESSION_STEERING_METHOD, } from "./AcpExtensions"; +import {ASYNC_TASK_STOP_METHOD} from "./async-tasks/AsyncTaskExtension"; const emptyExtensionParamsParser = z.preprocess( (params) => params ?? {}, @@ -44,6 +45,11 @@ const goalControlParamsParser = z.discriminatedUnion("action", [ }).passthrough(), ]); +const asyncTaskStopParamsParser = z.object({ + sessionId: z.string().trim().min(1), + asyncTaskId: z.string().trim().min(1), +}).passthrough(); + if (process.argv.includes("--version")) { console.log(`${packageJson.name} ${packageJson.version}`); process.exit(0); @@ -161,6 +167,7 @@ function startAcpServer() { .onRequest("authentication/logout", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/logout", ctx.params)) .onRequest(LEGACY_SET_SESSION_MODEL_METHOD, legacySetSessionModelParamsParser, (ctx) => getAgent().extMethod(LEGACY_SET_SESSION_MODEL_METHOD, ctx.params)) .onRequest(SESSION_STEERING_METHOD, sessionSteerParamsParser, (ctx) => getAgent().extMethod(SESSION_STEERING_METHOD, ctx.params)) + .onRequest(ASYNC_TASK_STOP_METHOD, asyncTaskStopParamsParser, (ctx) => getAgent().extMethod(ASYNC_TASK_STOP_METHOD, ctx.params)) .onRequest(GOAL_CONTROL_METHOD, goalControlParamsParser, (ctx) => getAgent().extMethod(GOAL_CONTROL_METHOD, ctx.params)) .connect(acpJsonStream); } diff --git a/src/subagents/AcpSubagents.ts b/src/subagents/AcpSubagents.ts index c511b095..bb665410 100644 --- a/src/subagents/AcpSubagents.ts +++ b/src/subagents/AcpSubagents.ts @@ -7,6 +7,7 @@ import { AIR_NATIVE_SUBAGENT_SESSIONS_KEY, clientSupportsAirCapability, } from "../AirExtension"; +import type {AsyncTaskUpdate} from "../async-tasks/AcpAsyncTasks"; /** Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. */ export type SubagentSessionCapabilities = { @@ -36,7 +37,8 @@ export type SubagentStateUpdate = { export type AcpSessionUpdate = | SessionNotification["update"] | SubagentSpawnedUpdate - | SubagentStateUpdate; + | SubagentStateUpdate + | AsyncTaskUpdate; export type AcpSessionNotification = Omit & { update: AcpSessionUpdate; From a31635d1603b6d0044d3b28a14e8c3b1fcc3c7da Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Wed, 2 Sep 2026 23:53:46 +0400 Subject: [PATCH 2/6] fix: harden background terminal task lifecycle Reconcile lost terminal events and retain retryable task state. Route child tasks to their native sessions and cover lifecycle races. --- docs/async-tasks.md | 6 +- src/ACPSessionConnection.ts | 2 +- src/AcpSessionExtensions.ts | 21 ++ src/CodexAcpServer.ts | 26 +- src/CodexEventHandler.ts | 9 +- .../CodexACPAgent/async-tasks.test.ts | 306 ++++++++++++++++-- src/async-tasks/BackgroundTerminalApi.ts | 6 +- .../CodexBackgroundTerminalTasks.ts | 209 +++++++++--- src/subagents/AcpSubagents.ts | 19 -- 9 files changed, 486 insertions(+), 118 deletions(-) create mode 100644 src/AcpSessionExtensions.ts diff --git a/docs/async-tasks.md b/docs/async-tasks.md index 79f444d2..0e5ecc2f 100644 --- a/docs/async-tasks.md +++ b/docs/async-tasks.md @@ -12,12 +12,16 @@ The adapter emits no async task updates when the client does not advertise this The adapter uses `thread/backgroundTerminals/list` as the source of active processes. It maps each active process to `async_task_spawned`. -The command item ID is both the async task ID and the related tool call ID. The app-server process ID remains an internal control handle. +For a root command, the command item ID is both the async task ID and the related tool call ID. A child command prefixes its task ID with the child thread ID. The prefix keeps task IDs distinct across native subagent sessions. The related tool call ID remains the command item ID. + +The adapter publishes a child command on its native subagent session. The app-server process ID remains an internal control handle. The existing command card owns the command output. Therefore, a background terminal task sets `showInTranscript` to `false`. When the command ends, the adapter emits `async_task_state_update` with `completed` or `failed`. +The active-terminal list repairs a lost completion event. The adapter reports `stopped` when an announced terminal disappears from that list. + ## Stop request The client sends `_session/async_task/stop` with the ACP session ID and async task ID: diff --git a/src/ACPSessionConnection.ts b/src/ACPSessionConnection.ts index e29a6514..c04570ca 100644 --- a/src/ACPSessionConnection.ts +++ b/src/ACPSessionConnection.ts @@ -2,7 +2,7 @@ import * as acp from "@agentclientprotocol/sdk"; import { type AcpSessionUpdate, asSdkSessionNotification, -} from "./subagents/AcpSubagents"; +} from "./AcpSessionExtensions"; export type AcpClientConnection = Pick; diff --git a/src/AcpSessionExtensions.ts b/src/AcpSessionExtensions.ts new file mode 100644 index 00000000..8c720c16 --- /dev/null +++ b/src/AcpSessionExtensions.ts @@ -0,0 +1,21 @@ +import type {SessionNotification} from "@agentclientprotocol/sdk"; +import type {AsyncTaskUpdate} from "./async-tasks/AcpAsyncTasks"; +import type {SubagentSpawnedUpdate, SubagentStateUpdate} from "./subagents/AcpSubagents"; + +/** Session updates that are not available in the published ACP SDK yet. */ +export type AcpSessionUpdate = + | SessionNotification["update"] + | SubagentSpawnedUpdate + | SubagentStateUpdate + | AsyncTaskUpdate; + +type AcpSessionNotification = Omit & { + update: AcpSessionUpdate; +}; + +/** The only cast needed until the ACP SDK publishes the extension updates. */ +export function asSdkSessionNotification( + notification: AcpSessionNotification, +): SessionNotification { + return notification as SessionNotification; +} diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index ecb4a010..fdb38593 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1563,11 +1563,7 @@ export class CodexAcpServer { private publishAsyncTasksAsync(sessionState: SessionState, sessionGeneration: number): void { if (!this.sessionPublishIsCurrent(sessionState, sessionGeneration)) return; - void sessionState.asyncTasks.sync().catch((error) => { - if (this.sessionPublishIsCurrent(sessionState, sessionGeneration)) { - logger.error(`Failed to list background terminals for ${sessionState.sessionId}`, error); - } - }); + sessionState.asyncTasks.refresh(); } private async publishCurrentGoalBestEffort( @@ -2591,26 +2587,6 @@ export class CodexAcpServer { await this.codexAcpClient.subscribeToSessionEvents(params.sessionId, async (event) => { await observeInteraction(event); - const startedCommand = event.method === "item/started" - && event.params.threadId === sessionState.sessionId - && event.params.item.type === "commandExecution" - ? event.params.item - : null; - if (startedCommand !== null) { - sessionState.asyncTasks.observeCommandStarted(startedCommand); - } - if (event.method === "item/completed" - && event.params.threadId === sessionState.sessionId - && event.params.item.type === "commandExecution") { - await sessionState.asyncTasks.observeCommandCompleted(event.params.item); - } - if ((event.method === "item/started" && event.params.threadId === sessionState.sessionId && startedCommand === null) - || (event.method === "turn/completed" && event.params.threadId === sessionState.sessionId)) { - this.publishAsyncTasksAsync( - sessionState, - this.getSessionGeneration(sessionState.sessionId), - ); - } if (!promptNotificationsActive) { await promptEventHandler.handleSessionScopedNotification(event); return; diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index f496c387..d27de148 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -372,9 +372,14 @@ export class CodexEventHandler { for (const buffered of this.subagents.takeBufferedNotifications()) { await this.handleNotification(buffered); } - if (handledBySubagents) { - return; + // The subagent router owns child turn completion, but async tasks also use that boundary for reconciliation. + if (!handledBySubagents || notification.method === "turn/completed") { + await this.sessionState.asyncTasks.handleNotification( + notification, + this.subagents.notificationSessionId(notification), + ); } + if (handledBySubagents) return; if (this.subagents.shouldIgnore(notification)) { return; } diff --git a/src/__tests__/CodexACPAgent/async-tasks.test.ts b/src/__tests__/CodexACPAgent/async-tasks.test.ts index d9b96dd4..e840b80a 100644 --- a/src/__tests__/CodexACPAgent/async-tasks.test.ts +++ b/src/__tests__/CodexACPAgent/async-tasks.test.ts @@ -1,9 +1,10 @@ import {describe, expect, it, vi} from "vitest"; import type {ThreadItem} from "../../app-server/v2"; -import {ACPSessionConnection} from "../../ACPSessionConnection"; +import {ACPSessionConnection, type UpdateSessionEvent} from "../../ACPSessionConnection"; import type {CodexAppServerClient} from "../../CodexAppServerClient"; import {CodexBackgroundTerminalTasks} from "../../async-tasks/CodexBackgroundTerminalTasks"; import {ASYNC_TASK_STOP_METHOD} from "../../async-tasks/AsyncTaskExtension"; +import {CodexSubagentEventRouter} from "../../subagents/CodexSubagentEventRouter"; import type { ThreadBackgroundTerminal, ThreadBackgroundTerminalsListResponse, @@ -58,7 +59,7 @@ describe("Codex background terminal tasks", () => { const fixture = createFixture(); fixture.list.mockResolvedValue(page([terminal()])); - fixture.tasks.observeCommandStarted(command()); + await fixture.tasks.handleNotification(started(command()), "thread-1"); await fixture.tasks.sync(); expect(fixture.updates).toEqual([{ @@ -78,8 +79,8 @@ describe("Codex background terminal tasks", () => { fixture.list.mockResolvedValue(page([])); const item = command(); - fixture.tasks.observeCommandStarted(item); - await fixture.tasks.observeCommandCompleted({...item, status: "completed", exitCode: 0}); + await fixture.tasks.handleNotification(started(item), "thread-1"); + await fixture.tasks.handleNotification(completed({...item, status: "completed", exitCode: 0}), "thread-1"); await fixture.tasks.sync(); expect(fixture.updates).toEqual([]); @@ -90,9 +91,9 @@ describe("Codex background terminal tasks", () => { fixture.list.mockResolvedValue(page([terminal()])); const item = command(); - fixture.tasks.observeCommandStarted(item); + await fixture.tasks.handleNotification(started(item), "thread-1"); await fixture.tasks.sync(); - await fixture.tasks.observeCommandCompleted({...item, status: "failed", exitCode: 1}); + await fixture.tasks.handleNotification(completed({...item, status: "failed", exitCode: 1}), "thread-1"); expect(fixture.updates.at(-1)).toEqual({ sessionUpdate: "async_task_state_update", @@ -102,6 +103,44 @@ describe("Codex background terminal tasks", () => { }); }); + it("retries a terminal update that the client rejected", async () => { + let rejectTerminalUpdate = true; + const fixture = createFixture(true, async update => { + if (update.sessionUpdate === "async_task_state_update" && rejectTerminalUpdate) { + rejectTerminalUpdate = false; + throw new Error("client disconnected"); + } + }); + fixture.list.mockResolvedValue(page([terminal()])); + const completedItem = {...command(), status: "completed" as const, exitCode: 0}; + await fixture.tasks.sync(); + + await expect(fixture.tasks.handleNotification(completed(completedItem), "thread-1")) + .rejects.toThrow("client disconnected"); + await expect(fixture.tasks.handleNotification(completed(completedItem), "thread-1")) + .resolves.toBeUndefined(); + + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_state_update")) + .toEqual([expect.objectContaining({state: "completed"})]); + }); + + it("finishes an announced task that disappears from the live terminal list", async () => { + const fixture = createFixture(); + fixture.list + .mockResolvedValueOnce(page([terminal()])) + .mockResolvedValueOnce(page([])); + + await fixture.tasks.sync(); + await fixture.tasks.sync(); + + expect(fixture.updates.at(-1)).toEqual({ + sessionUpdate: "async_task_state_update", + asyncTaskId: "command-1", + state: "stopped", + toolCallId: "command-1", + }); + }); + it("stops one task through the app-server process id", async () => { const fixture = createFixture(); fixture.list.mockResolvedValue(page([terminal()])); @@ -123,6 +162,61 @@ describe("Codex background terminal tasks", () => { await expect(fixture.tasks.stop("command-1")).resolves.toBe(false); }); + it("keeps a task stoppable when termination is rejected or fails", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + fixture.terminate + .mockResolvedValueOnce({terminated: false}) + .mockRejectedValueOnce(new Error("terminate failed")) + .mockResolvedValueOnce({terminated: true}); + await fixture.tasks.sync(); + + await expect(fixture.tasks.stop("command-1")).resolves.toBe(false); + await expect(fixture.tasks.stop("command-1")).rejects.toThrow("terminate failed"); + await expect(fixture.tasks.stop("command-1")).resolves.toBe(true); + + expect(fixture.terminate).toHaveBeenCalledTimes(3); + }); + + it("keeps a task stoppable when its stopped update fails", async () => { + let rejectStoppedUpdate = true; + const fixture = createFixture(true, async update => { + if (update.sessionUpdate === "async_task_state_update" && rejectStoppedUpdate) { + rejectStoppedUpdate = false; + throw new Error("client disconnected"); + } + }); + fixture.list.mockResolvedValue(page([terminal()])); + fixture.terminate.mockResolvedValue({terminated: true}); + await fixture.tasks.sync(); + + await expect(fixture.tasks.stop("command-1")).rejects.toThrow("client disconnected"); + await expect(fixture.tasks.stop("command-1")).resolves.toBe(true); + + expect(fixture.terminate).toHaveBeenCalledTimes(2); + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_state_update")) + .toEqual([expect.objectContaining({state: "stopped"})]); + }); + + it("does not overwrite completion that races with a stop request", async () => { + const fixture = createFixture(); + const termination = deferred<{terminated: boolean}>(); + fixture.list.mockResolvedValue(page([terminal()])); + fixture.terminate.mockReturnValue(termination.promise); + await fixture.tasks.sync(); + + const stop = fixture.tasks.stop("command-1"); + await fixture.tasks.handleNotification( + completed({...command(), status: "completed", exitCode: 0}), + "thread-1", + ); + termination.resolve({terminated: true}); + + await expect(stop).resolves.toBe(true); + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_state_update")) + .toEqual([expect.objectContaining({state: "completed"})]); + }); + it("routes the AIR stop request to the session task runtime", async () => { const fixture = createCodexMockTestFixture(); const sessionState = createTestSessionState({sessionId: "thread-1"}); @@ -148,20 +242,131 @@ describe("Codex background terminal tasks", () => { expect(fixture.list).toHaveBeenNthCalledWith(1, { threadId: "thread-1", cursor: null, - limit: 64, }); expect(fixture.list).toHaveBeenNthCalledWith(2, { threadId: "thread-1", cursor: "42", - limit: 64, }); expect(fixture.updates).toHaveLength(2); }); + it("rejects a repeated background terminal cursor", async () => { + const fixture = createFixture(); + fixture.list + .mockResolvedValueOnce(page([], "repeated")) + .mockResolvedValueOnce(page([], "repeated")); + + await expect(fixture.tasks.sync()).rejects.toThrow("repeated background terminal cursor"); + + expect(fixture.list).toHaveBeenCalledTimes(2); + }); + + it("runs a trailing refresh when a list request is already in flight", async () => { + const fixture = createFixture(); + const listing = deferred(); + fixture.list + .mockReturnValueOnce(listing.promise) + .mockResolvedValueOnce(page([terminal()])); + + const first = fixture.tasks.sync(); + const second = fixture.tasks.sync(); + listing.resolve(page([terminal()])); + await Promise.all([first, second]); + + expect(fixture.list).toHaveBeenCalledTimes(2); + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_spawned")).toHaveLength(1); + }); + + it("does not publish an in-flight list result after clear", async () => { + const fixture = createFixture(); + const listing = deferred(); + fixture.list.mockReturnValue(listing.promise); + + const sync = fixture.tasks.sync(); + fixture.tasks.clear(); + listing.resolve(page([terminal()])); + await sync; + + expect(fixture.updates).toEqual([]); + }); + + it("publishes a child terminal on its native subagent session", async () => { + const fixture = createCodexMockTestFixture(); + await fixture.getCodexAcpAgent().initialize({ + protocolVersion: 1, + clientCapabilities: { + _meta: {jetbrains: {air: {version: 1, capabilities: ["asyncTasks", "nativeSubagentSessions"]}}}, + }, + }); + const sessionState = createTestSessionState({sessionId: "thread-1"}); + const session = new ACPSessionConnection(fixture.getAcpConnection(), sessionState.sessionId); + sessionState.subagents = new CodexSubagentEventRouter(sessionState.sessionId, true, session); + sessionState.asyncTasks = new CodexBackgroundTerminalTasks( + true, + sessionState.sessionId, + fixture.getCodexAppServerClient(), + session, + ); + // @ts-expect-error - register the local session for session-generation checks + fixture.getCodexAcpAgent().sessions.set(sessionState.sessionId, sessionState); + let childListCount = 0; + vi.spyOn(fixture.getCodexAppServerClient(), "threadBackgroundTerminalsList") + .mockImplementation(async params => page( + params.threadId === "child-1" && childListCount++ === 0 ? [terminal()] : [], + )); + + await setupPromptAndSendNotifications(fixture, sessionState.sessionId, sessionState, [ + childSpawned(), + childMaterialized(), + started(command(), "child-1"), + started({type: "reasoning", id: "reasoning-1", summary: [], content: []}, "child-1"), + turnCompleted("child-1"), + ]); + + await vi.waitFor(() => { + const spawned = fixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0]) + .find(event => event.update.sessionUpdate === "async_task_spawned"); + expect(spawned).toMatchObject({ + sessionId: "child-1", + update: {asyncTaskId: "child-1:command-1", toolCallId: "command-1"}, + }); + const stopped = fixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0]) + .find(event => event.update.sessionUpdate === "async_task_state_update"); + expect(stopped).toMatchObject({ + sessionId: "child-1", + update: {asyncTaskId: "child-1:command-1", state: "stopped"}, + }); + }); + + const terminate = vi.spyOn(fixture.getCodexAppServerClient(), "threadBackgroundTerminalsTerminate") + .mockResolvedValue({terminated: true}); + await expect(fixture.getCodexAcpAgent().extMethod(ASYNC_TASK_STOP_METHOD, { + sessionId: sessionState.sessionId, + asyncTaskId: "child-1:command-1", + })).resolves.toEqual({stopped: false}); + expect(terminate).not.toHaveBeenCalled(); + }); + + it("routes a child task stop to its owning Codex thread", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + fixture.terminate.mockResolvedValue({terminated: true}); + await fixture.tasks.handleNotification(started(command(), "child-1"), "child-1"); + await fixture.tasks.sync("child-1", "child-1"); + + await expect(fixture.tasks.stop("child-1:command-1")).resolves.toBe(true); + + expect(fixture.terminate).toHaveBeenCalledWith({threadId: "child-1", processId: "42"}); + }); + it("does nothing when the client did not negotiate async tasks", async () => { const fixture = createFixture(false); - fixture.tasks.observeCommandStarted(command()); + await fixture.tasks.handleNotification(started(command()), "thread-1"); await fixture.tasks.sync(); await expect(fixture.tasks.stop("command-1")).resolves.toBe(false); @@ -171,8 +376,11 @@ describe("Codex background terminal tasks", () => { }); }); -function createFixture(enabled = true) { - const updates: unknown[] = []; +function createFixture( + enabled = true, + beforeUpdate?: (update: UpdateSessionEvent) => void | Promise, +) { + const updates: UpdateSessionEvent[] = []; const list = vi.fn<() => Promise>(); const terminate = vi.fn(); const appServer = { @@ -181,7 +389,9 @@ function createFixture(enabled = true) { } as unknown as CodexAppServerClient; const session = new ACPSessionConnection({ notify: vi.fn(async (_method, params) => { - updates.push((params as {update: unknown}).update); + const update = (params as {update: UpdateSessionEvent}).update; + await beforeUpdate?.(update); + updates.push(update); }), request: vi.fn(), }, "thread-1"); @@ -198,10 +408,6 @@ function terminal(overrides: Partial = {}): ThreadBack itemId: "command-1", processId: "42", command: "python -m http.server", - cwd: "/workspace", - osPid: null, - cpuPercent: null, - rssKb: null, ...overrides, }; } @@ -228,14 +434,78 @@ function command(): CommandExecutionItem { }; } -function started(item: ThreadItem) { +function started(item: ThreadItem, threadId = "thread-1") { return { method: "item/started" as const, params: { - threadId: "thread-1", + threadId, turnId: "turn-id", startedAtMs: 0, item, }, }; } + +function completed(item: ThreadItem, threadId = "thread-1") { + return { + method: "item/completed" as const, + params: { + threadId, + turnId: "turn-id", + completedAtMs: 0, + item, + }, + }; +} + +function turnCompleted(threadId: string) { + return { + method: "turn/completed" as const, + params: { + threadId, + turn: { + id: "turn-id", + items: [], + itemsView: "notLoaded" as const, + status: "completed" as const, + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }, + }; +} + +function childSpawned() { + return started({ + type: "collabAgentToolCall", + id: "spawn-child", + tool: "spawnAgent", + status: "inProgress", + senderThreadId: "thread-1", + receiverThreadIds: ["child-1"], + prompt: "Run a server", + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status: "running", message: null}}, + }); +} + +function childMaterialized() { + return started({ + type: "subAgentActivity", + id: "child-activity", + kind: "started", + agentThreadId: "child-1", + agentPath: "/root/worker", + }); +} + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise(innerResolve => { + resolve = innerResolve; + }); + return {promise, resolve}; +} diff --git a/src/async-tasks/BackgroundTerminalApi.ts b/src/async-tasks/BackgroundTerminalApi.ts index 849936a1..d3238c2f 100644 --- a/src/async-tasks/BackgroundTerminalApi.ts +++ b/src/async-tasks/BackgroundTerminalApi.ts @@ -1,12 +1,8 @@ -/** Experimental app-server API types that `generate-ts` does not export yet. */ +/** The fields used from an API that stable `generate-ts` output omits. */ export type ThreadBackgroundTerminal = { itemId: string; processId: string; command: string; - cwd: string; - osPid: number | null; - cpuPercent: number | null; - rssKb: number | null; }; export type ThreadBackgroundTerminalsListParams = { diff --git a/src/async-tasks/CodexBackgroundTerminalTasks.ts b/src/async-tasks/CodexBackgroundTerminalTasks.ts index ffbf1729..90d00b84 100644 --- a/src/async-tasks/CodexBackgroundTerminalTasks.ts +++ b/src/async-tasks/CodexBackgroundTerminalTasks.ts @@ -1,12 +1,17 @@ +import type {ServerNotification} from "../app-server"; import type {ThreadItem} from "../app-server/v2"; import type {ACPSessionConnection} from "../ACPSessionConnection"; import type {CodexAppServerClient} from "../CodexAppServerClient"; +import {logger} from "../Logger"; import type {ThreadBackgroundTerminal} from "./BackgroundTerminalApi"; type CommandExecutionItem = Extract; type TerminalState = "completed" | "failed" | "stopped"; type Task = { + threadId: string; + sessionId: string; + asyncTaskId: string; processId: string; itemId: string; command: string; @@ -14,73 +19,110 @@ type Task = { state: "running" | "stopping" | TerminalState; }; +type PendingSync = { + requested: boolean; + sessionId: string; + promise: Promise; +}; + /** Maps Codex-owned background terminals to the AIR async task extension. */ export class CodexBackgroundTerminalTasks { - private readonly tasksById = new Map(); + private readonly tasks = new Map(); + private readonly syncs = new Map(); private disposed = false; constructor( readonly enabled: boolean, - private readonly threadId: string, + private readonly rootSessionId: string, private readonly appServer: CodexAppServerClient, private readonly session: ACPSessionConnection, ) {} - observeCommandStarted(item: CommandExecutionItem): void { + async handleNotification(notification: ServerNotification, sessionId: string): Promise { + if (!this.isActive()) return; + const threadId = notificationThreadId(notification); + if (threadId === null) return; + + if (notification.method === "item/started" && notification.params.item.type === "commandExecution") { + this.observeCommandStarted(notification.params.item, threadId, sessionId); + return; + } + if (notification.method === "item/completed" && notification.params.item.type === "commandExecution") { + await this.observeCommandCompleted(notification.params.item, threadId); + return; + } + if (notification.method === "item/started" || notification.method === "turn/completed") { + this.refresh(threadId, sessionId); + } + } + + private observeCommandStarted( + item: CommandExecutionItem, + threadId: string, + sessionId: string, + ): void { if (!this.isActive() || item.processId === null) return; - this.remember({ + this.remember(threadId, sessionId, { itemId: item.id, processId: item.processId, command: item.command, }); } - async observeCommandCompleted(item: CommandExecutionItem): Promise { + private async observeCommandCompleted( + item: CommandExecutionItem, + threadId: string, + ): Promise { if (!this.isActive()) return; - await this.finish(item.id, item.status === "completed" ? "completed" : "failed"); + const task = this.tasks.get(wireTaskId(this.rootSessionId, threadId, item.id)); + if (task) await this.finish(task, item.status === "completed" ? "completed" : "failed"); + } + + refresh(threadId: string = this.rootSessionId, sessionId: string = this.rootSessionId): void { + void this.sync(threadId, sessionId).catch((error) => { + if (this.isActive()) logger.error(`Failed to list background terminals for ${threadId}`, error); + }); } - async sync(): Promise { + async sync( + threadId: string = this.rootSessionId, + sessionId: string = this.rootSessionId, + ): Promise { if (!this.isActive()) return; - for (const terminal of await this.listAll()) { - if (!this.isActive()) return; - const task = this.remember(terminal); - if (!task.announced && task.state === "running") { - task.announced = true; - try { - await this.session.update({ - sessionUpdate: "async_task_spawned", - asyncTaskId: task.itemId, - name: task.command, - taskType: "shell", - description: task.command, - showInTranscript: false, - canStop: true, - toolCallId: task.itemId, - }); - } catch (error) { - task.announced = false; - throw error; - } - } + const current = this.syncs.get(threadId); + if (current) { + current.requested = true; + current.sessionId = sessionId; + return await current.promise; } + + const pending: PendingSync = { + requested: false, + sessionId, + promise: Promise.resolve(), + }; + pending.promise = this.syncUntilCurrent(threadId, pending).finally(() => { + if (this.syncs.get(threadId) === pending) this.syncs.delete(threadId); + }); + this.syncs.set(threadId, pending); + await pending.promise; } async stop(taskId: string): Promise { if (!this.isActive()) return false; - const task = this.tasksById.get(taskId); - if (!task || task.state !== "running") return false; + const task = this.tasks.get(taskId); + if (!task || !task.announced || task.state !== "running") return false; task.state = "stopping"; try { const response = await this.appServer.threadBackgroundTerminalsTerminate({ - threadId: this.threadId, + threadId: task.threadId, processId: task.processId, }); if (!response.terminated) { if (task.state === "stopping") task.state = "running"; return false; } - await this.finish(taskId, "stopped"); + await this.finish(task, "stopped"); return true; } catch (error) { if (task.state === "stopping") task.state = "running"; @@ -90,54 +132,118 @@ export class CodexBackgroundTerminalTasks { clear(): void { this.disposed = true; - this.tasksById.clear(); + this.tasks.clear(); + this.syncs.clear(); + } + + private async syncThread(threadId: string, sessionId: string): Promise { + const terminals = await this.listAll(threadId); + if (!this.isActive()) return; + + const liveTaskIds = new Set(); + for (const terminal of terminals) { + if (!this.isActive()) return; + liveTaskIds.add(terminal.itemId); + const task = this.remember(threadId, sessionId, terminal); + if (!task.announced && task.state === "running") await this.announce(task); + } + + for (const task of this.tasks.values()) { + if (task.threadId === threadId + && task.announced + && (task.state === "running" || task.state === "stopping") + && !liveTaskIds.has(task.itemId)) { + await this.finish(task, "stopped"); + } + } + } + + private async syncUntilCurrent(threadId: string, pending: PendingSync): Promise { + do { + pending.requested = false; + await this.syncThread(threadId, pending.sessionId); + } while (pending.requested && this.isActive()); + } + + private async announce(task: Task): Promise { + task.announced = true; + try { + await this.session.update({ + sessionUpdate: "async_task_spawned", + asyncTaskId: task.asyncTaskId, + name: task.command, + taskType: "shell", + description: task.command, + showInTranscript: false, + canStop: true, + toolCallId: task.itemId, + }, task.sessionId); + } catch (error) { + task.announced = false; + throw error; + } } - private remember(terminal: Pick): Task { - const existing = this.tasksById.get(terminal.itemId); + private remember( + threadId: string, + sessionId: string, + terminal: ThreadBackgroundTerminal, + ): Task { + const asyncTaskId = wireTaskId(this.rootSessionId, threadId, terminal.itemId); + const existing = this.tasks.get(asyncTaskId); if (existing) { existing.processId = terminal.processId; return existing; } const task: Task = { + threadId, + sessionId, + asyncTaskId, processId: terminal.processId, itemId: terminal.itemId, command: terminal.command, announced: false, state: "running", }; - this.tasksById.set(task.itemId, task); + this.tasks.set(asyncTaskId, task); return task; } - private async finish(taskId: string, state: TerminalState): Promise { - const task = this.tasksById.get(taskId); - if (!task || (task.state !== "running" && task.state !== "stopping")) return; + private async finish(task: Task, state: TerminalState): Promise { + if (task.state !== "running" && task.state !== "stopping") return; + const previousState = task.state; task.state = state; - if (task.announced) { + if (!task.announced) return; + + try { await this.session.update({ sessionUpdate: "async_task_state_update", - asyncTaskId: task.itemId, + asyncTaskId: task.asyncTaskId, state, toolCallId: task.itemId, - }); + }, task.sessionId); + } catch (error) { + if (task.state === state) task.state = previousState; + throw error; } } - private async listAll(): Promise { + private async listAll(threadId: string): Promise { const terminals: ThreadBackgroundTerminal[] = []; const seenCursors = new Set(); let cursor: string | null = null; do { const response = await this.appServer.threadBackgroundTerminalsList({ - threadId: this.threadId, + threadId, cursor, - limit: 64, }); terminals.push(...response.data); cursor = response.nextCursor; - if (cursor !== null && !seenCursors.add(cursor)) { - throw new Error("Codex returned a repeated background terminal cursor"); + if (cursor !== null) { + if (seenCursors.has(cursor)) { + throw new Error("Codex returned a repeated background terminal cursor"); + } + seenCursors.add(cursor); } } while (cursor !== null); return terminals; @@ -147,3 +253,12 @@ export class CodexBackgroundTerminalTasks { return this.enabled && !this.disposed; } } + +function wireTaskId(rootSessionId: string, threadId: string, itemId: string): string { + return threadId === rootSessionId ? itemId : `${threadId}:${itemId}`; +} + +function notificationThreadId(notification: ServerNotification): string | null { + const threadId = (notification.params as {threadId?: unknown}).threadId; + return typeof threadId === "string" ? threadId : null; +} diff --git a/src/subagents/AcpSubagents.ts b/src/subagents/AcpSubagents.ts index bb665410..0b56196a 100644 --- a/src/subagents/AcpSubagents.ts +++ b/src/subagents/AcpSubagents.ts @@ -1,13 +1,11 @@ import type { ClientCapabilities, SessionCapabilities, - SessionNotification, } from "@agentclientprotocol/sdk"; import { AIR_NATIVE_SUBAGENT_SESSIONS_KEY, clientSupportsAirCapability, } from "../AirExtension"; -import type {AsyncTaskUpdate} from "../async-tasks/AcpAsyncTasks"; /** Temporary typed surface for agentclientprotocol/agent-client-protocol#1992. */ export type SubagentSessionCapabilities = { @@ -34,16 +32,6 @@ export type SubagentStateUpdate = { _meta?: Record | null; }; -export type AcpSessionUpdate = - | SessionNotification["update"] - | SubagentSpawnedUpdate - | SubagentStateUpdate - | AsyncTaskUpdate; - -export type AcpSessionNotification = Omit & { - update: AcpSessionUpdate; -}; - export type SubagentAwareSessionCapabilities = SessionCapabilities & { subagents?: Record; }; @@ -60,10 +48,3 @@ export function clientSupportsSubagents( return clientSupportsAirCapability(capabilities, AIR_NATIVE_SUBAGENT_SESSIONS_KEY); } - -/** The only cast needed until the TypeScript SDK publishes PR #1992. */ -export function asSdkSessionNotification( - notification: AcpSessionNotification, -): SessionNotification { - return notification as SessionNotification; -} From fe036b87328617c778e16262247075353208cd56 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 3 Sep 2026 12:48:58 +0400 Subject: [PATCH 3/6] fix: complete background terminal lifecycle Keep task control valid after app-server replacement. Restore child tasks after history replay and publish task updates before child completion. Retry failed reconciliation and close unfinished tasks after process exit. --- docs/async-tasks.md | 6 + src/AirExtension.ts | 1 + src/CodexAcpServer.ts | 43 +++++++- src/CodexEventHandler.ts | 14 ++- .../CodexACPAgent/async-tasks.test.ts | 103 ++++++++++++++---- .../CodexACPAgent/load-session.test.ts | 47 ++++++-- src/__tests__/CodexACPAgent/providers.test.ts | 10 +- .../CodexBackgroundTerminalTasks.ts | 78 ++++++++++++- 8 files changed, 255 insertions(+), 47 deletions(-) diff --git a/docs/async-tasks.md b/docs/async-tasks.md index 0e5ecc2f..88f0e98a 100644 --- a/docs/async-tasks.md +++ b/docs/async-tasks.md @@ -12,6 +12,8 @@ The adapter emits no async task updates when the client does not advertise this The adapter uses `thread/backgroundTerminals/list` as the source of active processes. It maps each active process to `async_task_spawned`. +Before the spawn update, the adapter marks the command with `_meta.jetbrains.air.asyncTasks.backgrounded`. AIR can then keep the command card active without duplicating its output. + For a root command, the command item ID is both the async task ID and the related tool call ID. A child command prefixes its task ID with the child thread ID. The prefix keeps task IDs distinct across native subagent sessions. The related tool call ID remains the command item ID. The adapter publishes a child command on its native subagent session. The app-server process ID remains an internal control handle. @@ -22,6 +24,10 @@ When the command ends, the adapter emits `async_task_state_update` with `complet The active-terminal list repairs a lost completion event. The adapter reports `stopped` when an announced terminal disappears from that list. +Session loading restores root and child tasks after it replays their command history. A provider restart stops old tasks and moves task control to the new app-server client. + +If the app-server exits, the adapter reports each unfinished task as `failed`. + ## Stop request The client sends `_session/async_task/stop` with the ACP session ID and async task ID: diff --git a/src/AirExtension.ts b/src/AirExtension.ts index 5ef7eaa9..03b12749 100644 --- a/src/AirExtension.ts +++ b/src/AirExtension.ts @@ -16,6 +16,7 @@ export const AIR_SESSION_FAILURE_KEY = "sessionFailure"; export const AIR_AGENT_FILE_CHANGE_REPORT_KEY = "agentFileChangeReport"; export const AIR_NATIVE_SUBAGENT_SESSIONS_KEY = "nativeSubagentSessions"; export const AIR_ASYNC_TASKS_KEY = "asyncTasks"; +export const AIR_ASYNC_TASKS_BACKGROUNDED_KEY = "backgrounded"; export const AIR_AGENT_FILE_CHANGE_REPORT_REQUEST_KEY = "agentFileChangeReportRequest"; export const AIR_EXTENSION_VERSION = 1; diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index fdb38593..8f296acc 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -391,6 +391,9 @@ export class CodexAcpServer { case SESSION_STEERING_METHOD: return await this.executeOrQueueSteeringRequest(this.parseSessionSteerParams(methodRequest.params)); case ASYNC_TASK_STOP_METHOD: { + if (this.providerUpdate !== null) { + await this.providerUpdate; + } const sessionState = this.sessions.get(methodRequest.params.sessionId); if (!sessionState) return {stopped: false}; return { @@ -668,7 +671,7 @@ export class CodexAcpServer { sessionState.cwd, () => sessionState.sessionTitleSource, ); - this.sessions.set(sessionId, sessionState); + this.installSessionState(sessionState); resumeSubscribed = false; const canPublishSessionUpdates = operation !== "fork"; @@ -727,6 +730,11 @@ export class CodexAcpServer { ); } + private installSessionState(sessionState: SessionState): void { + this.sessions.get(sessionState.sessionId)?.asyncTasks.clear(); + this.sessions.set(sessionState.sessionId, sessionState); + } + private getAuthProviderForAuthenticateRequest(request: acp.AuthenticateRequest): string | null { if (isCodexAuthRequest(request) && request.methodId === "gateway") { return "custom-gateway"; @@ -747,6 +755,7 @@ export class CodexAcpServer { } = await this.getOrCreateSessionWithHistory(params); await this.streamThreadHistory(sessionId, thread); + await this.getSessionState(sessionId).asyncTasks.reconcile(); logger.log("Session loaded", { sessionId: sessionId, @@ -994,6 +1003,7 @@ export class CodexAcpServer { } logger.log("Restarting Codex app-server for provider update", {sessionCount: this.sessions.size}); + await this.finishAllAsyncTasks("stopped", "before the provider restart"); const replacement = await this.restartCodexClient(); apply(replacement); if (this.initializeRequest === null) { @@ -1005,6 +1015,7 @@ export class CodexAcpServer { const resumeErrors: unknown[] = []; for (const session of this.sessions.values()) { + session.asyncTasks.setAppServer(replacement.appServerClient); try { await replacement.resumeSession({ sessionId: session.sessionId, @@ -1013,6 +1024,7 @@ export class CodexAcpServer { mcpServers: session.mcpServers ?? [], }); session.authProvider = replacement.getModelProvider(); + session.asyncTasks.refresh(); logger.log("Resumed session after provider restart", {sessionId: session.sessionId}); } catch (error) { resumeErrors.push(error); @@ -1736,9 +1748,8 @@ export class CodexAcpServer { sessionState.cwd, () => sessionState.sessionTitleSource, ); - this.sessions.set(sessionId, sessionState); + this.installSessionState(sessionState); subscribed = false; - this.publishAsyncTasksAsync(sessionState, requestedSessionGeneration); if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { this.pendingMcpStartupSessions.set(sessionId, { @@ -1847,6 +1858,15 @@ export class CodexAcpServer { new Set([...ancestry, item.agentThreadId]), threadCache, ); + try { + await sessionState.asyncTasks.recover( + item.agentThreadId, + childSessionId, + commandItemIds(childTurn.items), + ); + } catch (error) { + logger.error(`Failed to restore background terminals for ${item.agentThreadId}`, error); + } } } } @@ -3080,6 +3100,7 @@ export class CodexAcpServer { throw new RequestError(requestErrorCode, `VC++ redistributable should be installed`); } if (exitCode !== null) { + await this.finishAllAsyncTasks("failed", "after the Codex process exited"); const stderr = this.getRecentStderr().trim(); const detail = stderr ? `:\n${stderr}` : ""; throw new RequestError(requestErrorCode, `Codex process has exited with code ${exitCode}${detail}`); @@ -3088,6 +3109,16 @@ export class CodexAcpServer { } } + private async finishAllAsyncTasks(state: "failed" | "stopped", reason: string): Promise { + for (const session of this.sessions.values()) { + try { + await session.asyncTasks.finishAll(state); + } catch (error) { + logger.error(`Failed to finish background terminal tasks ${reason}`, error); + } + } + } + async cancel(params: acp.CancelNotification): Promise { const sessionState = this.sessions.get(params.sessionId); if (!sessionState) { @@ -3157,6 +3188,12 @@ function mergeHistoryUpdates( return merged; } +function commandItemIds(items: ThreadItem[]): Set { + return new Set(items + .filter((item): item is Extract => item.type === "commandExecution") + .map(item => item.id)); +} + function historyUpdateKey(update: UpdateSessionEvent): string | null { switch (update.sessionUpdate) { case "user_message_chunk": diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index d27de148..fb0e524f 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -368,16 +368,18 @@ export class CodexEventHandler { async handleNotification(notification: ServerNotification) { await this.flushPendingErrors(); + const asyncTaskSessionId = this.subagents.notificationSessionId(notification); + const handledAsyncTasksFirst = notification.method === "turn/completed" + && asyncTaskSessionId !== this.sessionState.sessionId; + if (handledAsyncTasksFirst) { + await this.sessionState.asyncTasks.handleNotification(notification, asyncTaskSessionId); + } const handledBySubagents = await this.subagents.handle(notification); for (const buffered of this.subagents.takeBufferedNotifications()) { await this.handleNotification(buffered); } - // The subagent router owns child turn completion, but async tasks also use that boundary for reconciliation. - if (!handledBySubagents || notification.method === "turn/completed") { - await this.sessionState.asyncTasks.handleNotification( - notification, - this.subagents.notificationSessionId(notification), - ); + if (!handledAsyncTasksFirst && !handledBySubagents) { + await this.sessionState.asyncTasks.handleNotification(notification, asyncTaskSessionId); } if (handledBySubagents) return; if (this.subagents.shouldIgnore(notification)) { diff --git a/src/__tests__/CodexACPAgent/async-tasks.test.ts b/src/__tests__/CodexACPAgent/async-tasks.test.ts index e840b80a..57721aaf 100644 --- a/src/__tests__/CodexACPAgent/async-tasks.test.ts +++ b/src/__tests__/CodexACPAgent/async-tasks.test.ts @@ -62,16 +62,23 @@ describe("Codex background terminal tasks", () => { await fixture.tasks.handleNotification(started(command()), "thread-1"); await fixture.tasks.sync(); - expect(fixture.updates).toEqual([{ - sessionUpdate: "async_task_spawned", - asyncTaskId: "command-1", - name: "python -m http.server", - taskType: "shell", - description: "python -m http.server", - showInTranscript: false, - canStop: true, - toolCallId: "command-1", - }]); + expect(fixture.updates).toEqual([ + { + sessionUpdate: "tool_call_update", + toolCallId: "command-1", + _meta: {jetbrains: {air: {asyncTasks: {backgrounded: true}}}}, + }, + { + sessionUpdate: "async_task_spawned", + asyncTaskId: "command-1", + name: "python -m http.server", + taskType: "shell", + description: "python -m http.server", + showInTranscript: false, + canStop: true, + toolCallId: "command-1", + }, + ]); }); it("does not publish a command that completes before it becomes background work", async () => { @@ -162,6 +169,38 @@ describe("Codex background terminal tasks", () => { await expect(fixture.tasks.stop("command-1")).resolves.toBe(false); }); + it("uses the replacement app-server client for stop requests", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + await fixture.tasks.sync(); + const replacementTerminate = vi.fn().mockResolvedValue({terminated: true}); + fixture.tasks.setAppServer({ + threadBackgroundTerminalsTerminate: replacementTerminate, + } as unknown as CodexAppServerClient); + + await expect(fixture.tasks.stop("command-1")).resolves.toBe(true); + + expect(fixture.terminate).not.toHaveBeenCalled(); + expect(replacementTerminate).toHaveBeenCalledWith({threadId: "thread-1", processId: "42"}); + }); + + it("fails every announced task after the app-server exits", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([ + terminal(), + terminal({itemId: "command-2", processId: "84"}), + ])); + await fixture.tasks.sync(); + + await fixture.tasks.finishAll("failed"); + + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_state_update")) + .toEqual([ + expect.objectContaining({asyncTaskId: "command-1", state: "failed"}), + expect.objectContaining({asyncTaskId: "command-2", state: "failed"}), + ]); + }); + it("keeps a task stoppable when termination is rejected or fails", async () => { const fixture = createFixture(); fixture.list.mockResolvedValue(page([terminal()])); @@ -247,7 +286,7 @@ describe("Codex background terminal tasks", () => { threadId: "thread-1", cursor: "42", }); - expect(fixture.updates).toHaveLength(2); + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_spawned")).toHaveLength(2); }); it("rejects a repeated background terminal cursor", async () => { @@ -277,6 +316,22 @@ describe("Codex background terminal tasks", () => { expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_spawned")).toHaveLength(1); }); + it("runs a trailing refresh after the in-flight request fails", async () => { + const fixture = createFixture(); + const listing = deferred(); + fixture.list + .mockReturnValueOnce(listing.promise) + .mockResolvedValueOnce(page([terminal()])); + + const first = fixture.tasks.sync(); + const second = fixture.tasks.sync(); + listing.reject(new Error("temporary list failure")); + await Promise.all([first, second]); + + expect(fixture.list).toHaveBeenCalledTimes(2); + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_spawned")).toHaveLength(1); + }); + it("does not publish an in-flight list result after clear", async () => { const fixture = createFixture(); const listing = deferred(); @@ -319,7 +374,6 @@ describe("Codex background terminal tasks", () => { childSpawned(), childMaterialized(), started(command(), "child-1"), - started({type: "reasoning", id: "reasoning-1", summary: [], content: []}, "child-1"), turnCompleted("child-1"), ]); @@ -332,14 +386,17 @@ describe("Codex background terminal tasks", () => { sessionId: "child-1", update: {asyncTaskId: "child-1:command-1", toolCallId: "command-1"}, }); - const stopped = fixture.getAcpConnectionEvents([]) + const childTerminalIndex = fixture.getAcpConnectionEvents([]) .filter(event => event.method === "sessionUpdate") .map(event => event.args[0]) - .find(event => event.update.sessionUpdate === "async_task_state_update"); - expect(stopped).toMatchObject({ - sessionId: "child-1", - update: {asyncTaskId: "child-1:command-1", state: "stopped"}, - }); + .findIndex(event => event.update.sessionUpdate === "subagent_state_update" + && event.update.subagentSessionId === "child-1"); + const taskSpawnIndex = fixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0]) + .findIndex(event => event.update.sessionUpdate === "async_task_spawned"); + expect(taskSpawnIndex).toBeGreaterThanOrEqual(0); + expect(childTerminalIndex).toBeGreaterThan(taskSpawnIndex); }); const terminate = vi.spyOn(fixture.getCodexAppServerClient(), "threadBackgroundTerminalsTerminate") @@ -347,8 +404,8 @@ describe("Codex background terminal tasks", () => { await expect(fixture.getCodexAcpAgent().extMethod(ASYNC_TASK_STOP_METHOD, { sessionId: sessionState.sessionId, asyncTaskId: "child-1:command-1", - })).resolves.toEqual({stopped: false}); - expect(terminate).not.toHaveBeenCalled(); + })).resolves.toEqual({stopped: true}); + expect(terminate).toHaveBeenCalledWith({threadId: "child-1", processId: "42"}); }); it("routes a child task stop to its owning Codex thread", async () => { @@ -504,8 +561,10 @@ function childMaterialized() { function deferred() { let resolve!: (value: T) => void; - const promise = new Promise(innerResolve => { + let reject!: (reason?: unknown) => void; + const promise = new Promise((innerResolve, innerReject) => { resolve = innerResolve; + reject = innerReject; }); - return {promise, resolve}; + return {promise, resolve, reject}; } diff --git a/src/__tests__/CodexACPAgent/load-session.test.ts b/src/__tests__/CodexACPAgent/load-session.test.ts index 6811f532..a223d925 100644 --- a/src/__tests__/CodexACPAgent/load-session.test.ts +++ b/src/__tests__/CodexACPAgent/load-session.test.ts @@ -90,14 +90,31 @@ describe("CodexACPAgent - loadSession", () => { agentPath: "/root/orphan_child", }, ]); - const child = makeThread("child-history", [{ - type: "agentMessage", - id: "child-history-message-1", - text: "Persisted first-generation output", - phase: null, - memoryCitation: null, - delivery: null, - }]); + const child = makeThread("child-history", [ + { + type: "commandExecution", + id: "child-command-1", + pluginId: null, + scriptPath: null, + command: "python -m http.server", + cwd: "/workspace", + processId: "42", + source: "unifiedExecStartup", + status: "inProgress", + commandActions: [], + aggregatedOutput: null, + exitCode: null, + durationMs: null, + }, + { + type: "agentMessage", + id: "child-history-message-1", + text: "Persisted first-generation output", + phase: null, + memoryCitation: null, + delivery: null, + }, + ]); const firstChildTurn = child.turns[0]!; child.turns.push({ id: "turn-child-history-2", @@ -129,11 +146,17 @@ describe("CodexACPAgent - loadSession", () => { if (threadId === "orphan-history") return Promise.reject(new Error("missing child history")); return Promise.resolve({thread: threadId === root.id ? root : child}); }); + appServer.threadBackgroundTerminalsList = vi.fn().mockImplementation(({threadId}) => Promise.resolve({ + data: threadId === "child-history" + ? [{itemId: "child-command-1", processId: "42", command: "python -m http.server"}] + : [], + nextCursor: null, + })); await agent.initialize({ protocolVersion: 1, clientCapabilities: { - _meta: {jetbrains: {air: {version: 1, capabilities: ["nativeSubagentSessions"]}}}, + _meta: {jetbrains: {air: {version: 1, capabilities: ["nativeSubagentSessions", "asyncTasks"]}}}, }, }); await agent.loadSession({sessionId: root.id, cwd: "/workspace", mcpServers: []}); @@ -144,12 +167,18 @@ describe("CodexACPAgent - loadSession", () => { const firstSpawnIndex = updates.findIndex(({update}) => update.subagentSessionId === "child-history" && update.sessionUpdate === "subagent_spawned"); const firstOutputIndex = updates.findIndex(({update}) => update.messageId === "child-history-message-1"); + const childTaskIndex = updates.findIndex(({update}) => update.sessionUpdate === "async_task_spawned" + && update.asyncTaskId === "child-history:child-command-1"); + const firstTerminalIndex = updates.findIndex(({update}) => update.sessionUpdate === "subagent_state_update" + && update.subagentSessionId === "child-history"); const secondSpawnIndex = updates.findIndex(({update}) => update.subagentSessionId === "child-history:generation:2" && update.sessionUpdate === "subagent_spawned"); const secondOutputIndex = updates.findIndex(({update}) => update.messageId === "child-history-message-2"); const orphanTerminalIndex = updates.findIndex(({update}) => update.subagentSessionId === "orphan-history" && update.state === "disconnected"); expect(firstOutputIndex).toBeGreaterThan(firstSpawnIndex); + expect(childTaskIndex).toBeGreaterThan(firstSpawnIndex); + expect(firstTerminalIndex).toBeGreaterThan(childTaskIndex); expect(secondSpawnIndex).toBeGreaterThan(firstOutputIndex); expect(secondOutputIndex).toBeGreaterThan(secondSpawnIndex); expect(orphanTerminalIndex).toBeGreaterThan(secondOutputIndex); diff --git a/src/__tests__/CodexACPAgent/providers.test.ts b/src/__tests__/CodexACPAgent/providers.test.ts index 45bc744c..79dada4c 100644 --- a/src/__tests__/CodexACPAgent/providers.test.ts +++ b/src/__tests__/CodexACPAgent/providers.test.ts @@ -211,8 +211,12 @@ describe("Configurable LLM providers (providers/*)", () => { const agent = fixture.getCodexAcpAgent(); await agent.initialize({protocolVersion: acp.PROTOCOL_VERSION}); const sessions = (agent as unknown as {sessions: Map>}).sessions; - sessions.set("thread-1", createTestSessionState({sessionId: "thread-1", cwd: "/one"})); - sessions.set("thread-2", createTestSessionState({sessionId: "thread-2", cwd: "/two"})); + const firstSession = createTestSessionState({sessionId: "thread-1", cwd: "/one"}); + const secondSession = createTestSessionState({sessionId: "thread-2", cwd: "/two"}); + const firstSessionSetAppServer = vi.spyOn(firstSession.asyncTasks, "setAppServer"); + const secondSessionSetAppServer = vi.spyOn(secondSession.asyncTasks, "setAppServer"); + sessions.set("thread-1", firstSession); + sessions.set("thread-2", secondSession); await agent.setProvider({ providerId: OPENAI_PROVIDER_ID, @@ -224,6 +228,8 @@ describe("Configurable LLM providers (providers/*)", () => { expect(restart).toHaveBeenCalledTimes(1); expect(firstGatewayReplacement.getModelProvider()).toBe(CUSTOM_GATEWAY_PROVIDER_ID); expect(firstGatewayResume).toHaveBeenCalledTimes(2); + expect(firstSessionSetAppServer).toHaveBeenCalledWith(firstGatewayReplacement.appServerClient); + expect(secondSessionSetAppServer).toHaveBeenCalledWith(firstGatewayReplacement.appServerClient); expect(firstGatewayResume).toHaveBeenCalledWith(expect.objectContaining({sessionId: "thread-1", cwd: "/one"})); expect(firstGatewayResume).toHaveBeenCalledWith(expect.objectContaining({sessionId: "thread-2", cwd: "/two"})); diff --git a/src/async-tasks/CodexBackgroundTerminalTasks.ts b/src/async-tasks/CodexBackgroundTerminalTasks.ts index 90d00b84..c984f0cf 100644 --- a/src/async-tasks/CodexBackgroundTerminalTasks.ts +++ b/src/async-tasks/CodexBackgroundTerminalTasks.ts @@ -2,6 +2,12 @@ import type {ServerNotification} from "../app-server"; import type {ThreadItem} from "../app-server/v2"; import type {ACPSessionConnection} from "../ACPSessionConnection"; import type {CodexAppServerClient} from "../CodexAppServerClient"; +import { + AIR_ASYNC_TASKS_BACKGROUNDED_KEY, + AIR_ASYNC_TASKS_KEY, + AIR_META_KEY, + JETBRAINS_META_KEY, +} from "../AirExtension"; import {logger} from "../Logger"; import type {ThreadBackgroundTerminal} from "./BackgroundTerminalApi"; @@ -34,7 +40,7 @@ export class CodexBackgroundTerminalTasks { constructor( readonly enabled: boolean, private readonly rootSessionId: string, - private readonly appServer: CodexAppServerClient, + private appServer: CodexAppServerClient, private readonly session: ACPSessionConnection, ) {} @@ -51,7 +57,11 @@ export class CodexBackgroundTerminalTasks { await this.observeCommandCompleted(notification.params.item, threadId); return; } - if (notification.method === "item/started" || notification.method === "turn/completed") { + if (notification.method === "turn/completed") { + await this.reconcile(threadId, sessionId); + return; + } + if (notification.method === "item/started") { this.refresh(threadId, sessionId); } } @@ -79,9 +89,45 @@ export class CodexBackgroundTerminalTasks { } refresh(threadId: string = this.rootSessionId, sessionId: string = this.rootSessionId): void { - void this.sync(threadId, sessionId).catch((error) => { + void this.reconcile(threadId, sessionId); + } + + async reconcile(threadId: string = this.rootSessionId, sessionId: string = this.rootSessionId): Promise { + try { + await this.sync(threadId, sessionId); + } catch (error) { if (this.isActive()) logger.error(`Failed to list background terminals for ${threadId}`, error); - }); + } + } + + setAppServer(appServer: CodexAppServerClient): void { + this.appServer = appServer; + } + + async recover(threadId: string, sessionId: string, itemIds: ReadonlySet): Promise { + if (!this.isActive() || itemIds.size === 0) return; + const terminals = await this.listAll(threadId); + for (const terminal of terminals) { + if (!this.isActive()) return; + if (!itemIds.has(terminal.itemId)) continue; + const task = this.remember(threadId, sessionId, terminal); + if (!task.announced && task.state === "running") await this.announce(task); + } + } + + async finishAll(state: TerminalState): Promise { + if (!this.isActive()) return; + const errors: unknown[] = []; + for (const task of this.tasks.values()) { + try { + await this.finish(task, state); + } catch (error) { + errors.push(error); + } + } + if (errors.length > 0) { + throw new AggregateError(errors, `Failed to finish ${errors.length} background terminal task(s)`); + } } async sync( @@ -159,15 +205,37 @@ export class CodexBackgroundTerminalTasks { } private async syncUntilCurrent(threadId: string, pending: PendingSync): Promise { + let hasFailure = false; + let failure: unknown; do { pending.requested = false; - await this.syncThread(threadId, pending.sessionId); + try { + await this.syncThread(threadId, pending.sessionId); + hasFailure = false; + } catch (error) { + hasFailure = true; + failure = error; + } } while (pending.requested && this.isActive()); + if (hasFailure) throw failure; } private async announce(task: Task): Promise { task.announced = true; try { + await this.session.update({ + sessionUpdate: "tool_call_update", + toolCallId: task.itemId, + _meta: { + [JETBRAINS_META_KEY]: { + [AIR_META_KEY]: { + [AIR_ASYNC_TASKS_KEY]: { + [AIR_ASYNC_TASKS_BACKGROUNDED_KEY]: true, + }, + }, + }, + }, + }, task.sessionId); await this.session.update({ sessionUpdate: "async_task_spawned", asyncTaskId: task.asyncTaskId, From 5a74acad7fac496d5ff81b29a0fbeabe19ece91d Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 3 Sep 2026 13:16:29 +0400 Subject: [PATCH 4/6] fix: serialize background terminal lifecycle Publish each task spawn before its terminal state. Fence old app-server queries during provider replacement, fail tasks on process exit, and reconcile every child terminal path. --- src/CodexAcpServer.ts | 17 ++ src/CodexEventHandler.ts | 15 +- .../CodexACPAgent/async-tasks.test.ts | 138 ++++++++++++- src/__tests__/CodexACPAgent/providers.test.ts | 4 + src/__tests__/acp-test-utils.ts | 14 +- .../CodexBackgroundTerminalTasks.ts | 192 ++++++++++++------ src/subagents/CodexSubagentEventRouter.ts | 33 +++ 7 files changed, 341 insertions(+), 72 deletions(-) diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 8f296acc..0b9b1d5b 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -261,6 +261,7 @@ export class CodexAcpServer { private readonly goalControlGenerations: Map; private readonly permissionLifecycleContexts: WeakMap; private readonly codexProcessState: CodexProcessState | null; + private codexProcessGeneration = 0; private initializeRequest: acp.InitializeRequest | null = null; private providerUpdate: Promise | null = null; @@ -295,6 +296,7 @@ export class CodexAcpServer { this.terminalOutputMode = "terminal_output_delta"; this.booleanConfigOptionsSupported = false; this.availableCommands = this.createAvailableCommands(codexAcpClient); + this.observeCodexProcess(); } private createAvailableCommands(client: CodexAcpClient): CodexCommands { @@ -1003,6 +1005,9 @@ export class CodexAcpServer { } logger.log("Restarting Codex app-server for provider update", {sessionCount: this.sessions.size}); + for (const session of this.sessions.values()) { + session.asyncTasks.prepareForAppServerReplacement(); + } await this.finishAllAsyncTasks("stopped", "before the provider restart"); const replacement = await this.restartCodexClient(); apply(replacement); @@ -1056,6 +1061,16 @@ export class CodexAcpServer { }); } + private observeCodexProcess(): void { + const process = this.codexProcessState?.connection.process; + if (!process) return; + const generation = ++this.codexProcessGeneration; + process.once("exit", () => { + if (generation !== this.codexProcessGeneration) return; + void this.finishAllAsyncTasks("failed", "after the Codex process exited"); + }); + } + private async restartCodexClient(): Promise { const state = this.codexProcessState; if (state === null) { @@ -1063,6 +1078,7 @@ export class CodexAcpServer { } const previous = state.connection; + this.codexProcessGeneration += 1; const exited = previous.process.exitCode === null ? once(previous.process, "exit") : Promise.resolve(); @@ -1079,6 +1095,7 @@ export class CodexAcpServer { state.stderr = ""; state.connection = startCodexConnection(state.codexPath); this.captureStderr(); + this.observeCodexProcess(); return new CodexAcpClient( new CodexAppServerClient(state.connection.connection), state.config, diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index fb0e524f..262ed686 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -368,18 +368,19 @@ export class CodexEventHandler { async handleNotification(notification: ServerNotification) { await this.flushPendingErrors(); - const asyncTaskSessionId = this.subagents.notificationSessionId(notification); - const handledAsyncTasksFirst = notification.method === "turn/completed" - && asyncTaskSessionId !== this.sessionState.sessionId; - if (handledAsyncTasksFirst) { - await this.sessionState.asyncTasks.handleNotification(notification, asyncTaskSessionId); + const closingChildren = this.subagents.closingChildSessions(notification); + for (const child of closingChildren) { + await this.sessionState.asyncTasks.reconcile(child.threadId, child.sessionId); } const handledBySubagents = await this.subagents.handle(notification); for (const buffered of this.subagents.takeBufferedNotifications()) { await this.handleNotification(buffered); } - if (!handledAsyncTasksFirst && !handledBySubagents) { - await this.sessionState.asyncTasks.handleNotification(notification, asyncTaskSessionId); + if (!handledBySubagents) { + await this.sessionState.asyncTasks.handleNotification( + notification, + this.subagents.notificationSessionId(notification), + ); } if (handledBySubagents) return; if (this.subagents.shouldIgnore(notification)) { diff --git a/src/__tests__/CodexACPAgent/async-tasks.test.ts b/src/__tests__/CodexACPAgent/async-tasks.test.ts index 57721aaf..4e51efdf 100644 --- a/src/__tests__/CodexACPAgent/async-tasks.test.ts +++ b/src/__tests__/CodexACPAgent/async-tasks.test.ts @@ -1,7 +1,9 @@ import {describe, expect, it, vi} from "vitest"; +import {EventEmitter} from "node:events"; import type {ThreadItem} from "../../app-server/v2"; import {ACPSessionConnection, type UpdateSessionEvent} from "../../ACPSessionConnection"; import type {CodexAppServerClient} from "../../CodexAppServerClient"; +import type {CodexConnection} from "../../CodexJsonRpcConnection"; import {CodexBackgroundTerminalTasks} from "../../async-tasks/CodexBackgroundTerminalTasks"; import {ASYNC_TASK_STOP_METHOD} from "../../async-tasks/AsyncTaskExtension"; import {CodexSubagentEventRouter} from "../../subagents/CodexSubagentEventRouter"; @@ -110,6 +112,31 @@ describe("Codex background terminal tasks", () => { }); }); + it("publishes a spawn before a completion that races with discovery", async () => { + const markerStarted = deferred(); + const releaseMarker = deferred(); + const fixture = createFixture(true, async update => { + if (update.sessionUpdate !== "tool_call_update") return; + markerStarted.resolve(); + await releaseMarker.promise; + }); + fixture.list.mockResolvedValue(page([terminal()])); + + const sync = fixture.tasks.sync(); + await markerStarted.promise; + const completion = fixture.tasks.handleNotification( + completed({...command(), status: "completed", exitCode: 0}), + "thread-1", + ); + releaseMarker.resolve(); + await Promise.all([sync, completion]); + + const spawnIndex = fixture.updates.findIndex(update => update.sessionUpdate === "async_task_spawned"); + const terminalIndex = fixture.updates.findIndex(update => update.sessionUpdate === "async_task_state_update"); + expect(spawnIndex).toBeGreaterThanOrEqual(0); + expect(terminalIndex).toBeGreaterThan(spawnIndex); + }); + it("retries a terminal update that the client rejected", async () => { let rejectTerminalUpdate = true; const fixture = createFixture(true, async update => { @@ -201,6 +228,52 @@ describe("Codex background terminal tasks", () => { ]); }); + it("finishes tasks while app-server queries are suspended for replacement", async () => { + const fixture = createFixture(); + fixture.list.mockResolvedValue(page([terminal()])); + await fixture.tasks.sync(); + + fixture.tasks.prepareForAppServerReplacement(); + await fixture.tasks.finishAll("stopped"); + + expect(fixture.updates.at(-1)).toMatchObject({ + sessionUpdate: "async_task_state_update", + asyncTaskId: "command-1", + state: "stopped", + }); + }); + + it("fails announced tasks as soon as the app-server process exits", async () => { + const process = Object.assign(new EventEmitter(), { + stderr: new EventEmitter(), + stdin: {end: vi.fn()}, + exitCode: null, + }) as unknown as CodexConnection["process"]; + const fixture = createCodexMockTestFixture(undefined, process); + const sessionState = createTestSessionState({sessionId: "thread-1"}); + sessionState.asyncTasks = new CodexBackgroundTerminalTasks( + true, + sessionState.sessionId, + fixture.getCodexAppServerClient(), + new ACPSessionConnection(fixture.getAcpConnection(), sessionState.sessionId), + ); + vi.spyOn(fixture.getCodexAppServerClient(), "threadBackgroundTerminalsList") + .mockResolvedValue(page([terminal()])); + await sessionState.asyncTasks.sync(); + // @ts-expect-error - register the local session for process lifecycle checks + fixture.getCodexAcpAgent().sessions.set(sessionState.sessionId, sessionState); + + process.emit("exit", 1); + + await vi.waitFor(() => { + const terminalUpdate = fixture.getAcpConnectionEvents([]) + .filter(event => event.method === "sessionUpdate") + .map(event => event.args[0].update) + .find(update => update.sessionUpdate === "async_task_state_update"); + expect(terminalUpdate).toMatchObject({asyncTaskId: "command-1", state: "failed"}); + }); + }); + it("keeps a task stoppable when termination is rejected or fails", async () => { const fixture = createFixture(); fixture.list.mockResolvedValue(page([terminal()])); @@ -226,13 +299,13 @@ describe("Codex background terminal tasks", () => { } }); fixture.list.mockResolvedValue(page([terminal()])); - fixture.terminate.mockResolvedValue({terminated: true}); + fixture.terminate.mockResolvedValueOnce({terminated: true}); await fixture.tasks.sync(); await expect(fixture.tasks.stop("command-1")).rejects.toThrow("client disconnected"); await expect(fixture.tasks.stop("command-1")).resolves.toBe(true); - expect(fixture.terminate).toHaveBeenCalledTimes(2); + expect(fixture.terminate).toHaveBeenCalledOnce(); expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_state_update")) .toEqual([expect.objectContaining({state: "stopped"})]); }); @@ -332,6 +405,30 @@ describe("Codex background terminal tasks", () => { expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_spawned")).toHaveLength(1); }); + it("discards an in-flight list result after app-server replacement", async () => { + const fixture = createFixture(); + const oldListing = deferred(); + fixture.list.mockReturnValue(oldListing.promise); + const replacementList = vi.fn().mockResolvedValue(page([ + terminal({itemId: "command-2", processId: "84"}), + ])); + const replacement = { + threadBackgroundTerminalsList: replacementList, + threadBackgroundTerminalsTerminate: vi.fn(), + } as unknown as CodexAppServerClient; + + const oldSync = fixture.tasks.sync(); + fixture.tasks.prepareForAppServerReplacement(); + fixture.tasks.setAppServer(replacement); + const replacementSync = fixture.tasks.sync(); + oldListing.resolve(page([terminal()])); + await Promise.all([oldSync, replacementSync]); + + expect(replacementList).toHaveBeenCalledOnce(); + expect(fixture.updates.filter(update => update.sessionUpdate === "async_task_spawned")) + .toEqual([expect.objectContaining({asyncTaskId: "command-2"})]); + }); + it("does not publish an in-flight list result after clear", async () => { const fixture = createFixture(); const listing = deferred(); @@ -345,7 +442,11 @@ describe("Codex background terminal tasks", () => { expect(fixture.updates).toEqual([]); }); - it("publishes a child terminal on its native subagent session", async () => { + it.each([ + ["turn completion", () => turnCompleted("child-1")], + ["activity interruption", childInterrupted], + ["collaboration completion", childCompletedByCollaboration], + ])("publishes a child terminal before %s closes its session", async (_name, terminalNotification) => { const fixture = createCodexMockTestFixture(); await fixture.getCodexAcpAgent().initialize({ protocolVersion: 1, @@ -374,7 +475,7 @@ describe("Codex background terminal tasks", () => { childSpawned(), childMaterialized(), started(command(), "child-1"), - turnCompleted("child-1"), + terminalNotification(), ]); await vi.waitFor(() => { @@ -559,11 +660,36 @@ function childMaterialized() { }); } +function childInterrupted() { + return started({ + type: "subAgentActivity", + id: "child-activity-terminal", + kind: "interrupted", + agentThreadId: "child-1", + agentPath: "/root/worker", + }); +} + +function childCompletedByCollaboration() { + return started({ + type: "collabAgentToolCall", + id: "wait-child", + tool: "wait", + status: "completed", + senderThreadId: "thread-1", + receiverThreadIds: ["child-1"], + prompt: null, + model: null, + reasoningEffort: null, + agentsStates: {"child-1": {status: "completed", message: null}}, + }); +} + function deferred() { - let resolve!: (value: T) => void; + let resolve!: (value?: T) => void; let reject!: (reason?: unknown) => void; const promise = new Promise((innerResolve, innerReject) => { - resolve = innerResolve; + resolve = innerResolve as (value?: T) => void; reject = innerReject; }); return {promise, resolve, reject}; diff --git a/src/__tests__/CodexACPAgent/providers.test.ts b/src/__tests__/CodexACPAgent/providers.test.ts index 79dada4c..5a724039 100644 --- a/src/__tests__/CodexACPAgent/providers.test.ts +++ b/src/__tests__/CodexACPAgent/providers.test.ts @@ -213,6 +213,8 @@ describe("Configurable LLM providers (providers/*)", () => { const sessions = (agent as unknown as {sessions: Map>}).sessions; const firstSession = createTestSessionState({sessionId: "thread-1", cwd: "/one"}); const secondSession = createTestSessionState({sessionId: "thread-2", cwd: "/two"}); + const firstSessionPrepare = vi.spyOn(firstSession.asyncTasks, "prepareForAppServerReplacement"); + const secondSessionPrepare = vi.spyOn(secondSession.asyncTasks, "prepareForAppServerReplacement"); const firstSessionSetAppServer = vi.spyOn(firstSession.asyncTasks, "setAppServer"); const secondSessionSetAppServer = vi.spyOn(secondSession.asyncTasks, "setAppServer"); sessions.set("thread-1", firstSession); @@ -228,6 +230,8 @@ describe("Configurable LLM providers (providers/*)", () => { expect(restart).toHaveBeenCalledTimes(1); expect(firstGatewayReplacement.getModelProvider()).toBe(CUSTOM_GATEWAY_PROVIDER_ID); expect(firstGatewayResume).toHaveBeenCalledTimes(2); + expect(firstSessionPrepare).toHaveBeenCalledOnce(); + expect(secondSessionPrepare).toHaveBeenCalledOnce(); expect(firstSessionSetAppServer).toHaveBeenCalledWith(firstGatewayReplacement.appServerClient); expect(secondSessionSetAppServer).toHaveBeenCalledWith(firstGatewayReplacement.appServerClient); expect(firstGatewayResume).toHaveBeenCalledWith(expect.objectContaining({sessionId: "thread-1", cwd: "/one"})); diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts index b74ec042..259f2f03 100644 --- a/src/__tests__/acp-test-utils.ts +++ b/src/__tests__/acp-test-utils.ts @@ -2,8 +2,8 @@ import * as acp from "@agentclientprotocol/sdk"; import type {CreateElicitationResponse, McpServerStdio, RequestPermissionResponse} from "@agentclientprotocol/sdk"; import {CodexAcpClient} from '../CodexAcpClient'; import {CodexAppServerClient, type CodexConnectionEvent} from '../CodexAppServerClient'; -import {startCodexConnection} from "../CodexJsonRpcConnection"; -import {CodexAcpServer, type SessionState} from "../CodexAcpServer"; +import {type CodexConnection, startCodexConnection} from "../CodexJsonRpcConnection"; +import {CodexAcpServer, type CodexProcessState, type SessionState} from "../CodexAcpServer"; import {ACPSessionConnection, type AcpClientConnection} from "../ACPSessionConnection"; import type {ServerNotification} from "../app-server"; import type {MessageConnection} from "vscode-jsonrpc/node"; @@ -88,6 +88,7 @@ export interface ConnectionConfig { connection: MessageConnection; getExitCode: () => number | null; acpConnection?: AcpConnectionConfig; + codexProcessState?: CodexProcessState; } export function createBaseTestFixture(config: ConnectionConfig): TestFixture { @@ -107,6 +108,7 @@ export function createBaseTestFixture(config: ConnectionConfig): TestFixture { undefined, config.getExitCode, undefined, + config.codexProcessState, ); const transportEvents: CodexConnectionEvent[] = []; @@ -269,6 +271,7 @@ export interface CodexMockTestFixture extends TestFixture { */ export function createCodexMockTestFixture( restartCodexClient?: () => Promise, + process?: CodexConnection["process"], ): CodexMockTestFixture { let unhandledNotificationHandler: ((notification: any) => void) | null = null; const requestHandlers = new Map Promise>(); @@ -317,6 +320,13 @@ export function createCodexMockTestFixture( const baseFixture = createBaseTestFixture({ connection: mockCodexConnection, getExitCode: () => null, + ...(process ? {codexProcessState: { + connection: {connection: mockCodexConnection, process}, + codexPath: undefined, + config: undefined, + modelProvider: undefined, + stderr: "", + }} : {}), acpConnection: { connection: acpConnection, events: acpConnectionEvents, diff --git a/src/async-tasks/CodexBackgroundTerminalTasks.ts b/src/async-tasks/CodexBackgroundTerminalTasks.ts index c984f0cf..1fa3bbe2 100644 --- a/src/async-tasks/CodexBackgroundTerminalTasks.ts +++ b/src/async-tasks/CodexBackgroundTerminalTasks.ts @@ -21,7 +21,10 @@ type Task = { processId: string; itemId: string; command: string; - announced: boolean; + publication: "unpublished" | "publishing" | "published"; + announcement: Promise | null; + terminalUpdate: Promise | null; + terminalPublished: boolean; state: "running" | "stopping" | TerminalState; }; @@ -31,10 +34,17 @@ type PendingSync = { promise: Promise; }; +type TerminalSnapshot = { + generation: number; + terminals: ThreadBackgroundTerminal[]; +}; + /** Maps Codex-owned background terminals to the AIR async task extension. */ export class CodexBackgroundTerminalTasks { private readonly tasks = new Map(); private readonly syncs = new Map(); + private appServerGeneration = 0; + private appServerQueriesEnabled = true; private disposed = false; constructor( @@ -102,16 +112,24 @@ export class CodexBackgroundTerminalTasks { setAppServer(appServer: CodexAppServerClient): void { this.appServer = appServer; + this.appServerGeneration += 1; + this.appServerQueriesEnabled = true; + } + + prepareForAppServerReplacement(): void { + this.appServerGeneration += 1; + this.appServerQueriesEnabled = false; } async recover(threadId: string, sessionId: string, itemIds: ReadonlySet): Promise { - if (!this.isActive() || itemIds.size === 0) return; - const terminals = await this.listAll(threadId); - for (const terminal of terminals) { - if (!this.isActive()) return; + if (!this.canQueryAppServer() || itemIds.size === 0) return; + const snapshot = await this.listAll(threadId); + if (snapshot === null) return; + for (const terminal of snapshot.terminals) { + if (!this.queryIsCurrent(snapshot.generation)) return; if (!itemIds.has(terminal.itemId)) continue; const task = this.remember(threadId, sessionId, terminal); - if (!task.announced && task.state === "running") await this.announce(task); + if (task.publication === "unpublished" && task.state === "running") await this.announce(task); } } @@ -134,7 +152,7 @@ export class CodexBackgroundTerminalTasks { threadId: string = this.rootSessionId, sessionId: string = this.rootSessionId, ): Promise { - if (!this.isActive()) return; + if (!this.canQueryAppServer()) return; const current = this.syncs.get(threadId); if (current) { current.requested = true; @@ -157,7 +175,12 @@ export class CodexBackgroundTerminalTasks { async stop(taskId: string): Promise { if (!this.isActive()) return false; const task = this.tasks.get(taskId); - if (!task || !task.announced || task.state !== "running") return false; + if (!task || task.publication !== "published") return false; + if (task.state === "stopped" && !task.terminalPublished) { + await this.publishTerminalState(task); + return true; + } + if (task.state !== "running") return false; task.state = "stopping"; try { const response = await this.appServer.threadBackgroundTerminalsTerminate({ @@ -178,25 +201,28 @@ export class CodexBackgroundTerminalTasks { clear(): void { this.disposed = true; + this.appServerGeneration += 1; + this.appServerQueriesEnabled = false; this.tasks.clear(); this.syncs.clear(); } private async syncThread(threadId: string, sessionId: string): Promise { - const terminals = await this.listAll(threadId); - if (!this.isActive()) return; + const snapshot = await this.listAll(threadId); + if (snapshot === null || !this.queryIsCurrent(snapshot.generation)) return; const liveTaskIds = new Set(); - for (const terminal of terminals) { - if (!this.isActive()) return; + for (const terminal of snapshot.terminals) { + if (!this.queryIsCurrent(snapshot.generation)) return; liveTaskIds.add(terminal.itemId); const task = this.remember(threadId, sessionId, terminal); - if (!task.announced && task.state === "running") await this.announce(task); + if (task.publication === "unpublished" && task.state === "running") await this.announce(task); } for (const task of this.tasks.values()) { + if (!this.queryIsCurrent(snapshot.generation)) return; if (task.threadId === threadId - && task.announced + && task.publication === "published" && (task.state === "running" || task.state === "stopping") && !liveTaskIds.has(task.itemId)) { await this.finish(task, "stopped"); @@ -216,42 +242,64 @@ export class CodexBackgroundTerminalTasks { hasFailure = true; failure = error; } - } while (pending.requested && this.isActive()); + } while (pending.requested && this.canQueryAppServer()); if (hasFailure) throw failure; } private async announce(task: Task): Promise { - task.announced = true; + if (task.publication === "published") return; + if (task.announcement !== null) { + await task.announcement; + return; + } + + task.publication = "publishing"; + const announcement = this.publishAnnouncement(task); + task.announcement = announcement; try { - await this.session.update({ - sessionUpdate: "tool_call_update", - toolCallId: task.itemId, - _meta: { - [JETBRAINS_META_KEY]: { - [AIR_META_KEY]: { - [AIR_ASYNC_TASKS_KEY]: { - [AIR_ASYNC_TASKS_BACKGROUNDED_KEY]: true, - }, - }, - }, - }, - }, task.sessionId); - await this.session.update({ - sessionUpdate: "async_task_spawned", - asyncTaskId: task.asyncTaskId, - name: task.command, - taskType: "shell", - description: task.command, - showInTranscript: false, - canStop: true, - toolCallId: task.itemId, - }, task.sessionId); + await announcement; + } finally { + if (task.announcement === announcement) task.announcement = null; + } + if (isTerminalState(task.state)) await this.publishTerminalState(task); + } + + private async publishAnnouncement(task: Task): Promise { + try { + await this.publishSpawn(task); + task.publication = "published"; } catch (error) { - task.announced = false; + task.publication = "unpublished"; throw error; } } + private async publishSpawn(task: Task): Promise { + await this.session.update({ + sessionUpdate: "tool_call_update", + toolCallId: task.itemId, + _meta: { + [JETBRAINS_META_KEY]: { + [AIR_META_KEY]: { + [AIR_ASYNC_TASKS_KEY]: { + [AIR_ASYNC_TASKS_BACKGROUNDED_KEY]: true, + }, + }, + }, + }, + }, task.sessionId); + await this.session.update({ + sessionUpdate: "async_task_spawned", + asyncTaskId: task.asyncTaskId, + name: task.command, + taskType: "shell", + description: task.command, + showInTranscript: false, + canStop: true, + toolCallId: task.itemId, + }, task.sessionId); + } + private remember( threadId: string, sessionId: string, @@ -270,7 +318,10 @@ export class CodexBackgroundTerminalTasks { processId: terminal.processId, itemId: terminal.itemId, command: terminal.command, - announced: false, + publication: "unpublished", + announcement: null, + terminalUpdate: null, + terminalPublished: false, state: "running", }; this.tasks.set(asyncTaskId, task); @@ -278,33 +329,48 @@ export class CodexBackgroundTerminalTasks { } private async finish(task: Task, state: TerminalState): Promise { - if (task.state !== "running" && task.state !== "stopping") return; - const previousState = task.state; - task.state = state; - if (!task.announced) return; + if (task.state === "running" || task.state === "stopping") { + task.state = state; + } else if (task.state !== state) { + return; + } + if (task.announcement !== null) await task.announcement; + if (task.publication === "published") await this.publishTerminalState(task); + } + private async publishTerminalState(task: Task): Promise { + if (!isTerminalState(task.state) || task.terminalPublished) return; + if (task.terminalUpdate !== null) { + await task.terminalUpdate; + return; + } + const terminalUpdate = this.session.update({ + sessionUpdate: "async_task_state_update", + asyncTaskId: task.asyncTaskId, + state: task.state, + toolCallId: task.itemId, + }, task.sessionId); + task.terminalUpdate = terminalUpdate; try { - await this.session.update({ - sessionUpdate: "async_task_state_update", - asyncTaskId: task.asyncTaskId, - state, - toolCallId: task.itemId, - }, task.sessionId); - } catch (error) { - if (task.state === state) task.state = previousState; - throw error; + await terminalUpdate; + task.terminalPublished = true; + } finally { + if (task.terminalUpdate === terminalUpdate) task.terminalUpdate = null; } } - private async listAll(threadId: string): Promise { + private async listAll(threadId: string): Promise { + const appServer = this.appServer; + const generation = this.appServerGeneration; const terminals: ThreadBackgroundTerminal[] = []; const seenCursors = new Set(); let cursor: string | null = null; do { - const response = await this.appServer.threadBackgroundTerminalsList({ + const response = await appServer.threadBackgroundTerminalsList({ threadId, cursor, }); + if (!this.queryIsCurrent(generation)) return null; terminals.push(...response.data); cursor = response.nextCursor; if (cursor !== null) { @@ -314,12 +380,24 @@ export class CodexBackgroundTerminalTasks { seenCursors.add(cursor); } } while (cursor !== null); - return terminals; + return {generation, terminals}; } private isActive(): boolean { return this.enabled && !this.disposed; } + + private canQueryAppServer(): boolean { + return this.isActive() && this.appServerQueriesEnabled; + } + + private queryIsCurrent(generation: number): boolean { + return this.canQueryAppServer() && generation === this.appServerGeneration; + } +} + +function isTerminalState(state: Task["state"]): state is TerminalState { + return state === "completed" || state === "failed" || state === "stopped"; } function wireTaskId(rootSessionId: string, threadId: string, itemId: string): string { diff --git a/src/subagents/CodexSubagentEventRouter.ts b/src/subagents/CodexSubagentEventRouter.ts index 6dc32640..8d232f12 100644 --- a/src/subagents/CodexSubagentEventRouter.ts +++ b/src/subagents/CodexSubagentEventRouter.ts @@ -29,6 +29,11 @@ type PendingSubagent = { droppedBufferedNotifications: number; }; +export type ClosingChildSession = { + threadId: string; + sessionId: string; +}; + /** Owns native lifecycle, child routing, waiting, and legacy activity deduplication. */ export class CodexSubagentEventRouter { private static readonly DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1000; @@ -180,6 +185,27 @@ export class CodexSubagentEventRouter { : this.rootSessionId; } + closingChildSessions(notification: ServerNotification): ClosingChildSession[] { + if (!this.supported) return []; + if (notification.method === "turn/completed") { + if (terminalStateFromTurn(notification.params.turn.status) === undefined) return []; + return this.closingChildSession(notification.params.threadId); + } + if (notification.method !== "item/started" && notification.method !== "item/completed") return []; + const item = notification.params.item; + if (item.type === "subAgentActivity") { + return item.kind === "interrupted" ? this.closingChildSession(item.agentThreadId) : []; + } + if (item.type !== "collabAgentToolCall") return []; + + const closing = new Map(); + for (const [threadId, state] of Object.entries(item.agentsStates)) { + if (!state || terminalStateOf(state.status) === undefined) continue; + for (const child of this.closingChildSession(threadId)) closing.set(threadId, child); + } + return [...closing.values()]; + } + takeBufferedNotifications(): ServerNotification[] { return this.replayQueue.splice(0); } @@ -272,6 +298,13 @@ export class CodexSubagentEventRouter { || this.terminalPendingSpawns.has(threadId)); } + private closingChildSession(threadId: string): ClosingChildSession[] { + const child = this.children.get(threadId); + return child && child.terminalState === undefined + ? [{threadId, sessionId: child.sessionId}] + : []; + } + private async materialize(childSessionId: string, path: string): Promise { if (this.children.has(childSessionId)) return; const pending = this.pendingSpawns.get(childSessionId); From 592810834a8a08a04f90f46564da6f66dddb6c42 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 3 Sep 2026 15:56:10 +0400 Subject: [PATCH 5/6] fix: omit duplicate async task descriptions The task name already contains the command text. Omit the matching description so clients do not render the command twice. --- src/__tests__/CodexACPAgent/async-tasks.test.ts | 1 - src/async-tasks/AcpAsyncTasks.ts | 2 +- src/async-tasks/CodexBackgroundTerminalTasks.ts | 1 - 3 files changed, 1 insertion(+), 3 deletions(-) diff --git a/src/__tests__/CodexACPAgent/async-tasks.test.ts b/src/__tests__/CodexACPAgent/async-tasks.test.ts index 4e51efdf..a4ced19a 100644 --- a/src/__tests__/CodexACPAgent/async-tasks.test.ts +++ b/src/__tests__/CodexACPAgent/async-tasks.test.ts @@ -75,7 +75,6 @@ describe("Codex background terminal tasks", () => { asyncTaskId: "command-1", name: "python -m http.server", taskType: "shell", - description: "python -m http.server", showInTranscript: false, canStop: true, toolCallId: "command-1", diff --git a/src/async-tasks/AcpAsyncTasks.ts b/src/async-tasks/AcpAsyncTasks.ts index 9262b14d..e658912a 100644 --- a/src/async-tasks/AcpAsyncTasks.ts +++ b/src/async-tasks/AcpAsyncTasks.ts @@ -5,7 +5,7 @@ export type AsyncTaskSpawnedUpdate = { asyncTaskId: string; name: string; taskType: string; - description: string; + description?: string; showInTranscript: boolean; canStop: boolean; outputFilePath?: string; diff --git a/src/async-tasks/CodexBackgroundTerminalTasks.ts b/src/async-tasks/CodexBackgroundTerminalTasks.ts index 1fa3bbe2..7a265718 100644 --- a/src/async-tasks/CodexBackgroundTerminalTasks.ts +++ b/src/async-tasks/CodexBackgroundTerminalTasks.ts @@ -293,7 +293,6 @@ export class CodexBackgroundTerminalTasks { asyncTaskId: task.asyncTaskId, name: task.command, taskType: "shell", - description: task.command, showInTranscript: false, canStop: true, toolCallId: task.itemId, From 3582f54c44104ae332c3c71f57ddeba577e5b452 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 3 Sep 2026 17:06:27 +0400 Subject: [PATCH 6/6] fix: reuse tool titles for background tasks Use the mapped command title as the async task name. Keep the raw command only in the tool input and terminal API. --- src/CodexEventHandler.ts | 20 +++++++++++++++---- .../CodexACPAgent/async-tasks.test.ts | 11 ++++++---- .../CodexBackgroundTerminalTasks.ts | 19 ++++++++++++------ 3 files changed, 36 insertions(+), 14 deletions(-) diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 262ed686..534fb222 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -376,17 +376,24 @@ export class CodexEventHandler { for (const buffered of this.subagents.takeBufferedNotifications()) { await this.handleNotification(buffered); } + const ignoredBySubagents = !handledBySubagents && this.subagents.shouldIgnore(notification); + let updateEvent: UpdateSessionEvent | null | undefined; + if (!handledBySubagents + && !ignoredBySubagents + && notification.method === "item/started" + && notification.params.item.type === "commandExecution") { + updateEvent = await this.createUpdateEvent(notification); + } if (!handledBySubagents) { await this.sessionState.asyncTasks.handleNotification( notification, this.subagents.notificationSessionId(notification), + toolCallTitle(updateEvent), ); } if (handledBySubagents) return; - if (this.subagents.shouldIgnore(notification)) { - return; - } - const updateEvent = await this.createUpdateEvent(notification); + if (ignoredBySubagents) return; + if (updateEvent === undefined) updateEvent = await this.createUpdateEvent(notification); if (updateEvent) { await this.session.update(updateEvent, this.subagents.notificationSessionId(notification)); } @@ -1334,3 +1341,8 @@ export class CodexEventHandler { return createGuardianApprovalReviewToolCall(params); } } + +function toolCallTitle(update: UpdateSessionEvent | null | undefined): string | undefined { + if (update?.sessionUpdate !== "tool_call") return undefined; + return update.title; +} diff --git a/src/__tests__/CodexACPAgent/async-tasks.test.ts b/src/__tests__/CodexACPAgent/async-tasks.test.ts index a4ced19a..e24c4076 100644 --- a/src/__tests__/CodexACPAgent/async-tasks.test.ts +++ b/src/__tests__/CodexACPAgent/async-tasks.test.ts @@ -37,11 +37,12 @@ describe("Codex background terminal tasks", () => { ); // @ts-expect-error - register the local session for session-generation checks fixture.getCodexAcpAgent().sessions.set(sessionState.sessionId, sessionState); + const rawCommand = "/bin/zsh -lc 'npm run build'"; vi.spyOn(fixture.getCodexAppServerClient(), "threadBackgroundTerminalsList") - .mockResolvedValue(page([terminal()])); + .mockResolvedValue(page([terminal({command: rawCommand})])); await setupPromptAndSendNotifications(fixture, sessionState.sessionId, sessionState, [ - started(command()), + started(command({command: rawCommand})), started({type: "reasoning", id: "reasoning-1", summary: [], content: []}), ]); @@ -52,6 +53,7 @@ describe("Codex background terminal tasks", () => { .filter(update => update.sessionUpdate === "async_task_spawned"); expect(taskUpdates).toEqual([expect.objectContaining({ asyncTaskId: "command-1", + name: "npm run build", toolCallId: "command-1", })]); }); @@ -61,7 +63,7 @@ describe("Codex background terminal tasks", () => { const fixture = createFixture(); fixture.list.mockResolvedValue(page([terminal()])); - await fixture.tasks.handleNotification(started(command()), "thread-1"); + await fixture.tasks.handleNotification(started(command()), "thread-1", "python -m http.server"); await fixture.tasks.sync(); expect(fixture.updates).toEqual([ @@ -573,7 +575,7 @@ function page(data: ThreadBackgroundTerminal[], nextCursor: string | null = null return {data, nextCursor}; } -function command(): CommandExecutionItem { +function command(overrides: Partial = {}): CommandExecutionItem { return { type: "commandExecution", id: "command-1", @@ -588,6 +590,7 @@ function command(): CommandExecutionItem { aggregatedOutput: null, exitCode: null, durationMs: null, + ...overrides, }; } diff --git a/src/async-tasks/CodexBackgroundTerminalTasks.ts b/src/async-tasks/CodexBackgroundTerminalTasks.ts index 7a265718..61ead906 100644 --- a/src/async-tasks/CodexBackgroundTerminalTasks.ts +++ b/src/async-tasks/CodexBackgroundTerminalTasks.ts @@ -20,7 +20,7 @@ type Task = { asyncTaskId: string; processId: string; itemId: string; - command: string; + name: string; publication: "unpublished" | "publishing" | "published"; announcement: Promise | null; terminalUpdate: Promise | null; @@ -54,13 +54,17 @@ export class CodexBackgroundTerminalTasks { private readonly session: ACPSessionConnection, ) {} - async handleNotification(notification: ServerNotification, sessionId: string): Promise { + async handleNotification( + notification: ServerNotification, + sessionId: string, + commandTitle?: string, + ): Promise { if (!this.isActive()) return; const threadId = notificationThreadId(notification); if (threadId === null) return; if (notification.method === "item/started" && notification.params.item.type === "commandExecution") { - this.observeCommandStarted(notification.params.item, threadId, sessionId); + this.observeCommandStarted(notification.params.item, threadId, sessionId, commandTitle); return; } if (notification.method === "item/completed" && notification.params.item.type === "commandExecution") { @@ -80,13 +84,14 @@ export class CodexBackgroundTerminalTasks { item: CommandExecutionItem, threadId: string, sessionId: string, + commandTitle?: string, ): void { if (!this.isActive() || item.processId === null) return; this.remember(threadId, sessionId, { itemId: item.id, processId: item.processId, command: item.command, - }); + }, commandTitle); } private async observeCommandCompleted( @@ -291,7 +296,7 @@ export class CodexBackgroundTerminalTasks { await this.session.update({ sessionUpdate: "async_task_spawned", asyncTaskId: task.asyncTaskId, - name: task.command, + name: task.name, taskType: "shell", showInTranscript: false, canStop: true, @@ -303,11 +308,13 @@ export class CodexBackgroundTerminalTasks { threadId: string, sessionId: string, terminal: ThreadBackgroundTerminal, + commandTitle?: string, ): Task { const asyncTaskId = wireTaskId(this.rootSessionId, threadId, terminal.itemId); const existing = this.tasks.get(asyncTaskId); if (existing) { existing.processId = terminal.processId; + if (commandTitle !== undefined) existing.name = commandTitle; return existing; } const task: Task = { @@ -316,7 +323,7 @@ export class CodexBackgroundTerminalTasks { asyncTaskId, processId: terminal.processId, itemId: terminal.itemId, - command: terminal.command, + name: commandTitle ?? terminal.command, publication: "unpublished", announcement: null, terminalUpdate: null,