Skip to content

Commit 4b33199

Browse files
obviyusobviyus
committed
refactor(agents): converge run staleness policy onto diagnostic activity owner
One RUN_STALE_TAKEOVER_MS constant and one resolveRunStaleThresholdMs in diagnostic-run-activity.ts replace the hand-paired constants and duplicated blocked-tool-floor derivations in the reply and embedded run registries. Thresholds are byte-identical before and after; zero behavior change. Part of #104219 (seam 4).
1 parent c3b426f commit 4b33199

8 files changed

Lines changed: 70 additions & 39 deletions

‎src/agents/embedded-agent-runner/runs.ts‎

Lines changed: 2 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,10 @@ import {
1616
waitForReplyRunEndBySessionId,
1717
} from "../../auto-reply/reply/reply-run-registry.js";
1818
import {
19-
BLOCKED_TOOL_CALL_ABORT_FLOOR_MS,
2019
getDiagnosticSessionActivitySnapshot,
2120
markDiagnosticEmbeddedRunEnded,
2221
markDiagnosticEmbeddedRunStarted,
22+
resolveRunStaleThresholdMs,
2323
} from "../../logging/diagnostic-run-activity.js";
2424
import {
2525
diagnosticLogger as diag,
@@ -95,10 +95,6 @@ type PreparedEmbeddedAgentQueueMessage =
9595
handle: EmbeddedAgentQueueHandle;
9696
};
9797

