Skip to content

Commit bfc4e1d

Browse files
Evaobviyus
Eva
authored and
obviyus
committed
fix(telegram): restore active-run steering
1 parent 4895a00 commit bfc4e1d

7 files changed

Lines changed: 278 additions & 10 deletions

File tree

‎extensions/codex/src/app-server/run-attempt.steering.test.ts‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,30 @@ import {
1717
turnStartResult,
1818
} from "./run-attempt-test-harness.js";
1919

20+
const activeRunRegistrationMocks = vi.hoisted(() => ({
21+
clearActiveEmbeddedRun: vi.fn(),
22+
setActiveEmbeddedRun: vi.fn(),
23+
}));
24+
25+
vi.mock("openclaw/plugin-sdk/agent-harness-runtime", async (importOriginal) => {
26+
const actual = await importOriginal<typeof import("openclaw/plugin-sdk/agent-harness-runtime")>();
27+
return {
28+
...actual,
29+
clearActiveEmbeddedRun: (
30+
...args: Parameters<typeof actual.clearActiveEmbeddedRun>
31+
): ReturnType<typeof actual.clearActiveEmbeddedRun> => {
32+
activeRunRegistrationMocks.clearActiveEmbeddedRun(...args);
33+
return actual.clearActiveEmbeddedRun(...args);
34+
},
35+
setActiveEmbeddedRun: (
36+
...args: Parameters<typeof actual.setActiveEmbeddedRun>
37+
): ReturnType<typeof actual.setActiveEmbeddedRun> => {
38+
activeRunRegistrationMocks.setActiveEmbeddedRun(...args);
39+
return actual.setActiveEmbeddedRun(...args);
40+
},
41+
};
42+
});
43+
2044
setupRunAttemptTestHooks();
2145

2246
let steeringSessionIndex = 0;
@@ -121,6 +145,52 @@ describe("runCodexAppServerAttempt steering", () => {
121145
await run;
122146
});
123147

