Skip to content

Commit 8cac327

Browse files
dutifulbobosolmaz
dutifulbob
and
osolmaz
authored
ACP: recover hung bound turns (openclaw#51816)
* ACP: add hung-turn starvation repro * ACP: recover hung bound turns * ACP: preserve timed-out session handles --------- Co-authored-by: Onur <[email protected]>
1 parent 5c05347 commit 8cac327

2 files changed

Lines changed: 454 additions & 21 deletions

File tree

‎src/acp/control-plane/manager.core.ts‎

Lines changed: 281 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
12
import type { OpenClawConfig } from "../../config/config.js";
23
import { logVerbose } from "../../globals.js";
34
import { normalizeAgentId } from "../../routing/session-key.js";
@@ -20,6 +21,7 @@ import type {
2021
AcpRuntime,
2122
AcpRuntimeCapabilities,
2223
AcpRuntimeHandle,
24+
AcpRuntimeSessionMode,
2325
AcpRuntimeStatus,
2426
} from "../runtime/types.js";
2527
import { reconcileManagerRuntimeSessionIdentifiers } from "./manager.identity-reconcile.js";
@@ -70,6 +72,10 @@ import {
7072
} from "./runtime-options.js";
7173
import { SessionActorQueue } from "./session-actor-queue.js";
7274

75+
const ACP_TURN_TIMEOUT_GRACE_MS = 1_000;
76+
const ACP_TURN_TIMEOUT_CLEANUP_GRACE_MS = 2_000;
77+
const ACP_TURN_TIMEOUT_REASON = "turn-timeout";
78+
7379
export class AcpSessionManager {
7480
private readonly actorQueue = new SessionActorQueue();
7581
private readonly actorTailBySession = this.actorQueue.getTailMapForTesting();
@@ -621,6 +627,7 @@ export class AcpSessionManager {
621627
let activeTurnStarted = false;
622628
let sawTurnOutput = false;
623629
let retryFreshHandle = false;
630+
let skipPostTurnCleanup = false;
624631
try {
625632
const ensured = await this.ensureRuntimeHandle({
626633
cfg: input.cfg,
@@ -667,26 +674,58 @@ export class AcpSessionManager {
667674
input.signal && typeof AbortSignal.any === "function"
668675
? AbortSignal.any([input.signal, internalAbortController.signal])
669676
: internalAbortController.signal;
670-
for await (const event of runtime.runTurn({
671-
handle,
672-
text: input.text,
673-
attachments: input.attachments,
674-
mode: input.mode,
675-
requestId: input.requestId,
676-
signal: combinedSignal,
677-
})) {
678-
if (event.type === "error") {
679-
streamError = new AcpRuntimeError(
680-
normalizeAcpErrorCode(event.code),
681-
event.message?.trim() || "ACP turn failed before completion.",
682-
);
683-
} else if (event.type === "text_delta" || event.type === "tool_call") {
684-
sawTurnOutput = true;
677+
const eventGate = { open: true };
678+
const turnPromise = (async () => {
679+
for await (const event of runtime.runTurn({
680+
handle,
681+
text: input.text,
682+
attachments: input.attachments,
683+
mode: input.mode,
684+
requestId: input.requestId,
685+
signal: combinedSignal,
686+
})) {
687+
if (!eventGate.open) {
688+
continue;
689+
}
690+
if (event.type === "error") {
691+
streamError = new AcpRuntimeError(
692+
normalizeAcpErrorCode(event.code),
693+
event.message?.trim() || "ACP turn failed before completion.",
694+
);
695+
} else if (event.type === "text_delta" || event.type === "tool_call") {
696+
sawTurnOutput = true;
697+
}
698+
if (input.onEvent) {
699+
await input.onEvent(event);
700+
}
685701
}
686-
if (input.onEvent) {
687-
await input.onEvent(event);
702+
if (eventGate.open && streamError) {
703+
throw streamError;
688704
}
689-
}
705+
})();
706+
const turnTimeoutMs = this.resolveTurnTimeoutMs({
707+
cfg: input.cfg,
708+
meta,
709+
});
710+
const sessionMode = meta.mode;
711+
await this.awaitTurnWithTimeout({
712+
sessionKey,
713+
turnPromise,
714+
timeoutMs: turnTimeoutMs + ACP_TURN_TIMEOUT_GRACE_MS,
715+
timeoutLabelMs: turnTimeoutMs,
716+
onTimeout: async () => {
717+
eventGate.open = false;
718+
skipPostTurnCleanup = true;
719+
if (!activeTurn) {
720+
return;
721+
}
722+
await this.cleanupTimedOutTurn({
723+
sessionKey,
724+
activeTurn,
725+
mode: sessionMode,
726+
});
727+
},
728+
});
690729
if (streamError) {
691730
throw streamError;
692731
}
@@ -735,7 +774,14 @@ export class AcpSessionManager {
735774
if (activeTurn && this.activeTurnBySession.get(actorKey) === activeTurn) {
736775
this.activeTurnBySession.delete(actorKey);
737776
}
738-
if (!retryFreshHandle && runtime && handle && meta && meta.mode !== "oneshot") {
777+
if (
778+
!retryFreshHandle &&
779+
!skipPostTurnCleanup &&
780+
runtime &&
781+
handle &&
782+
meta &&
783+
meta.mode !== "oneshot"
784+
) {
739785
({ handle } = await this.reconcileRuntimeSessionIdentifiers({
740786
cfg: input.cfg,
741787
sessionKey,
@@ -745,7 +791,14 @@ export class AcpSessionManager {
745791
failOnStatusError: false,
746792
}));
747793
}
748-
if (!retryFreshHandle && runtime && handle && meta && meta.mode === "oneshot") {
794+
if (
795+
!retryFreshHandle &&
796+
!skipPostTurnCleanup &&
797+
runtime &&
798+
handle &&
799+
meta &&
800+
meta.mode === "oneshot"
801+
) {
749802
try {
750803
await runtime.close({
751804
handle,
@@ -767,6 +820,191 @@ export class AcpSessionManager {
767820
});
768821
}
769822

823+
private resolveTurnTimeoutMs(params: { cfg: OpenClawConfig; meta: SessionAcpMeta }): number {
824+
const runtimeTimeoutSeconds = resolveRuntimeOptionsFromMeta(params.meta).timeoutSeconds;
825+
if (
826+
typeof runtimeTimeoutSeconds === "number" &&
827+
Number.isFinite(runtimeTimeoutSeconds) &&
828+
runtimeTimeoutSeconds > 0
829+
) {
830+
return Math.max(1_000, Math.round(runtimeTimeoutSeconds * 1_000));
831+
}
832+
return resolveAgentTimeoutMs({
833+
cfg: params.cfg,
834+
minMs: 1_000,
835+
});
836+
}
837+
838+
private async awaitTurnWithTimeout<T>(params: {
839+
sessionKey: string;
840+
turnPromise: Promise<T>;
841+
timeoutMs: number;
842+
timeoutLabelMs: number;
843+
onTimeout: () => Promise<void>;
844+
}): Promise<T> {
845+
const observedTurnPromise: Promise<
846+
| {
847+
kind: "value";
848+
value: T;
849+
}
850+
| {
851+
kind: "error";
852+
error: unknown;
853+
}
854+
> = params.turnPromise.then(
855+
(value) => ({
856+
kind: "value" as const,
857+
value,
858+
}),
859+
(error) => ({
860+
kind: "error" as const,
861+
error,
862+
}),
863+
);
864+
865+
if (params.timeoutMs <= 0) {
866+
const outcome = await observedTurnPromise;
867+
if (outcome.kind === "error") {
868+
throw outcome.error;
869+
}
870+
return outcome.value;
871+
}
872+
873+
const timeoutToken = Symbol("acp-turn-timeout");
874+
let timer: NodeJS.Timeout | undefined;
875+
const timeoutPromise = new Promise<typeof timeoutToken>((resolve) => {
876+
timer = setTimeout(() => resolve(timeoutToken), params.timeoutMs);
877+
timer.unref?.();
878+
});
879+
880+
try {
881+
const outcome = await Promise.race([observedTurnPromise, timeoutPromise]);
882+
if (outcome === timeoutToken) {
883+
void observedTurnPromise.then((lateOutcome) => {
884+
if (lateOutcome.kind === "error") {
885+
logVerbose(
886+
`acp-manager: detached late turn error after timeout for ${params.sessionKey}: ${String(lateOutcome.error)}`,
887+
);
888+
}
889+
});
890+
await params.onTimeout();
891+
throw new AcpRuntimeError(
892+
"ACP_TURN_FAILED",
893+
`ACP turn timed out after ${Math.max(1, Math.round(params.timeoutLabelMs / 1_000))}s.`,
894+
);
895+
}
896+
if (outcome.kind === "error") {
897+
throw outcome.error;
898+
}
899+
return outcome.value;
900+
} finally {
901+
if (timer) {
902+
clearTimeout(timer);
903+
}
904+
}
905+
}
906+
907+
private async cleanupTimedOutTurn(params: {
908+
sessionKey: string;
909+
activeTurn: ActiveTurnState;
910+
mode: AcpRuntimeSessionMode;
911+
}): Promise<void> {
912+
params.activeTurn.abortController.abort();
913+
if (!params.activeTurn.cancelPromise) {
914+
params.activeTurn.cancelPromise = params.activeTurn.runtime.cancel({
915+
handle: params.activeTurn.handle,
916+
reason: ACP_TURN_TIMEOUT_REASON,
917+
});
918+
}
919+
const cancelFinished = await this.awaitCleanupWithGrace({
920+
sessionKey: params.sessionKey,
921+
label: "cancel",
922+
promise: params.activeTurn.cancelPromise,
923+
});
924+
if (params.mode !== "oneshot") {
925+
return;
926+
}
927+
const closePromise = params.activeTurn.runtime.close({
928+
handle: params.activeTurn.handle,
929+
reason: ACP_TURN_TIMEOUT_REASON,
930+
});
931+
const closeFinished = await this.awaitCleanupWithGrace({
932+
sessionKey: params.sessionKey,
933+
label: "close",
934+
promise: closePromise,
935+
});
936+
if (cancelFinished && closeFinished) {
937+
this.clearCachedRuntimeStateIfHandleMatches({
938+
sessionKey: params.sessionKey,
939+
handle: params.activeTurn.handle,
940+
});
941+
return;
942+
}
943+
void Promise.allSettled([params.activeTurn.cancelPromise, closePromise]).then(() => {
944+
this.clearCachedRuntimeStateIfHandleMatches({
945+
sessionKey: params.sessionKey,
946+
handle: params.activeTurn.handle,
947+
});
948+
});
949+
}
950+
951+
private async awaitCleanupWithGrace(params: {
952+
sessionKey: string;
953+
label: "cancel" | "close";
954+
promise: Promise<unknown>;
955+
}): Promise<boolean> {
956+
const observedCleanupPromise: Promise<
957+
| {
958+
kind: "done";
959+
}
960+
| {
961+
kind: "error";
962+
error: unknown;
963+
}
964+
> = params.promise.then(
965+
() => ({
966+
kind: "done" as const,
967+
}),
968+
(error) => ({
969+
kind: "error" as const,
970+
error,
971+
}),
972+
);
973+
const timeoutToken = Symbol(`acp-timeout-${params.label}`);
974+
let timer: NodeJS.Timeout | undefined;
975+
const timeoutPromise = new Promise<typeof timeoutToken>((resolve) => {
976+
timer = setTimeout(() => resolve(timeoutToken), ACP_TURN_TIMEOUT_CLEANUP_GRACE_MS);
977+
timer.unref?.();
978+
});
979+
980+
try {
981+
const outcome = await Promise.race([observedCleanupPromise, timeoutPromise]);
982+
if (outcome === timeoutToken) {
983+
void observedCleanupPromise.then((lateOutcome) => {
984+
if (lateOutcome.kind === "error") {
985+
logVerbose(
986+
`acp-manager: detached timed-out turn ${params.label} cleanup failed for ${params.sessionKey}: ${String(lateOutcome.error)}`,
987+
);
988+
}
989+
});
990+
logVerbose(
991+
`acp-manager: timed-out turn ${params.label} cleanup exceeded ${ACP_TURN_TIMEOUT_CLEANUP_GRACE_MS}ms for ${params.sessionKey}`,
992+
);
993+
return false;
994+
}
995+
if (outcome.kind === "error") {
996+
logVerbose(
997+
`acp-manager: timed-out turn ${params.label} cleanup failed for ${params.sessionKey}: ${String(outcome.error)}`,
998+
);
999+
}
1000+
return true;
1001+
} finally {
1002+
if (timer) {
1003+
clearTimeout(timer);
1004+
}
1005+
}
1006+
}
1007+
7701008
async cancelSession(params: {
7711009
cfg: OpenClawConfig;
7721010
sessionKey: string;
@@ -1418,4 +1656,27 @@ export class AcpSessionManager {
14181656
private clearCachedRuntimeState(sessionKey: string): void {
14191657
this.runtimeCache.clear(normalizeActorKey(sessionKey));
14201658
}
1659+
1660+
private clearCachedRuntimeStateIfHandleMatches(params: {
1661+
sessionKey: string;
1662+
handle: AcpRuntimeHandle;
1663+
}): void {
1664+
const cached = this.getCachedRuntimeState(params.sessionKey);
1665+
if (!cached || !this.runtimeHandlesMatch(cached.handle, params.handle)) {
1666+
return;
1667+
}
1668+
this.clearCachedRuntimeState(params.sessionKey);
1669+
}
1670+
1671+
private runtimeHandlesMatch(a: AcpRuntimeHandle, b: AcpRuntimeHandle): boolean {
1672+
return (
1673+
a.sessionKey === b.sessionKey &&
1674+
a.backend === b.backend &&
1675+
a.runtimeSessionName === b.runtimeSessionName &&
1676+
(a.cwd ?? "") === (b.cwd ?? "") &&
1677+
(a.acpxRecordId ?? "") === (b.acpxRecordId ?? "") &&
1678+
(a.backendSessionId ?? "") === (b.backendSessionId ?? "") &&
1679+
(a.agentSessionId ?? "") === (b.agentSessionId ?? "")
1680+
);
1681+
}
14211682
}

0 commit comments

Comments
 (0)