98-
// Paired with REPLY_RUN_STALE_TAKEOVER_MS in the reply registry; src/agents
99-
// keeps its own constant to avoid importing auto-reply policy into this owner.
100-
const EMBEDDED_STEER_STALE_CAPTURE_MS = 10 * 60_000;
101-
10298
function createQueueFailureOutcome(
10399
sessionId: string,
104100
reason: EmbeddedAgentQueueFailureReason,
@@ -490,16 +486,9 @@ function prepareEmbeddedAgentQueueMessage(
490486
return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "not_streaming") };
491487
}
492488
const activity = getDiagnosticSessionActivitySnapshot({ sessionId });
493-
// Quiet tool phases stay steerable until the blocked-tool floor: refusing at
494-
// the shorter window would push the message into admission takeover of a run
495-
// the diagnostic layer still considers healthy.
496-
const steerStaleCaptureMs =
497-
activity.activeWorkKind === "tool_call"
498-
? Math.max(EMBEDDED_STEER_STALE_CAPTURE_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)
499-
: EMBEDDED_STEER_STALE_CAPTURE_MS;
500489
if (
501490
typeof activity.lastProgressAgeMs === "number" &&
502-
activity.lastProgressAgeMs > steerStaleCaptureMs
491+
activity.lastProgressAgeMs > resolveRunStaleThresholdMs(activity)
503492
) {
504493
diag.debug(`queue message failed: sessionId=${sessionId} reason=stale_run`);
505494
return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "stale_run") };

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
22
import type { OpenClawConfig } from "../../config/types.openclaw.js";
3+
import { RUN_STALE_TAKEOVER_MS } from "../../logging/diagnostic-run-activity.js";
34
import type { ReplyPayload } from "../types.js";
45
import {
56
createDispatcher,
@@ -9,7 +10,6 @@ import {
910
resetPluginTtsAndThreadMocks,
1011
runtimePluginMocks,
1112
} from "./dispatch-from-config.shared.test-harness.js";
12-
import { REPLY_RUN_STALE_TAKEOVER_MS } from "./reply-run-registry.js";
1313
import { buildTestCtx } from "./test-ctx.js";
1414

1515
let dispatchReplyFromConfig: typeof import("./dispatch-from-config.js").dispatchReplyFromConfig;
@@ -117,7 +117,7 @@ describe("dispatchReplyFromConfig stale visible admission recovery", () => {
117117
activeOperation.setPhase("running");
118118
const replyResolver = vi.fn(async () => ({ text: "telegram reply" }) satisfies ReplyPayload);
119119
const dispatchParams = createVisibleDispatchParams(replyResolver);
120-
vi.setSystemTime(startedAt + REPLY_RUN_STALE_TAKEOVER_MS + 1);
120+
vi.setSystemTime(startedAt + RUN_STALE_TAKEOVER_MS + 1);
121121

122122
const result = await dispatchReplyFromConfig(dispatchParams);
123123

‎src/auto-reply/reply/reply-run-registry.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { createAgentRunRestartAbortError } from "../../agents/run-termination.js
44
import {
55
getDiagnosticSessionActivitySnapshot,
66
resetDiagnosticRunActivityForTest,
7+
RUN_STALE_TAKEOVER_MS,
78
} from "../../logging/diagnostic-run-activity.js";
89
import { diagnosticLogger } from "../../logging/diagnostic-runtime.js";
910
import { MAX_TIMER_TIMEOUT_MS } from "../../shared/number-coercion.js";
@@ -18,7 +19,6 @@ import {
1819
isReplyRunAbortableForSignal,
1920
queueReplyRunMessage,
2021
REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS,
21-
REPLY_RUN_STALE_TAKEOVER_MS,
2222
REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS,
2323
replyRunRegistry,
2424
runAfterReplyOperationClear,
@@ -901,7 +901,7 @@ describe("reply run registry", () => {
901901
});
902902
operation.setPhase("running");
903903

904-
vi.advanceTimersByTime(REPLY_RUN_STALE_TAKEOVER_MS + 1);
904+
vi.advanceTimersByTime(RUN_STALE_TAKEOVER_MS + 1);
905905

906906
expect(queueReplyRunMessage("session-running", "stale")).toBe(false);
907907
expect(queueMessage).not.toHaveBeenCalled();

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

Lines changed: 5 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,10 @@ import {
66
} from "../../agents/run-termination.js";
77
import { createAbortError } from "../../infra/abort-signal.js";
88
import {
9-
BLOCKED_TOOL_CALL_ABORT_FLOOR_MS,
109
getDiagnosticSessionActivitySnapshot,
1110
markDiagnosticEmbeddedRunEnded,
1211
markDiagnosticEmbeddedRunStarted,
12+
resolveRunStaleThresholdMs,
1313
} from "../../logging/diagnostic-run-activity.js";
1414
import { diagnosticLogger as diag } from "../../logging/diagnostic-runtime.js";
1515
import type { UserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.types.js";
@@ -218,9 +218,6 @@ export const REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS = 15_000;
218218
// Terminal results must release the lane even if the owner never resumes.
219219
// Without this, abort/failure can leave the session wedged until process restart.
220220
export const REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS = 60_000;
221-
// Visible human turns may reclaim only runs with no real progress for this window.
222-
// Timers and user-message injection never refresh activity; agent events do.
223-
export const REPLY_RUN_STALE_TAKEOVER_MS = 10 * 60_000;
224221

225222
export type ReplyOperationStaleReason = "terminal_unreleased" | "no_activity" | "stuck_recovery";
226223

@@ -886,25 +883,16 @@ export function expireStaleReplyRunBySessionId(
886883
return operation ? expireStaleReplyOperation(operation, reason) : false;
887884
}
888885

889-
/**
890-
* Effective staleness window for an operation. Quiet-but-alive tool phases get
891-
* the diagnostic blocked-tool floor: a human message must not reclaim a healthy
892-
* long tool that stuck recovery itself would not touch yet.
893-
*/
894-
export function resolveReplyRunStaleThresholdMs(operation: ReplyOperation): number {
886+
// lastActivityAtMs is refreshed by agent events only; timers and user-message
887+
// injection never refresh it, so quiet runs age toward reclaim.
888+
export function isReplyRunEvidenceStale(operation: ReplyOperation): boolean {
895889
const activity = getDiagnosticSessionActivitySnapshot({
896890
sessionId: operation.sessionId,
897891
sessionKey: operation.key,
898892
});
899-
return activity.activeWorkKind === "tool_call"
900-
? Math.max(REPLY_RUN_STALE_TAKEOVER_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)
901-
: REPLY_RUN_STALE_TAKEOVER_MS;
902-
}
903-
904-
export function isReplyRunEvidenceStale(operation: ReplyOperation): boolean {
905893
return (
906894
!operation.result &&
907-
Date.now() - operation.lastActivityAtMs > resolveReplyRunStaleThresholdMs(operation)
895+
Date.now() - operation.lastActivityAtMs > resolveRunStaleThresholdMs(activity)
908896
);
909897
}
910898

‎src/auto-reply/reply/reply-turn-admission.test.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"
66
import {
77
markDiagnosticToolStartedForTest,
88
resetDiagnosticRunActivityForTest,
9+
RUN_STALE_TAKEOVER_MS,
910
} from "../../logging/diagnostic-run-activity.js";
1011
import {
1112
interruptSessionWorkAdmissions,
@@ -14,7 +15,6 @@ import {
1415
import {
1516
createReplyOperation,
1617
REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS,
17-
REPLY_RUN_STALE_TAKEOVER_MS,
1818
REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS,
1919
replyRunRegistry,
2020
runAfterReplyOperationClear,
@@ -947,7 +947,7 @@ describe("reply turn admission", () => {
947947
isStreaming: () => true,
948948
});
949949
active.setPhase("running");
950-
vi.setSystemTime(startedAt + REPLY_RUN_STALE_TAKEOVER_MS + 1);
950+
vi.setSystemTime(startedAt + RUN_STALE_TAKEOVER_MS + 1);
951951

952952
const result = await admitReplyTurn({
953953
sessionKey: "agent:main:telegram:topic:stale-visible",
@@ -1084,7 +1084,7 @@ describe("reply turn admission", () => {
10841084
isStreaming: () => true,
10851085
});
10861086
active.setPhase("running");
1087-
vi.setSystemTime(startedAt + REPLY_RUN_STALE_TAKEOVER_MS + 1);
1087+
vi.setSystemTime(startedAt + RUN_STALE_TAKEOVER_MS + 1);
10881088

10891089
const admission = admitReplyTurn({
10901090
sessionKey: `agent:main:telegram:topic:stale-${kind}`,

‎src/auto-reply/reply/reply-turn-admission.ts‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,10 @@
22
import { resolveSessionWorkStartError } from "../../config/sessions/lifecycle.js";
33
import { loadSessionEntry } from "../../config/sessions/session-accessor.js";
44
import type { SessionEntry } from "../../config/sessions/types.js";
5+
import {
6+
getDiagnosticSessionActivitySnapshot,
7+
resolveRunStaleThresholdMs,
8+
} from "../../logging/diagnostic-run-activity.js";
59
import {
610
beginSessionWorkAdmission,
711
type SessionWorkAdmissionLease,
@@ -12,7 +16,6 @@ import {
1216
isReplyRunEvidenceStale,
1317
REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS,
1418
REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS,
15-
resolveReplyRunStaleThresholdMs,
1619
replyRunRegistry,
1720
ReplyRunAlreadyActiveError,
1821
ReplyRunFollowupAdmissionBlockedError,
@@ -78,9 +81,13 @@ function resolveVisibleActiveWaitMs(operation: ReplyOperation | undefined): numb
7881
return REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS;
7982
}
8083
const ageMs = Date.now() - operation.lastActivityAtMs;
84+
const activity = getDiagnosticSessionActivitySnapshot({
85+
sessionId: operation.sessionId,
86+
sessionKey: operation.key,
87+
});
8188
const remainingMs = operation.result
8289
? REPLY_RUN_TERMINAL_SETTLE_TIMEOUT_MS - ageMs
83-
: resolveReplyRunStaleThresholdMs(operation) - ageMs;
90+
: resolveRunStaleThresholdMs(activity) - ageMs;
8491
return Math.min(REPLY_RUN_IDLE_SETTLE_TIMEOUT_MS, Math.max(1, remainingMs));
8592
}
8693

‎src/logging/diagnostic-run-activity.test.ts‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
// Unit tests for shared run-staleness threshold policy.
2+
import { describe, expect, it } from "vitest";
3+
import {
4+
BLOCKED_TOOL_CALL_ABORT_FLOOR_MS,
5+
resolveRunStaleThresholdMs,
6+
RUN_STALE_TAKEOVER_MS,
7+
} from "./diagnostic-run-activity.js";
8+
9+
describe("resolveRunStaleThresholdMs", () => {
10+
it.each([
11+
{
12+
name: "default window when no active work",
13+
activity: {},
14+
expected: RUN_STALE_TAKEOVER_MS,
15+
},
16+
{
17+
name: "default window for model_call",
18+
activity: { activeWorkKind: "model_call" as const },
19+
expected: RUN_STALE_TAKEOVER_MS,
20+
},
21+
{
22+
name: "default window for embedded_run",
23+
activity: { activeWorkKind: "embedded_run" as const },
24+
expected: RUN_STALE_TAKEOVER_MS,
25+
},
26+
{
27+
name: "blocked-tool floor for tool_call",
28+
activity: { activeWorkKind: "tool_call" as const },
29+
expected: Math.max(RUN_STALE_TAKEOVER_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS),
30+
},
31+
])("$name", ({ activity, expected }) => {
32+
expect(resolveRunStaleThresholdMs(activity)).toBe(expected);
33+
});
34+
});

‎src/logging/diagnostic-run-activity.ts‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,9 @@ type DiagnosticRunProgressActivityEvent = Pick<
6161
// steer gates): lowering it reopens #88870, removing it reopens #96168.
6262
export const BLOCKED_TOOL_CALL_ABORT_FLOOR_MS = 15 * 60_000;
6363

64+
// Default quiet-run reclaim window for steer/takeover. Evidence clocks stay local.
65+
export const RUN_STALE_TAKEOVER_MS = 10 * 60_000;
66+
6467
export type DiagnosticSessionActivitySnapshot = {
6568
activeWorkKind?: DiagnosticSessionActiveWorkKind;
6669
hasActiveEmbeddedRun?: boolean;
@@ -71,6 +74,16 @@ export type DiagnosticSessionActivitySnapshot = {
7174
lastProgressReason?: string;
7275
};
7376

77+
// Quiet-but-alive tool phases get the blocked-tool floor so a human message
78+
// cannot reclaim a healthy long tool that stuck recovery would not touch yet.
79+
export function resolveRunStaleThresholdMs(
80+
activity: Pick<DiagnosticSessionActivitySnapshot, "activeWorkKind">,
81+
): number {
82+
return activity.activeWorkKind === "tool_call"
83+
? Math.max(RUN_STALE_TAKEOVER_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)
84+
: RUN_STALE_TAKEOVER_MS;
85+
}
86+
7487
const activityByRef = new Map<string, SessionActivity>();
7588
const activityByRunId = new Map<string, SessionActivity>();
7689
let embeddedRunSequence = 0;

0 commit comments

Comments
 (0)