From 6f1076351cbef3a3e7fe1df6199dc8f1dff6a583 Mon Sep 17 00:00:00 2001 From: Josh Avant <830519+joshavant@users.noreply.github.com> Date: Fri, 26 Jun 2026 22:42:45 -0500 Subject: [PATCH] fix: defer active implicit session rollover (#97164) --- .../reply/get-reply-run.media-only.test.ts | 27 +++ src/auto-reply/reply/get-reply-run.ts | 12 ++ src/auto-reply/reply/session.test.ts | 172 ++++++++++++++++++ src/auto-reply/reply/session.ts | 21 ++- 4 files changed, 229 insertions(+), 3 deletions(-) diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts index 595cb3921de9..03f6992a6459 100644 --- a/src/auto-reply/reply/get-reply-run.media-only.test.ts +++ b/src/auto-reply/reply/get-reply-run.media-only.test.ts @@ -1622,6 +1622,33 @@ describe("runPreparedReply media-only handling", () => { } }); + it("rebinds a queued pre-dispatch reply operation after session rollover", async () => { + const operation = createReplyOperation({ + sessionId: "session-before-rollover", + sessionKey: "session-key", + resetTriggered: false, + }); + + try { + await expect( + runPreparedReply( + baseParams({ + isNewSession: true, + sessionId: "session-after-rollover", + opts: { replyOperation: operation } as never, + }), + ), + ).resolves.toEqual({ text: "ok" }); + + const call = requireLastRunReplyAgentCall(); + expect(operation.sessionId).toBe("session-after-rollover"); + expect(call.replyOperation).toBe(operation); + expect(call.followupRun.run.sessionId).toBe("session-after-rollover"); + } finally { + operation.complete(); + } + }); + it("does not interrupt its provided pre-dispatch reply operation for reset turns", async () => { const queueSettings = await import("./queue/settings-runtime.js"); const embeddedAgentRuntime = await import("../../agents/embedded-agent.runtime.js"); diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts index c7c0421927e1..8bf7dabad620 100644 --- a/src/auto-reply/reply/get-reply-run.ts +++ b/src/auto-reply/reply/get-reply-run.ts @@ -937,6 +937,18 @@ export async function runPreparedReply( } const internalOpts = opts as InternalGetReplyOptions | undefined; const providedReplyOperation = internalOpts?.replyOperation; + if ( + providedReplyOperation !== undefined && + providedReplyOperation.result === null && + providedReplyOperation.phase === "queued" && + sessionId !== undefined && + sessionId !== providedReplyOperation.sessionId + ) { + // Dispatch reserves a queued operation before session init. If stale init + // rotates the session, move the reservation so later steer/abort paths + // target the session that will actually run. + providedReplyOperation.updateSessionId(sessionId); + } const isOwnPreDispatchOperationSession = (candidateSessionId: string | undefined): boolean => providedReplyOperation !== undefined && providedReplyOperation.result === null && diff --git a/src/auto-reply/reply/session.test.ts b/src/auto-reply/reply/session.test.ts index d7e4a8f24c6e..a0df25c52da9 100644 --- a/src/auto-reply/reply/session.test.ts +++ b/src/auto-reply/reply/session.test.ts @@ -32,6 +32,7 @@ import { createSessionConversationTestRegistry } from "../../test-utils/session- import { drainFormattedSystemEvents } from "./session-updates.js"; import { persistSessionUsageUpdate } from "./session-usage.js"; import { initSessionState } from "./session.js"; +import { replyRunRegistry } from "./reply-run-registry.js"; const sessionForkMocks = vi.hoisted(() => ({ forkSessionFromParent: vi.fn(), @@ -3709,6 +3710,177 @@ describe("initSessionState preserves behavior overrides across /new and /reset", } }); + it("defers implicit daily rollover while the same session has an active run", async () => { + vi.useFakeTimers(); + const existingSessionId = "active-stale-session"; + let operation: ReturnType | undefined; + try { + vi.setSystemTime(new Date(2026, 0, 18, 5, 0, 0)); + const storePath = await createStorePath("openclaw-active-stale-archive-"); + const sessionKey = "agent:main:telegram:dm:active-stale-user"; + const transcriptPath = path.join(path.dirname(storePath), `${existingSessionId}.jsonl`); + const sessionStartedAt = new Date(2026, 0, 18, 3, 0, 0).getTime(); + + await writeSessionStoreFast(storePath, { + [sessionKey]: { + sessionId: existingSessionId, + updatedAt: sessionStartedAt, + sessionStartedAt, + }, + }); + await fs.writeFile(transcriptPath, '{"type":"message"}\n', "utf8"); + operation = replyRunRegistry.begin({ + sessionKey, + sessionId: existingSessionId, + resetTriggered: false, + }); + operation.setPhase("running"); + + const cfg = { session: { store: storePath } } as OpenClawConfig; + const result = await initSessionState({ + ctx: { + Body: "hello while active", + RawBody: "hello while active", + CommandBody: "hello while active", + From: "user-active-stale", + To: "bot", + ChatType: "direct", + SessionKey: sessionKey, + Provider: "telegram", + Surface: "telegram", + }, + cfg, + commandAuthorized: true, + }); + + expect(result.isNewSession).toBe(false); + expect(result.resetTriggered).toBe(false); + expect(result.sessionId).toBe(existingSessionId); + expect(result.previousSessionEntry).toBeUndefined(); + expect(result.sessionEntry.sessionStartedAt).toBe(sessionStartedAt); + expect(await fs.stat(transcriptPath).catch(() => null)).not.toBeNull(); + const archived = (await fs.readdir(path.dirname(storePath))).filter((entry) => + entry.startsWith(`${existingSessionId}.jsonl.reset.`), + ); + expect(archived).toHaveLength(0); + } finally { + operation?.complete(); + vi.useRealTimers(); + } + }); + + it("does not defer stale archival for the current turn's queued reservation", async () => { + vi.useFakeTimers(); + let operation: ReturnType | undefined; + try { + vi.setSystemTime(new Date(2026, 0, 18, 5, 0, 0)); + const storePath = await createStorePath("openclaw-queued-stale-archive-"); + const sessionKey = "agent:main:telegram:dm:queued-stale-user"; + const existingSessionId = "queued-stale-session"; + const transcriptPath = path.join(path.dirname(storePath), `${existingSessionId}.jsonl`); + + await writeSessionStoreFast(storePath, { + [sessionKey]: { + sessionId: existingSessionId, + updatedAt: new Date(2026, 0, 18, 3, 0, 0).getTime(), + }, + }); + await fs.writeFile(transcriptPath, '{"type":"message"}\n', "utf8"); + operation = replyRunRegistry.begin({ + sessionKey, + sessionId: existingSessionId, + resetTriggered: false, + }); + + const cfg = { session: { store: storePath } } as OpenClawConfig; + const result = await initSessionState({ + ctx: { + Body: "hello after boundary", + RawBody: "hello after boundary", + CommandBody: "hello after boundary", + From: "user-queued-stale", + To: "bot", + ChatType: "direct", + SessionKey: sessionKey, + Provider: "telegram", + Surface: "telegram", + }, + cfg, + commandAuthorized: true, + }); + + expect(operation.phase).toBe("queued"); + expect(result.isNewSession).toBe(true); + expect(result.resetTriggered).toBe(false); + expect(result.sessionId).not.toBe(existingSessionId); + expect(result.previousSessionEntry?.sessionId).toBe(existingSessionId); + expect(await fs.stat(transcriptPath).catch(() => null)).toBeNull(); + const archived = (await fs.readdir(path.dirname(storePath))).filter((entry) => + entry.startsWith(`${existingSessionId}.jsonl.reset.`), + ); + expect(archived).toHaveLength(1); + } finally { + operation?.complete(); + vi.useRealTimers(); + } + }); + + it("does not defer stale archival for a different active session id", async () => { + vi.useFakeTimers(); + let operation: ReturnType | undefined; + try { + vi.setSystemTime(new Date(2026, 0, 18, 5, 0, 0)); + const storePath = await createStorePath("openclaw-active-other-stale-archive-"); + const sessionKey = "agent:main:telegram:dm:active-other-stale-user"; + const existingSessionId = "inactive-stale-session"; + const transcriptPath = path.join(path.dirname(storePath), `${existingSessionId}.jsonl`); + + await writeSessionStoreFast(storePath, { + [sessionKey]: { + sessionId: existingSessionId, + updatedAt: new Date(2026, 0, 18, 3, 0, 0).getTime(), + }, + }); + await fs.writeFile(transcriptPath, '{"type":"message"}\n', "utf8"); + operation = replyRunRegistry.begin({ + sessionKey, + sessionId: "different-active-session", + resetTriggered: false, + }); + operation.setPhase("running"); + + const cfg = { session: { store: storePath } } as OpenClawConfig; + const result = await initSessionState({ + ctx: { + Body: "hello after boundary", + RawBody: "hello after boundary", + CommandBody: "hello after boundary", + From: "user-active-other-stale", + To: "bot", + ChatType: "direct", + SessionKey: sessionKey, + Provider: "telegram", + Surface: "telegram", + }, + cfg, + commandAuthorized: true, + }); + + expect(result.isNewSession).toBe(true); + expect(result.resetTriggered).toBe(false); + expect(result.sessionId).not.toBe(existingSessionId); + expect(result.previousSessionEntry?.sessionId).toBe(existingSessionId); + expect(await fs.stat(transcriptPath).catch(() => null)).toBeNull(); + const archived = (await fs.readdir(path.dirname(storePath))).filter((entry) => + entry.startsWith(`${existingSessionId}.jsonl.reset.`), + ); + expect(archived).toHaveLength(1); + } finally { + operation?.complete(); + vi.useRealTimers(); + } + }); + it("keeps provider-owned CLI sessions on implicit daily reset boundaries", async () => { vi.useFakeTimers(); try { diff --git a/src/auto-reply/reply/session.ts b/src/auto-reply/reply/session.ts index 58cbd8701cd2..0e1d63c253da 100644 --- a/src/auto-reply/reply/session.ts +++ b/src/auto-reply/reply/session.ts @@ -78,6 +78,7 @@ import { resolveLastChannelRaw, resolveLastToRaw, } from "./session-delivery.js"; +import { replyRunRegistry } from "./reply-run-registry.js"; import { createReplySessionEntryHandle, type ReplySessionEntryHandle, @@ -543,11 +544,25 @@ async function initSessionStateAttemptLocked( (entryFreshness?.fresh ?? false) || (softResetAllowed && canReuseExistingEntry)) && !terminalMainTranscriptNewerThanRegistry); + const activeReplyOperation = replyRunRegistry.get(sessionKey); + const deferImplicitRolloverForActiveRun = + !resetTriggered && + !freshEntry && + canReuseExistingEntry && + entryFreshness?.fresh === false && + entryFreshness.staleReason != null && + activeReplyOperation?.phase !== "queued" && + activeReplyOperation?.sessionId === entry?.sessionId; + // Implicit daily/idle rollover must not rename a transcript while that exact + // session's active writer is still running. Admission will steer/wait/queue; + // queued pre-dispatch reservations still let the current turn roll over. + const effectiveFreshEntry = deferImplicitRolloverForActiveRun ? true : freshEntry; // Capture the current session entry before any reset so its transcript can be // archived afterward. We need to do this for both explicit resets (/new, /reset) // and for scheduled/daily resets where the session has become stale (!freshEntry). // Without this, daily-reset transcripts are left as orphaned files on disk (#35481). - const previousSessionEntry = (resetTriggered || !freshEntry) && entry ? { ...entry } : undefined; + const previousSessionEntry = + (resetTriggered || !effectiveFreshEntry) && entry ? { ...entry } : undefined; const previousSessionEndReason = resetTriggered ? resolveExplicitSessionEndReason(matchedResetTriggerLower) : resolveStaleSessionEndReason({ @@ -562,7 +577,7 @@ async function initSessionStateAttemptLocked( clearSessionResetRuntimeState([sessionKey, previousSessionEntry.sessionId]); } - if (!isNewSession && freshEntry && canReuseExistingEntry) { + if (!isNewSession && effectiveFreshEntry && canReuseExistingEntry) { sessionId = entry.sessionId; systemSent = entry.systemSent ?? false; abortedLastRun = entry.abortedLastRun ?? false; @@ -633,7 +648,7 @@ async function initSessionStateAttemptLocked( } } - const baseEntry = !isNewSession && freshEntry ? entry : undefined; + const baseEntry = !isNewSession && effectiveFreshEntry ? entry : undefined; const usageFamilyKey = previousSessionEntry ? (previousSessionEntry.usageFamilyKey ?? sessionKey) : baseEntry?.usageFamilyKey;