148+
it("passes session files through active Codex app-server registration for command lookup", async () => {
149+
const { requests, waitForMethod, completeTurn } = createStartedThreadHarness();
150+
const params = createSteeringParams();
151+
activeRunRegistrationMocks.setActiveEmbeddedRun.mockClear();
152+
activeRunRegistrationMocks.clearActiveEmbeddedRun.mockClear();
153+
154+
const run = runCodexAppServerAttempt(params);
155+
await waitForMethod("turn/start");
156+
157+
expect(activeRunRegistrationMocks.setActiveEmbeddedRun).toHaveBeenCalledWith(
158+
params.sessionId,
159+
expect.anything(),
160+
params.sessionKey,
161+
params.sessionFile,
162+
);
163+
164+
await waitAndQueueActiveRunMessage(params.sessionId, "session-file registered", {
165+
debounceMs: 0,
166+
});
167+
168+
await vi.waitFor(
169+
() =>
170+
expect(requests.filter((entry) => entry.method === "turn/steer")).toEqual([
171+
{
172+
method: "turn/steer",
173+
params: {
174+
threadId: "thread-1",
175+
expectedTurnId: "turn-1",
176+
input: [{ type: "text", text: "session-file registered", text_elements: [] }],
177+
},
178+
},
179+
]),
180+
fastWait,
181+
);
182+
183+
await completeTurn({ threadId: "thread-1", turnId: "turn-1" });
184+
await run;
185+
186+
expect(activeRunRegistrationMocks.clearActiveEmbeddedRun).toHaveBeenCalledWith(
187+
params.sessionId,
188+
expect.anything(),
189+
params.sessionKey,
190+
params.sessionFile,
191+
);
192+
});
193+
124194
it("flushes batched default queued steering during normal turn cleanup", async () => {
125195
const { requests, waitForMethod, completeTurn } = createStartedThreadHarness();
126196
const params = createSteeringParams();

‎extensions/codex/src/app-server/run-attempt.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2886,12 +2886,13 @@ export async function runCodexAppServerAttempt(
28862886
queueMessage: async (text: string, optionsLocal?: CodexSteeringQueueOptions) =>
28872887
activeSteeringQueue.queue(text, optionsLocal),
28882888
isStreaming: () => !completed && !runAbortController.signal.aborted,
2889+
isStopped: () => completed || timedOut || runAbortController.signal.aborted,
28892890
isCompacting: () => projectorRef.current?.isCompacting() ?? false,
28902891
sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,
28912892
cancel: () => runAbortController.abort("cancelled"),
28922893
abort: () => runAbortController.abort("aborted"),
28932894
};
2894-
setActiveEmbeddedRun(params.sessionId, handle, params.sessionKey);
2895+
setActiveEmbeddedRun(params.sessionId, handle, params.sessionKey, params.sessionFile);
28952896
const notifyUserMessagePersisted = createCodexAppServerUserMessagePersistenceNotifier(params);
28962897
void mirrorPromptAtTurnStartBestEffort({
28972898
params,
@@ -3281,7 +3282,7 @@ export async function runCodexAppServerAttempt(
32813282
runAbortController.signal.removeEventListener("abort", abortListener);
32823283
params.abortSignal?.removeEventListener("abort", abortFromUpstream);
32833284
steeringQueueRef.current?.cancel();
3284-
clearActiveEmbeddedRun(params.sessionId, handle, params.sessionKey);
3285+
clearActiveEmbeddedRun(params.sessionId, handle, params.sessionKey, params.sessionFile);
32853286
}
32863287
}
32873288

‎extensions/telegram/src/sequential-key.test.ts‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,18 @@ describe("getTelegramSequentialKey", () => {
7979
{ message: mockMessage({ chat: mockChat({ id: 123 }), text: "/stop" }) },
8080
"telegram:123:control",
8181
],
82+
[
83+
{ message: mockMessage({ chat: mockChat({ id: 123 }), text: "/steer keep going" }) },
84+
"telegram:123:control",
85+
],
86+
[
87+
{ message: mockMessage({ chat: mockChat({ id: 123 }), text: "/tell use the cache" }) },
88+
"telegram:123:control",
89+
],
90+
[
91+
{ message: mockMessage({ chat: mockChat({ id: 123 }), text: "/queue status" }) },
92+
"telegram:123:control",
93+
],
8294
[
8395
{
8496
message: mockMessage({
@@ -90,6 +102,41 @@ describe("getTelegramSequentialKey", () => {
90102
},
91103
"telegram:-100:control",
92104
],
105+
[
106+
{
107+
message: mockMessage({
108+
chat: mockChat({ id: -100, type: "supergroup", is_forum: true }),
109+
is_topic_message: true,
110+
message_thread_id: 5907,
111+
text: "/steer@vacs_tars_bot keep going",
112+
}),
113+
},
114+
"telegram:-100:control",
115+
],
116+
[
117+
{
118+
me: { username: "openclaw_bot" } as never,
119+
message: mockMessage({
120+
chat: mockChat({ id: -100, type: "supergroup", is_forum: true }),
121+
is_topic_message: true,
122+
message_thread_id: 5907,
123+
text: "/tell@openclaw_bot keep going!",
124+
}),
125+
},
126+
"telegram:-100:control",
127+
],
128+
[
129+
{
130+
me: { username: "openclaw_bot" } as never,
131+
message: mockMessage({
132+
chat: mockChat({ id: -100, type: "supergroup", is_forum: true }),
133+
is_topic_message: true,
134+
message_thread_id: 5907,
135+
text: "/queue@some_other_bot status",
136+
}),
137+
},
138+
"telegram:-100:topic:5907",
139+
],
93140
[
94141
{
95142
me: { username: "openclaw_bot" } as never,

‎extensions/telegram/src/sequential-key.ts‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ const TELEGRAM_READ_ONLY_STATUS_COMMAND_KEYS = new Set([
2525
"whoami",
2626
]);
2727

28+
const TELEGRAM_ACTIVE_RUN_CONTROL_COMMAND_KEYS = new Set(["queue", "steer"]);
29+
2830
type TelegramSequentialKeyContext = {
2931
chat?: { id?: number };
3032
me?: UserFromGetMe;
@@ -80,6 +82,50 @@ function isTelegramTargetedStopCommand(rawText?: string, botUsername?: string):
8082
return match[1]?.toLowerCase() === normalizedBotUsername;
8183
}
8284

85+
function resolveTelegramCommandAliasForControlLane(
86+
rawText?: string,
87+
botUsername?: string,
88+
): string | undefined {
89+
const trimmed = rawText?.trim();
90+
if (!trimmed?.startsWith("/")) {
91+
return undefined;
92+
}
93+
94+
const targetedMatch = trimmed.match(
95+
/^\/([A-Za-z0-9_-]+)(?:@([A-Za-z0-9_]+))?(?:$|\s|[.!?…,，。;；:：'"’”)\]}])/iu,
96+
);
97+
const targetBotUsername = targetedMatch?.[2]?.trim().toLowerCase();
98+
const normalizedBotUsername = botUsername?.trim().toLowerCase();
99+
if (targetBotUsername && normalizedBotUsername && targetBotUsername !== normalizedBotUsername) {
100+
return undefined;
101+
}
102+
103+
if (targetBotUsername && !normalizedBotUsername) {
104+
const commandAlias = `/${targetedMatch?.[1]?.toLowerCase() ?? ""}`;
105+
return commandAlias === "/" ? undefined : commandAlias;
106+
}
107+
108+
return (
109+
maybeResolveTextAlias(
110+
normalizeCommandBody(trimmed, botUsername ? { botUsername } : undefined),
111+
) ?? undefined
112+
);
113+
}
114+
115+
function isTelegramActiveRunControlLaneText(params: {
116+
rawText?: string;
117+
botUsername?: string;
118+
}): boolean {
119+
const alias = resolveTelegramCommandAliasForControlLane(params.rawText, params.botUsername);
120+
if (!alias) {
121+
return false;
122+
}
123+
const command = listChatCommands().find((entry) =>
124+
entry.textAliases.some((candidate) => candidate.trim().toLowerCase() === alias),
125+
);
126+
return command ? TELEGRAM_ACTIVE_RUN_CONTROL_COMMAND_KEYS.has(command.key) : false;
127+
}
128+
83129
export function isTelegramControlLaneText(params: {
84130
rawText?: string;
85131
botUsername?: string;
@@ -95,6 +141,9 @@ export function isTelegramControlLaneText(params: {
95141
if (isTelegramTargetedStopCommand(params.rawText, params.botUsername)) {
96142
return true;
97143
}
144+
if (isTelegramActiveRunControlLaneText(params)) {
145+
return true;
146+
}
98147
return isTelegramReadOnlyControlLaneText(params);
99148
}
100149

‎src/auto-reply/reply/commands-steer.runtime.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,4 +5,5 @@ export {
55
queueEmbeddedAgentMessage,
66
queueEmbeddedAgentMessageWithOutcomeAsync,
77
resolveActiveEmbeddedRunSessionId,
8+
resolveActiveEmbeddedRunSessionIdBySessionFile,
89
} from "../../agents/embedded-agent-runner/runs.js";

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

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ const steerRuntimeMocks = vi.hoisted(() => ({
88
isEmbeddedAgentRunActive: vi.fn(),
99
queueEmbeddedAgentMessageWithOutcomeAsync: vi.fn(),
1010
resolveActiveEmbeddedRunSessionId: vi.fn(),
11+
resolveActiveEmbeddedRunSessionIdBySessionFile: vi.fn(),
1112
}));
1213

1314
vi.mock("./commands-steer.runtime.js", () => steerRuntimeMocks);
@@ -38,6 +39,9 @@ describe("handleSteerCommand", () => {
3839
gatewayHealth: "live",
3940
});
4041
steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockReset().mockReturnValue(undefined);
42+
steerRuntimeMocks.resolveActiveEmbeddedRunSessionIdBySessionFile
43+
.mockReset()
44+
.mockReturnValue(undefined);
4145
});
4246

4347
it("queues steering for the active current text-command session", async () => {
@@ -107,6 +111,72 @@ describe("handleSteerCommand", () => {
107111
);
108112
});
109113

114+
it("resolves an active run from the target session file before stored session id fallback", async () => {
115+
steerRuntimeMocks.resolveActiveEmbeddedRunSessionIdBySessionFile.mockReturnValue(
116+
"session-file-active",
117+
);
118+
119+
const params = buildParams("/steer check the active file");
120+
params.ctx.CommandSource = "native";
121+
params.ctx.CommandTargetSessionKey = "agent:main:telegram:topic:5907";
122+
params.sessionKey = "agent:main:telegram:control";
123+
params.sessionStore = {
124+
"agent:main:telegram:topic:5907": {
125+
sessionId: "stored-session-id",
126+
sessionFile: "/tmp/openclaw-topic-5907.jsonl",
127+
updatedAt: Date.now(),
128+
},
129+
};
130+
131+
await handleSteerCommand(params, true);
132+
133+
expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenCalledWith(
134+
"agent:main:telegram:topic:5907",
135+
);
136+
expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionIdBySessionFile).toHaveBeenCalledWith(
137+
"/tmp/openclaw-topic-5907.jsonl",
138+
);
139+
expect(steerRuntimeMocks.isEmbeddedAgentRunActive).not.toHaveBeenCalledWith(
140+
"stored-session-id",
141+
);
142+
expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
143+
"session-file-active",
144+
"check the active file",
145+
{
146+
steeringMode: "all",
147+
debounceMs: 0,
148+
},
149+
);
150+
});
151+
152+
it("falls back from a slash-lane command session to an active direct sibling", async () => {
153+
steerRuntimeMocks.resolveActiveEmbeddedRunSessionId.mockImplementation((key: string) =>
154+
key === "agent:main:telegram:direct:123" ? "session-direct-active" : undefined,
155+
);
156+
157+
const params = buildParams("/steer use the active direct lane");
158+
params.sessionKey = "agent:main:telegram:slash:123";
159+
160+
await handleSteerCommand(params, true);
161+
162+
expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenNthCalledWith(
163+
1,
164+
"agent:main:telegram:slash:123",
165+
);
166+
expect(steerRuntimeMocks.resolveActiveEmbeddedRunSessionId).toHaveBeenNthCalledWith(
167+
2,
168+
"agent:main:telegram:direct:123",
169+
);
170+
expect(steerRuntimeMocks.queueEmbeddedAgentMessageWithOutcomeAsync).toHaveBeenCalledWith(
171+
"session-direct-active",
172+
"use the active direct lane",
173+
{
174+
steeringMode: "all",
175+
debounceMs: 0,
176+
},
177+
);
178+
});
179+
110180
it("returns usage for an empty steer command", async () => {
111181
const result = await handleSteerCommand(buildParams("/steer"), true);
112182

0 commit comments

Comments
 (0)