Skip to content

Commit 3ad7777

Browse files
vincentkocvincentkoc
authored
fix: carry 2026.7.1 stability repairs into main (#101043)
* fix(telegram): throttle reconnect delivery drains (cherry picked from commit 7182c74) * fix(agents): honor aborts during session lock acquisition (cherry picked from commit 6443c2a) * fix(cli): prefer installed launcher package roots (cherry picked from commit 37e1b32) * fix(e2e): allow Linux onboarding to finish (cherry picked from commit 8819255) * fix(e2e): scrub single-plugin allowlists on Windows (cherry picked from commit bee5dee) * fix(e2e): preserve Windows config shape during updates (cherry picked from commit a580a7f)
1 parent a34b8a8 commit 3ad7777

13 files changed

Lines changed: 268 additions & 61 deletions

‎extensions/telegram/src/polling-session.test.ts‎

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1329,7 +1329,11 @@ describe("TelegramPollingSession", () => {
13291329
};
13301330
createTelegramBotMock.mockReturnValueOnce(bot);
13311331
let onMessage:
1332-
| ((message: { type: "poll-success"; finishedAt: number; count: number }) => void)
1332+
| ((
1333+
message:
1334+
| { type: "poll-success"; finishedAt: number; count: number }
1335+
| { type: "poll-error"; finishedAt: number; message: string },
1336+
) => void)
13331337
| undefined;
13341338
let stopWorker: (() => void) | undefined;
13351339
const workerDone = new Promise<void>((resolve) => {
@@ -1359,9 +1363,21 @@ describe("TelegramPollingSession", () => {
13591363

13601364
const runPromise = session.runUntilAbort();
13611365
await vi.waitFor(() => expect(init).toHaveBeenCalledTimes(1));
1362-
onMessage?.({ type: "poll-success", finishedAt: Date.now(), count: 0 });
1366+
onMessage?.({ type: "poll-success", finishedAt: 10_000, count: 0 });
1367+
onMessage?.({ type: "poll-success", finishedAt: 10_001, count: 0 });
13631368

13641369
await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(1));
1370+
await new Promise<void>((resolve) => {
1371+
setTimeout(resolve, 0);
1372+
});
1373+
onMessage?.({ type: "poll-success", finishedAt: 15_000, count: 0 });
1374+
await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(2));
1375+
onMessage?.({ type: "poll-error", finishedAt: 15_001, message: "offline" });
1376+
await new Promise<void>((resolve) => {
1377+
setTimeout(resolve, 0);
1378+
});
1379+
onMessage?.({ type: "poll-success", finishedAt: 15_002, count: 0 });
1380+
await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(3));
13651381

13661382
abort.abort();
13671383
await runPromise;
@@ -4666,7 +4682,7 @@ describe("TelegramPollingSession", () => {
46664682
await runPromise;
46674683
});
46684684

4669-
it("drains Telegram delivery queue after each getUpdates success", async () => {
4685+
it("throttles healthy delivery drains and re-arms after polling errors", async () => {
46704686
const abort = new AbortController();
46714687
const botStop = vi.fn(async () => undefined);
46724688
const runnerStop = vi.fn(async () => undefined);
@@ -4690,6 +4706,27 @@ describe("TelegramPollingSession", () => {
46904706
{ offset: 2 },
46914707
);
46924708

4709+
await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(1));
4710+
await apiMiddleware(
4711+
vi.fn(async () => []),
4712+
"getUpdates",
4713+
{ offset: 3 },
4714+
);
4715+
await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(1));
4716+
await expect(
4717+
apiMiddleware(
4718+
vi.fn(async () => {
4719+
throw new Error("offline");
4720+
}),
4721+
"getUpdates",
4722+
{ offset: 4 },
4723+
),
4724+
).rejects.toThrow("offline");
4725+
await apiMiddleware(
4726+
vi.fn(async () => []),
4727+
"getUpdates",
4728+
{ offset: 5 },
4729+
);
46934730
await vi.waitFor(() => expect(drainPendingDeliveriesMock).toHaveBeenCalledTimes(2));
46944731

46954732
abort.abort();

‎extensions/telegram/src/polling-session.ts‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,7 @@ const TELEGRAM_GET_UPDATES_CONFLICT_HINT =
128128

129129
const DEFAULT_POLL_STALL_THRESHOLD_MS = 120_000;
130130
const MIN_POLL_STALL_THRESHOLD_MS = 30_000;
131+
const TELEGRAM_DELIVERY_DRAIN_INTERVAL_MS = 5_000;
131132
const MAX_POLL_STALL_THRESHOLD_MS = 600_000;
132133
const POLL_WATCHDOG_INTERVAL_MS = 30_000;
133134
const POLL_STOP_GRACE_MS = 15_000;
@@ -333,6 +334,7 @@ export class TelegramPollingSession {
333334
#spooledUpdateHandlerTimeoutMs: number;
334335
#spooledUpdateHandlerAbortGraceMs: number;
335336
#deliveryDrainInFlight = false;
337+
#nextDeliveryDrainAt = 0;
336338

337339
constructor(private readonly opts: TelegramPollingSessionOpts) {
338340
this.#transportState = new TelegramPollingTransportState({
@@ -473,6 +475,20 @@ export class TelegramPollingSession {
473475
});
474476
}
475477

478+
#maybeDrainPendingDeliveries(finishedAt: number) {
479+
if (finishedAt < this.#nextDeliveryDrainAt) {
480+
return;
481+
}
482+
// Match the queue's first retry window. This keeps healthy polling useful
483+
// as a recovery driver without reopening the drain on every long poll.
484+
this.#nextDeliveryDrainAt = finishedAt + TELEGRAM_DELIVERY_DRAIN_INTERVAL_MS;
485+
this.#drainPendingDeliveriesAfterReconnect();
486+
}
487+
488+
#rearmPendingDeliveryDrain() {
489+
this.#nextDeliveryDrainAt = 0;
490+
}
491+
476492
async #createPollingBot(): Promise<TelegramBot | undefined> {
477493
const fetchAbortController = new AbortController();
478494
this.#activeFetchAbort = fetchAbortController;
@@ -1203,11 +1219,12 @@ export class TelegramPollingSession {
12031219
if (!restartRequested && stalledBacklogKeys.size === 0) {
12041220
this.#status.notePollSuccess(message.finishedAt);
12051221
}
1206-
this.#drainPendingDeliveriesAfterReconnect();
1222+
this.#maybeDrainPendingDeliveries(message.finishedAt);
12071223
pollState.outcome = `ok:${message.count}`;
12081224
return;
12091225
}
12101226
if (message.type === "poll-error") {
1227+
this.#rearmPendingDeliveryDrain();
12111228
liveness.noteGetUpdatesError(new Error(message.message), message.finishedAt);
12121229
liveness.noteGetUpdatesFinished();
12131230
pollState.outcome = "error";
@@ -1455,7 +1472,7 @@ export class TelegramPollingSession {
14551472
onPollSuccess: (finishedAt) => {
14561473
this.#noteHealthyPollingCycle();
14571474
this.#status.notePollSuccess(finishedAt);
1458-
this.#drainPendingDeliveriesAfterReconnect();
1475+
this.#maybeDrainPendingDeliveries(finishedAt);
14591476
},
14601477
});
14611478
bot.api.config.use(async (prev, method, payload, signal) => {
@@ -1469,6 +1486,7 @@ export class TelegramPollingSession {
14691486
liveness.noteGetUpdatesSuccess(result);
14701487
return result;
14711488
} catch (err) {
1489+
this.#rearmPendingDeliveryDrain();
14721490
liveness.noteGetUpdatesError(err);
14731491
throw err;
14741492
} finally {

‎scripts/e2e/parallels/linux-smoke.ts‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -331,7 +331,7 @@ class LinuxSmoke extends SmokeRunController<LinuxOptions> {
331331
);
332332
this.status.freshVersion = await this.extractLastVersion("fresh.install-main");
333333
await this.phase("fresh.verify-main-version", 90, () => this.verifyTargetVersion());
334-
await this.phase("fresh.onboard-ref", 180, () => this.runRefOnboard());
334+
await this.phase("fresh.onboard-ref", 420, () => this.runRefOnboard());
335335
await this.phase("fresh.inject-bad-plugin", 90, () =>
336336
this.maybeInjectBadPluginFixture("fresh"),
337337
);
@@ -366,7 +366,7 @@ class LinuxSmoke extends SmokeRunController<LinuxOptions> {
366366
await this.phase("upgrade.inject-bad-plugin", 90, () =>
367367
this.maybeInjectBadPluginFixture("upgrade"),
368368
);
369-
await this.phase("upgrade.onboard-ref", 180, () => this.runRefOnboard());
369+
await this.phase("upgrade.onboard-ref", 420, () => this.runRefOnboard());
370370
await this.phase("upgrade.gateway-start", 240, () => this.startGatewayBackground());
371371
await this.phase("upgrade.bad-plugin-diagnostic", 90, () =>
372372
this.maybeVerifyBadPluginDiagnostic("upgrade"),
@@ -514,9 +514,7 @@ if command -v curl >/dev/null 2>&1; then
514514
url,
515515
)} -o ${shellQuote(outputPath)}
516516
else
517-
wget -q --timeout=10 --read-timeout=120 --tries=3 -O ${shellQuote(outputPath)} ${shellQuote(
518-
url,
519-
)}
517+
wget -q --timeout=10 --read-timeout=120 --tries=3 -O ${shellQuote(outputPath)} ${shellQuote(url)}
520518
fi`);
521519
}
522520

‎scripts/e2e/parallels/npm-update-scripts.ts‎

Lines changed: 38 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -275,51 +275,47 @@ ${windowsScopedEnvFunction}
275275
function Remove-FuturePluginEntries {
276276
$configPath = Join-Path $env:USERPROFILE '.openclaw\\openclaw.json'
277277
if (-not (Test-Path $configPath)) { return }
278-
try { $config = Get-Content $configPath -Raw | ConvertFrom-Json } catch { return }
279-
$plugins = Get-OpenClawJsonProperty $config 'plugins'
280-
if ($null -eq $plugins) { return }
281-
$entries = Get-OpenClawJsonProperty $plugins 'entries'
282-
if ($null -ne $entries) {
283-
foreach ($pluginId in @('feishu', 'whatsapp', 'openai')) {
284-
Remove-OpenClawJsonProperty $entries $pluginId
278+
$nodeScript = @'
279+
const fs = require("node:fs");
280+
const configPath = process.argv[2];
281+
let config;
282+
try {
283+
config = JSON.parse(fs.readFileSync(configPath, "utf8").replace(/^\\uFEFF/u, ""));
284+
} catch {
285+
process.exit(0);
286+
}
287+
const plugins = config.plugins;
288+
if (!plugins || typeof plugins !== "object") {
289+
process.exit(0);
290+
}
291+
const futurePluginIds = new Set(["feishu", "whatsapp", "openai"]);
292+
let changed = false;
293+
if (plugins.entries && typeof plugins.entries === "object") {
294+
for (const pluginId of futurePluginIds) {
295+
if (Object.hasOwn(plugins.entries, pluginId)) {
296+
delete plugins.entries[pluginId];
297+
changed = true;
285298
}
286299
}
287-
$allow = Get-OpenClawJsonProperty $plugins 'allow'
288-
if ($allow -is [array]) {
289-
Set-OpenClawJsonProperty $plugins 'allow' @($allow | Where-Object { $_ -notin @('feishu', 'whatsapp', 'openai') })
290-
}
291-
$config | ConvertTo-Json -Depth 100 | Set-Content -Path $configPath -Encoding UTF8
292-
}
293-
function Get-OpenClawJsonProperty {
294-
param([object]$Object, [string]$Name)
295-
if ($null -eq $Object) { return $null }
296-
if ($Object -is [System.Collections.IDictionary]) { return $Object[$Name] }
297-
$property = $Object.PSObject.Properties[$Name]
298-
if ($null -eq $property) { return $null }
299-
return $property.Value
300-
}
301-
function Set-OpenClawJsonProperty {
302-
param([object]$Object, [string]$Name, [object]$Value)
303-
if ($Object -is [System.Collections.IDictionary]) {
304-
$Object[$Name] = $Value
305-
return
306-
}
307-
$property = $Object.PSObject.Properties[$Name]
308-
if ($null -ne $property) {
309-
$property.Value = $Value
310-
return
311-
}
312-
$Object | Add-Member -NotePropertyName $Name -NotePropertyValue $Value
313-
}
314-
function Remove-OpenClawJsonProperty {
315-
param([object]$Object, [string]$Name)
316-
if ($null -eq $Object) { return }
317-
if ($Object -is [System.Collections.IDictionary]) {
318-
if ($Object.Contains($Name)) { $Object.Remove($Name) }
319-
return
300+
}
301+
if (Array.isArray(plugins.allow)) {
302+
const allow = plugins.allow.filter((pluginId) => !futurePluginIds.has(pluginId));
303+
if (allow.length !== plugins.allow.length) {
304+
plugins.allow = allow;
305+
changed = true;
320306
}
321-
if ($null -ne $Object.PSObject.Properties[$Name]) {
322-
$Object.PSObject.Properties.Remove($Name)
307+
}
308+
if (changed) {
309+
fs.writeFileSync(configPath, JSON.stringify(config, null, 2) + "\\n");
310+
}
311+
'@
312+
$nodeScriptPath = Join-Path ([System.IO.Path]::GetTempPath()) ('openclaw-future-plugin-scrub-' + [guid]::NewGuid().ToString('N') + '.cjs')
313+
try {
314+
$nodeScript | Set-Content -Path $nodeScriptPath -Encoding UTF8
315+
& node.exe $nodeScriptPath $configPath
316+
if ($LASTEXITCODE -ne 0) { throw "failed to scrub future plugin entries" }
317+
} finally {
318+
Remove-Item $nodeScriptPath -Force -ErrorAction SilentlyContinue
323319
}
324320
}
325321
function Stop-OpenClawGatewayProcesses {

‎src/agents/embedded-agent-runner/run/attempt.abort-race.test.ts‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
2+
import {
3+
cleanupTempPaths,
4+
createContextEngineAttemptRunner,
5+
createContextEngineBootstrapAndAssemble,
6+
getHoisted,
7+
preloadRunEmbeddedAttemptForTests,
8+
resetEmbeddedAttemptHarness,
9+
} from "./attempt.spawn-workspace.test-support.js";
10+
11+
const hoisted = getHoisted();
12+
const tempPaths: string[] = [];
13+
14+
describe("runEmbeddedAttempt abort races", () => {
15+
beforeAll(async () => {
16+
await preloadRunEmbeddedAttemptForTests();
17+
});
18+
19+
beforeEach(() => {
20+
resetEmbeddedAttemptHarness();
21+
});
22+
23+
afterEach(async () => {
24+
await cleanupTempPaths(tempPaths);
25+
tempPaths.length = 0;
26+
});
27+
28+
it("stops before session creation when aborted during eager lock acquisition", async () => {
29+
const abortController = new AbortController();
30+
const prompt = vi.fn(async () => {});
31+
const abortError = new Error("stopped during lock acquisition");
32+
abortError.name = "AbortError";
33+
let markLockRequested!: () => void;
34+
let observedSignal: AbortSignal | undefined;
35+
const lockRequested = new Promise<void>((resolve) => {
36+
markLockRequested = resolve;
37+
});
38+
hoisted.acquireSessionWriteLockMock.mockImplementationOnce(async (params) => {
39+
observedSignal = params.signal;
40+
markLockRequested();
41+
await new Promise<void>((resolve) => {
42+
params.signal?.addEventListener("abort", () => resolve(), { once: true });
43+
});
44+
throw params.signal?.reason;
45+
});
46+
47+
const attempt = createContextEngineAttemptRunner({
48+
contextEngine: createContextEngineBootstrapAndAssemble(),
49+
sessionKey: "agent:main:telegram:direct:123",
50+
tempPaths,
51+
sessionPrompt: prompt,
52+
attemptOverrides: {
53+
abortSignal: abortController.signal,
54+
},
55+
});
56+
await lockRequested;
57+
abortController.abort(abortError);
58+
59+
await expect(attempt).rejects.toBe(abortError);
60+
61+
expect(hoisted.createAgentSessionMock).not.toHaveBeenCalled();
62+
expect(prompt).not.toHaveBeenCalled();
63+
expect(observedSignal).toBe(abortController.signal);
64+
});
65+
});

‎src/agents/embedded-agent-runner/run/attempt.session-lock.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1176,21 +1176,23 @@ export type EmbeddedAttemptSessionLockController = {
11761176

11771177
export async function createEmbeddedAttemptSessionLockController(params: {
11781178
acquireSessionWriteLock: AcquireSessionWriteLock;
1179+
initialAcquireSignal?: AbortSignal;
11791180
lockOptions: LockOptions;
11801181
mergePromptReleasedSessionEntries?: (
11811182
entries: readonly PromptReleasedSessionEntry[],
11821183
) => Promise<PromptReleasedSessionMergeResult | void> | PromptReleasedSessionMergeResult | void;
11831184
reloadPromptReleasedSessionFile?: () => Promise<void> | void;
11841185
}): Promise<EmbeddedAttemptSessionLockController> {
1185-
const acquireLock = async (): Promise<SessionLock> =>
1186+
const acquireLock = async (signal?: AbortSignal): Promise<SessionLock> =>
11861187
await params.acquireSessionWriteLock({
11871188
sessionFile: params.lockOptions.sessionFile,
11881189
timeoutMs: params.lockOptions.timeoutMs,
11891190
staleMs: params.lockOptions.staleMs,
11901191
maxHoldMs: params.lockOptions.maxHoldMs,
1192+
...(signal ? { signal } : {}),
11911193
});
11921194

1193-
let heldLock: SessionLock | undefined = await acquireLock();
1195+
let heldLock: SessionLock | undefined = await acquireLock(params.initialAcquireSignal);
11941196
const activeWriteLock = new AsyncLocalStorage<ActiveWriteLockState>();
11951197
let ownedPublicationQueue: Promise<void> = Promise.resolve();
11961198
let fenceFingerprint: SessionFileFingerprint | undefined;

‎src/agents/embedded-agent-runner/run/attempt.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2088,6 +2088,7 @@ export async function runEmbeddedAttempt(
20882088
let sessionManager: ReturnType<typeof guardSessionManager> | undefined;
20892089
const sessionLockController = await createEmbeddedAttemptSessionLockController({
20902090
acquireSessionWriteLock,
2091+
initialAcquireSignal: params.abortSignal,
20912092
lockOptions: {
20922093
sessionFile: params.sessionFile,
20932094
...sessionWriteLockOptions,
@@ -2123,6 +2124,9 @@ export async function runEmbeddedAttempt(
21232124
sessionLockController.withSessionWriteLock(operation),
21242125
);
21252126
armExternalAbortSignal();
2127+
// The signal can fire while the eager session lock is being acquired.
2128+
// Recheck after arming so a stopped run never reaches session creation or provider prompt.
2129+
await throwIfAttemptAbortSignalFiredAfterPrepCleanup();
21262130

21272131
let session: Awaited<ReturnType<typeof createAgentSession>>["session"] | undefined;
21282132
let removeToolResultContextGuard: (() => void) | undefined;

‎src/agents/session-write-lock.test.ts‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -244,6 +244,29 @@ describe("acquireSessionWriteLock", () => {
244244
});
245245
});
246246

247+
it("cancels a contended infinite acquisition without leaving a lock waiter", async () => {
248+
await withTempSessionLockFile(async ({ sessionFile }) => {
249+
const heldLock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
250+
const abortController = new AbortController();
251+
const abortError = new Error("stop requested");
252+
abortError.name = "AbortError";
253+
const pendingLock = acquireSessionWriteLock({
254+
sessionFile,
255+
timeoutMs: Number.POSITIVE_INFINITY,
256+
signal: abortController.signal,
257+
});
258+
await new Promise<void>((resolve) => {
259+
setTimeout(resolve, 20);
260+
});
261+
abortController.abort(abortError);
262+
263+
await expect(pendingLock).rejects.toBe(abortError);
264+
await heldLock.release();
265+
const nextLock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
266+
await nextLock.release();
267+
});
268+
});
269+
247270
it("does not reenter locks by default through symlinked session paths", async () => {
248271
await withSymlinkedSessionPaths(async ({ sessionReal, sessionLink }) => {
249272
const lock = await acquireSessionWriteLock({ sessionFile: sessionReal, timeoutMs: 500 });

0 commit comments

Comments
 (0)