Skip to content

Commit 6f10763

Browse files
joshavantjoshavant
authored
fix: defer active implicit session rollover (#97164)
1 parent 898ca97 commit 6f10763

4 files changed

Lines changed: 229 additions & 3 deletions

File tree

‎src/auto-reply/reply/get-reply-run.media-only.test.ts‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1622,6 +1622,33 @@ describe("runPreparedReply media-only handling", () => {
16221622
}
16231623
});
16241624

1625+
it("rebinds a queued pre-dispatch reply operation after session rollover", async () => {
1626+
const operation = createReplyOperation({
1627+
sessionId: "session-before-rollover",
1628+
sessionKey: "session-key",
1629+
resetTriggered: false,
1630+
});
1631+
1632+
try {
1633+
await expect(
1634+
runPreparedReply(
1635+
baseParams({
1636+
isNewSession: true,
1637+
sessionId: "session-after-rollover",
1638+
opts: { replyOperation: operation } as never,
1639+
}),
1640+
),
1641+
).resolves.toEqual({ text: "ok" });
1642+
1643+
const call = requireLastRunReplyAgentCall();
1644+
expect(operation.sessionId).toBe("session-after-rollover");
1645+
expect(call.replyOperation).toBe(operation);
1646+
expect(call.followupRun.run.sessionId).toBe("session-after-rollover");
1647+
} finally {
1648+
operation.complete();
1649+
}
1650+
});
1651+
16251652
it("does not interrupt its provided pre-dispatch reply operation for reset turns", async () => {
16261653
const queueSettings = await import("./queue/settings-runtime.js");
16271654
const embeddedAgentRuntime = await import("../../agents/embedded-agent.runtime.js");

‎src/auto-reply/reply/get-reply-run.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -937,6 +937,18 @@ export async function runPreparedReply(
937937
}
938938
const internalOpts = opts as InternalGetReplyOptions | undefined;
939939
const providedReplyOperation = internalOpts?.replyOperation;
940+
if (
941+
providedReplyOperation !== undefined &&
942+
providedReplyOperation.result === null &&
943+
providedReplyOperation.phase === "queued" &&
944+
sessionId !== undefined &&
945+
sessionId !== providedReplyOperation.sessionId
946+
) {
947+
// Dispatch reserves a queued operation before session init. If stale init
948+
// rotates the session, move the reservation so later steer/abort paths
949+
// target the session that will actually run.
950+
providedReplyOperation.updateSessionId(sessionId);
951+
}
940952
const isOwnPreDispatchOperationSession = (candidateSessionId: string | undefined): boolean =>
941953
providedReplyOperation !== undefined &&
942954
providedReplyOperation.result === null &&

‎src/auto-reply/reply/session.test.ts‎

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ import { createSessionConversationTestRegistry } from "../../test-utils/session-
3232
import { drainFormattedSystemEvents } from "./session-updates.js";
3333
import { persistSessionUsageUpdate } from "./session-usage.js";
3434
import { initSessionState } from "./session.js";
35+
import { replyRunRegistry } from "./reply-run-registry.js";
3536

3637
const sessionForkMocks = vi.hoisted(() => ({
3738
forkSessionFromParent: vi.fn(),
@@ -3709,6 +3710,177 @@ describe("initSessionState preserves behavior overrides across /new and /reset",
37093710
}
37103711
});
37113712

3713+
it("defers implicit daily rollover while the same session has an active run", async () => {
3714+
vi.useFakeTimers();
3715+
const existingSessionId = "active-stale-session";
3716+
let operation: ReturnType<typeof replyRunRegistry.begin> | undefined;
3717+
try {
3718+
vi.setSystemTime(new Date(2026, 0, 18, 5, 0, 0));
3719+
const storePath = await createStorePath("openclaw-active-stale-archive-");
3720+
const sessionKey = "agent:main:telegram:dm:active-stale-user";
3721+
const transcriptPath = path.join(path.dirname(storePath), `${existingSessionId}.jsonl`);
3722+
const sessionStartedAt = new Date(2026, 0, 18, 3, 0, 0).getTime();
3723+
3724+
await writeSessionStoreFast(storePath, {
3725+
[sessionKey]: {
3726+
sessionId: existingSessionId,
3727+
updatedAt: sessionStartedAt,
3728+
sessionStartedAt,
3729+
},
3730+
});
3731+
await fs.writeFile(transcriptPath, '{"type":"message"}\n', "utf8");
3732+
operation = replyRunRegistry.begin({
3733+
sessionKey,
3734+
sessionId: existingSessionId,
3735+
resetTriggered: false,
3736+
});
3737+
operation.setPhase("running");
3738+
3739+
const cfg = { session: { store: storePath } } as OpenClawConfig;
3740+
const result = await initSessionState({
3741+
ctx: {
3742+
Body: "hello while active",
3743+
RawBody: "hello while active",
3744+
CommandBody: "hello while active",
3745+
From: "user-active-stale",
3746+
To: "bot",
3747+
ChatType: "direct",
3748+
SessionKey: sessionKey,
3749+
Provider: "telegram",
3750+
Surface: "telegram",
3751+
},
3752+
cfg,
3753+
commandAuthorized: true,
3754+
});
3755+
3756+
expect(result.isNewSession).toBe(false);
3757+
expect(result.resetTriggered).toBe(false);
3758+
expect(result.sessionId).toBe(existingSessionId);
3759+
expect(result.previousSessionEntry).toBeUndefined();
3760+
expect(result.sessionEntry.sessionStartedAt).toBe(sessionStartedAt);
3761+
expect(await fs.stat(transcriptPath).catch(() => null)).not.toBeNull();
3762+
const archived = (await fs.readdir(path.dirname(storePath))).filter((entry) =>
3763+
entry.startsWith(`${existingSessionId}.jsonl.reset.`),
3764+
);
3765+
expect(archived).toHaveLength(0);
3766+
} finally {
3767+
operation?.complete();
3768+
vi.useRealTimers();
3769+
}
3770+
});
3771+
3772+
it("does not defer stale archival for the current turn's queued reservation", async () => {
3773+
vi.useFakeTimers();
3774+
let operation: ReturnType<typeof replyRunRegistry.begin> | undefined;
3775+
try {
3776+
vi.setSystemTime(new Date(2026, 0, 18, 5, 0, 0));
3777+
const storePath = await createStorePath("openclaw-queued-stale-archive-");
3778+
const sessionKey = "agent:main:telegram:dm:queued-stale-user";
3779+
const existingSessionId = "queued-stale-session";
3780+
const transcriptPath = path.join(path.dirname(storePath), `${existingSessionId}.jsonl`);
3781+
3782+
await writeSessionStoreFast(storePath, {
3783+
[sessionKey]: {
3784+
sessionId: existingSessionId,
3785+
updatedAt: new Date(2026, 0, 18, 3, 0, 0).getTime(),
3786+
},
3787+
});
3788+
await fs.writeFile(transcriptPath, '{"type":"message"}\n', "utf8");
3789+
operation = replyRunRegistry.begin({
3790+
sessionKey,
3791+
sessionId: existingSessionId,
3792+
resetTriggered: false,
3793+
});
3794+
3795+
const cfg = { session: { store: storePath } } as OpenClawConfig;
3796+
const result = await initSessionState({
3797+
ctx: {
3798+
Body: "hello after boundary",
3799+
RawBody: "hello after boundary",
3800+
CommandBody: "hello after boundary",
3801+
From: "user-queued-stale",
3802+
To: "bot",
3803+
ChatType: "direct",
3804+
SessionKey: sessionKey,
3805+
Provider: "telegram",
3806+
Surface: "telegram",
3807+
},
3808+
cfg,
3809+
commandAuthorized: true,
3810+
});
3811+
3812+
expect(operation.phase).toBe("queued");
3813+
expect(result.isNewSession).toBe(true);
3814+
expect(result.resetTriggered).toBe(false);
3815+
expect(result.sessionId).not.toBe(existingSessionId);
3816+
expect(result.previousSessionEntry?.sessionId).toBe(existingSessionId);
3817+
expect(await fs.stat(transcriptPath).catch(() => null)).toBeNull();
3818+
const archived = (await fs.readdir(path.dirname(storePath))).filter((entry) =>
3819+
entry.startsWith(`${existingSessionId}.jsonl.reset.`),
3820+
);
3821+
expect(archived).toHaveLength(1);
3822+
} finally {
3823+
operation?.complete();
3824+
vi.useRealTimers();
3825+
}
3826+
});
3827+
3828+
it("does not defer stale archival for a different active session id", async () => {
3829+
vi.useFakeTimers();
3830+
let operation: ReturnType<typeof replyRunRegistry.begin> | undefined;
3831+
try {
3832+
vi.setSystemTime(new Date(2026, 0, 18, 5, 0, 0));
3833+
const storePath = await createStorePath("openclaw-active-other-stale-archive-");
3834+
const sessionKey = "agent:main:telegram:dm:active-other-stale-user";
3835+
const existingSessionId = "inactive-stale-session";
3836+
const transcriptPath = path.join(path.dirname(storePath), `${existingSessionId}.jsonl`);
3837+
3838+
await writeSessionStoreFast(storePath, {
3839+
[sessionKey]: {
3840+
sessionId: existingSessionId,
3841+
updatedAt: new Date(2026, 0, 18, 3, 0, 0).getTime(),
3842+
},
3843+
});
3844+
await fs.writeFile(transcriptPath, '{"type":"message"}\n', "utf8");
3845+
operation = replyRunRegistry.begin({
3846+
sessionKey,
3847+
sessionId: "different-active-session",
3848+
resetTriggered: false,
3849+
});
3850+
operation.setPhase("running");
3851+
3852+
const cfg = { session: { store: storePath } } as OpenClawConfig;
3853+
const result = await initSessionState({
3854+
ctx: {
3855+
Body: "hello after boundary",
3856+
RawBody: "hello after boundary",
3857+
CommandBody: "hello after boundary",
3858+
From: "user-active-other-stale",
3859+
To: "bot",
3860+
ChatType: "direct",
3861+
SessionKey: sessionKey,
3862+
Provider: "telegram",
3863+
Surface: "telegram",
3864+
},
3865+
cfg,
3866+
commandAuthorized: true,
3867+
});
3868+
3869+
expect(result.isNewSession).toBe(true);
3870+
expect(result.resetTriggered).toBe(false);
3871+
expect(result.sessionId).not.toBe(existingSessionId);
3872+
expect(result.previousSessionEntry?.sessionId).toBe(existingSessionId);
3873+
expect(await fs.stat(transcriptPath).catch(() => null)).toBeNull();
3874+
const archived = (await fs.readdir(path.dirname(storePath))).filter((entry) =>
3875+
entry.startsWith(`${existingSessionId}.jsonl.reset.`),
3876+
);
3877+
expect(archived).toHaveLength(1);
3878+
} finally {
3879+
operation?.complete();
3880+
vi.useRealTimers();
3881+
}
3882+
});
3883+
37123884
it("keeps provider-owned CLI sessions on implicit daily reset boundaries", async () => {
37133885
vi.useFakeTimers();
37143886
try {

‎src/auto-reply/reply/session.ts‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@ import {
7878
resolveLastChannelRaw,
7979
resolveLastToRaw,
8080
} from "./session-delivery.js";
81+
import { replyRunRegistry } from "./reply-run-registry.js";
8182
import {
8283
createReplySessionEntryHandle,
8384
type ReplySessionEntryHandle,
@@ -543,11 +544,25 @@ async function initSessionStateAttemptLocked(
543544
(entryFreshness?.fresh ?? false) ||
544545
(softResetAllowed && canReuseExistingEntry)) &&
545546
!terminalMainTranscriptNewerThanRegistry);
547+
const activeReplyOperation = replyRunRegistry.get(sessionKey);
548+
const deferImplicitRolloverForActiveRun =
549+
!resetTriggered &&
550+
!freshEntry &&
551+
canReuseExistingEntry &&
552+
entryFreshness?.fresh === false &&
553+
entryFreshness.staleReason != null &&
554+
activeReplyOperation?.phase !== "queued" &&
555+
activeReplyOperation?.sessionId === entry?.sessionId;
556+
// Implicit daily/idle rollover must not rename a transcript while that exact
557+
// session's active writer is still running. Admission will steer/wait/queue;
558+
// queued pre-dispatch reservations still let the current turn roll over.
559+
const effectiveFreshEntry = deferImplicitRolloverForActiveRun ? true : freshEntry;
546560
// Capture the current session entry before any reset so its transcript can be
547561
// archived afterward. We need to do this for both explicit resets (/new, /reset)
548562
// and for scheduled/daily resets where the session has become stale (!freshEntry).
549563
// Without this, daily-reset transcripts are left as orphaned files on disk (#35481).
550-
const previousSessionEntry = (resetTriggered || !freshEntry) && entry ? { ...entry } : undefined;
564+
const previousSessionEntry =
565+
(resetTriggered || !effectiveFreshEntry) && entry ? { ...entry } : undefined;
551566
const previousSessionEndReason = resetTriggered
552567
? resolveExplicitSessionEndReason(matchedResetTriggerLower)
553568
: resolveStaleSessionEndReason({
@@ -562,7 +577,7 @@ async function initSessionStateAttemptLocked(
562577
clearSessionResetRuntimeState([sessionKey, previousSessionEntry.sessionId]);
563578
}
564579

565-
if (!isNewSession && freshEntry && canReuseExistingEntry) {
580+
if (!isNewSession && effectiveFreshEntry && canReuseExistingEntry) {
566581
sessionId = entry.sessionId;
567582
systemSent = entry.systemSent ?? false;
568583
abortedLastRun = entry.abortedLastRun ?? false;
@@ -633,7 +648,7 @@ async function initSessionStateAttemptLocked(
633648
}
634649
}
635650

636-
const baseEntry = !isNewSession && freshEntry ? entry : undefined;
651+
const baseEntry = !isNewSession && effectiveFreshEntry ? entry : undefined;
637652
const usageFamilyKey = previousSessionEntry
638653
? (previousSessionEntry.usageFamilyKey ?? sessionKey)
639654
: baseEntry?.usageFamilyKey;

0 commit comments

Comments
 (0)