diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index d7d466ad..e9fdfe85 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1416,6 +1416,7 @@ export class CodexAcpServer { } await this.trackCodexTurnStart(sessionState, event); await this.trackSteerLanding(sessionState, event); + this.handleSkillsChanged(sessionState, event); // Codex-started turns carry no `userMessage` item; `handleSessionScopedNotification` // already drops that item type, so nothing is synthesized here. await baselineEventHandler.handleSessionScopedNotification(event); @@ -1782,6 +1783,9 @@ export class CodexAcpServer { this.turnStartQueueTail.delete(params.sessionId); this.queuedV2PromptCancellers.delete(params.sessionId); this.steeringQueues.delete(params.sessionId); + if (sessionState) { + this.availableCommands.cancelPendingSkillsPublish(sessionState); + } } this.endSessionCloseFence(params.sessionId); } @@ -2960,6 +2964,24 @@ export class CodexAcpServer { await this.publishGoalSnapshot(sessionState, snapshot, force, false); } + /** + * Lists the skills of a session again when Codex reports that its skill files changed. + * + * The notification carries no thread, so app-server sends it to every subscribed session, and each one lists + * its own cwds. A skill that appears while a session is open - a new file, a new symlink, a changed link + * target - becomes a command right away instead of only after the next turn. + */ + private handleSkillsChanged(sessionState: SessionState, event: ServerNotification): void { + if (event.method !== "skills/changed") { + return; + } + const sessionGeneration = this.getSessionGeneration(sessionState.sessionId); + this.availableCommands.publishOnSkillsChanged( + sessionState, + () => this.sessionPublishIsCurrent(sessionState, sessionGeneration), + ); + } + private sessionPublishIsCurrent(sessionState: SessionState, sessionGeneration: number): boolean { return this.sessions.get(sessionState.sessionId) === sessionState && this.getSessionGeneration(sessionState.sessionId) === sessionGeneration @@ -4368,6 +4390,7 @@ export class CodexAcpServer { // notifications for as long as no later prompt replaces it. await this.trackCodexTurnStart(sessionState, event); await this.trackSteerLanding(sessionState, event); + this.handleSkillsChanged(sessionState, event); if (pendingInsertion !== undefined && isInsertedUserMessage(event, params.sessionId, pendingInsertion.clientUserMessageId)) { await resolvePendingInsertion(); @@ -4722,11 +4745,13 @@ export class CodexAcpServer { await clearRecoveredSessionFailure(eventHandler); - // Codex sends no notification for a new skill file. A skill that appeared during the turn becomes a - // slash command after it. Never await: the prompt response does not wait for the skill list. + // A skill that appeared during the turn becomes a slash command after it, even when the + // `skills/changed` notification of its file was missed. Never await: the prompt response does not + // wait for the skill list. + const turnSessionGeneration = this.getSessionGeneration(sessionState.sessionId); void this.availableCommands.publish( sessionState, - () => this.sessions.get(sessionState.sessionId) === sessionState, + () => this.sessionPublishIsCurrent(sessionState, turnSessionGeneration), true, ); diff --git a/src/CodexCommands.ts b/src/CodexCommands.ts index 673f741f..43c59319 100644 --- a/src/CodexCommands.ts +++ b/src/CodexCommands.ts @@ -46,6 +46,15 @@ export type CommandHandleOptions = { export type LogoutHandler = () => void | Promise; +/** How long a `skills/changed` burst settles before the commands are listed again. */ +const SKILLS_CHANGED_PUBLISH_DELAY_MS = 300; + +/** How long a `skills/changed` burst may hold the publish back before it goes out anyway. */ +const SKILLS_CHANGED_PUBLISH_MAX_DELAY_MS = 2_000; + +/** A publish that waits out a burst of skill file changes, and the moment it stops waiting for more. */ +type PendingSkillsPublish = {timer: ReturnType; deadline: number}; + export class CodexCommands { private readonly connection: AcpClientConnection; private readonly codexAcpClient: CodexAcpClient; @@ -54,6 +63,10 @@ export class CodexCommands { private readonly mcpCommand: McpCommand; /** The commands that each session got last, to skip a publish that changes nothing. */ private readonly published = new WeakMap(); + /** The pending publish of each session whose skills changed, so one burst of file changes lists once. */ + private readonly pendingSkillsPublishes = new WeakMap(); + /** The publish each session started last, so a slower earlier one does not overwrite its list. */ + private readonly publishRevisions = new WeakMap(); constructor( connection: AcpClientConnection, @@ -69,6 +82,39 @@ export class CodexCommands { this.mcpCommand = new McpCommand(codexAcpClient, runWithProcessCheck, mcpServerSignIn); } + /** + * Lists the skills again and publishes the commands once the skill files settle. + * + * Codex watches the skill files and reports every change as `skills/changed`, naming nothing that changed, so + * the list is the only way to learn what a session has now. An editor that writes a skill, or a checkout that + * adds one, fires several of those notifications, so the publish waits out the burst. Each session keeps one + * pending publish, and a publish that changes no command sends nothing. + */ + publishOnSkillsChanged(sessionState: SessionState, shouldPublish: () => boolean): void { + const pending = this.pendingSkillsPublishes.get(sessionState); + clearTimeout(pending?.timer); + // A checkout, or a watcher that keeps writing, would otherwise push the publish out forever. + const deadline = pending?.deadline ?? Date.now() + SKILLS_CHANGED_PUBLISH_MAX_DELAY_MS; + const delay = Math.max(0, Math.min(SKILLS_CHANGED_PUBLISH_DELAY_MS, deadline - Date.now())); + const timer = setTimeout(() => { + this.pendingSkillsPublishes.delete(sessionState); + void this.publish(sessionState, shouldPublish, true); + }, delay); + // The adapter must not stay alive for a publish that no one waits for. + timer.unref?.(); + this.pendingSkillsPublishes.set(sessionState, {timer, deadline}); + } + + /** Drops the publish a closing session was still waiting out. */ + cancelPendingSkillsPublish(sessionState: SessionState): void { + const pending = this.pendingSkillsPublishes.get(sessionState); + if (pending === undefined) { + return; + } + clearTimeout(pending.timer); + this.pendingSkillsPublishes.delete(sessionState); + } + /** Sends the available commands. With `onlyChanges`, it sends nothing when the commands did not change. */ async publish( sessionState: SessionState, @@ -79,11 +125,17 @@ export class CodexCommands { if (!shouldPublish()) { return; } + const revision = (this.publishRevisions.get(sessionState) ?? 0) + 1; + this.publishRevisions.set(sessionState, revision); const skillsResponse = await this.runWithProcessCheck(() => this.codexAcpClient.listSkills(this.createSkillsListParams(sessionState))); const availableCommands = this.buildAvailableCommands( skillsResponse?.data ?? [], sessionState.clientCapabilities.airClient, ); + // A publish that started later already listed the skills, and its list is the newer one. + if (this.publishRevisions.get(sessionState) !== revision) { + return; + } if (availableCommands.length === 0 || !shouldPublish()) { return; } diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index c11f74eb..293b8fd7 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -697,6 +697,9 @@ export class CodexEventHandler { case "account/gatewayOAuth/changed": // Gateway login state is account-scoped and has no ACP session update counterpart. return null; + case "skills/changed": + // The server lists the skills again and publishes the commands. The change itself is no update. + return null; // ignored events case "thread/deleted": case "thread/reverted": @@ -733,7 +736,6 @@ export class CodexEventHandler { case "thread/realtime/closed": case "windowsSandbox/setupCompleted": case "account/login/completed": - case "skills/changed": case "mcpServer/oauthLogin/completed": case "externalAgentConfig/import/completed": case "rawResponseItem/completed": diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 19a969c3..6bdf47d4 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -2013,6 +2013,36 @@ describe('ACP server test', { timeout: 40_000 }, () => { expect(published()).toBe(2); }); + // The event-driven side of `skills/changed` lives in `skills-changed.test.ts`; this covers only the cap + // that keeps a burst from holding the publish back forever. + it('publishes a never-ending burst of skill file changes anyway', async () => { + vi.useFakeTimers(); + try { + const mockFixture = createCodexMockTestFixture(); + const codexAcpAgent = mockFixture.getCodexAcpAgent(); + const listSkills = vi.spyOn(mockFixture.getCodexAcpClient(), "listSkills") + .mockResolvedValue({data: [{cwd: "/workspace", skills: [{ + name: "build", description: "build", shortDescription: "build", + path: "/workspace", scope: "user" as const, enabled: true, pluginId: null, + }], errors: []}]}); + const sessionState = createTestSessionState({sessionId: "session-id", cwd: "/workspace"}); + // @ts-expect-error - exercising private state + codexAcpAgent.sessions.set(sessionState.sessionId, sessionState); + // @ts-expect-error - exercising private helper + const skillsChanged = () => codexAcpAgent.handleSkillsChanged(sessionState, {method: "skills/changed", params: {}}); + + // A writer that never pauses for 300 ms: a trailing debounce alone would never fire. + for (let elapsed = 0; elapsed < 2_500; elapsed += 200) { + skillsChanged(); + await vi.advanceTimersByTimeAsync(200); + } + + expect(listSkills).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + it('handles builtin slash command locally', async () => { const mockFixture = createCodexMockTestFixture(); const codexAcpAgent = mockFixture.getCodexAcpAgent(); diff --git a/src/__tests__/CodexACPAgent/skills-changed.test.ts b/src/__tests__/CodexACPAgent/skills-changed.test.ts new file mode 100644 index 00000000..61c36d1f --- /dev/null +++ b/src/__tests__/CodexACPAgent/skills-changed.test.ts @@ -0,0 +1,130 @@ +import {afterEach, describe, expect, it} from 'vitest'; +import {connectSession, settle, type PromptSession, type TranscriptEntry} from './v2-prompt-harness'; +import type * as acpV2 from '@agentclientprotocol/sdk/experimental/v2'; +import type {ServerNotification} from '../../app-server'; + +/** + * Codex reports every change under a skill root as `skills/changed`, naming nothing that changed. The adapter + * lists the skills again and publishes the slash commands, so a skill that appears while a session is open - a + * new file, a new symlink, a changed link target - becomes a command right away instead of only after the next + * completed turn. + */ +describe('skills/changed republishes the available commands', () => { + let closeClient: (() => void) | null = null; + + afterEach(() => { + closeClient?.(); + closeClient = null; + }); + + const skillsChanged = (): ServerNotification => + ({method: "skills/changed", params: {}} as unknown as ServerNotification); + + const skills = (...names: string[]) => ({ + data: [{ + cwd: "/workspace", + errors: [], + skills: names.map(name => ({ + name, + description: name, + shortDescription: name, + path: `/workspace/.agents/skills/${name}`, + scope: "project" as const, + enabled: true, + pluginId: null, + })), + }], + }); + + const commandNames = (transcript: TranscriptEntry[]): string[][] => + transcript.flatMap(entry => + "sessionUpdate" in entry && entry.sessionUpdate.sessionUpdate === "available_commands_update" + ? [(entry.sessionUpdate.availableCommands as acpV2.AvailableCommand[]).map(command => command.name)] + : []); + + /** Waits out the publish the notification debounces, plus the wire round-trip behind it. */ + const settleSkillsPublish = async (client: PromptSession) => { + await new Promise(resolve => setTimeout(resolve, 400)); + await settle(); + }; + + it('publishes a skill that appeared before any turn has run', async () => { + const client = await connectSession(); + closeClient = () => client.connection.close(); + client.setCodexResponse("skills/list", async () => skills("deploy")); + const start = client.transcript.length; + + client.emit(skillsChanged()); + await settleSkillsPublish(client); + + expect(commandNames(client.transcript.slice(start)).at(-1)).toContain("$deploy"); + }); + + it('publishes once for a burst of changes, and nothing when the skills are the same', async () => { + const client = await connectSession(); + closeClient = () => client.connection.close(); + let listed = 0; + client.setCodexResponse("skills/list", async () => { + listed += 1; + return skills("deploy"); + }); + + client.emit(skillsChanged()); + client.emit(skillsChanged()); + client.emit(skillsChanged()); + await settleSkillsPublish(client); + + expect(listed).toBe(1); + const start = client.transcript.length; + + // The same skills: the list is read again, but the client already has those commands. + client.emit(skillsChanged()); + await settleSkillsPublish(client); + + expect(listed).toBe(2); + expect(commandNames(client.transcript.slice(start))).toEqual([]); + }); + + it('keeps the newer list when a slower earlier publish answers last', async () => { + const client = await connectSession(); + closeClient = () => client.connection.close(); + let listed = 0; + client.setCodexResponse("skills/list", async () => { + listed += 1; + if (listed === 1) { + // The first list is answered long after the second one, and it carries the older skills. + await new Promise(resolve => setTimeout(resolve, 1_000)); + return skills("build"); + } + return skills("build", "deploy"); + }); + const start = client.transcript.length; + + client.emit(skillsChanged()); + await new Promise(resolve => setTimeout(resolve, 400)); + client.emit(skillsChanged()); + await new Promise(resolve => setTimeout(resolve, 1_500)); + await settle(); + + expect(listed).toBe(2); + expect(commandNames(client.transcript.slice(start)).at(-1)).toContain("$deploy"); + }); + + it('publishes nothing for a session that closed before its change settled', async () => { + const client = await connectSession(); + closeClient = () => client.connection.close(); + let listed = 0; + client.setCodexResponse("skills/list", async () => { + listed += 1; + return skills("deploy"); + }); + const listedAtClose = () => listed; + + client.emit(skillsChanged()); + await client.request("session/close", {sessionId: "thread-1"}); + const before = listedAtClose(); + await settleSkillsPublish(client); + + expect(listed).toBe(before); + }); +});