mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-25 03:45:46 -06:00
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>
This commit is contained in:
Vincent Koc
@@ -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,
|
||||
|
||||
@@ -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<typeof createQueuedTaskRunOrNull>[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<void>((resolve) => {
|
||||
releaseForeground = resolve;
|
||||
});
|
||||
let releaseMaintenance: (() => void) | undefined;
|
||||
let maintenanceCalls = 0;
|
||||
const maintain = vi.fn(async () => {
|
||||
maintenanceCalls += 1;
|
||||
if (maintenanceCalls === 1) {
|
||||
await new Promise<void>((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<Parameters<typeof runContextEngineMaintenance>[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<void>((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<void>((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<Parameters<typeof runContextEngineMaintenance>[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();
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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,
|
||||
|
||||
@@ -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<typeof buildEmbeddedRunPayloads>
|
||||
>(() => []);
|
||||
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"),
|
||||
}));
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
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<void>((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 }));
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user