Skip to content

Commit 348c468

Browse files
obviyusobviyus
committed
fix(agents): respect the blocked-tool floor in stale takeover and steer gates
1 parent e1f1ea5 commit 348c468

8 files changed

Lines changed: 179 additions & 82 deletions

File tree

‎TRACK_B1_BRIEF.md‎

Lines changed: 0 additions & 70 deletions
This file was deleted.

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

Lines changed: 58 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,10 @@ import {
1111
isReplyRunActiveForSessionId,
1212
} from "../../auto-reply/reply/reply-run-registry.js";
1313
import { setDiagnosticsEnabledForProcess } from "../../infra/diagnostic-events.js";
14-
import { resetDiagnosticRunActivityForTest } from "../../logging/diagnostic-run-activity.js";
14+
import {
15+
markDiagnosticToolStartedForTest,
16+
resetDiagnosticRunActivityForTest,
17+
} from "../../logging/diagnostic-run-activity.js";
1518
import {
1619
getDiagnosticSessionState,
1720
resetDiagnosticSessionStateForTest,
@@ -541,6 +544,60 @@ describe("embedded-agent runner run registry", () => {
541544
}
542545
});
543546

547+
it("keeps steering into a quiet tool phase until the blocked-tool floor", () => {
548+
vi.useFakeTimers();
549+
try {
550+
const queueMessage = vi.fn(async () => {});
551+
setActiveEmbeddedRun("session-quiet-tool-steer", createRunHandle({ queueMessage }));
552+
markDiagnosticToolStartedForTest({
553+
sessionId: "session-quiet-tool-steer",
554+
toolName: "exec",
555+
toolCallId: "tool-quiet-steer",
556+
});
557+
558+
vi.advanceTimersByTime(12 * 60_000);
559+
expect(
560+
queueEmbeddedAgentMessageWithOutcome("session-quiet-tool-steer", "status?").queued,
561+
).toBe(true);
562+
563+
vi.advanceTimersByTime(4 * 60_000);
564+
const late = queueEmbeddedAgentMessageWithOutcome("session-quiet-tool-steer", "status?");
565+
expect(late).toMatchObject({ queued: false, reason: "stale_run" });
566+
} finally {
567+
vi.useRealTimers();
568+
}
569+
});
570+
571+
it("refuses reply-backed steering with stale registry evidence as stale_run", () => {
572+
vi.useFakeTimers();
573+
try {
574+
const operation = createReplyOperation({
575+
sessionKey: "agent:main:cli-stale-steer",
576+
sessionId: "session-cli-stale-steer",
577+
resetTriggered: false,
578+
});
579+
operation.attachBackend({
580+
kind: "cli",
581+
cancel: () => {},
582+
isStreaming: () => true,
583+
});
584+
operation.setPhase("running");
585+
586+
vi.advanceTimersByTime(10 * 60_000 + 1);
587+
const outcome = queueEmbeddedAgentMessageWithOutcome("session-cli-stale-steer", "hello");
588+
589+
expect(outcome).toEqual({
590+
queued: false,
591+
sessionId: "session-cli-stale-steer",
592+
reason: "stale_run",
593+
gatewayHealth: "live",
594+
});
595+
operation.complete();
596+
} finally {
597+
vi.useRealTimers();
598+
}
599+
});
600+
544601
it("accepts embedded steering with fresh or missing diagnostic evidence", () => {
545602
const freshQueueMessage = vi.fn(async () => {});
546603
setActiveEmbeddedRun(

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

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import {
66
abortReplyRunBySessionId,
77
expireStaleReplyRunBySessionId,
88
forceClearReplyRunBySessionId,
9+
isReplyRunEvidenceStaleBySessionId,
910
isReplyRunActiveForSessionId,
1011
isReplyRunAbortableForCompaction,
1112
isReplyRunStreamingForSessionId,
@@ -14,6 +15,7 @@ import {
1415
waitForReplyRunEndBySessionId,
1516
} from "../../auto-reply/reply/reply-run-registry.js";
1617
import {
18+
BLOCKED_TOOL_CALL_ABORT_FLOOR_MS,
1719
getDiagnosticSessionActivitySnapshot,
1820
markDiagnosticEmbeddedRunEnded,
1921
markDiagnosticEmbeddedRunStarted,
@@ -448,6 +450,13 @@ function prepareEmbeddedAgentQueueMessage(
448450
): PreparedEmbeddedAgentQueueMessage {
449451
const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
450452
if (!handle) {
453+
// A stale reply-backed run must produce the same closed reason as the
454+
// embedded gate so announce delivery falls through to direct instead of
455+
// reading the wedged op as active and dropping the handoff.
456+
if (isReplyRunEvidenceStaleBySessionId(sessionId)) {
457+
diag.debug(`queue message failed: sessionId=${sessionId} reason=stale_run`);
458+
return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "stale_run") };
459+
}
451460
const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text, options);
452461
if (queuedReplyRunMessage) {
453462
logMessageQueued({ sessionId, source: "embedded-agent-runner" });
@@ -479,9 +488,16 @@ function prepareEmbeddedAgentQueueMessage(
479488
return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "not_streaming") };
480489
}
481490
const activity = getDiagnosticSessionActivitySnapshot({ sessionId });
491+
// Quiet tool phases stay steerable until the blocked-tool floor: refusing at
492+
// the shorter window would push the message into admission takeover of a run
493+
// the diagnostic layer still considers healthy.
494+
const steerStaleCaptureMs =
495+
activity.activeWorkKind === "tool_call"
496+
? Math.max(EMBEDDED_STEER_STALE_CAPTURE_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)
497+
: EMBEDDED_STEER_STALE_CAPTURE_MS;
482498
if (
483499
typeof activity.lastProgressAgeMs === "number" &&
484-
activity.lastProgressAgeMs > EMBEDDED_STEER_STALE_CAPTURE_MS
500+
activity.lastProgressAgeMs > steerStaleCaptureMs
485501
) {
486502
diag.debug(`queue message failed: sessionId=${sessionId} reason=stale_run`);
487503
return { kind: "complete", outcome: createQueueFailureOutcome(sessionId, "stale_run") };

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

Lines changed: 30 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ import {
66
} from "../../agents/run-termination.js";
77
import { createAbortError } from "../../infra/abort-signal.js";
88
import {
9+
BLOCKED_TOOL_CALL_ABORT_FLOOR_MS,
10+
getDiagnosticSessionActivitySnapshot,
911
markDiagnosticEmbeddedRunEnded,
1012
markDiagnosticEmbeddedRunStarted,
1113
} from "../../logging/diagnostic-run-activity.js";
@@ -851,6 +853,33 @@ export function expireStaleReplyRunBySessionId(
851853
return operation ? expireStaleReplyOperation(operation, reason) : false;
852854
}
853855

856+
/**
857+
* Effective staleness window for an operation. Quiet-but-alive tool phases get
858+
* the diagnostic blocked-tool floor: a human message must not reclaim a healthy
859+
* long tool that stuck recovery itself would not touch yet.
860+
*/
861+
export function resolveReplyRunStaleThresholdMs(operation: ReplyOperation): number {
862+
const activity = getDiagnosticSessionActivitySnapshot({
863+
sessionId: operation.sessionId,
864+
sessionKey: operation.key,
865+
});
866+
return activity.activeWorkKind === "tool_call"
867+
? Math.max(REPLY_RUN_STALE_TAKEOVER_MS, BLOCKED_TOOL_CALL_ABORT_FLOOR_MS)
868+
: REPLY_RUN_STALE_TAKEOVER_MS;
869+
}
870+
871+
export function isReplyRunEvidenceStale(operation: ReplyOperation): boolean {
872+
return (
873+
!operation.result &&
874+
Date.now() - operation.lastActivityAtMs > resolveReplyRunStaleThresholdMs(operation)
875+
);
876+
}
877+
878+
export function isReplyRunEvidenceStaleBySessionId(sessionId: string): boolean {
879+
const operation = resolveReplyRunForCurrentSessionId(sessionId);
880+
return operation ? isReplyRunEvidenceStale(operation) : false;
881+
}
882+
854883
export const replyRunRegistry: ReplyRunRegistry = {
855884
begin(params) {
856885
return createReplyOperation(params);
@@ -979,7 +1008,7 @@ export function queueReplyRunMessage(
9791008
}
9801009
// Steering into an evidence-dead run swallows the human message that would
9811010
// otherwise trigger stale takeover through normal reply admission.
982-
if (Date.now() - operation.lastActivityAtMs > REPLY_RUN_STALE_TAKEOVER_MS) {
1011+
if (isReplyRunEvidenceStale(operation)) {
9831012
return false;
9841013
}
9851014
if (!isReplyBackendMessageInjectable(backend)) {

0 commit comments

Comments
 (0)