From c1df7aa08be26d3df1773613cf53e155d871822b Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Wed, 17 Jun 2026 01:30:09 +0800 Subject: [PATCH] fix(context-engine): avoid turn-maintenance lane livelock (#93727) * fix(context-engine): resolve deferred turn-maintenance livelock Co-authored-by: Vishnu <268122714+baghvn@users.noreply.github.com> * fix(clownfish): address review for gitcrawl-451-autonomous-terminal-gap (1) Co-authored-by: Vishnu <268122714+baghvn@users.noreply.github.com> * fix(clownfish): address review for gitcrawl-451-autonomous-terminal-gap (1) Co-authored-by: Vishnu <268122714+baghvn@users.noreply.github.com> * fix(clownfish): address review for gitcrawl-451-autonomous-terminal-gap (1) Co-authored-by: Vishnu <268122714+baghvn@users.noreply.github.com> --------- Co-authored-by: openclaw-clownfish[bot] <280122609+openclaw-clownfish[bot]@users.noreply.github.com> Co-authored-by: Vishnu <268122714+baghvn@users.noreply.github.com> --- src/agents/cli-runner.ts | 4 + .../context-engine-maintenance.test.ts | 276 +++++++----------- .../context-engine-maintenance.ts | 48 +-- .../run.overflow-compaction.harness.ts | 17 +- .../run.overflow-compaction.test.ts | 78 +++++ src/agents/embedded-agent-runner/run.ts | 12 +- 6 files changed, 214 insertions(+), 221 deletions(-) diff --git a/src/agents/cli-runner.ts b/src/agents/cli-runner.ts index 8e9c0c4ae876..94c6e1049e10 100644 --- a/src/agents/cli-runner.ts +++ b/src/agents/cli-runner.ts @@ -28,6 +28,7 @@ import type { PreparedCliRunContext, RunCliAgentParams } from "./cli-runner/type import { claudeCliSessionTranscriptHasContent as claudeCliSessionTranscriptHasContentImpl } from "./command/attempt-execution.helpers.js"; import { classifyFailoverReason, isFailoverErrorMessage } from "./embedded-agent-helpers.js"; import type { EmbeddedAgentRunResult } from "./embedded-agent-runner.js"; +import { waitForDeferredTurnMaintenanceForSession } from "./embedded-agent-runner/context-engine-maintenance.js"; import { buildEmbeddedRunPayloads } from "./embedded-agent-runner/run/payloads.js"; import { FailoverError, isFailoverError, resolveFailoverStatus } from "./failover-error.js"; import { @@ -480,6 +481,9 @@ export async function runPreparedCliAgent( const hasAgentEndHooks = hookRunner?.hasHooks("agent_end") === true; const hasBeforeAgentRunHooks = hookRunner?.hasHooks("before_agent_run") === true; const needsHookHistory = hasLlmInputHooks || hasAgentEndHooks || hasBeforeAgentRunHooks; + // Prior turn maintenance can rewrite transcript entries after finalization. + // Reads for the next same-session inference must observe that rewrite. + await waitForDeferredTurnMaintenanceForSession(params.sessionKey ?? params.sessionId); const historyMessages = needsHookHistory ? await loadCliSessionHistoryMessages({ sessionId: params.sessionId, diff --git a/src/agents/embedded-agent-runner/context-engine-maintenance.test.ts b/src/agents/embedded-agent-runner/context-engine-maintenance.test.ts index d750ab00b1e0..dc90800b4bb8 100644 --- a/src/agents/embedded-agent-runner/context-engine-maintenance.test.ts +++ b/src/agents/embedded-agent-runner/context-engine-maintenance.test.ts @@ -35,6 +35,7 @@ const rewriteTranscriptEntriesInSessionFileMock = vi.fn(async (_params?: unknown let buildContextEngineMaintenanceRuntimeContext: typeof import("./context-engine-maintenance.js").buildContextEngineMaintenanceRuntimeContext; let createDeferredTurnMaintenanceAbortSignal: typeof import("./context-engine-maintenance.js").createDeferredTurnMaintenanceAbortSignal; let resetDeferredTurnMaintenanceStateForTest: typeof import("./context-engine-maintenance.js").resetDeferredTurnMaintenanceStateForTest; +let waitForDeferredTurnMaintenanceForSession: typeof import("./context-engine-maintenance.js").waitForDeferredTurnMaintenanceForSession; function createQueuedTaskRun(params: Parameters[0]): TaskRecord { // Task creation can legally return null for invalid inputs; tests here always @@ -118,6 +119,7 @@ async function loadFreshContextEngineMaintenanceModuleForTest() { createDeferredTurnMaintenanceAbortSignal, resetDeferredTurnMaintenanceStateForTest, runContextEngineMaintenance, + waitForDeferredTurnMaintenanceForSession, } = await import("./context-engine-maintenance.js")); resetDeferredTurnMaintenanceStateForTest(); } @@ -243,7 +245,7 @@ describe("buildContextEngineMaintenanceRuntimeContext", () => { expect(rewriteTranscriptEntriesInSessionFileMock).not.toHaveBeenCalled(); }); - it("defers file rewrites onto the session lane when requested", async () => { + it("lets background file rewrites run without the session lane", async () => { vi.useFakeTimers(); try { resetCommandQueueStateForTest(); @@ -275,7 +277,6 @@ describe("buildContextEngineMaintenanceRuntimeContext", () => { sessionId: "session-rewrite-handoff", sessionKey, sessionFile: "/tmp/session-rewrite-handoff.jsonl", - deferTranscriptRewriteToSessionLane: true, }); const rewritePromise = runtimeContext.rewriteTranscriptEntries?.({ @@ -285,20 +286,19 @@ describe("buildContextEngineMaintenanceRuntimeContext", () => { }); expect(rewritePromise?.["then"]).toBeTypeOf("function"); - await flushAsyncWork(); - expect(rewriteTranscriptEntriesInSessionFileMock).not.toHaveBeenCalled(); - - if (!releaseForeground) { - throw new Error("Expected foreground turn release callback to be initialized"); - } - releaseForeground(); await expect(rewritePromise!).resolves.toEqual({ changed: true, bytesFreed: 123, rewrittenEntries: 2, }); - expect(events).toEqual(["foreground-start", "foreground-end", "rewrite"]); + expect(events).toEqual(["foreground-start", "rewrite"]); + + if (!releaseForeground) { + throw new Error("Expected foreground turn release callback to be initialized"); + } + releaseForeground(); await foregroundTurn; + expect(events).toEqual(["foreground-start", "rewrite", "foreground-end"]); } finally { vi.useRealTimers(); } @@ -614,7 +614,27 @@ describe("runContextEngineMaintenance", () => { }); expect(result).toBeUndefined(); - expect(maintain).not.toHaveBeenCalled(); + await waitForAssertion(() => expect(maintain).toHaveBeenCalledTimes(1)); + await waitForAssertion(() => + expect(rewriteTranscriptEntriesInSessionFileMock).toHaveBeenCalledWith({ + sessionFile: "/tmp/session.jsonl", + sessionId: "session-1", + sessionKey, + config: { session: { writeLock: { acquireTimeoutMs: 91_000 } } }, + request: { + replacements: [ + { + entryId: "entry-1", + message: castAgentMessage({ + role: "assistant", + content: [{ type: "text", text: "done" }], + timestamp: 2, + }), + }, + ], + }, + }), + ); const queuedTasks = listTasksForOwnerKey(sessionKey).filter( (task) => task.taskKind === TURN_MAINTENANCE_TASK_KIND, @@ -635,7 +655,6 @@ describe("runContextEngineMaintenance", () => { throw new Error("Expected foreground turn release callback to be initialized"); } releaseForeground(); - await waitForAssertion(() => expect(maintain).toHaveBeenCalledTimes(1)); const maintainParams = firstMaintainParams(maintain); expectRecordFields(maintainParams, { sessionId: "session-1", @@ -648,25 +667,10 @@ describe("runContextEngineMaintenance", () => { tokenBudget: 2048, currentTokenCount: 1536, }); - expect(rewriteTranscriptEntriesInSessionFileMock).toHaveBeenCalledWith({ - sessionFile: "/tmp/session.jsonl", - sessionId: "session-1", - sessionKey, - config: { session: { writeLock: { acquireTimeoutMs: 91_000 } } }, - request: { - replacements: [ - { - entryId: "entry-1", - message: castAgentMessage({ - role: "assistant", - content: [{ type: "text", text: "done" }], - timestamp: 2, - }), - }, - ], - }, - }); + await waitForAssertion(() => + expect(getTaskById(queuedTasks[0].taskId)?.status).toBe("succeeded"), + ); const completedTask = getTaskById(queuedTasks[0].taskId); const completedTaskRecord = requireRecord(completedTask, "completed task"); expect(completedTaskRecord.status).toBe("succeeded"); @@ -690,20 +694,21 @@ describe("runContextEngineMaintenance", () => { resetTaskFlowRegistryForTests({ persist: false }); const sessionKey = "agent:main:session-2"; - const sessionLane = resolveSessionLane(sessionKey); - let releaseForeground: (() => void) | undefined; - const foregroundTurn = enqueueCommandInLane(sessionLane, async () => { - await new Promise((resolve) => { - releaseForeground = resolve; - }); + let releaseMaintenance: (() => void) | undefined; + let maintenanceCalls = 0; + const maintain = vi.fn(async () => { + maintenanceCalls += 1; + if (maintenanceCalls === 1) { + await new Promise((resolve) => { + releaseMaintenance = resolve; + }); + } + return { + changed: false, + bytesFreed: 0, + rewrittenEntries: 0, + }; }); - await Promise.resolve(); - - const maintain = vi.fn(async () => ({ - changed: false, - bytesFreed: 0, - rewrittenEntries: 0, - })); const backgroundEngine = { info: { @@ -720,40 +725,39 @@ describe("runContextEngineMaintenance", () => { maintain, } as NonNullable[0]["contextEngine"]>; - await Promise.all([ - runContextEngineMaintenance({ - contextEngine: backgroundEngine, - sessionId: "session-2", - sessionKey, - sessionFile: "/tmp/session-2.jsonl", - reason: "turn", - }), - runContextEngineMaintenance({ - contextEngine: backgroundEngine, - sessionId: "session-2", - sessionKey, - sessionFile: "/tmp/session-2.jsonl", - reason: "turn", - }), - ]); + await runContextEngineMaintenance({ + contextEngine: backgroundEngine, + sessionId: "session-2", + sessionKey, + sessionFile: "/tmp/session-2.jsonl", + reason: "turn", + }); + await waitForAssertion(() => expect(maintain).toHaveBeenCalledTimes(1)); + await runContextEngineMaintenance({ + contextEngine: backgroundEngine, + sessionId: "session-2", + sessionKey, + sessionFile: "/tmp/session-2.jsonl", + reason: "turn", + }); const queuedTasks = listTasksForOwnerKey(sessionKey).filter( (task) => task.taskKind === TURN_MAINTENANCE_TASK_KIND, ); expect(queuedTasks).toHaveLength(1); - if (!releaseForeground) { - throw new Error("Expected foreground turn release callback to be initialized"); + if (!releaseMaintenance) { + throw new Error("Expected maintenance release callback to be initialized"); } - releaseForeground(); + releaseMaintenance(); await waitForAssertion(() => expect(maintain).toHaveBeenCalledTimes(2)); - const completedTasks = listTasksForOwnerKey(sessionKey).filter( - (task) => task.taskKind === TURN_MAINTENANCE_TASK_KIND, + await waitForAssertion(() => + expect( + listTasksForOwnerKey(sessionKey) + .filter((task) => task.taskKind === TURN_MAINTENANCE_TASK_KIND) + .map((task) => task.status), + ).toEqual(["succeeded", "succeeded"]), ); - expect(completedTasks).toHaveLength(2); - expect(completedTasks.every((task) => task.status === "succeeded")).toBe(true); - - await foregroundTurn; } finally { vi.useRealTimers(); } @@ -1243,7 +1247,7 @@ describe("runContextEngineMaintenance", () => { }); }); - it("lets foreground turns win while deferred maintenance is waiting", async () => { + it("starts deferred maintenance while the foreground session lane stays busy", async () => { await withStateDirEnv("openclaw-turn-maintenance-", async () => { vi.useFakeTimers(); try { @@ -1301,6 +1305,18 @@ describe("runContextEngineMaintenance", () => { events.push("foreground-2-end"); }); + await waitForAssertion(() => + expect(events).toEqual(["foreground-1-start", "maintenance-start"]), + ); + expect(maintain).toHaveBeenCalledTimes(1); + await waitForAssertion(() => + expect( + listTasksForOwnerKey(sessionKey).find( + (task) => task.taskKind === TURN_MAINTENANCE_TASK_KIND, + )?.status, + ).toBe("succeeded"), + ); + if (!releaseFirstForeground) { throw new Error("Expected first foreground release callback to be initialized"); } @@ -1308,13 +1324,12 @@ describe("runContextEngineMaintenance", () => { await waitForAssertion(() => expect(events).toEqual([ "foreground-1-start", + "maintenance-start", "foreground-1-end", "foreground-2-start", "foreground-2-end", - "maintenance-start", ]), ); - expect(maintain).toHaveBeenCalledTimes(1); await Promise.all([firstForeground, secondForeground]); } finally { @@ -1323,7 +1338,7 @@ describe("runContextEngineMaintenance", () => { }); }); - it("lets a foreground turn run before a deferred maintenance transcript rewrite", async () => { + it("waits at the same-session read checkpoint before deferred maintenance rewrites", async () => { await withStateDirEnv("openclaw-turn-maintenance-", async () => { vi.useFakeTimers(); try { @@ -1400,8 +1415,9 @@ describe("runContextEngineMaintenance", () => { await waitForAssertion(() => expect(events).toContain("maintenance-start")); const foregroundTurn = enqueueCommandInLane(sessionLane, async () => { - events.push("foreground-start"); - events.push("foreground-end"); + events.push("foreground-before-read-checkpoint"); + await waitForDeferredTurnMaintenanceForSession(sessionKey); + events.push("foreground-read"); }); if (!allowRewrite) { @@ -1412,11 +1428,11 @@ describe("runContextEngineMaintenance", () => { await waitForAssertion(() => expect(events).toEqual([ "maintenance-start", - "foreground-start", - "foreground-end", + "foreground-before-read-checkpoint", "maintenance-before-rewrite", "rewrite", "maintenance-after-rewrite", + "foreground-read", ]), ); @@ -1488,20 +1504,17 @@ describe("runContextEngineMaintenance", () => { resetSystemEventsForTest(); const sessionKey = "agent:main:session-long"; - const sessionLane = resolveSessionLane(sessionKey); - let releaseForeground: (() => void) | undefined; - const foregroundTurn = enqueueCommandInLane(sessionLane, async () => { + let releaseMaintenance: (() => void) | undefined; + const maintain = vi.fn(async () => { await new Promise((resolve) => { - releaseForeground = resolve; + releaseMaintenance = resolve; }); + return { + changed: false, + bytesFreed: 0, + rewrittenEntries: 0, + }; }); - await Promise.resolve(); - - const maintain = vi.fn(async () => ({ - changed: false, - bytesFreed: 0, - rewrittenEntries: 0, - })); const backgroundEngine = { info: { id: "test", @@ -1525,6 +1538,7 @@ describe("runContextEngineMaintenance", () => { reason: "turn", }); + await waitForAssertion(() => expect(maintain).toHaveBeenCalledTimes(1)); await vi.advanceTimersByTimeAsync(11_000); await waitForAssertion(() => expectSystemEventContaining( @@ -1533,98 +1547,16 @@ describe("runContextEngineMaintenance", () => { ), ); - if (!releaseForeground) { - throw new Error("Expected foreground turn release callback to be initialized"); + if (!releaseMaintenance) { + throw new Error("Expected maintenance release callback to be initialized"); } - releaseForeground(); + releaseMaintenance(); await waitForAssertion(() => expectSystemEventContaining( sessionKey, "Background task done: Context engine turn maintenance", ), ); - - await foregroundTurn; - } finally { - vi.useRealTimers(); - } - }); - }); - - it("throttles deferred wait notices while the session lane stays busy", async () => { - await withStateDirEnv("openclaw-turn-maintenance-", async () => { - vi.useFakeTimers(); - try { - resetCommandQueueStateForTest(); - resetTaskRegistryForTests({ persist: false }); - resetTaskFlowRegistryForTests({ persist: false }); - resetSystemEventsForTest(); - - const sessionKey = "agent:main:session-throttle"; - const sessionLane = resolveSessionLane(sessionKey); - let releaseForeground: (() => void) | undefined; - const foregroundTurn = enqueueCommandInLane(sessionLane, async () => { - await new Promise((resolve) => { - releaseForeground = resolve; - }); - }); - await Promise.resolve(); - - const backgroundEngine = { - info: { - id: "test", - name: "Test Engine", - turnMaintenanceMode: "background" as const, - }, - ingest: async () => ({ ingested: true }), - assemble: async ({ messages }: { messages: unknown[] }) => ({ - messages, - estimatedTokens: 0, - }), - compact: async () => ({ ok: true, compacted: false }), - maintain: vi.fn(async () => ({ - changed: false, - bytesFreed: 0, - rewrittenEntries: 0, - })), - } as NonNullable[0]["contextEngine"]>; - - await runContextEngineMaintenance({ - contextEngine: backgroundEngine, - sessionId: "session-throttle", - sessionKey, - sessionFile: "/tmp/session-throttle.jsonl", - reason: "turn", - }); - - await vi.advanceTimersByTimeAsync(11_000); - await waitForAssertion(() => - expect( - peekSystemEvents(sessionKey).filter((event) => - event.includes("Background task update: Context engine turn maintenance."), - ), - ).toHaveLength(1), - ); - - await vi.advanceTimersByTimeAsync(9_000); - expect( - peekSystemEvents(sessionKey).filter((event) => - event.includes("Background task update: Context engine turn maintenance."), - ), - ).toHaveLength(2); - - await vi.advanceTimersByTimeAsync(1_000); - expect( - peekSystemEvents(sessionKey).filter((event) => - event.includes("Background task update: Context engine turn maintenance."), - ), - ).toHaveLength(2); - - if (!releaseForeground) { - throw new Error("Expected foreground turn release callback to be initialized"); - } - releaseForeground(); - await foregroundTurn; } finally { vi.useRealTimers(); } diff --git a/src/agents/embedded-agent-runner/context-engine-maintenance.ts b/src/agents/embedded-agent-runner/context-engine-maintenance.ts index 96f6c725d74f..3180c568af99 100644 --- a/src/agents/embedded-agent-runner/context-engine-maintenance.ts +++ b/src/agents/embedded-agent-runner/context-engine-maintenance.ts @@ -10,12 +10,10 @@ import type { ContextEngineMaintenanceResult, ContextEngineRuntimeContext, } from "../../context-engine/types.js"; -import { sleepWithAbort } from "../../infra/backoff.js"; import { formatErrorMessage } from "../../infra/errors.js"; import { enqueueCommandInLane, GatewayDrainingError, - getQueueSize, isGatewayDraining, } from "../../process/command-queue.js"; import { @@ -33,7 +31,6 @@ import { } from "../../tasks/task-owner-access.js"; import { findActiveSessionTask } from "../session-async-task-status.js"; import { resolveContextEngineCapabilities } from "./context-engine-capabilities.js"; -import { resolveSessionLane } from "./lanes.js"; import { log } from "./logger.js"; import { rewriteTranscriptEntriesInSessionFile, @@ -44,7 +41,6 @@ const TURN_MAINTENANCE_TASK_KIND = "context_engine_turn_maintenance"; const TURN_MAINTENANCE_TASK_LABEL = "Context engine turn maintenance"; const TURN_MAINTENANCE_TASK_TASK = "Deferred context-engine maintenance after turn."; const TURN_MAINTENANCE_LANE_PREFIX = "context-engine-turn-maintenance:"; -const TURN_MAINTENANCE_WAIT_POLL_MS = 100; const TURN_MAINTENANCE_LONG_WAIT_MS = 10_000; const DEFERRED_TURN_MAINTENANCE_ABORT_STATE_KEY = Symbol.for( "openclaw.contextEngineTurnMaintenanceAbortState", @@ -210,6 +206,14 @@ export function resetDeferredTurnMaintenanceStateForTest(): void { delete processLike[DEFERRED_TURN_MAINTENANCE_ABORT_STATE_KEY]; } +export async function waitForDeferredTurnMaintenanceForSession(sessionKey?: string): Promise { + const normalizedSessionKey = normalizeSessionKey(sessionKey); + if (!normalizedSessionKey) { + return; + } + await activeDeferredTurnMaintenanceRuns.get(normalizedSessionKey)?.promise; +} + function markDeferredTurnMaintenanceTaskScheduleFailure(params: { sessionKey: string; taskId: string; @@ -305,7 +309,6 @@ export function buildContextEngineMaintenanceRuntimeContext(params: { runtimeContext?: ContextEngineRuntimeContext; agentId?: string; allowDeferredCompactionExecution?: boolean; - deferTranscriptRewriteToSessionLane?: boolean; config?: OpenClawConfig; purpose?: string; contextEnginePluginId?: string; @@ -342,13 +345,6 @@ export function buildContextEngineMaintenanceRuntimeContext(params: { config: params.config, request, }); - const rewriteSessionKey = normalizeSessionKey(params.sessionKey ?? params.sessionId); - if (params.deferTranscriptRewriteToSessionLane && rewriteSessionKey) { - return await enqueueCommandInLane( - resolveSessionLane(rewriteSessionKey), - async () => await rewriteTranscriptEntriesInFile(), - ); - } return await rewriteTranscriptEntriesInFile(); }, }; @@ -384,7 +380,6 @@ async function executeContextEngineMaintenance(params: { runtimeContext: params.runtimeContext, agentId: params.agentId, allowDeferredCompactionExecution: params.executionMode === "background", - deferTranscriptRewriteToSessionLane: params.executionMode === "background", config: params.config, purpose: `context-engine.${params.reason}.maintenance`, contextEnginePluginId: resolveContextEngineOwnerPluginId(params.contextEngine), @@ -433,33 +428,6 @@ async function runDeferredTurnMaintenanceWorker(params: { }; try { - const sessionLane = resolveSessionLane(params.sessionKey); - const startedWaitingAt = Date.now(); - let lastWaitNoticeAt = 0; - - for (;;) { - while (getQueueSize(sessionLane) > 0) { - const now = Date.now(); - if ( - now - startedWaitingAt >= TURN_MAINTENANCE_LONG_WAIT_MS && - now - lastWaitNoticeAt >= TURN_MAINTENANCE_LONG_WAIT_MS - ) { - lastWaitNoticeAt = now; - surfaceMaintenanceUpdate( - "Waiting for the session lane to go idle.", - surfacedUserNotice - ? "Still waiting for the session lane to go idle." - : "Deferred maintenance is waiting for the session lane to go idle.", - ); - } - await sleepWithAbort(TURN_MAINTENANCE_WAIT_POLL_MS, shutdownAbort.abortSignal); - } - await Promise.resolve(); - if (getQueueSize(sessionLane) === 0) { - break; - } - } - const runningAt = Date.now(); startTaskRunByRunId({ runId: params.runId, diff --git a/src/agents/embedded-agent-runner/run.overflow-compaction.harness.ts b/src/agents/embedded-agent-runner/run.overflow-compaction.harness.ts index 5a74185f3c8d..396d3535e67f 100644 --- a/src/agents/embedded-agent-runner/run.overflow-compaction.harness.ts +++ b/src/agents/embedded-agent-runner/run.overflow-compaction.harness.ts @@ -16,6 +16,7 @@ import type { PluginHookBeforeModelResolveResult, PluginHookBeforePromptBuildResult, } from "../../plugins/types.js"; +import { resetCommandQueueStateForTest } from "../../process/command-queue.js"; import type { FailoverReason } from "../embedded-agent-helpers/types.js"; import { clearAgentHarnesses, registerAgentHarness } from "../harness/registry.js"; import type { buildEmbeddedRunPayloads } from "./run/payloads.js"; @@ -146,6 +147,9 @@ export const mockedBuildEmbeddedRunPayloads = vi.fn< ) => ReturnType >(() => []); export const mockedRunContextEngineMaintenance = vi.fn(async () => undefined); +export const mockedWaitForDeferredTurnMaintenanceForSession = vi.fn( + async (_sessionKey?: string) => undefined, +); export const mockedSessionLikelyHasOversizedToolResults = vi.fn(() => false); export const mockedResolveLiveToolResultMaxChars = vi.fn(() => 32_000); type MockTruncateOversizedToolResultsResult = { @@ -295,6 +299,7 @@ export const overflowBaseRunParams = { /** Reset every mocked runner dependency to the default successful no-op state. */ export function resetRunOverflowCompactionHarnessMocks(): void { + resetCommandQueueStateForTest(); clearAgentHarnesses(); registerAgentHarness({ id: "codex", @@ -357,6 +362,8 @@ export function resetRunOverflowCompactionHarnessMocks(): void { mockedBuildEmbeddedRunPayloads.mockReturnValue([]); mockedRunContextEngineMaintenance.mockReset(); mockedRunContextEngineMaintenance.mockResolvedValue(undefined); + mockedWaitForDeferredTurnMaintenanceForSession.mockReset(); + mockedWaitForDeferredTurnMaintenanceForSession.mockResolvedValue(undefined); mockedSessionLikelyHasOversizedToolResults.mockReset(); mockedSessionLikelyHasOversizedToolResults.mockReturnValue(false); mockedResolveLiveToolResultMaxChars.mockReset(); @@ -670,6 +677,7 @@ export async function loadRunOverflowCompactionHarness(): Promise<{ vi.doMock("./context-engine-maintenance.js", () => ({ runContextEngineMaintenance: mockedRunContextEngineMaintenance, + waitForDeferredTurnMaintenanceForSession: mockedWaitForDeferredTurnMaintenanceForSession, })); vi.doMock("./model.js", () => ({ @@ -700,11 +708,6 @@ export async function loadRunOverflowCompactionHarness(): Promise<{ resolveContextWindowInfo: mockedResolveContextWindowInfo, })); - vi.doMock("../../process/command-queue.js", () => ({ - enqueueCommandInLane: vi.fn((_lane: string, task: () => unknown) => task()), - clearCommandLane: vi.fn(() => 0), - })); - vi.doMock("../../utils/message-channel.js", () => ({ isMarkdownCapableMessageChannel: vi.fn(() => true), })); @@ -723,8 +726,8 @@ export async function loadRunOverflowCompactionHarness(): Promise<{ })); vi.doMock("./lanes.js", () => ({ - resolveSessionLane: vi.fn(() => "session-lane"), - resolveEmbeddedSessionLane: vi.fn(() => "session-lane"), + resolveSessionLane: vi.fn((key: string) => `session:${key}`), + resolveEmbeddedSessionLane: vi.fn((key: string) => `session:${key}`), resolveGlobalLane: vi.fn(() => "global-lane"), })); diff --git a/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts b/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts index 7cb077e788c3..1ff4faa0e92b 100644 --- a/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts +++ b/src/agents/embedded-agent-runner/run.overflow-compaction.test.ts @@ -47,6 +47,7 @@ import { mockedRunEmbeddedAttempt, mockedSessionLikelyHasOversizedToolResults, mockedTruncateOversizedToolResultsInSession, + mockedWaitForDeferredTurnMaintenanceForSession, overflowBaseRunParams, resetRunOverflowCompactionHarnessMocks, } from "./run.overflow-compaction.harness.js"; @@ -236,6 +237,16 @@ function expectRuntimePlanFields( } } +async function waitForRunEvent(events: string[], expected: string): Promise { + for (let attempt = 0; attempt < 20; attempt += 1) { + if (events.includes(expected)) { + return; + } + await Promise.resolve(); + } + throw new Error(`Expected run event ${expected}; saw ${events.join(", ")}`); +} + describe("runEmbeddedAgent overflow compaction trigger routing", () => { beforeAll(async () => { ({ runEmbeddedAgent } = await loadRunOverflowCompactionHarness()); @@ -286,6 +297,73 @@ describe("runEmbeddedAgent overflow compaction trigger routing", () => { }); }); + it("waits for same-session deferred maintenance before the attempt reads session state", async () => { + const events: string[] = []; + mockedWaitForDeferredTurnMaintenanceForSession.mockImplementationOnce(async (sessionKey) => { + events.push(`wait:${sessionKey}`); + }); + mockedRunEmbeddedAttempt.mockImplementationOnce(async () => { + events.push("attempt"); + return makeAttemptResult({ promptError: null }); + }); + + await runEmbeddedAgent({ + ...overflowBaseRunParams, + runId: "run-wait-deferred-maintenance", + sessionKey: "agent:main:session-wait-deferred-maintenance", + }); + + expect(events).toEqual(["wait:agent:main:session-wait-deferred-maintenance", "attempt"]); + }); + + it("does not hold the global run lane while waiting for another session's deferred maintenance", async () => { + const events: string[] = []; + let releaseSessionA: (() => void) | undefined; + mockedWaitForDeferredTurnMaintenanceForSession.mockImplementation(async (sessionKey) => { + events.push(`wait:${sessionKey}`); + if (sessionKey !== "agent:main:session-a") { + return; + } + await new Promise((resolve) => { + releaseSessionA = resolve; + }); + }); + mockedRunEmbeddedAttempt.mockImplementation(async (params) => { + events.push(`attempt:${(params as { sessionKey?: string }).sessionKey}`); + return makeAttemptResult({ promptError: null }); + }); + + const sessionARun = runEmbeddedAgent({ + ...overflowBaseRunParams, + runId: "run-deferred-maintenance-session-a", + sessionKey: "agent:main:session-a", + }); + await waitForRunEvent(events, "wait:agent:main:session-a"); + + await runEmbeddedAgent({ + ...overflowBaseRunParams, + runId: "run-deferred-maintenance-session-b", + sessionKey: "agent:main:session-b", + }); + + expect(events).toEqual([ + "wait:agent:main:session-a", + "wait:agent:main:session-b", + "attempt:agent:main:session-b", + ]); + if (!releaseSessionA) { + throw new Error("Expected session A maintenance release callback to be initialized"); + } + releaseSessionA(); + await sessionARun; + expect(events).toEqual([ + "wait:agent:main:session-a", + "wait:agent:main:session-b", + "attempt:agent:main:session-b", + "attempt:agent:main:session-a", + ]); + }); + it("uses the lightweight auth profile store during reply startup", async () => { mockedRunEmbeddedAttempt.mockResolvedValueOnce(makeAttemptResult({ promptError: null })); diff --git a/src/agents/embedded-agent-runner/run.ts b/src/agents/embedded-agent-runner/run.ts index 100a9474063f..f06ee6b2cfbb 100644 --- a/src/agents/embedded-agent-runner/run.ts +++ b/src/agents/embedded-agent-runner/run.ts @@ -130,7 +130,10 @@ import { resolveCompactionTimeoutMs, } from "./compaction-safety-timeout.js"; import { resolveContextEngineCapabilities } from "./context-engine-capabilities.js"; -import { runContextEngineMaintenance } from "./context-engine-maintenance.js"; +import { + runContextEngineMaintenance, + waitForDeferredTurnMaintenanceForSession, +} from "./context-engine-maintenance.js"; import { hasMessagingToolDeliveryEvidence, hasOutboundDeliveryEvidence, @@ -735,7 +738,12 @@ async function runEmbeddedAgentInternal( throwIfAborted(); - return enqueueSession(() => { + return enqueueSession(async () => { + throwIfAborted(); + // Same-session reads below must see any prior deferred transcript rewrite. + // Checkpoint before the global lane so unrelated sessions can still start + // while this session waits on its own maintenance lane. + await waitForDeferredTurnMaintenanceForSession(params.sessionKey); throwIfAborted(); return enqueueGlobal(async () => { throwIfAborted();