diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 3a02c45b23f..b32f9fa07f7 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts @@ -24,6 +24,7 @@ import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts"; const ROOT = wireFixture.rootThreadId; const [CHILD_A, CHILD_B] = wireFixture.childThreadIds as [string, string]; +const MEMORY = "memory-consolidation-thread"; /** * The captured sequence, extended with the shapes the live capture didn't @@ -174,15 +175,44 @@ describe("CodexSessionRuntime collab integration", () => { const turnStartedB = byIndex.find((entry) => isTurnStarted(entry, CHILD_B)); const registrationA = byIndex.find((entry) => isRegistration(entry, CHILD_A)); const registrationB = byIndex.find((entry) => isRegistration(entry, CHILD_B)); + const rootThreadStarted = byIndex.find((entry) => entry.method === "thread/started"); assert.isDefined(turnStartedA); assert.isDefined(turnStartedB); assert.isDefined(registrationA); assert.isDefined(registrationB); + assert.isDefined(rootThreadStarted); + const memoryThreadStarted = { + ...rootThreadStarted, + params: { + thread: { + ...rootThreadStarted.params.thread, + id: MEMORY, + sessionId: MEMORY, + source: "unknown", + threadSource: "memory_consolidation", + }, + }, + }; + const memoryTurnStarted = { + ...turnStartedA, + params: { + ...turnStartedA.params, + threadId: MEMORY, + turn: { ...turnStartedA.params.turn, id: "memory-consolidation-turn" }, + }, + }; const script = { rootThreadId: ROOT, holdTurnOpen: true, hangInterruptFor: CHILD_A, - notifications: [turnStartedA, registrationA, registrationB, turnStartedB], + notifications: [ + turnStartedA, + registrationA, + memoryThreadStarted, + memoryTurnStarted, + registrationB, + turnStartedB, + ], }; // @effect-diagnostics-next-line preferSchemaOverJson:off NodeFS.writeFileSync(scriptPath, JSON.stringify(script), "utf8"); @@ -240,6 +270,10 @@ describe("CodexSessionRuntime collab integration", () => { "pre-registration child A must still receive the interrupt RPC", ); assert.isTrue(interruptedThreads.has(CHILD_B), "registered child B must be interrupted"); + assert.isTrue( + interruptedThreads.has(MEMORY), + "memory consolidation must be interrupted without appearing in chat", + ); assert.isTrue(interruptedThreads.has(ROOT), "parent turn must be interrupted last"); yield* runtime.close; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index d7346a0e0db..796166b2fec 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -7,6 +7,7 @@ import { describe } from "vite-plus/test"; import { DEFAULT_MODEL, ThreadId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; import * as CodexRpc from "effect-codex-app-server/rpc"; +import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { buildCodexDeveloperInstructions, @@ -18,6 +19,7 @@ import { buildTurnStartParams, hasConfiguredMcpServer, isRecoverableThreadResumeError, + makeMemoryConsolidationNotificationFilter, openCodexThread, } from "./CodexSessionRuntime.ts"; const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); @@ -318,6 +320,134 @@ describe("hasConfiguredMcpServer", () => { }); }); +function makeThreadStartedNotification( + threadId: string, + source: EffectCodexSchema.V2ThreadStartedNotification["thread"]["source"], + threadSource?: string, +) { + return { + method: "thread/started" as const, + params: { + thread: { + cliVersion: "0.0.0", + createdAt: 0, + cwd: "/tmp/project", + ephemeral: true, + id: threadId, + modelProvider: "openai", + preview: "", + sessionId: threadId, + source, + status: { type: "idle" as const }, + ...(threadSource ? { threadSource } : {}), + turns: [], + updatedAt: 0, + }, + }, + }; +} + +describe("makeMemoryConsolidationNotificationFilter", () => { + it("suppresses memory consolidation without hiding other Codex subagents", () => { + const shouldSuppress = makeMemoryConsolidationNotificationFilter(); + + NodeAssert.equal( + shouldSuppress( + makeThreadStartedNotification("memory-thread", "unknown", "memory_consolidation"), + ), + true, + ); + NodeAssert.equal( + shouldSuppress({ + method: "item/agentMessage/delta", + params: { + delta: "internal memory update", + itemId: "memory-message", + threadId: "memory-thread", + turnId: "memory-turn", + }, + }), + true, + ); + NodeAssert.equal( + shouldSuppress({ + method: "warning", + params: { + message: "internal warning", + threadId: "memory-thread", + }, + }), + true, + ); + NodeAssert.equal( + shouldSuppress({ + method: "item/agentMessage/delta", + params: { + delta: "normal reply", + itemId: "root-message", + threadId: "root-thread", + turnId: "root-turn", + }, + }), + false, + ); + + NodeAssert.equal( + shouldSuppress( + makeThreadStartedNotification("legacy-memory-thread", { + subAgent: "memory_consolidation", + }), + ), + true, + ); + + for (const source of [ + { subAgent: "review" as const }, + { subAgent: "compact" as const }, + { + subAgent: { + thread_spawn: { + depth: 1, + parent_thread_id: "root-thread", + }, + }, + }, + ]) { + NodeAssert.equal( + shouldSuppress(makeThreadStartedNotification("visible-subagent", source)), + false, + ); + } + }); + + it("forgets memory consolidation threads after they close", () => { + const shouldSuppress = makeMemoryConsolidationNotificationFilter(); + shouldSuppress( + makeThreadStartedNotification("memory-thread", "unknown", "memory_consolidation"), + ); + + NodeAssert.equal( + shouldSuppress({ + method: "thread/closed", + params: { threadId: "memory-thread" }, + }), + true, + ); + NodeAssert.equal( + shouldSuppress({ + method: "item/agentMessage/delta", + params: { + delta: "later message", + itemId: "later-message", + threadId: "memory-thread", + turnId: "later-turn", + }, + }), + false, + ); + }); +}); + describe("codexSessionAppServerArgs", () => { it("keeps the app-server subcommand when explicit args are provided", () => { NodeAssert.deepStrictEqual(codexSessionAppServerArgs(["-c", "model=gpt-5"], undefined), [ diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 57a1162dd08..d4208920922 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -541,6 +541,45 @@ function readNotificationThreadId(notification: CodexServerNotification): string } } +export function makeMemoryConsolidationNotificationFilter(): ( + notification: CodexServerNotification, +) => boolean { + const threadIds = new Set(); + + return (notification) => { + if (notification.method === "thread/started") { + const thread = notification.params.thread; + const source = thread.source; + if ( + thread.threadSource === "memory_consolidation" || + (typeof source === "object" && + source !== null && + "subAgent" in source && + source.subAgent === "memory_consolidation") + ) { + threadIds.add(thread.id); + return true; + } + } + + const params = notification.params; + const threadId = + notification.method === "thread/started" + ? notification.params.thread.id + : "threadId" in params && typeof params.threadId === "string" + ? params.threadId + : undefined; + if (!threadId || !threadIds.has(threadId)) { + return false; + } + + if (notification.method === "thread/closed") { + threadIds.delete(threadId); + } + return true; + }; +} + function readRouteFields(notification: CodexServerNotification): { readonly turnId: TurnId | undefined; readonly itemId: ProviderItemId | undefined; @@ -857,6 +896,7 @@ export const makeCodexSessionRuntime = ( const collabChildAgentsRef = yield* Ref.make(new Map()); /** Child provider-thread id → its currently running provider turn id. */ const collabChildLiveTurnsRef = yield* Ref.make(new Map()); + const suppressMemoryConsolidationNotification = makeMemoryConsolidationNotificationFilter(); const closedRef = yield* Ref.make(false); // `~` is not shell-expanded when env vars are set via @@ -1249,6 +1289,9 @@ export const makeCodexSessionRuntime = ( const handleRawNotification = (notification: CodexServerNotification) => Effect.gen(function* () { + const isMemoryConsolidationNotification = + suppressMemoryConsolidationNotification(notification); + const payload = notification.params; const route = readRouteFields(notification); const collabReceiverTurns = yield* Ref.get(collabReceiverTurnsRef); @@ -1326,6 +1369,10 @@ export const makeCodexSessionRuntime = ( return; } + if (isMemoryConsolidationNotification) { + return; + } + let requestId: ApprovalRequestId | undefined; let requestKind: ProviderRequestKind | undefined; let turnId = childParentTurnId ?? route.turnId;