Skip to content

Commit c682919

Browse files
vincentkocvincentkoc
committed
fix(gateway): notify session changes from goal commands
1 parent 6324abb commit c682919

15 files changed

Lines changed: 527 additions & 100 deletions

‎src/auto-reply/dispatch.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import {
2020
resolveCommandTurnTargetSessionKey,
2121
} from "./command-turn-context.js";
2222
import { withReplyDispatcher } from "./dispatch-dispatcher.js";
23+
import type { CommandSessionMetadataChange } from "./reply/command-session-metadata.js";
2324
import { dispatchReplyFromConfig } from "./reply/dispatch-from-config.js";
2425
import type { DispatchFromConfigResult } from "./reply/dispatch-from-config.types.js";
2526
import type { GetReplyFromConfig } from "./reply/get-reply.types.js";
@@ -477,6 +478,7 @@ export async function dispatchInboundMessage(params: {
477478
toolsAllow?: string[];
478479
replyOptions?: Omit<GetReplyOptions, "onBlockReply">;
479480
replyResolver?: GetReplyFromConfig;
481+
onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void;
480482
}): Promise<DispatchInboundResult> {
481483
const replyOptions = applyRuntimeToolsAllow(params.replyOptions, params.toolsAllow);
482484
const finalized = measureDiagnosticsTimelineSpanSync(
@@ -512,6 +514,7 @@ export async function dispatchInboundMessage(params: {
512514
dispatcher: params.dispatcher,
513515
replyOptions,
514516
replyResolver: params.replyResolver,
517+
onSessionMetadataChanges: params.onSessionMetadataChanges,
515518
}),
516519
{
517520
phase: "agent-turn",
@@ -531,6 +534,7 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: {
531534
toolsAllow?: string[];
532535
replyOptions?: Omit<GetReplyOptions, "onBlockReply">;
533536
replyResolver?: GetReplyFromConfig;
537+
onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void;
534538
}): Promise<DispatchInboundResult> {
535539
const finalized = finalizeInboundContext(params.ctx);
536540
const foregroundReplyFence = beginForegroundReplyFence(finalized);
@@ -597,6 +601,7 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: {
597601
...params.replyOptions,
598602
...replyOptions,
599603
},
604+
onSessionMetadataChanges: params.onSessionMetadataChanges,
600605
});
601606
} finally {
602607
try {

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

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
// Tracks session metadata mutations made by command handlers during a turn.
2+
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
3+
import type { HandleCommandsParams } from "./commands-types.js";
4+
5+
export type CommandSessionMetadataChange = {
6+
sessionKey: string;
7+
agentId?: string;
8+
reason: "command-metadata";
9+
};
10+
11+
const commandSessionMetadataChanges = new WeakMap<object, CommandSessionMetadataChange[]>();
12+
13+
function addChange(target: object, change: CommandSessionMetadataChange): void {
14+
const changes = commandSessionMetadataChanges.get(target) ?? [];
15+
if (
16+
!changes.some(
17+
(candidate) =>
18+
candidate.sessionKey === change.sessionKey &&
19+
candidate.agentId === change.agentId &&
20+
candidate.reason === change.reason,
21+
)
22+
) {
23+
changes.push(change);
24+
}
25+
commandSessionMetadataChanges.set(target, changes);
26+
}
27+
28+
export function markCommandSessionMetadataChanged(
29+
params: Pick<HandleCommandsParams, "agentId" | "ctx" | "rootCtx" | "sessionKey">,
30+
): void {
31+
const sessionKey = normalizeOptionalString(params.sessionKey);
32+
if (!sessionKey) {
33+
return;
34+
}
35+
const change: CommandSessionMetadataChange = {
36+
sessionKey,
37+
...(params.agentId ? { agentId: params.agentId } : {}),
38+
reason: "command-metadata",
39+
};
40+
const targets = new Set<object>();
41+
if (params.rootCtx && typeof params.rootCtx === "object") {
42+
targets.add(params.rootCtx);
43+
}
44+
if (params.ctx && typeof params.ctx === "object") {
45+
targets.add(params.ctx);
46+
}
47+
for (const target of targets) {
48+
addChange(target, change);
49+
}
50+
}
51+
52+
export function takeCommandSessionMetadataChanges(
53+
target: object,
54+
): CommandSessionMetadataChange[] | undefined {
55+
const changes = commandSessionMetadataChanges.get(target);
56+
commandSessionMetadataChanges.delete(target);
57+
return changes && changes.length > 0 ? changes : undefined;
58+
}
59+
60+
export function takeCommandSessionMetadataChangesFromTargets(
61+
targets: Iterable<object>,
62+
): CommandSessionMetadataChange[] | undefined {
63+
const changes: CommandSessionMetadataChange[] = [];
64+
const seen = new Set<string>();
65+
for (const target of new Set(targets)) {
66+
for (const change of takeCommandSessionMetadataChanges(target) ?? []) {
67+
const key = JSON.stringify([change.sessionKey, change.agentId ?? null, change.reason]);
68+
if (seen.has(key)) {
69+
continue;
70+
}
71+
seen.add(key);
72+
changes.push(change);
73+
}
74+
}
75+
return changes.length > 0 ? changes : undefined;
76+
}

‎src/auto-reply/reply/commands-goal.test.ts‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import path from "node:path";
55
import { afterEach, describe, expect, it } from "vitest";
66
import { getSessionEntry, upsertSessionEntry } from "../../config/sessions.js";
77
import type { OpenClawConfig } from "../../config/types.openclaw.js";
8+
import { takeCommandSessionMetadataChanges } from "./command-session-metadata.js";
89
import {
910
formatGoalContinuationPrompt,
1011
handleGoalCommand,
@@ -110,6 +111,9 @@ describe("goal commands", () => {
110111
expect(params.command.commandBodyNormalized).toBe("build a 3d game");
111112
expect((params.ctx as { BodyForAgent?: string }).BodyForAgent).toBe("build a 3d game");
112113
expect(getSessionEntry({ storePath, sessionKey })?.goal?.objective).toBe("build a 3d game");
114+
expect(takeCommandSessionMetadataChanges(params.ctx)).toEqual([
115+
{ sessionKey, reason: "command-metadata" },
116+
]);
113117
});
114118

115119
it("wraps command-prefixed goal objectives before continuing", async () => {
@@ -216,4 +220,39 @@ describe("goal commands", () => {
216220
expect(directives.hasFastDirective).toBe(false);
217221
expect(getSessionEntry({ storePath, sessionKey })?.goal?.status).toBe("active");
218222
});
223+
224+
it("renders status without persisting derived budget state", async () => {
225+
const storePath = await createStorePath();
226+
await upsertSessionEntry({
227+
storePath,
228+
sessionKey,
229+
entry: {
230+
sessionId: "sess-main",
231+
updatedAt: 1,
232+
totalTokens: 25,
233+
totalTokensFresh: true,
234+
goal: {
235+
schemaVersion: 1,
236+
id: "goal-1",
237+
objective: "finish the migration",
238+
status: "active",
239+
createdAt: 1,
240+
updatedAt: 1,
241+
tokenStart: 0,
242+
tokenStartFresh: true,
243+
tokenBudget: 20,
244+
tokensUsed: 0,
245+
continuationTurns: 0,
246+
},
247+
},
248+
});
249+
250+
const params = buildGoalParams("/goal status", storePath);
251+
const result = await handleGoalCommand(params, true);
252+
253+
expect(result?.shouldContinue).toBe(false);
254+
expect(result?.reply?.text).toContain("Status: budget_limited");
255+
expect(getSessionEntry({ storePath, sessionKey })?.goal?.status).toBe("active");
256+
expect(takeCommandSessionMetadataChanges(params.ctx)).toBeUndefined();
257+
});
219258
});

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import {
1212
updateSessionGoalStatus,
1313
} from "../../config/sessions.js";
1414
import { rejectUnauthorizedCommand } from "./command-gates.js";
15+
import { markCommandSessionMetadataChanged } from "./command-session-metadata.js";
1516
import type {
1617
CommandHandler,
1718
CommandHandlerResult,
@@ -170,6 +171,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
170171
const snapshot = await getSessionGoal({
171172
sessionKey: params.sessionKey,
172173
storePath: params.storePath,
174+
fallbackEntry: params.sessionEntry,
175+
persist: false,
173176
});
174177
syncGoalSessionEntry(params);
175178
return goalReply(formatSessionGoalStatus(snapshot.goal));
@@ -188,6 +191,7 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
188191
fallbackEntry: params.sessionEntry,
189192
});
190193
syncGoalSessionEntry(params);
194+
markCommandSessionMetadataChanged(params);
191195
applyGoalContinuationPrompt(params, formatGoalContinuationPrompt(goal.objective));
192196
return goalContinuation();
193197
}
@@ -199,6 +203,7 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
199203
...(parsed.text ? { note: parsed.text } : {}),
200204
});
201205
syncGoalSessionEntry(params);
206+
markCommandSessionMetadataChanged(params);
202207
return goalReply(`Goal paused: ${goal.objective}`);
203208
}
204209
case "resume": {
@@ -209,6 +214,7 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
209214
...(parsed.text ? { note: parsed.text } : {}),
210215
});
211216
syncGoalSessionEntry(params);
217+
markCommandSessionMetadataChanged(params);
212218
const message = formatGoalResumeContinuationPrompt(parsed.text);
213219
applyGoalContinuationPrompt(params, message);
214220
return goalContinuation();
@@ -222,6 +228,7 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
222228
...(parsed.text ? { note: parsed.text } : {}),
223229
});
224230
syncGoalSessionEntry(params);
231+
markCommandSessionMetadataChanged(params);
225232
return goalReply(`Goal complete: ${goal.objective}\nTokens used: ${goal.tokensUsed}`);
226233
}
227234
case "block":
@@ -233,6 +240,7 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
233240
...(parsed.text ? { note: parsed.text } : {}),
234241
});
235242
syncGoalSessionEntry(params);
243+
markCommandSessionMetadataChanged(params);
236244
return goalReply(`Goal blocked: ${goal.objective}`);
237245
}
238246
case "clear": {
@@ -241,6 +249,9 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
241249
storePath: params.storePath,
242250
});
243251
syncGoalSessionEntry(params);
252+
if (removed) {
253+
markCommandSessionMetadataChanged(params);
254+
}
244255
return goalReply(removed ? "Goal cleared." : "No goal to clear.");
245256
}
246257
default:

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

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ import { createInternalHookEventPayload } from "../../test-utils/internal-hook-e
3535
import { getReplyPayloadMetadata } from "../reply-payload.js";
3636
import type { MsgContext } from "../templating.js";
3737
import { setReplyPayloadMetadata, type GetReplyOptions, type ReplyPayload } from "../types.js";
38+
import { markCommandSessionMetadataChanged } from "./command-session-metadata.js";
3839
import { PROVIDER_CONVERSATION_STATE_ERROR_USER_MESSAGE } from "./provider-request-error-classifier.js";
3940
import {
4041
createReplyDispatcher,
@@ -1042,6 +1043,56 @@ describe("dispatchReplyFromConfig", () => {
10421043
);
10431044
});
10441045

