Skip to content

Commit 49e6f5a

Browse files
authored
refactor(auto-reply): add lifecycle storage seams (#93685)
* refactor(auto-reply): add lifecycle storage seams * fix(auto-reply): remove unused transcript replay shim
1 parent 95c87e3 commit 49e6f5a

8 files changed

Lines changed: 341 additions & 96 deletions

scripts/check-session-accessor-boundary.mjs

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
9696
"src/auto-reply/reply/agent-runner-cli-dispatch.ts",
9797
"src/auto-reply/reply/agent-runner-execution.ts",
9898
"src/auto-reply/reply/agent-runner-memory.ts",
99+
"src/auto-reply/reply/agent-runner-session-reset.ts",
99100
"src/auto-reply/reply/agent-runner.ts",
100101
"src/auto-reply/reply/body.ts",
101102
"src/auto-reply/reply/commands-acp/lifecycle.ts",
@@ -106,6 +107,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
106107
"src/auto-reply/reply/followup-runner.ts",
107108
"src/auto-reply/reply/get-reply.ts",
108109
"src/auto-reply/reply/model-selection.ts",
110+
"src/auto-reply/reply/session.ts",
109111
"src/auto-reply/reply/session-reset-model.ts",
110112
"src/auto-reply/reply/session-updates.ts",
111113
"src/auto-reply/reply/session-usage.ts",

src/auto-reply/reply/agent-runner-session-reset.ts

Lines changed: 12 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,13 @@
11
// Handles session reset requests produced during agent runner execution.
2-
import fs from "node:fs";
32
import type { SessionEntry } from "../../config/sessions.js";
43
import {
54
resolveAgentIdFromSessionKey,
6-
resolveSessionFilePath,
7-
resolveSessionFilePathOptions,
85
resolveSessionTranscriptPath,
9-
updateSessionStore,
106
} from "../../config/sessions.js";
7+
import { persistSessionResetLifecycle } from "../../config/sessions/session-accessor.js";
118
import { generateSecureUuid } from "../../infra/secure-random.js";
129
import { defaultRuntime } from "../../runtime.js";
1310
import { refreshQueuedFollowupSession, type FollowupRun } from "./queue.js";
14-
import { replayRecentUserAssistantMessages } from "./session-transcript-replay.js";
1511

1612
type ResetSessionOptions = {
1713
failureLabel: string;
@@ -21,15 +17,15 @@ type ResetSessionOptions = {
2117

2218
const deps = {
2319
generateSecureUuid,
24-
updateSessionStore,
20+
persistSessionResetLifecycle,
2521
refreshQueuedFollowupSession,
2622
error: (message: string) => defaultRuntime.error(message),
2723
};
2824

2925
export function setAgentRunnerSessionResetTestDeps(overrides?: Partial<typeof deps>): void {
3026
Object.assign(deps, {
3127
generateSecureUuid,
32-
updateSessionStore,
28+
persistSessionResetLifecycle,
3329
refreshQueuedFollowupSession,
3430
error: (message: string) => defaultRuntime.error(message),
3531
...overrides,
@@ -95,21 +91,21 @@ export async function resetReplyRunSession(params: {
9591
nextEntry.sessionFile = nextSessionFile;
9692
params.activeSessionStore[params.sessionKey] = nextEntry;
9793
try {
98-
await deps.updateSessionStore(params.storePath, (store) => {
99-
store[params.sessionKey!] = nextEntry;
94+
await deps.persistSessionResetLifecycle({
95+
agentId,
96+
cleanupPreviousTranscript: params.options.cleanupTranscripts,
97+
nextEntry,
98+
nextSessionFile,
99+
previousEntry: prevEntry,
100+
previousSessionId: prevSessionId,
101+
sessionKey: params.sessionKey,
102+
storePath: params.storePath,
100103
});
101104
} catch (err) {
102105
deps.error(
103106
`Failed to persist session reset after ${params.options.failureLabel} (${params.sessionKey}): ${String(err)}`,
104107
);
105108
}
106-
// Silent rotations (compaction/role-ordering) fire without user intent, so
107-
// preserve recent user/assistant turns for direct-chat continuity.
108-
await replayRecentUserAssistantMessages({
109-
sourceTranscript: prevEntry.sessionFile,
110-
targetTranscript: nextSessionFile,
111-
newSessionId: nextSessionId,
112-
});
113109
params.followupRun.run.sessionId = nextSessionId;
114110
params.followupRun.run.sessionFile = nextSessionFile;
115111
deps.refreshQueuedFollowupSession({
@@ -121,24 +117,5 @@ export async function resetReplyRunSession(params: {
121117
params.onActiveSessionEntry(nextEntry);
122118
params.onNewSession(nextSessionId, nextSessionFile);
123119
deps.error(params.options.buildLogMessage(nextSessionId));
124-
if (params.options.cleanupTranscripts && prevSessionId) {
125-
const transcriptCandidates = new Set<string>();
126-
const resolved = resolveSessionFilePath(
127-
prevSessionId,
128-
prevEntry,
129-
resolveSessionFilePathOptions({ agentId, storePath: params.storePath }),
130-
);
131-
if (resolved) {
132-
transcriptCandidates.add(resolved);
133-
}
134-
transcriptCandidates.add(resolveSessionTranscriptPath(prevSessionId, agentId));
135-
for (const candidate of transcriptCandidates) {
136-
try {
137-
fs.unlinkSync(candidate);
138-
} catch {
139-
// Best-effort cleanup.
140-
}
141-
}
142-
}
143120
return true;
144121
}

src/auto-reply/reply/session-transcript-replay.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ import fs from "node:fs/promises";
33
import os from "node:os";
44
import path from "node:path";
55
import { afterEach, beforeEach, describe, expect, it } from "vitest";
6-
import { replayRecentUserAssistantMessages } from "./session-transcript-replay.js";
6+
import { replayRecentUserAssistantMessages } from "../../config/sessions/transcript-replay.js";
77

88
const DEFAULT_REPLAY_MAX_MESSAGES = 6;
99

src/auto-reply/reply/session.ts

Lines changed: 29 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,11 @@ import {
3030
resolveThreadFlag,
3131
type SessionFreshness,
3232
} from "../../config/sessions/reset.js";
33+
import { persistSessionRolloverLifecycle } from "../../config/sessions/session-accessor.js";
3334
import { resolveAndPersistSessionFile } from "../../config/sessions/session-file.js";
3435
import { resolveSessionKey } from "../../config/sessions/session-key.js";
3536
import { resolveMaintenanceConfigFromInput } from "../../config/sessions/store-maintenance.js";
36-
import { loadSessionStore, updateSessionStore } from "../../config/sessions/store.js";
37+
import { loadSessionStore } from "../../config/sessions/store.js";
3738
import { parseSessionThreadInfoFast } from "../../config/sessions/thread-info.js";
3839
import {
3940
DEFAULT_RESET_TRIGGERS,
@@ -54,7 +55,6 @@ import { getGlobalHookRunner } from "../../plugins/hook-runner-global.js";
5455
import type { PluginHookSessionEndReason } from "../../plugins/hook-types.js";
5556
import { isAcpSessionKey, normalizeMainKey } from "../../routing/session-key.js";
5657
import { isInterSessionInputProvenance } from "../../sessions/input-provenance.js";
57-
import { createLazyImportLoader } from "../../shared/lazy-promise.js";
5858
import {
5959
normalizeDeliveryChannelRoute,
6060
normalizeSessionDeliveryFields,
@@ -78,13 +78,6 @@ import { buildSessionEndHookPayload, buildSessionStartHookPayload } from "./sess
7878
import { clearSessionResetRuntimeState } from "./session-reset-cleanup.js";
7979

8080
const log = createSubsystemLogger("session-init");
81-
const sessionArchiveRuntimeLoader = createLazyImportLoader(
82-
() => import("../../gateway/session-archive.runtime.js"),
83-
);
84-
85-
function loadSessionArchiveRuntime() {
86-
return sessionArchiveRuntimeLoader.load();
87-
}
8881

8982
function stripThreadFromSessionRoute(route: SessionEntry["route"]): SessionEntry["route"] {
9083
const normalized = normalizeDeliveryChannelRoute(route);
@@ -836,56 +829,36 @@ export async function initSessionState(params: {
836829
}
837830
// Preserve per-session overrides while resetting compaction state on /new.
838831
sessionStore[sessionKey] = { ...sessionStore[sessionKey], ...sessionEntry };
839-
await updateSessionStore(
840-
storePath,
841-
(store) => {
842-
// Preserve per-session overrides while resetting compaction state on /new.
843-
store[sessionKey] = { ...store[sessionKey], ...sessionEntry };
844-
if (retiredLegacyMainDelivery) {
845-
store[retiredLegacyMainDelivery.key] = retiredLegacyMainDelivery.entry;
846-
}
847-
},
848-
{
849-
activeSessionKey: sessionKey,
850-
maintenanceConfig,
851-
onWarn: (warning) =>
852-
deliverSessionMaintenanceWarning({
853-
cfg,
854-
sessionKey,
855-
entry: sessionEntry,
856-
warning,
857-
}),
858-
},
859-
);
860832

861833
// Archive old transcript so it doesn't accumulate on disk (#14869).
862-
let previousSessionTranscript: {
863-
sessionFile?: string;
864-
transcriptArchived?: boolean;
865-
} = {};
834+
const rollover = await persistSessionRolloverLifecycle({
835+
activeSessionKey: sessionKey,
836+
agentId,
837+
maintenanceConfig,
838+
onArchiveError: (error, sourcePath) => {
839+
log.warn(
840+
`failed to archive previous session transcript ${sourcePath} for session ${previousSessionEntry?.sessionId}`,
841+
{ error: String(error) },
842+
);
843+
},
844+
onMaintenanceWarning: (warning) =>
845+
deliverSessionMaintenanceWarning({
846+
cfg,
847+
sessionKey,
848+
entry: sessionEntry,
849+
warning,
850+
}),
851+
previousEntry: previousSessionEntry,
852+
retiredEntry: retiredLegacyMainDelivery,
853+
sessionEntry,
854+
sessionKey,
855+
storePath,
856+
});
857+
sessionEntry = rollover.sessionEntry;
858+
sessionStore[sessionKey] = { ...sessionStore[sessionKey], ...sessionEntry };
859+
const previousSessionTranscript = rollover.previousSessionTranscript;
860+
866861
if (previousSessionEntry?.sessionId) {
867-
const { archiveSessionTranscriptsDetailed, resolveStableSessionEndTranscript } =
868-
await loadSessionArchiveRuntime();
869-
const archivedTranscripts = archiveSessionTranscriptsDetailed({
870-
sessionId: previousSessionEntry.sessionId,
871-
storePath,
872-
sessionFile: previousSessionEntry.sessionFile,
873-
agentId,
874-
reason: "reset",
875-
onArchiveError: (error, sourcePath) => {
876-
log.warn(
877-
`failed to archive previous session transcript ${sourcePath} for session ${previousSessionEntry.sessionId}`,
878-
{ error: String(error) },
879-
);
880-
},
881-
});
882-
previousSessionTranscript = resolveStableSessionEndTranscript({
883-
sessionId: previousSessionEntry.sessionId,
884-
storePath,
885-
sessionFile: previousSessionEntry.sessionFile,
886-
agentId,
887-
archivedTranscripts,
888-
});
889862
await retireSessionMcpRuntime({
890863
sessionId: previousSessionEntry.sessionId,
891864
reason: "reply-session-rollover",

src/config/sessions/session-accessor.test.ts

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ import {
1717
loadSessionEntry,
1818
loadTranscriptEvents,
1919
patchSessionEntry,
20+
persistSessionResetLifecycle,
21+
persistSessionRolloverLifecycle,
2022
persistSessionTranscriptTurn,
2123
purgeDeletedAgentSessionEntries,
2224
publishTranscriptUpdate,
@@ -597,6 +599,118 @@ describe("session accessor file-backed seam", () => {
597599
expect(fs.readdirSync(siblingDir)).toEqual(["sibling-lifecycle.jsonl"]);
598600
});
599601

602+
it("persists reset lifecycle entry changes with transcript replay and cleanup", async () => {
603+
const now = Date.now();
604+
const sessionKey = "agent:main:main";
605+
const previousTranscript = path.join(tempDir, "previous-session.jsonl");
606+
const nextTranscript = path.join(tempDir, "next-session.jsonl");
607+
const previousEntry: SessionEntry = {
608+
sessionFile: previousTranscript,
609+
sessionId: "previous-session",
610+
updatedAt: now,
611+
};
612+
const nextEntry: SessionEntry = {
613+
sessionFile: nextTranscript,
614+
sessionId: "next-session",
615+
updatedAt: now + 1,
616+
};
617+
fs.writeFileSync(
618+
previousTranscript,
619+
[
620+
JSON.stringify({ type: "session", id: "previous-session" }),
621+
JSON.stringify({
622+
id: "msg-user",
623+
message: { role: "user", content: "hello" },
624+
parentId: null,
625+
timestamp: "2026-06-16T00:00:00.000Z",
626+
type: "message",
627+
}),
628+
JSON.stringify({
629+
id: "msg-assistant",
630+
message: { role: "assistant", content: "hi" },
631+
parentId: "msg-user",
632+
timestamp: "2026-06-16T00:00:01.000Z",
633+
type: "message",
634+
}),
635+
].join("\n") + "\n",
636+
"utf-8",
637+
);
638+
await upsertSessionEntry({ sessionKey, storePath }, previousEntry);
639+
640+
const result = await persistSessionResetLifecycle({
641+
agentId: "main",
642+
cleanupPreviousTranscript: true,
643+
nextEntry,
644+
nextSessionFile: nextTranscript,
645+
previousEntry,
646+
previousSessionId: previousEntry.sessionId,
647+
sessionKey,
648+
storePath,
649+
});
650+
651+
expect(result.replayedMessages).toBe(2);
652+
expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject(nextEntry);
653+
expect(fs.existsSync(previousTranscript)).toBe(false);
654+
expect(fs.readFileSync(nextTranscript, "utf-8")).toContain('"content":"hello"');
655+
});
656+
657+
it("persists rollover entries and returns archived previous transcript info", async () => {
658+
const now = Date.now();
659+
const sessionKey = "agent:main:telegram:dm:user";
660+
const retiredKey = "agent:main:main";
661+
const previousTranscript = path.join(tempDir, "previous-rollover.jsonl");
662+
const previousEntry: SessionEntry = {
663+
sessionFile: previousTranscript,
664+
sessionId: "previous-rollover",
665+
updatedAt: now,
666+
};
667+
const nextEntry: SessionEntry = {
668+
sessionFile: path.join(tempDir, "next-rollover.jsonl"),
669+
sessionId: "next-rollover",
670+
updatedAt: now + 1,
671+
};
672+
fs.writeFileSync(previousTranscript, '{"type":"session","id":"previous-rollover"}\n', "utf-8");
673+
await upsertSessionEntry({ sessionKey, storePath }, previousEntry);
674+
await upsertSessionEntry(
675+
{ sessionKey: retiredKey, storePath },
676+
{
677+
lastChannel: "telegram",
678+
lastTo: "user",
679+
sessionId: "legacy-main",
680+
updatedAt: now,
681+
},
682+
);
683+
684+
const result = await persistSessionRolloverLifecycle({
685+
activeSessionKey: sessionKey,
686+
agentId: "main",
687+
previousEntry,
688+
retiredEntry: {
689+
key: retiredKey,
690+
entry: {
691+
sessionId: "legacy-main",
692+
updatedAt: now,
693+
},
694+
},
695+
sessionEntry: nextEntry,
696+
sessionKey,
697+
storePath,
698+
});
699+
700+
expect(result.sessionEntry).toMatchObject(nextEntry);
701+
expect(result.previousSessionTranscript.transcriptArchived).toBe(true);
702+
expect(result.previousSessionTranscript.sessionFile).toContain(
703+
"previous-rollover.jsonl.reset.",
704+
);
705+
expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject(nextEntry);
706+
expect(loadSessionEntry({ sessionKey: retiredKey, storePath })).toEqual({
707+
sessionId: "legacy-main",
708+
updatedAt: expect.any(Number),
709+
});
710+
expect(fs.existsSync(previousTranscript)).toBe(false);
711+
expect(fs.existsSync(result.previousSessionTranscript.sessionFile ?? "")).toBe(true);
712+
});
713+
600714
it("loads and appends transcript events through a session scope", async () => {
601715
const scope = {
602716
sessionFile: transcriptPath,

0 commit comments

Comments
 (0)