From 2108d8e39afbfc40d118f34f6c0f297fb57e3862 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Sat, 31 Oct 2026 00:25:27 +0100 Subject: [PATCH] fix: preserve sibling session metadata namespace fields --- src/ACPSessionConnection.ts | 18 ++- src/AcpV2SessionInfo.ts | 6 +- src/AirExtension.ts | 21 ++- src/CodexAcpServer.ts | 33 ++-- src/CodexEventHandler.ts | 25 ++- src/SessionIndexService.ts | 7 +- src/ThreadGoalSnapshot.ts | 8 +- src/__tests__/AcpV2SessionInfo.test.ts | 11 ++ .../session-metadata-writers.test.ts | 145 ++++++++++++++++++ 9 files changed, 234 insertions(+), 40 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/session-metadata-writers.test.ts diff --git a/src/ACPSessionConnection.ts b/src/ACPSessionConnection.ts index 8688d4684..f495da6d6 100644 --- a/src/ACPSessionConnection.ts +++ b/src/ACPSessionConnection.ts @@ -10,7 +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 {AcpV2SessionInfo, type SessionInfoFields, type SessionMetadataPatch} from "./AcpV2SessionInfo"; import {type AcpClientDelivery, clientNotificationSessionId, clientRequestSessionId} from "./AcpClientDelivery"; export type AcpClientConnection = Pick; @@ -120,8 +120,8 @@ export class AcpV2Connection { return this.sessionInfoFor(sessionId).snapshot(title); } - patchSessionMetadata(sessionId: string, change: SessionMetadataPatch): Promise { - return this.sessionInfoFor(sessionId).patchMetadata(change); + patchSessionMetadata(sessionId: string, change: SessionMetadataPatch, fields: SessionInfoFields = {}): Promise { + return this.sessionInfoFor(sessionId).patchMetadata(change, fields); } startForegroundItem(sessionId: string, itemId: string): Promise { @@ -137,12 +137,16 @@ export class AcpV2Connection { } forgetSession(sessionId: string): void { - this.foreground.get(sessionId)?.dispose(); - this.foreground.delete(sessionId); + this.forgetForegroundState(sessionId); this.sessionInfo.get(sessionId)?.dispose(); this.sessionInfo.delete(sessionId); } + forgetForegroundState(sessionId: string): void { + this.foreground.get(sessionId)?.dispose(); + this.foreground.delete(sessionId); + } + private sessionInfoFor(sessionId: string, metadata: Record | null = null): AcpV2SessionInfo { if (this.sessionDelivery && !this.sessionDelivery.canPublishSession(sessionId)) { const closed = new AcpV2SessionInfo(async () => {}); @@ -343,8 +347,8 @@ 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 { - return this.v2Connection?.patchSessionMetadata(this.sessionId, change) ?? Promise.resolve(); + patchSessionMetadata(change: SessionMetadataPatch, fields: SessionInfoFields = {}): Promise { + return this.v2Connection?.patchSessionMetadata(this.sessionId, change, fields) ?? Promise.resolve(); } /** Replay shares live report filtering, with v2 entity initialization after conversion. */ diff --git a/src/AcpV2SessionInfo.ts b/src/AcpV2SessionInfo.ts index 749fd312f..d4662b008 100644 --- a/src/AcpV2SessionInfo.ts +++ b/src/AcpV2SessionInfo.ts @@ -4,6 +4,7 @@ import {isRecord} from "./permissions/json"; type Update = acp.SessionInfoUpdate & {sessionUpdate: "session_info_update"}; type NotificationMetadata = acp.UpdateSessionNotification["_meta"]; export type SessionMetadataPatch = (current: Readonly>) => Record | null; +export type SessionInfoFields = Omit; /** Owns current session metadata and orders live patches against resume resynchronization. */ export class AcpV2SessionInfo { @@ -28,12 +29,13 @@ export class AcpV2SessionInfo { }); } - patchMetadata(change: SessionMetadataPatch): Promise { + patchMetadata(change: SessionMetadataPatch, fields: SessionInfoFields = {}): Promise { + const copy: SessionInfoFields = JSON.parse(JSON.stringify(fields)); return this.enqueue(async () => { const current = JSON.parse(JSON.stringify(Object.fromEntries(this.metadata))) as Record; const patch = change(current); if (patch === null) return; - const update: Update = JSON.parse(JSON.stringify({sessionUpdate: "session_info_update", _meta: patch})); + const update: Update = JSON.parse(JSON.stringify({...copy, sessionUpdate: "session_info_update", _meta: patch})); this.applyMetadata(update._meta); await this.publish(update); }); diff --git a/src/AirExtension.ts b/src/AirExtension.ts index 50c22f3e7..a09e0b146 100644 --- a/src/AirExtension.ts +++ b/src/AirExtension.ts @@ -1,4 +1,5 @@ import type {ClientCapabilities} from "@agentclientprotocol/sdk/unstable"; +import type {ACPSessionConnection} from "./ACPSessionConnection"; /** * Wire names for the versioned JetBrains AIR ACP extension. @@ -46,6 +47,14 @@ export function withAirMeta( meta: Record | null | undefined, key: string, value: unknown, +): Record { + return withAirMetadata(meta, {[key]: value}); +} + +/** Compose AIR-owned fields; this is a producer operation, not a protocol metadata merge. */ +export function withAirMetadata( + meta: Record | null | undefined, + fields: Record, ): Record { const root = asRecord(meta); const jetbrains = asRecord(root[JETBRAINS_META_KEY]); @@ -57,12 +66,22 @@ export function withAirMeta( [AIR_META_KEY]: { ...air, [AIR_EXTENSION_VERSION_KEY]: AIR_EXTENSION_VERSION, - [key]: value, + ...fields, }, }, }; } +/** A v2 session patch replaces the whole JetBrains value, so compose it at publication time. */ +export function publishAirSessionMetadata(session: ACPSessionConnection, key: string, value: unknown): Promise { + if (session.protocolVersion === 2) { + return session.patchSessionMetadata(current => ({ + [JETBRAINS_META_KEY]: withAirMeta(current, key, value)[JETBRAINS_META_KEY], + })); + } + return session.update({sessionUpdate: "session_info_update", _meta: withAirMeta(undefined, key, value)}); +} + /** * The metadata of a key that exists only for AIR. * AIR gets `_meta.jetbrains.air.`. Every other client gets no metadata. diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index b83503524..722988938 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -165,7 +165,7 @@ import { createUserMessageChunk, } from "./ContentChunks"; import { - goalSessionInfoUpdate, + publishGoalSessionInfo, sameThreadGoalSnapshot, type ThreadGoalSnapshot, toThreadGoalSnapshot, @@ -217,6 +217,7 @@ import { AIR_SESSION_FAILURE_KEY, clientSupportsAirCapability, JETBRAINS_META_KEY, + publishAirSessionMetadata, } from "./AirExtension"; import {SessionIndexService} from "./SessionIndexService"; import type {SessionIndexTitleState} from "./SessionIndexTitles"; @@ -1238,7 +1239,7 @@ export class CodexAcpServer { private endSessionCloseFence(sessionId: string): void { const fence = this.closingSessions.get(sessionId); if (!fence || --fence.count > 0) return; - for (const closed of fence.closedSessions) this.clientDelivery.closeSession(closed); + for (const closed of fence.closedSessions) this.sealSessionDelivery(closed); this.closingSessions.delete(sessionId); } @@ -1247,10 +1248,15 @@ export class CodexAcpServer { const fence = this.closingSessions.get(sessionId); for (const closed of [sessionId, ...children]) { if (fence) fence.closedSessions.add(closed); - else this.clientDelivery.closeSession(closed); + else this.sealSessionDelivery(closed); } } + private sealSessionDelivery(sessionId: string): void { + this.clientDelivery.closeSession(sessionId); + AcpV2Connection.of(this.connection)?.forgetSession(sessionId); + } + private getSessionGeneration(sessionId: string): number { return this.sessionGenerations.get(sessionId) ?? 0; } @@ -2092,7 +2098,7 @@ export class CodexAcpServer { this.sessions.delete(params.sessionId); this.asyncQuestions.get(params.sessionId)?.dispose(); this.asyncQuestions.delete(params.sessionId); - AcpV2Connection.of(this.connection)?.forgetSession(params.sessionId); + AcpV2Connection.of(this.connection)?.forgetForegroundState(params.sessionId); this.mcpSessionStartup.close(params.sessionId); this.codexAcpClient.appServerClient.mcpStartup.forgetThread(params.sessionId); this.pendingTurnStarts.delete(params.sessionId); @@ -3399,9 +3405,8 @@ export class CodexAcpServer { return; } sessionState.currentGoal = snapshot; - const update = goalSessionInfoUpdate(snapshot, sessionState.clientCapabilities.airClient); - if (update === null) return; - await new ACPSessionConnection(this.connection, sessionState.sessionId).update(update); + await publishGoalSessionInfo(new ACPSessionConnection(this.connection, sessionState.sessionId), + snapshot, sessionState.clientCapabilities.airClient); } /** The models of a root session, for the configuration of its native subagent sessions. */ @@ -3982,17 +3987,7 @@ export class CodexAcpServer { } try { const session = new ACPSessionConnection(this.connection, sessionState.sessionId); - await session.update({ - sessionUpdate: "session_info_update", - _meta: { - [JETBRAINS_META_KEY]: { - [AIR_META_KEY]: { - [AIR_EXTENSION_VERSION_KEY]: AIR_EXTENSION_VERSION, - [AIR_AGENT_FILE_CHANGE_REPORT_KEY]: report, - }, - }, - }, - }); + await publishAirSessionMetadata(session, AIR_AGENT_FILE_CHANGE_REPORT_KEY, report); } catch (error) { logger.error("Failed to publish agent file-change report", error); } @@ -4751,7 +4746,7 @@ export class CodexAcpServer { ).finally(() => { signal?.removeEventListener("abort", dropPendingRequest); reservation.release(); - if (!this.sessions.has(sessionId)) AcpV2Connection.of(this.connection)?.forgetSession(sessionId); + if (!this.sessions.has(sessionId)) AcpV2Connection.of(this.connection)?.forgetForegroundState(sessionId); }); }); } diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 6a15f096e..31e0f790e 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -63,6 +63,7 @@ import { AIR_META_KEY, AIR_SESSION_FAILURE_KEY, JETBRAINS_META_KEY, + withAirMetadata, } from "./AirExtension"; import type {SubagentEventRouter} from "./subagents/SubagentEventRouter"; import {StandardSubagentEventRouter} from "./subagents/StandardSubagentEventRouter"; @@ -335,7 +336,7 @@ export class CodexEventHandler { await this.finishCompactionsForNotification(notification); if (this.isAuthenticationRequiredError(notification.params.error.codexErrorInfo)) return; if (notification.params.willRetry) { - await this.session.update(this.createSessionFailureUpdate(this.recordRetryWarning(notification.params, false))); + await this.publishUpdate(this.createSessionFailureUpdate(this.recordRetryWarning(notification.params, false))); return; } const failure = this.recordSessionFailure( @@ -346,7 +347,7 @@ export class CodexEventHandler { undefined, false, ); - await this.session.update(this.createSessionFailureUpdate(failure)); + await this.publishUpdate(this.createSessionFailureUpdate(failure)); } async flushPendingErrors(): Promise { @@ -357,7 +358,7 @@ export class CodexEventHandler { for (const error of errors) { const update = await this.createErrorEvent(error); if (update) { - await this.session.update(update); + await this.publishUpdate(update); } } } @@ -484,7 +485,7 @@ export class CodexEventHandler { if (ignoredBySubagents) return; if (updateEvent === undefined) updateEvent = await this.createUpdateEvent(notification); if (updateEvent) { - await this.session.update(updateEvent, this.subagents.notificationSessionId(notification)); + await this.publishUpdate(updateEvent, this.subagents.notificationSessionId(notification)); } } @@ -796,6 +797,22 @@ export class CodexEventHandler { } } + /** These fact builders produce partial owned namespaces, unlike raw metadata patches. */ + private publishUpdate(update: UpdateSessionEvent, sessionId = this.sessionState.sessionId): Promise { + if (this.session.protocolVersion !== 2 || update.sessionUpdate !== "session_info_update" || update._meta == null) { + return this.session.update(update, sessionId); + } + const {_meta, sessionUpdate: _tag, ...fields} = update; + return this.session.forSession(sessionId).patchSessionMetadata(current => { + const metadata = {..._meta}; + const codex = recordOrNull(_meta["codex"]); + if (codex) metadata["codex"] = {...recordOrNull(current["codex"]), ...codex}; + const air = recordOrNull(recordOrNull(_meta[JETBRAINS_META_KEY])?.[AIR_META_KEY]); + if (air) metadata[JETBRAINS_META_KEY] = withAirMetadata(current, air)[JETBRAINS_META_KEY]; + return metadata; + }, fields); + } + private createCodexSessionInfoUpdate(codexMetadata: Record): UpdateSessionEvent { return { sessionUpdate: "session_info_update", diff --git a/src/SessionIndexService.ts b/src/SessionIndexService.ts index 6cd7792d2..b0b56c9ec 100644 --- a/src/SessionIndexService.ts +++ b/src/SessionIndexService.ts @@ -8,7 +8,7 @@ import path from "node:path"; import type * as acp from "@agentclientprotocol/sdk/unstable"; import {RequestError} from "@agentclientprotocol/sdk/unstable"; import {ACPSessionConnection, type AcpClientConnection} from "./ACPSessionConnection"; -import {clientSupportsAirCapability, withAirMeta} from "./AirExtension"; +import {clientSupportsAirCapability, publishAirSessionMetadata} from "./AirExtension"; import type {ServerNotification} from "./app-server"; import type {CodexAcpClient} from "./CodexAcpClient"; import type {SessionState} from "./CodexAcpServer"; @@ -274,10 +274,7 @@ export class SessionIndexService { }); // Tell the client why it closed before ending this operation's delivery fence. if (archived ? open : this.host.session(sessionId) !== undefined) { - await new ACPSessionConnection(this.host.connection(), sessionId).update({ - sessionUpdate: "session_info_update", - _meta: withAirMeta(undefined, AIR_ARCHIVED_KEY, archived), - }); + await publishAirSessionMetadata(new ACPSessionConnection(this.host.connection(), sessionId), AIR_ARCHIVED_KEY, archived); } } finally { if (archived) this.host.endSessionCloseFence(sessionId); diff --git a/src/ThreadGoalSnapshot.ts b/src/ThreadGoalSnapshot.ts index 516880345..54a4d7c7a 100644 --- a/src/ThreadGoalSnapshot.ts +++ b/src/ThreadGoalSnapshot.ts @@ -1,5 +1,5 @@ -import type {UpdateSessionEvent} from "./ACPSessionConnection"; -import {AIR_GOAL_KEY, withAirMeta} from "./AirExtension"; +import type {ACPSessionConnection, UpdateSessionEvent} from "./ACPSessionConnection"; +import {AIR_GOAL_KEY, publishAirSessionMetadata, withAirMeta} from "./AirExtension"; import {GOAL_CONTROL_METHOD, type GoalSnapshot, type GoalStatus} from "./GoalExtension"; import type {ThreadGoal} from "./app-server/v2"; @@ -55,3 +55,7 @@ export function goalSessionInfoUpdate(goal: ThreadGoalSnapshot | null, airClient _meta: withAirMeta(undefined, AIR_GOAL_KEY, goal), }; } + +export function publishGoalSessionInfo(session: ACPSessionConnection, goal: ThreadGoalSnapshot | null, airClient: boolean): Promise { + return airClient ? publishAirSessionMetadata(session, AIR_GOAL_KEY, goal) : Promise.resolve(); +} diff --git a/src/__tests__/AcpV2SessionInfo.test.ts b/src/__tests__/AcpV2SessionInfo.test.ts index 0c30dfc0c..adc661111 100644 --- a/src/__tests__/AcpV2SessionInfo.test.ts +++ b/src/__tests__/AcpV2SessionInfo.test.ts @@ -74,6 +74,17 @@ describe("V2 session metadata", () => { expect(sent.at(-1)?._meta).toEqual({ns: {version: 1, added: true}}); }); + it("preserves accompanying session info fields while composing metadata", async () => { + const sent: Update[] = []; + const info = new AcpV2SessionInfo(async update => { sent.push(update); }); + const fields = {title: "Current", updatedAt: "2026-10-30T12:00:00Z"}; + const pending = info.patchMetadata(() => ({current: true}), fields); + fields.title = "Changed after publication"; + await pending; + expect(sent).toEqual([{title: "Current", updatedAt: "2026-10-30T12:00:00Z", + sessionUpdate: "session_info_update", _meta: {current: true}}]); + }); + it("stops queued and in-flight restoration after disposal", async () => { const sent: Update[] = []; const clearing = deferred(); diff --git a/src/__tests__/CodexACPAgent/session-metadata-writers.test.ts b/src/__tests__/CodexACPAgent/session-metadata-writers.test.ts new file mode 100644 index 000000000..de495d333 --- /dev/null +++ b/src/__tests__/CodexACPAgent/session-metadata-writers.test.ts @@ -0,0 +1,145 @@ +import * as acp from "@agentclientprotocol/sdk/unstable"; +import {describe, expect, it, vi} from "vitest"; +import {ACPSessionConnection, AcpV2Connection} from "../../ACPSessionConnection"; +import {publishAirSessionMetadata} from "../../AirExtension"; +import {CodexEventHandler} from "../../CodexEventHandler"; +import {SESSION_ARCHIVE_METHOD} from "../../SessionIndex"; +import {connectSession, createThread, sessionId, type PromptSession} from "./v2-prompt-harness"; + +const airCapabilities = {_meta: {jetbrains: {air: { + version: 1, capabilities: ["sessionFailure", "agentFileChangeReport", "sessionIndex"], +}}}}; +const goal = {threadId: sessionId, objective: "Current goal", status: "active" as const, + tokenBudget: null, tokensUsed: 1, timeUsedSeconds: 1, createdAt: 100, updatedAt: 101}; + +function metadata(h: PromptSession): Record { + let current: Record = {}; + for (const {sessionId: id, update} of h.scopedUpdates) { + if (id !== sessionId || update.sessionUpdate !== "session_info_update") continue; + if (update._meta === null) current = {}; + else if (update._meta && typeof update._meta === "object") { + for (const [key, value] of Object.entries(update._meta)) { + if (value === null) delete current[key]; + else current[key] = value; + } + } + } + return current; +} + +async function flush(h: PromptSession) { + await h.waitForNotifications(); + await h.request("session/list", {}); +} + +describe.each([1, 2] as const)("ACP v%s owned session metadata", version => { + it("composes goals, reports, failures and archive without changing v1 payloads", async () => { + const h = await connectSession(version, {clientCapabilities: airCapabilities, v1ClientCapabilities: airCapabilities}); + try { + // @ts-expect-error - seed unrelated metadata through the shared client path + const outbound = h.agent.connection as acp.AgentContext; + await outbound.notify(acp.methods.client.session.update, {sessionId, update: { + sessionUpdate: "session_info_update", + _meta: {unrelated: true, jetbrains: {other: {literal: null}, air: {version: 1, sibling: true}}}, + }}); + h.emit({method: "thread/goal/updated", params: {threadId: sessionId, turnId: null, goal}}); + await flush(h); + if (version === 2) { + expect(metadata(h)["jetbrains"]).toMatchObject({other: {literal: null}, air: {sibling: true, goal: {objective: goal.objective}}}); + } else { + expect(metadata(h)["jetbrains"].air.sibling).toBeUndefined(); + } + const publishedGoal = metadata(h)["jetbrains"].air.goal; + await h.request("session/prompt", {sessionId, prompt: [{type: "text", text: "/plan"}], + _meta: {jetbrains: {air: {agentFileChangeReportRequest: {version: 1, requestId: "report"}}}}, + }); + await h.promptRunFinished(); + await flush(h); + const report = metadata(h)["jetbrains"].air.agentFileChangeReport; + expect(report).toMatchObject({requestId: "report"}); + if (version === 2) expect(metadata(h)["jetbrains"].air.goal).toEqual(publishedGoal); + else expect(metadata(h)["jetbrains"].air.goal).toBeUndefined(); + + h.emit({method: "error", params: {threadId: sessionId, turnId: "background-turn", willRetry: true, + error: {message: "Retrying", codexErrorInfo: null, additionalDetails: null, misalignment: null}}}); + await flush(h); + expect(metadata(h)["jetbrains"].air.sessionFailure).toMatchObject({severity: "warning"}); + if (version === 2) expect(metadata(h)["jetbrains"].air.agentFileChangeReport).toEqual(report); + else expect(metadata(h)["jetbrains"].air.agentFileChangeReport).toBeUndefined(); + const failure = metadata(h)["jetbrains"].air.sessionFailure; + + h.emit({method: "thread/goal/cleared", params: {threadId: sessionId}}); + await flush(h); + expect(metadata(h)["jetbrains"].air.goal).toBeNull(); + if (version === 2) expect(metadata(h)["jetbrains"].air.sessionFailure).toEqual(failure); + else expect(metadata(h)["jetbrains"].air.sessionFailure).toBeUndefined(); + + h.setCodexResponse("thread/read", async () => ({thread: {...createThread(), path: "/codex/sessions/rollout.jsonl"}})); + await h.request(SESSION_ARCHIVE_METHOD, {sessionId}); + const final = metadata(h); + expect(final["jetbrains"].air.archived).toBe(true); + if (version === 2) { + expect(final["jetbrains"]).toMatchObject({other: {literal: null}, air: {sibling: true, goal: null}}); + expect(final["jetbrains"].air.agentFileChangeReport).toEqual(report); + expect(final["jetbrains"].air.sessionFailure).toEqual(failure); + } else { + expect(final["jetbrains"]).toEqual({air: {version: 1, archived: true}}); + } + expect(final["unrelated"]).toBe(true); + const boundary = h.scopedUpdates.length; + await publishAirSessionMetadata(new ACPSessionConnection(outbound, sessionId), "late", true); + await flush(h); + expect(h.scopedUpdates).toHaveLength(boundary); + } finally { h.connection.close(); } + }); + + it("preserves independent Codex fields across status and archive events on v2", async () => { + const h = await connectSession(version); + try { + h.emit({method: "thread/archived", params: {threadId: sessionId}}); + h.emit({method: "thread/status/changed", params: {threadId: sessionId, status: {type: "idle"}}}); + await flush(h); + expect(metadata(h)["codex"]).toEqual(version === 2 + ? {archived: true, threadStatus: {type: "idle"}} : {threadStatus: {type: "idle"}}); + h.emit({method: "thread/unarchived", params: {threadId: sessionId}}); + await flush(h); + expect(metadata(h)["codex"]).toEqual(version === 2 + ? {archived: false, threadStatus: {type: "idle"}} : {archived: false}); + } finally { h.connection.close(); } + }); +}); + +describe("V2 namespace transactions", () => { + it("composes concurrent AIR writers at publication time", async () => { + const h = await connectSession(2); + try { + // @ts-expect-error - exercise the writer on the actual shared client connection + const session = new ACPSessionConnection(h.agent.connection, sessionId); + await Promise.all([ + publishAirSessionMetadata(session, "first", {literal: null}), + publishAirSessionMetadata(session, "second", true), + ]); + await flush(h); + expect(metadata(h)["jetbrains"]).toEqual({air: {version: 1, first: {literal: null}, second: true}}); + } finally { h.connection.close(); } + }); + + it("composes after routing without copying parent metadata into a child", async () => { + const h = await connectSession(2); + try { + // @ts-expect-error - seed both metadata scopes through the shared connection + const outbound = h.agent.connection as acp.AgentContext; + const v2 = AcpV2Connection.of(outbound)!; + await v2.client.notify("session/update", {sessionId, update: {sessionUpdate: "session_info_update", _meta: {codex: {parent: true}}}}); + await v2.client.notify("session/update", {sessionId: "child", update: {sessionUpdate: "session_info_update", _meta: {codex: {child: true}}}}); + const state = h.agent.getSessionState(sessionId); + vi.spyOn(state.subagents, "notificationSessionId").mockReturnValue("child"); + const handler = new CodexEventHandler(outbound, state, false, "epoch", state.subagents, undefined, false, false, false); + await handler.handleNotification({method: "thread/status/changed", params: {threadId: sessionId, status: {type: "idle"}}}); + await flush(h); + const child = h.scopedUpdates.filter(entry => entry.sessionId === "child").at(-1)?.update; + expect(child?._meta).toEqual({codex: {child: true, threadStatus: {type: "idle"}}}); + expect(metadata(h)["codex"]).toEqual({parent: true}); + } finally { h.connection.close(); } + }); +});