1046+
it("returns session metadata changes marked during reply resolution", async () => {
1047+
setNoAbort();
1048+
const sessionKey = "agent:main:main";
1049+
const dispatcher = createDispatcher();
1050+
const result = await dispatchReplyFromConfig({
1051+
ctx: buildTestCtx({
1052+
Provider: "telegram",
1053+
SessionKey: sessionKey,
1054+
}),
1055+
cfg: emptyConfig,
1056+
dispatcher,
1057+
replyResolver: async (ctx) => {
1058+
markCommandSessionMetadataChanged({ ctx, sessionKey });
1059+
return { text: "goal updated" };
1060+
},
1061+
});
1062+
1063+
expect(result.sessionMetadataChanges).toEqual([{ sessionKey, reason: "command-metadata" }]);
1064+
});
1065+
1066+
it("notifies session metadata changes before later dispatch errors", async () => {
1067+
setNoAbort();
1068+
const sessionKey = "agent:main:main";
1069+
const dispatcher = createDispatcher();
1070+
dispatcher.sendFinalReply = vi.fn(() => {
1071+
throw new Error("delivery failed");
1072+
});
1073+
const onSessionMetadataChanges = vi.fn();
1074+
1075+
await expect(
1076+
dispatchReplyFromConfig({
1077+
ctx: buildTestCtx({
1078+
Provider: "telegram",
1079+
SessionKey: sessionKey,
1080+
}),
1081+
cfg: emptyConfig,
1082+
dispatcher,
1083+
onSessionMetadataChanges,
1084+
replyResolver: async (ctx) => {
1085+
markCommandSessionMetadataChanged({ ctx, sessionKey });
1086+
return { text: "goal updated" };
1087+
},
1088+
}),
1089+
).rejects.toThrow("delivery failed");
1090+
1091+
expect(onSessionMetadataChanges).toHaveBeenCalledWith([
1092+
{ sessionKey, reason: "command-metadata" },
1093+
]);
1094+
});
1095+
10451096
it("skips pre-dispatch admission when the caller already aborted", async () => {
10461097
setNoAbort();
10471098
const sessionKey = "agent:main:telegram:group:-1003774691294:topic:3731";

0 commit comments

Comments
 (0)