Skip to content

Commit b1da734

Browse files
obviyusobviyus
committed
fix(logging): recover orphaned stalled lanes
Stalled sessions could leave a session lane occupied by an orphaned active task after the embedded run handle was gone. Diagnostics repeatedly reported the stuck state, but the lane stayed held and queued turns could remain blocked indefinitely. Release ownerless active lane tasks after the stale threshold, while guarding against fresh lane work and raised compaction safety windows so recovery does not double-run serialized session work or overlap transcript compaction. Stalled model-call recovery also no longer requires an in-process embedded run handle, so no-handle stalls can be reclaimed by the existing conservative recovery path. Refs #99847. Refs #94650.
1 parent ec28935 commit b1da734

5 files changed

Lines changed: 157 additions & 4 deletions

‎src/logging/diagnostic-session-recovery.ts‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,11 @@ export type StuckSessionRecoveryRequest = {
2929
* reclaimed. Honors an operator-raised threshold; falls back to a safe floor.
3030
*/
3131
staleActiveProgressAbortMs?: number;
32+
/**
33+
* Resolved compaction safety timeout. Ownerless lane recovery waits at least
34+
* this long plus settle grace so queued compaction cannot be double-run.
35+
*/
36+
compactionSafetyTimeoutMs?: number;
3237
};
3338

