fix: resynchronize current session metadata on ACP v2 resume

src/ACPSessionConnection.ts

@@ -10,6 +10,7 @@ import {toV2ReplayUpdates} from "./AcpV2Replay";
 import {toV1RequestPermissionResponse, toV2RequestPermissionRequest} from "./AcpV2Permissions";
 import {elicitationSessionId, toV1CreateElicitationResponse, toV2CreateElicitationRequest} from "./AcpV2Elicitation";
 import {AcpV2ForegroundState} from "./AcpV2ForegroundState";
+import {AcpV2SessionInfo, type SessionMetadataPatch} from "./AcpV2SessionInfo";
 import {type AcpClientDelivery, clientNotificationSessionId, clientRequestSessionId} from "./AcpClientDelivery";
 
 export type AcpClientConnection = Pick<acp.AgentContext, "notify" | "request">;
@@ -23,6 +24,7 @@ export class AcpV2Connection {
     readonly client: AcpV2ClientConnection;
     private isTurnRunning: (sessionId: string) => boolean = () => true;
     private readonly foreground = new Map<string, AcpV2ForegroundState>();
+    private readonly sessionInfo = new Map<string, AcpV2SessionInfo>();
     private sessionDelivery: AcpClientDelivery | null = null;
     private sendElicitationCompletion: (params: unknown) => Promise<void>;
 
@@ -34,6 +36,10 @@ export class AcpV2Connection {
                 if (method === acpV2.methods.client.elicitation.complete) return this.sendElicitationCompletion(params);
                 const sessionId = clientNotificationSessionId(method, params);
                 if (sessionId !== null && this.sessionDelivery && !this.sessionDelivery.canPublishSession(sessionId)) return Promise.resolve();
+                if (sessionId !== null) {
+                    const {update, _meta} = params as acpV2.UpdateSessionNotification;
+                    if (acpV2.SessionUpdate.isSessionInfoUpdate(update)) return this.sessionInfoFor(sessionId).update(update, _meta);
+                }
                 return client.notify(method as `_${string}`, params);
             },
             request: (method: string, params?: unknown, options?: acpV2.SendRequestOptions) => {
@@ -80,9 +86,10 @@ export class AcpV2Connection {
      * Sends a session update, rendered in the v2 wire shape, through the v2 `session/update`
      * binding. One update can render to several v2 updates (or none), sent in order.
      */
-    async updateSession(sessionId: string, update: AcpSessionUpdate): Promise<void> {
+    async updateSession(sessionId: string, update: AcpSessionUpdate, metadata?: acpV2.UpdateSessionNotification["_meta"]): Promise<void> {
         for (const v2Update of toV2SessionUpdates(update)) {
-            await this.client.notify(acpV2.methods.client.session.update, {sessionId, update: v2Update});
+            await this.client.notify(acpV2.methods.client.session.update,
+                {sessionId, update: v2Update, ...(metadata !== undefined ? {_meta: metadata} : {})});
         }
     }
 
@@ -99,6 +106,24 @@ export class AcpV2Connection {
         return this.foregroundFor(sessionId).snapshot();
     }
 
+    resetSessionInfo(sessionId: string, metadata: Record<string, unknown> | null): void {
+        this.sessionInfo.get(sessionId)?.dispose();
+        this.sessionInfo.delete(sessionId);
+        this.sessionInfoFor(sessionId, metadata);
+    }
+
+    refreshSessionInfo(sessionId: string, metadata: Record<string, unknown> | null): Promise<void> {
+        return metadata ? this.sessionInfoFor(sessionId).seedMetadata(metadata) : Promise.resolve();
+    }
+
+    snapshotSessionInfo(sessionId: string, title: string | null): Promise<void> {
+        return this.sessionInfoFor(sessionId).snapshot(title);
+    }
+
+    patchSessionMetadata(sessionId: string, change: SessionMetadataPatch): Promise<void> {
+        return this.sessionInfoFor(sessionId).patchMetadata(change);
+    }
+
     startForegroundItem(sessionId: string, itemId: string): Promise<void> {
         return this.foregroundFor(sessionId).startItem(itemId);
     }
@@ -114,6 +139,26 @@ export class AcpV2Connection {
     forgetSession(sessionId: string): void {
         this.foreground.get(sessionId)?.dispose();
         this.foreground.delete(sessionId);
+        this.sessionInfo.get(sessionId)?.dispose();
+        this.sessionInfo.delete(sessionId);
+    }
+
+    private sessionInfoFor(sessionId: string, metadata: Record<string, unknown> | null = null): AcpV2SessionInfo {
+        if (this.sessionDelivery && !this.sessionDelivery.canPublishSession(sessionId)) {
+            const closed = new AcpV2SessionInfo(async () => {});
+            closed.dispose();
+            return closed;
+        }
+        let info = this.sessionInfo.get(sessionId);
+        if (!info) {
+            info = new AcpV2SessionInfo((update, notificationMetadata) => {
+                if (this.sessionDelivery && !this.sessionDelivery.canPublishSession(sessionId)) return Promise.resolve();
+                return this.transportClient.notify(acpV2.methods.client.session.update,
+                    {sessionId, update, ...(notificationMetadata !== undefined ? {_meta: notificationMetadata} : {})});
+            }, metadata);
+            this.sessionInfo.set(sessionId, info);
+        }
+        return info;
     }
 
     private foregroundFor(sessionId: string, running = false): AcpV2ForegroundState {
@@ -216,7 +261,7 @@ export class AcpV2Connection {
                 }
                 if (method === acp.methods.client.session.update) {
                     const notification = params as AcpSessionNotification;
-                    return this.updateSession(notification.sessionId, notification.update);
+                    return this.updateSession(notification.sessionId, notification.update, notification._meta);
                 }
                 if (method === acp.methods.client.elicitation.complete) {
                     return this.completeElicitation(params as acp.CompleteElicitationNotification);
@@ -297,6 +342,11 @@ export class ACPSessionConnection {
 
     forSession(sessionId: string): ACPSessionConnection { return new ACPSessionConnection(this.connection, sessionId); }
 
+    /** Domain writers can rebuild an owned metadata value without changing v2's shallow patch semantics. */
+    patchSessionMetadata(change: SessionMetadataPatch): Promise<void> {
+        return this.v2Connection?.patchSessionMetadata(this.sessionId, change) ?? Promise.resolve();
+    }
+
     /** Replay shares live report filtering, with v2 entity initialization after conversion. */
     replayUpdate(update: UpdateSessionEvent, initialized: Set<string>): Promise<void> {
         return this.v2Connection

src/AcpV2SessionInfo.ts

@@ -0,0 +1,81 @@
+import type * as acp from "@agentclientprotocol/sdk/experimental/v2/unstable";
+import {isRecord} from "./permissions/json";
+
+type Update = acp.SessionInfoUpdate & {sessionUpdate: "session_info_update"};
+type NotificationMetadata = acp.UpdateSessionNotification["_meta"];
+export type SessionMetadataPatch = (current: Readonly<Record<string, unknown>>) => Record<string, unknown> | null;
+
+/** Owns current session metadata and orders live patches against resume resynchronization. */
+export class AcpV2SessionInfo {
+    private readonly metadata = new Map<string, unknown>();
+    private tail = Promise.resolve();
+    private disposed = false;
+
+    constructor(
+        private readonly send: (update: Update, notificationMetadata?: NotificationMetadata) => Promise<void>,
+        initialMetadata: Record<string, unknown> | null = null,
+    ) {
+        this.applyMetadata(JSON.parse(JSON.stringify({_meta: initialMetadata}))._meta);
+    }
+
+    update(update: Update, notificationMetadata?: NotificationMetadata): Promise<void> {
+        let copy: {update: Update; _meta?: NotificationMetadata};
+        try { copy = JSON.parse(JSON.stringify({update, _meta: notificationMetadata})); }
+        catch (error) { return Promise.reject(error); }
+        return this.enqueue(async () => {
+            this.applyMetadata(copy.update._meta);
+            await this.publish(copy.update, copy._meta);
+        });
+    }
+
+    patchMetadata(change: SessionMetadataPatch): Promise<void> {
+        return this.enqueue(async () => {
+            const current = JSON.parse(JSON.stringify(Object.fromEntries(this.metadata))) as Record<string, unknown>;
+            const patch = change(current);
+            if (patch === null) return;
+            const update: Update = JSON.parse(JSON.stringify({sessionUpdate: "session_info_update", _meta: patch}));
+            this.applyMetadata(update._meta);
+            await this.publish(update);
+        });
+    }
+
+    seedMetadata(metadata: Record<string, unknown>): Promise<void> {
+        const copy: unknown = JSON.parse(JSON.stringify({_meta: metadata}))._meta;
+        return this.enqueue(async () => { this.applyMetadata(copy); });
+    }
+
+    snapshot(title: string | null): Promise<void> {
+        return this.enqueue(async () => {
+            // The client may hold keys this connection has never seen.
+            await this.publish({sessionUpdate: "session_info_update", title, _meta: null});
+            if (this.metadata.size > 0) {
+                await this.publish({sessionUpdate: "session_info_update", _meta: Object.fromEntries(this.metadata)});
+            }
+        });
+    }
+
+    dispose(): void {
+        this.disposed = true;
+        this.metadata.clear();
+    }
+
+    private applyMetadata(patch: unknown): void {
+        if (patch === null) this.metadata.clear();
+        else if (isRecord(patch)) for (const [key, value] of Object.entries(patch)) {
+            if (value === null) this.metadata.delete(key);
+            else this.metadata.set(key, value);
+        }
+    }
+
+    private enqueue(operation: () => Promise<void>): Promise<void> {
+        const next = this.tail.then(async () => {
+            if (!this.disposed) await operation();
+        });
+        this.tail = next.catch(() => {});
+        return next;
+    }
+
+    private publish(update: Update, notificationMetadata?: NotificationMetadata): Promise<void> {
+        return this.disposed ? Promise.resolve() : this.send(update, notificationMetadata);
+    }
+}

src/CodexAcpServer.ts

@@ -1407,7 +1407,7 @@ export class CodexAcpServer {
             this.sessionIndex.titleWriter(sessionId),
         );
         sessionState.promptSuggestions = this.createPromptSuggestions(sessionState);
-        this.installSessionState(sessionState);
+        this.installSessionState(sessionState, sessionMetadata.threadStatus);
         AcpV2Connection.of(this.connection)?.restoreForegroundState(sessionId, sessionMetadata.threadStatus?.type === "active");
         this.recovery?.markLoaded(sessionState, openGeneration);
         if (settleTrackerReady) {
@@ -1593,13 +1593,24 @@ export class CodexAcpServer {
         );
     }
 
-    private installSessionState(sessionState: SessionState): void {
+    private installSessionState(sessionState: SessionState, threadStatus?: SessionMetadata["threadStatus"]): void {
         const previous = this.sessions.get(sessionState.sessionId);
         previous?.asyncTasks.clear();
         // The new state holds no suggestion, so the old one must drop its result in flight and clear what it published.
         void previous?.promptSuggestions?.clear();
         this.sessions.set(sessionState.sessionId, sessionState);
         this.clientDelivery.openSession(sessionState.sessionId);
+        const v2 = AcpV2Connection.of(this.connection);
+        if (v2) {
+            const metadata = threadStatus ? {codex: {threadStatus}} : null;
+            // Seeding is queued before new live events; metadata belongs to the active session lifetime.
+            if (previous) {
+                if (previous.sessionFailure) sessionState.sessionFailure = previous.sessionFailure;
+                void v2.refreshSessionInfo(sessionState.sessionId, metadata);
+            } else {
+                v2.resetSessionInfo(sessionState.sessionId, metadata);
+            }
+        }
     }
 
     /**
@@ -1962,10 +1973,7 @@ export class CodexAcpServer {
         // Drain live events released by setup/replay before taking the ordered snapshot.
         await this.codexAcpClient.waitForSessionNotifications(sessionId);
         const state = this.getSessionState(sessionId);
-        await new ACPSessionConnection(this.connection, sessionId).update({
-            sessionUpdate: "session_info_update",
-            title: state.sessionTitle,
-        });
+        await AcpV2Connection.of(this.connection)!.snapshotSessionInfo(sessionId, state.sessionTitle);
         await AcpV2Connection.of(this.connection)!.snapshotForegroundState(sessionId);
     }
 
@@ -3567,7 +3575,7 @@ export class CodexAcpServer {
             this.sessionIndex.titleWriter(sessionId),
         );
         sessionState.promptSuggestions = this.createPromptSuggestions(sessionState);
-        this.installSessionState(sessionState);
+        this.installSessionState(sessionState, thread.status);
         AcpV2Connection.of(this.connection)?.restoreForegroundState(sessionId, thread.status.type === "active");
         this.recovery?.markLoaded(sessionState, openGeneration);
         if (knownAccount === "pending") {

src/CodexEventHandler.ts

@@ -56,6 +56,7 @@ import {
 } from "./ContentChunks";
 import {goalSessionInfoUpdate, sameThreadGoalSnapshot, toThreadGoalSnapshot} from "./ThreadGoalSnapshot";
 import {logger} from "./Logger";
+import {recordOrNull} from "./permissions/json";
 import {
     AIR_EXTENSION_VERSION,
     AIR_EXTENSION_VERSION_KEY,
@@ -372,7 +373,17 @@ export class CodexEventHandler {
     }
 
     async clearSessionFailure(): Promise<void> {
+        const failure = this.sessionState.sessionFailure;
         delete this.sessionState.sessionFailure;
+        if (!failure || this.session.protocolVersion !== 2) return;
+        await this.session.patchSessionMetadata(metadata => {
+            const jetbrains = recordOrNull(metadata[JETBRAINS_META_KEY]);
+            const air = recordOrNull(jetbrains?.[AIR_META_KEY]);
+            const published = recordOrNull(air?.[AIR_SESSION_FAILURE_KEY]);
+            if (!air || published?.["id"] !== failure.id || published?.["revision"] !== failure.revision) return null;
+            const {[AIR_SESSION_FAILURE_KEY]: _failure, ...currentAir} = air;
+            return {[JETBRAINS_META_KEY]: {...jetbrains, [AIR_META_KEY]: currentAir}};
+        });
     }
 
     async completeSuccessfulTurn(turnId: string | null): Promise<void> {
@@ -381,7 +392,7 @@ export class CodexEventHandler {
         const active = this.sessionState.sessionFailure;
         if (active?.id !== this.activeFailureIdByScope.get(turnId) || active?.severity !== "warning") return;
         this.activeFailureIdByScope.delete(turnId);
-        delete this.sessionState.sessionFailure;
+        await this.clearSessionFailure();
     }
 
     async handleFailedTurn(turn: Turn): Promise<void> {
@@ -1199,7 +1210,7 @@ export class CodexEventHandler {
         if (activeId === undefined || this.failuresById.get(activeId)?.severity !== "warning") return;
         this.activeFailureIdByScope.delete(turnId);
         if (this.sessionState.sessionFailure?.id === activeId) {
-            delete this.sessionState.sessionFailure;
+            void this.clearSessionFailure().catch(error => logger.error("Failed to clear session failure metadata", error));
         }
     }