Skip to content

Commit 3e6b7a7

Browse files
steipeteTurboTheTurtle
steipete
and
TurboTheTurtle
committed
fix(reply): refresh transcript session bindings
Co-authored-by: Andy Ye <[email protected]>
1 parent 536aef9 commit 3e6b7a7

14 files changed

Lines changed: 205 additions & 16 deletions

‎src/auto-reply/reply/commands-reset-hooks.test.ts‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,9 +209,16 @@ describe("handleCommands reset hooks", () => {
209209
});
210210

211211
it("uses gateway session reset for bound ACP sessions", async () => {
212+
resetMocks.resetConfiguredBindingTargetInPlace.mockResolvedValue({
213+
ok: true,
214+
sessionKey: "agent:claude:acp:binding:discord:default:9373ab192b2317f4",
215+
sessionId: "session-after-acp-reset",
216+
storePath: "/tmp/claude-sessions.json",
217+
});
212218
resetMocks.resolveBoundAcpThreadSessionKey.mockReturnValue(
213219
"agent:claude:acp:binding:discord:default:9373ab192b2317f4",
214220
);
221+
const onSessionPrepared = vi.fn();
215222
const params = buildResetParams(
216223
"/reset",
217224
{
@@ -224,6 +231,7 @@ describe("handleCommands reset hooks", () => {
224231
CommandSource: "native",
225232
},
226233
);
234+
params.opts = { onSessionPrepared } as never;
227235

228236
const result = await maybeHandleResetCommand(params);
229237

@@ -245,6 +253,11 @@ describe("handleCommands reset hooks", () => {
245253
});
246254
expect(triggerInternalHookMock).not.toHaveBeenCalled();
247255
expect(params.command.resetHookTriggered).toBe(true);
256+
expect(onSessionPrepared).toHaveBeenCalledWith({
257+
sessionKey: "agent:claude:acp:binding:discord:default:9373ab192b2317f4",
258+
sessionId: "session-after-acp-reset",
259+
storePath: "/tmp/claude-sessions.json",
260+
});
248261
});
249262

250263
it("keeps tail dispatch after a bound ACP reset", async () => {

‎src/auto-reply/reply/commands-reset.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,13 @@ import { resolveBoundAcpThreadSessionKey } from "./commands-acp/targets.js";
99
import { emitResetCommandHooks, type ResetCommandAction } from "./commands-reset-hooks.js";
1010
import { parseSoftResetCommand } from "./commands-reset-mode.js";
1111
import type { CommandHandlerResult, HandleCommandsParams } from "./commands-types.js";
12+
import type { ReplySessionBinding } from "./get-reply.types.js";
1213
import { isResetAuthorizedForContext } from "./reset-authorization.js";
1314

15+
type InternalResetCommandOptions = NonNullable<HandleCommandsParams["opts"]> & {
16+
onSessionPrepared?: (binding: ReplySessionBinding) => void;
17+
};
18+
1419
function applyAcpResetTailContext(ctx: HandleCommandsParams["ctx"], resetTail: string): void {
1520
const mutableCtx = ctx as Record<string, unknown>;
1621
mutableCtx.Body = resetTail;
@@ -139,6 +144,13 @@ export async function maybeHandleResetCommand(
139144
logVerbose(`acp reset failed for ${boundAcpKey}: ${resetResult.error ?? "unknown error"}`);
140145
}
141146
if (resetResult.ok) {
147+
if (resetResult.sessionId) {
148+
(params.opts as InternalResetCommandOptions | undefined)?.onSessionPrepared?.({
149+
sessionKey: resetResult.sessionKey ?? boundAcpKey,
150+
sessionId: resetResult.sessionId,
151+
storePath: resetResult.storePath,
152+
});
153+
}
142154
params.command.resetHookTriggered = true;
143155
if (resetTail) {
144156
applyAcpResetTailContext(params.ctx, resetTail);

‎src/auto-reply/reply/dispatch-from-config.test.ts‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1271,6 +1271,57 @@ describe("dispatchReplyFromConfig", () => {
12711271
});
12721272
});
12731273

1274+
it("mirrors reset acknowledgements into the canonically prepared Slack session", async () => {
1275+
setNoAbort();
1276+
const dispatcher = createDispatcher();
1277+
const sessionKey = "Agent:Main:Slack:Channel:C123";
1278+
const preparedSessionKey = "agent:main:slack:channel:c123";
1279+
sessionStoreMocks.currentEntry = {
1280+
sessionId: "previous-session",
1281+
updatedAt: Date.now(),
1282+
};
1283+
transcriptMocks.appendAssistantMessageToSessionTranscript.mockClear();
1284+
1285+
await dispatchReplyFromConfig({
1286+
ctx: buildTestCtx({
1287+
Provider: "slack",
1288+
Surface: "slack",
1289+
OriginatingChannel: "slack",
1290+
OriginatingTo: "channel:C123",
1291+
ChatType: "group",
1292+
SessionKey: sessionKey,
1293+
MessageSid: "slack-reset-message",
1294+
}),
1295+
cfg: emptyConfig,
1296+
dispatcher,
1297+
replyResolver: async (_ctx, opts) => {
1298+
(
1299+
opts as GetReplyOptions & {
1300+
onSessionPrepared?: (binding: {
1301+
sessionKey?: string;
1302+
sessionId: string;
1303+
storePath?: string;
1304+
}) => void;
1305+
}
1306+
).onSessionPrepared?.({
1307+
sessionKey: preparedSessionKey,
1308+
sessionId: "new-session",
1309+
storePath: "/tmp/rotated-sessions.json",
1310+
});
1311+
return { text: "✅ New session started." };
1312+
},
1313+
});
1314+
1315+
expect(transcriptMocks.appendAssistantMessageToSessionTranscript).toHaveBeenCalledWith(
1316+
expect.objectContaining({
1317+
sessionKey,
1318+
expectedSessionId: "new-session",
1319+
storePath: "/tmp/rotated-sessions.json",
1320+
text: "✅ New session started.",
1321+
}),
1322+
);
1323+
});
1324+
12741325
it.each([
12751326
["embedded", { assistantMessageIndex: 7 }],
12761327
["CLI", { assistantTranscriptOwned: true }],

‎src/auto-reply/reply/dispatch-from-config.ts‎

Lines changed: 44 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ import { normalizeChatType } from "../../channels/chat-type.js";
4343
import { resolveChannelModelOverride } from "../../channels/model-overrides.js";
4444
import { shouldSuppressLocalExecApprovalPrompt } from "../../channels/plugins/exec-approval-local.js";
4545
import { applyMergePatch } from "../../config/merge-patch.js";
46+
import { normalizeExplicitSessionKey } from "../../config/sessions/explicit-session-key-normalization.js";
4647
import { resolveGroupSessionKey } from "../../config/sessions/group.js";
4748
import {
4849
appendAssistantMessageToSessionTranscript,
@@ -142,6 +143,7 @@ import type {
142143
} from "./dispatch-from-config.types.js";
143144
import { resolveEffectiveReplyRoute } from "./effective-reply-route.js";
144145
import { withFullRuntimeReplyConfig } from "./get-reply-fast-path.js";
146+
import type { ReplySessionBinding } from "./get-reply.types.js";
145147
import { claimInboundDedupe, commitInboundDedupe, releaseInboundDedupe } from "./inbound-dedupe.js";
146148
import { hasInboundAudio } from "./inbound-media.js";
147149
import { resolveOriginMessageProvider } from "./origin-routing.js";
@@ -180,6 +182,7 @@ type TranscriptMirror = SourceReplyTranscriptMirror & {
180182
};
181183
type InternalReplyResolverOptions = {
182184
onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void;
185+
onSessionPrepared?: (binding: ReplySessionBinding) => void;
183186
};
184187

185188
class DispatchReplyOperationAbortedError extends Error {
@@ -1225,6 +1228,34 @@ export async function dispatchReplyFromConfig(
12251228
const sessionStoreEntry = boundAcpDispatchSessionKey
12261229
? resolveSessionStoreLookup({ ...ctx, SessionKey: boundAcpDispatchSessionKey }, cfg)
12271230
: initialSessionStoreEntry;
1231+
let preparedSessionBinding: ReplySessionBinding | undefined =
1232+
sessionStoreEntry.sessionKey && sessionStoreEntry.entry?.sessionId
1233+
? {
1234+
sessionKey: sessionStoreEntry.sessionKey,
1235+
sessionId: sessionStoreEntry.entry.sessionId,
1236+
storePath: sessionStoreEntry.storePath,
1237+
}
1238+
: undefined;
1239+
const sessionKeysMatch = (left?: string, right?: string) =>
1240+
Boolean(
1241+
left &&
1242+
right &&
1243+
normalizeExplicitSessionKey(left, ctx) === normalizeExplicitSessionKey(right, ctx),
1244+
);
1245+
const notePreparedSession = (binding: ReplySessionBinding) => {
1246+
if (sessionKeysMatch(binding.sessionKey, sessionStoreEntry.sessionKey)) {
1247+
preparedSessionBinding = binding;
1248+
}
1249+
};
1250+
const resolvePreparedTranscriptBinding = (mirrorSessionKey?: string) => {
1251+
if (
1252+
!preparedSessionBinding ||
1253+
!sessionKeysMatch(mirrorSessionKey, preparedSessionBinding.sessionKey)
1254+
) {
1255+
return undefined;
1256+
}
1257+
return preparedSessionBinding;
1258+
};
12281259
const sessionAgentId = resolveSessionAgentId({
12291260
sessionKey: acpDispatchSessionKey,
12301261
config: cfg,
@@ -2250,14 +2281,16 @@ export async function dispatchReplyFromConfig(
22502281
await flushPendingCommentaryProgress();
22512282
throwIfFinalDeliveryAborted();
22522283
const payloadMetadata = getReplyPayloadMetadata(payload);
2284+
const sourceReplySessionBinding = resolvePreparedTranscriptBinding(
2285+
payloadMetadata?.sourceReplyTranscriptMirror?.sessionKey,
2286+
);
22532287
const sourceReplyTranscriptMirror = payloadMetadata?.sourceReplyTranscriptMirror
22542288
? {
22552289
...payloadMetadata.sourceReplyTranscriptMirror,
2256-
...(payloadMetadata.sourceReplyTranscriptMirror.sessionKey ===
2257-
sessionStoreEntry.sessionKey && sessionStoreEntry.entry?.sessionId
2258-
? { expectedSessionId: sessionStoreEntry.entry.sessionId }
2290+
...(sourceReplySessionBinding
2291+
? { expectedSessionId: sourceReplySessionBinding.sessionId }
22592292
: {}),
2260-
storePath: sessionStoreEntry.storePath,
2293+
storePath: sourceReplySessionBinding?.storePath ?? sessionStoreEntry.storePath,
22612294
}
22622295
: undefined;
22632296
const hasTranscriptOwner =
@@ -2309,6 +2342,9 @@ export async function dispatchReplyFromConfig(
23092342
const transcriptMirrorSourceId =
23102343
normalizeOptionalString(messageIdForHook) ??
23112344
normalizeOptionalString(params.replyOptions?.runId);
2345+
const transcriptMirrorSessionBinding = resolvePreparedTranscriptBinding(
2346+
transcriptMirrorSessionKey,
2347+
);
23122348
const transcriptMirror =
23132349
sourceReplyTranscriptMirror ??
23142350
(!hasTranscriptOwner &&
@@ -2319,11 +2355,10 @@ export async function dispatchReplyFromConfig(
23192355
{
23202356
sessionKey: transcriptMirrorSessionKey,
23212357
agentId: sessionAgentId,
2322-
...(transcriptMirrorSessionKey === sessionStoreEntry.sessionKey &&
2323-
sessionStoreEntry.entry?.sessionId
2324-
? { expectedSessionId: sessionStoreEntry.entry.sessionId }
2358+
...(transcriptMirrorSessionBinding
2359+
? { expectedSessionId: transcriptMirrorSessionBinding.sessionId }
23252360
: {}),
2326-
storePath: sessionStoreEntry.storePath,
2361+
storePath: transcriptMirrorSessionBinding?.storePath ?? sessionStoreEntry.storePath,
23272362
preferText: true,
23282363
idempotencyKey: transcriptMirrorSourceId
23292364
? `channel-final:${transcriptMirrorSourceId}:${options.deliveryId ?? "single"}`
@@ -2795,6 +2830,7 @@ export async function dispatchReplyFromConfig(
27952830
sourceReplyDeliveryMode,
27962831
...({
27972832
onSessionMetadataChanges: notifySessionMetadataChanges,
2833+
onSessionPrepared: notePreparedSession,
27982834
} satisfies InternalReplyResolverOptions),
27992835
onObservedReplyDelivery: markObservedReplyDelivery,
28002836
suppressToolErrorWarnings,

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1730,13 +1730,16 @@ describe("runPreparedReply media-only handling", () => {
17301730
async () => await authPromise.then(() => undefined),
17311731
);
17321732
vi.mocked(queueSettings.resolveQueueSettings).mockReturnValueOnce({ mode: "interrupt" });
1733+
const onSessionPrepared = vi.fn();
17331734

17341735
const runPromise = runPreparedReply(
17351736
baseParams({
17361737
isNewSession: false,
17371738
sessionId: "session-before-rotation",
17381739
sessionEntry: sessionStore["session-key"],
17391740
sessionStore,
1741+
storePath: "/tmp/sessions.json",
1742+
opts: { onSessionPrepared } as never,
17401743
}),
17411744
);
17421745

@@ -1768,6 +1771,11 @@ describe("runPreparedReply media-only handling", () => {
17681771
await expect(runPromise).resolves.toEqual({ text: "ok" });
17691772
const call = requireLastRunReplyAgentCall();
17701773
expect(call?.followupRun.run.sessionId).toBe("session-after-rotation");
1774+
expect(onSessionPrepared).toHaveBeenLastCalledWith({
1775+
sessionKey: "session-key",
1776+
sessionId: "session-after-rotation",
1777+
storePath: "/tmp/sessions.json",
1778+
});
17711779
});
17721780
it("reports still shutting down when a new owner appears after waiting", async () => {
17731781
vi.useFakeTimers();

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ import type { InlineDirectives } from "./directive-handling.js";
6969
import { isSystemEventProvider } from "./effective-reply-route.js";
7070
import { shouldUseReplyFastTestRuntime } from "./get-reply-fast-path.js";
7171
import { resolvePreparedReplyQueueState } from "./get-reply-run-queue.js";
72+
import type { ReplySessionBinding } from "./get-reply.types.js";
7273
import {
7374
buildDirectChatContext,
7475
buildGroupChatContext,
@@ -119,6 +120,7 @@ type InternalGetReplyOptions = GetReplyOptions & {
119120
* can differ from abortSignal when dispatch temporarily borrows an active lane.
120121
*/
121122
queuedFollowupAbortSignal?: AbortSignal;
123+
onSessionPrepared?: (binding: ReplySessionBinding) => void;
122124
};
123125

124126
type AgentDefaults = NonNullable<OpenClawConfig["agents"]>["defaults"];
@@ -940,6 +942,11 @@ export async function runPreparedReply(
940942
}).existing ?? sessionEntry)
941943
: sessionEntry;
942944
const latestSessionId = latestSessionEntry?.sessionId ?? sessionIdFinal;
945+
internalOpts?.onSessionPrepared?.({
946+
sessionKey,
947+
sessionId: latestSessionId,
948+
storePath,
949+
});
943950
return {
944951
sessionEntry: latestSessionEntry,
945952
sessionId: latestSessionId,
@@ -1330,8 +1337,7 @@ export async function runPreparedReply(
13301337
: {}),
13311338
},
13321339
timeoutMs,
1333-
runTimeoutOverrideMs:
1334-
opts?.timeoutOverrideSeconds !== undefined ? timeoutMs : undefined,
1340+
runTimeoutOverrideMs: opts?.timeoutOverrideSeconds !== undefined ? timeoutMs : undefined,
13351341
blockReplyBreak: resolvedBlockStreamingBreak,
13361342
ownerNumbers: command.ownerList.length > 0 ? command.ownerList : undefined,
13371343
inputProvenance,

‎src/auto-reply/reply/get-reply.fast-path.test.ts‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,34 @@ describe("getReplyFromConfig fast test bootstrap", () => {
222222
expectResolvedTelegramTimezone(mocks.resolveReplyDirectives);
223223
});
224224

225+
it("reports the prepared session binding after session bootstrap", async () => {
226+
vi.stubEnv("OPENCLAW_ALLOW_SLOW_REPLY_TESTS", "1");
227+
mocks.initSessionState.mockResolvedValue(
228+
createGetReplySessionState({
229+
sessionKey: "agent:main:slack:channel:C123",
230+
sessionId: "rotated-session",
231+
storePath: "/tmp/custom-sessions.json",
232+
}),
233+
);
234+
const onSessionPrepared = vi.fn();
235+
236+
await getReplyFromConfig(
237+
buildGetReplyCtx({
238+
Provider: "slack",
239+
Surface: "slack",
240+
SessionKey: "agent:main:slack:channel:C123",
241+
}),
242+
{ onSessionPrepared } as never,
243+
{} as OpenClawConfig,
244+
);
245+
246+
expect(onSessionPrepared).toHaveBeenCalledWith({
247+
sessionKey: "agent:main:slack:channel:C123",
248+
sessionId: "rotated-session",
249+
storePath: "/tmp/custom-sessions.json",
250+
});
251+
});
252+
225253
it("marks configs through withFastReplyConfig()", async () => {
226254
const cfg = withFastReplyConfig({ session: { store: "/tmp/sessions.json" } } as OpenClawConfig);
227255

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ import {
4545
import { handleInlineActions } from "./get-reply-inline-actions.js";
4646
import { maybeResolveNativeSlashCommandFastReply } from "./get-reply-native-slash-fast-path.js";
4747
import { runPreparedReply } from "./get-reply-run.js";
48+
import type { ReplySessionBinding } from "./get-reply.types.js";
4849
import { finalizeInboundContext } from "./inbound-context.js";
4950
import { hasInboundMedia } from "./inbound-media.js";
5051
import { emitPreAgentMessageHooks } from "./message-preprocess-hooks.js";
@@ -60,6 +61,10 @@ import { createTypingController } from "./typing.js";
6061

6162
type ResetCommandAction = "new" | "reset";
6263

64+
type InternalGetReplyOptions = GetReplyOptions & {
65+
onSessionPrepared?: (binding: ReplySessionBinding) => void;
66+
};
67+
6368
function classifyHeartbeatPendingFinalDelivery(text: string, ackMaxChars: number) {
6469
const stripped = stripHeartbeatToken(text, {
6570
mode: "heartbeat",
@@ -496,6 +501,11 @@ export async function getReplyFromConfig(
496501
} = sessionState;
497502
let { abortedLastRun } = sessionState;
498503
resolverTimingSessionKey = sessionKey ?? resolverTimingSessionKey;
504+
(resolvedOpts as InternalGetReplyOptions | undefined)?.onSessionPrepared?.({
505+
sessionKey,
506+
sessionId,
507+
storePath,
508+
});
499509

500510
if (sessionEntry?.pendingFinalDelivery && sessionEntry.pendingFinalDeliveryText) {
501511
const text = sanitizePendingFinalDeliveryText(sessionEntry.pendingFinalDeliveryText);

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,12 @@ import type { GetReplyOptions } from "../get-reply-options.types.js";
44
import type { ReplyPayload } from "../reply-payload.js";
55
import type { MsgContext } from "../templating.js";
66

7+
export type ReplySessionBinding = {
8+
sessionKey?: string;
9+
sessionId: string;
10+
storePath?: string;
11+
};
12+
713
/** Reply resolver signature used by dispatchers and tests for dependency injection. */
814
export type GetReplyFromConfig = (
915
ctx: MsgContext,

0 commit comments

Comments
 (0)