3439
export function resolveStuckSessionRecoveryRef(
@@ -60,7 +65,9 @@ export type StuckSessionRecoveryOutcome =
6065
| (DiagnosticSessionRecoveryBaseOutcome & {
6166
status: "released";
6267
action: "release_lane";
68+
reason?: "stale_lane_task";
6369
released: number;
70+
queuedCount?: number;
6471
})
6572
| (DiagnosticSessionRecoveryBaseOutcome & {
6673
status: "skipped";
@@ -137,7 +144,10 @@ export function formatRecoveryOutcome(outcome: StuckSessionRecoveryOutcome): str
137144
if ("released" in outcome) {
138145
fields.push(`released=${outcome.released}`);
139146
}
140-
if (outcome.status === "aborted" && outcome.queuedCount !== undefined) {
147+
if (
148+
(outcome.status === "aborted" || outcome.status === "released") &&
149+
outcome.queuedCount !== undefined
150+
) {
141151
fields.push(`queuedCount=${outcome.queuedCount}`);
142152
}
143153
if ("activeCount" in outcome && outcome.activeCount !== undefined) {

‎src/logging/diagnostic-stuck-session-recovery.runtime.test.ts‎

Lines changed: 103 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -547,6 +547,109 @@ describe("stuck session recovery", () => {
547547
]);
548548
});
549549

550+
it("releases stale unregistered lane work after the diagnostic abort floor", async () => {
551+
mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined);
552+
mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined);
553+
mocks.isEmbeddedAgentRunActive.mockReturnValue(false);
554+
mocks.isEmbeddedAgentRunHandleActive.mockReturnValue(false);
555+
mocks.getCommandLaneSnapshot.mockReturnValue({
556+
lane: "session:agent:main:main",
557+
queuedCount: 1,
558+
activeCount: 1,
559+
maxConcurrent: 1,
560+
draining: false,
561+
generation: 0,
562+
});
563+
mocks.resetCommandLane.mockReturnValue(1);
564+
565+
await recoverStuckDiagnosticSession({
566+
sessionId: "unregistered-work-session",
567+
sessionKey: "agent:main:main",
568+
ageMs: 300_000,
569+
queueDepth: 0,
570+
});
571+
572+
expect(mocks.abortEmbeddedAgentRun).not.toHaveBeenCalled();
573+
expect(mocks.forceClearEmbeddedAgentRun).not.toHaveBeenCalled();
574+
expect(mocks.resetCommandLane).toHaveBeenCalledWith("session:agent:main:main");
575+
expect(warnLogMessages()).toEqual([
576+
"stuck session recovery outcome: status=released action=release_lane sessionId=unregistered-work-session sessionKey=agent:main:main lane=session:agent:main:main reason=stale_lane_task released=1 queuedCount=1",
577+
]);
578+
});
579+
580+
it("does not release stale unregistered lane work when a fresh task appeared", async () => {
581+
mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined);
582+
mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined);
583+
mocks.isEmbeddedAgentRunActive.mockReturnValue(false);
584+
mocks.isEmbeddedAgentRunHandleActive.mockReturnValue(false);
585+
mocks.getCommandLaneSnapshot.mockReturnValue({
586+
lane: "session:agent:main:main",
587+
queuedCount: 1,
588+
activeCount: 1,
589+
maxConcurrent: 1,
590+
draining: false,
591+
generation: 0,
592+
});
593+
mocks.getCommandLaneActiveTaskIds.mockReturnValueOnce([101]).mockReturnValueOnce([202]);
594+
595+
await recoverStuckDiagnosticSession({
596+
sessionId: "unregistered-work-session",
597+
sessionKey: "agent:main:main",
598+
ageMs: 720_000,
599+
queueDepth: 0,
600+
});
601+
602+
expect(mocks.resetCommandLane).not.toHaveBeenCalled();
603+
expect(warnLogMessages()).toEqual([
604+
"stuck session recovery outcome: status=skipped action=keep_lane sessionId=unregistered-work-session sessionKey=agent:main:main lane=session:agent:main:main reason=active_lane_task laneActive=1 laneQueued=1",
605+
]);
606+
});
607+
608+
it("waits for the compaction safety window before releasing unregistered lane work", async () => {
609+
mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined);
610+
mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined);
611+
mocks.isEmbeddedAgentRunActive.mockReturnValue(false);
612+
mocks.isEmbeddedAgentRunHandleActive.mockReturnValue(false);
613+
mocks.getCommandLaneSnapshot.mockReturnValue({
614+
lane: "session:agent:main:main",
615+
queuedCount: 1,
616+
activeCount: 1,
617+
maxConcurrent: 1,
618+
draining: false,
619+
generation: 0,
620+
});
621+
622+
await recoverStuckDiagnosticSession({
623+
sessionId: "unregistered-work-session",
624+
sessionKey: "agent:main:main",
625+
ageMs: 300_000,
626+
queueDepth: 0,
627+
compactionSafetyTimeoutMs: 600_000,
628+
});
629+
630+
expect(mocks.resetCommandLane).not.toHaveBeenCalled();
631+
expect(warnLogMessages()).toEqual([
632+
"stuck session recovery outcome: status=skipped action=keep_lane sessionId=unregistered-work-session sessionKey=agent:main:main lane=session:agent:main:main reason=active_lane_task laneActive=1 laneQueued=1",
633+
]);
634+
635+
mocks.diag.warn.mockClear();
636+
mocks.resetCommandLane.mockClear();
637+
mocks.resetCommandLane.mockReturnValue(1);
638+
639+
await recoverStuckDiagnosticSession({
640+
sessionId: "unregistered-work-session",
641+
sessionKey: "agent:main:main",
642+
ageMs: 615_000,
643+
queueDepth: 0,
644+
compactionSafetyTimeoutMs: 600_000,
645+
});
646+
647+
expect(mocks.resetCommandLane).toHaveBeenCalledWith("session:agent:main:main");
648+
expect(warnLogMessages()).toEqual([
649+
"stuck session recovery outcome: status=released action=release_lane sessionId=unregistered-work-session sessionKey=agent:main:main lane=session:agent:main:main reason=stale_lane_task released=1 queuedCount=1",
650+
]);
651+
});
652+
550653
it("reports when recovery finds no active work to release", async () => {
551654
mocks.resolveActiveEmbeddedRunHandleSessionId.mockReturnValue(undefined);
552655
mocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(undefined);

‎src/logging/diagnostic-stuck-session-recovery.runtime.ts‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@ import { isDiagnosticSessionStateCurrent } from "./diagnostic-session-state.js";
3131
// Runtime repair path for diagnostic sessions that appear stuck in processing/waiting states.
3232
const STUCK_SESSION_ABORT_SETTLE_MS = 15_000;
3333
const STUCK_SESSION_PROGRESS_STALE_MS = 5 * 60_000;
34+
// Ownerless lane release shares the no-progress abort floor, then extends for
35+
// compaction because queued compaction owns the session lane without a run handle.
36+
const STALE_ACTIVE_LANE_TASK_RELEASE_MS = STUCK_SESSION_PROGRESS_STALE_MS;
3437
const recoveriesInFlight = new Set<string>();
3538

3639
/** Request parameters accepted by the stuck-session recovery runtime. */
@@ -43,6 +46,15 @@ function resolveStaleActiveProgressAbortMs(params: StuckSessionRecoveryParams):
4346
: STUCK_SESSION_PROGRESS_STALE_MS;
4447
}
4548

49+
function resolveStaleActiveLaneTaskReleaseMs(params: StuckSessionRecoveryParams): number {
50+
const compactionSafetyTimeoutMs = params.compactionSafetyTimeoutMs;
51+
const compactionReleaseMs =
52+
typeof compactionSafetyTimeoutMs === "number" && compactionSafetyTimeoutMs > 0
53+
? compactionSafetyTimeoutMs + STUCK_SESSION_ABORT_SETTLE_MS
54+
: 0;
55+
return Math.max(STALE_ACTIVE_LANE_TASK_RELEASE_MS, compactionReleaseMs);
56+
}
57+
4658
function isActiveRunProgressStale(params: {
4759
sessionId?: string;
4860
sessionKey?: string;
@@ -146,6 +158,7 @@ export async function recoverStuckDiagnosticSession(
146158
let drained = true;
147159
let forceCleared = false;
148160
const staleActiveProgressAbortMs = resolveStaleActiveProgressAbortMs(params);
161+
const staleActiveLaneTaskReleaseMs = resolveStaleActiveLaneTaskReleaseMs(params);
149162

150163
if (activeSessionId) {
151164
const reclaimStaleActiveRun =
@@ -237,6 +250,26 @@ export async function recoverStuckDiagnosticSession(
237250
if (!activeSessionId && sessionLane) {
238251
const laneSnapshot = getCommandLaneSnapshot(sessionLane);
239252
if (laneSnapshot.activeCount > 0) {
253+
const laneStartedFreshTask = getCommandLaneActiveTaskIds(sessionLane).some(
254+
(id) => !preAbortActiveTaskIds.has(id),
255+
);
256+
// Orphaned active lane tasks have no run handle to abort. Release only
257+
// after the ownerless-lane window and only if no fresh task appeared.
258+
if (!laneStartedFreshTask && params.ageMs >= staleActiveLaneTaskReleaseMs) {
259+
const released = resetCommandLane(sessionLane);
260+
const outcome: StuckSessionRecoveryOutcome = {
261+
status: "released",
262+
action: "release_lane",
263+
reason: "stale_lane_task",
264+
sessionId: params.sessionId,
265+
sessionKey: params.sessionKey,
266+
lane: sessionLane,
267+
released,
268+
queuedCount: laneSnapshot.queuedCount,
269+
};
270+
diag.warn(`stuck session recovery outcome: ${formatRecoveryOutcome(outcome)}`);
271+
return outcome;
272+
}
240273
const outcome: StuckSessionRecoveryOutcome = {
241274
status: "skipped",
242275
action: "keep_lane",

‎src/logging/diagnostic.test.ts‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -988,7 +988,7 @@ describe("stuck session diagnostics threshold", () => {
988988
);
989989
});
990990

991-
it("does not recover stale model calls without active embedded-run ownership", async () => {
991+
it("recovers stale model calls without active embedded-run ownership", async () => {
992992
const events: DiagnosticEventPayload[] = [];
993993
const recoverStuckSession = vi.fn();
994994
const stuckSessionWarnMs = 30_000;
@@ -1033,7 +1033,11 @@ describe("stuck session diagnostics threshold", () => {
10331033
lastProgressReason: "model_call:started",
10341034
},
10351035
);
1036-
expect(recoverStuckSession).not.toHaveBeenCalled();
1036+
expectRecoveryCall(
1037+
recoverStuckSession,
1038+
{ sessionId: "s1", sessionKey: "main", queueDepth: 0, allowActiveAbort: true },
1039+
["ageMs", "stateGeneration"],
1040+
);
10371041
});
10381042

10391043
it("does not recover a recent native tool call just because the session is old", async () => {

‎src/logging/diagnostic.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
// Diagnostic logger records structured runtime events, timings, and health snapshots.
22
import { monitorEventLoopDelay, performance } from "node:perf_hooks";
3+
import { resolveCompactionTimeoutMs } from "../agents/embedded-agent-runner/compaction-safety-timeout.js";
34
import { getRuntimeConfig } from "../config/config.js";
45
import { resolveAllAgentSessionStoreTargetsSync } from "../config/sessions/targets.js";
56
import type { OpenClawConfig } from "../config/types.openclaw.js";
@@ -541,7 +542,6 @@ function isStalledModelCallRecoveryEligible(params: {
541542
params.classification?.eventType === "session.stalled" &&
542543
params.classification.classification === "stalled_agent_run" &&
543544
params.classification.activeWorkKind === "model_call" &&
544-
params.activity?.hasActiveEmbeddedRun === true &&
545545
typeof lastProgressAgeMs === "number" &&
546546
lastProgressAgeMs >= params.stuckSessionAbortMs
547547
);
@@ -1219,6 +1219,7 @@ export function startDiagnosticHeartbeat(
12191219
}
12201220
const stuckSessionWarnMs = resolveStuckSessionWarnMs(heartbeatConfig);
12211221
const stuckSessionAbortMs = resolveStuckSessionAbortMs(heartbeatConfig, stuckSessionWarnMs);
1222+
const compactionSafetyTimeoutMs = resolveCompactionTimeoutMs(heartbeatConfig);
12221223
const now = Date.now();
12231224
pruneDiagnosticSessionStates(now, true);
12241225
const work = getDiagnosticWorkSnapshot(now);
@@ -1315,6 +1316,7 @@ export function startDiagnosticHeartbeat(
13151316
expectedState: state.state,
13161317
stateGeneration: state.generation,
13171318
staleActiveProgressAbortMs: stuckSessionAbortMs,
1319+
compactionSafetyTimeoutMs,
13181320
},
13191321
});
13201322
} else if (
@@ -1337,6 +1339,7 @@ export function startDiagnosticHeartbeat(
13371339
allowActiveAbort: true,
13381340
expectedState: state.state,
13391341
stateGeneration: state.generation,
1342+
compactionSafetyTimeoutMs,
13401343
},
13411344
});
13421345
}

0 commit comments

Comments
 (0)