Skip to content

Commit 9a4f0e4

Browse files
steipetesteipete
committed
fix(reply): wait for visible reply lane ownership
1 parent 0954542 commit 9a4f0e4

3 files changed

Lines changed: 31 additions & 165 deletions

File tree

‎src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts‎

Lines changed: 10 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -901,22 +901,14 @@ describe("dispatchReplyFromConfig ACP abort", () => {
901901
expect(getActiveReplyRunCount()).toBe(0);
902902
});
903903

904-
it("keeps borrowed active operation abort out of reply resolver runs", async () => {
904+
it("suppresses reply resolver runs after active source abort", async () => {
905905
const existingOperation = createReplyOperation({
906906
sessionKey: "agent:already-active-resolver",
907907
sessionId: "active-session",
908908
resetTriggered: false,
909909
});
910-
let resolverStarted!: () => void;
911-
const resolverStartedPromise = new Promise<void>((resolve) => {
912-
resolverStarted = resolve;
913-
});
914-
let releaseResolver!: () => void;
915-
const releaseResolverPromise = new Promise<void>((resolve) => {
916-
releaseResolver = resolve;
917-
});
918-
let resolverAbortSignal: AbortSignal | undefined;
919-
910+
existingOperation.setPhase("running");
911+
const replyResolver = vi.fn(async () => undefined);
920912
const dispatcher = createDispatcher();
921913
const ctx = buildTestCtx({
922914
Provider: "discord",
@@ -933,41 +925,30 @@ describe("dispatchReplyFromConfig ACP abort", () => {
933925
},
934926
} as OpenClawConfig,
935927
dispatcher,
936-
replyResolver: async (_resolverCtx, options) => {
937-
resolverAbortSignal = options?.abortSignal;
938-
resolverStarted();
939-
await releaseResolverPromise;
940-
return undefined;
941-
},
928+
replyResolver,
942929
});
943930

944-
await resolverStartedPromise;
945-
expect(resolverAbortSignal).toBeUndefined();
946931
expect(replyRunRegistry.abort("agent:already-active-resolver")).toBe(true);
947-
releaseResolver();
948932

949933
await expect(dispatchPromise).resolves.toMatchObject({
950934
queuedFinal: false,
951935
counts: { tool: 0, block: 0, final: 0 },
952936
});
953937
expect(existingOperation.result).toEqual({ kind: "aborted", code: "aborted_by_user" });
938+
expect(replyResolver).not.toHaveBeenCalled();
954939
expect(dispatcher.sendFinalReply).not.toHaveBeenCalled();
940+
existingOperation.complete();
955941
expect(getActiveReplyRunCount()).toBe(0);
956942
});
957943

958-
it("keeps caller abort active when reply resolver borrows an active source operation", async () => {
944+
it("keeps caller abort active while waiting for an active source operation", async () => {
959945
const existingOperation = createReplyOperation({
960946
sessionKey: "agent:already-active-caller-abort",
961947
sessionId: "active-session",
962948
resetTriggered: false,
963949
});
964950
const callerAbort = new AbortController();
965-
let resolverStarted!: () => void;
966-
const resolverStartedPromise = new Promise<void>((resolve) => {
967-
resolverStarted = resolve;
968-
});
969-
let resolverAbortSignal: AbortSignal | undefined;
970-
951+
const replyResolver = vi.fn(async () => ({ text: "late final should not send" }));
971952
const dispatcher = createDispatcher();
972953
const ctx = buildTestCtx({
973954
Provider: "discord",
@@ -985,30 +966,17 @@ describe("dispatchReplyFromConfig ACP abort", () => {
985966
} as OpenClawConfig,
986967
dispatcher,
987968
replyOptions: { abortSignal: callerAbort.signal },
988-
replyResolver: async (_resolverCtx, options) => {
989-
const signal = options?.abortSignal;
990-
resolverAbortSignal = signal;
991-
resolverStarted();
992-
if (!signal) {
993-
throw new Error("expected composed abort signal");
994-
}
995-
await new Promise<void>((resolve) => {
996-
signal.addEventListener("abort", () => resolve(), { once: true });
997-
});
998-
return { text: "late final should not send" };
999-
},
969+
replyResolver,
1000970
});
1001971

1002-
await resolverStartedPromise;
1003-
expect(resolverAbortSignal).toBeInstanceOf(AbortSignal);
1004-
expect(resolverAbortSignal).not.toBe(existingOperation.abortSignal);
1005972
callerAbort.abort();
1006973

1007974
await expect(dispatchPromise).resolves.toMatchObject({
1008975
queuedFinal: false,
1009976
counts: { tool: 0, block: 0, final: 0 },
1010977
});
1011978
expect(existingOperation.result).toBeNull();
979+
expect(replyResolver).not.toHaveBeenCalled();
1012980
expect(dispatcher.sendFinalReply).not.toHaveBeenCalled();
1013981
existingOperation.abortByUser();
1014982
expect(getActiveReplyRunCount()).toBe(0);

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

Lines changed: 0 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -954,124 +954,6 @@ describe("dispatchReplyFromConfig", () => {
954954
);
955955
});
956956

957-
it("lets queue-aware resolvers see active Telegram topic turns", async () => {
958-
setNoAbort();
959-
const sessionKey = "agent:main:telegram:group:-1003774691294:topic:3731";
960-
const activeOperation = createReplyOperation({
961-
sessionKey,
962-
sessionId: "heartbeat-session",
963-
resetTriggered: false,
964-
});
965-
activeOperation.setPhase("running");
966-
const dispatcher = createDispatcher();
967-
const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => {
968-
const internalOpts = opts as GetReplyOptions & {
969-
replyOperation?: { sessionId?: string };
970-
};
971-
expect(internalOpts.replyOperation).toBeUndefined();
972-
return { text: "visible after active run" } satisfies ReplyPayload;
973-
});
974-
const waitForIdleSpy = vi.spyOn(replyRunRegistry, "waitForIdle");
975-
976-
const dispatchPromise = dispatchReplyFromConfig({
977-
ctx: buildTestCtx({
978-
Provider: "telegram",
979-
Surface: "telegram",
980-
OriginatingChannel: "telegram",
981-
SessionKey: sessionKey,
982-
ChatType: "group",
983-
IsForum: true,
984-
MessageThreadId: 3731,
985-
TransportThreadId: 3731,
986-
To: "telegram:-1003774691294:topic:3731",
987-
BodyForAgent: "real user message while heartbeat is active",
988-
}),
989-
cfg: automaticGroupReplyConfig,
990-
dispatcher,
991-
replyResolver,
992-
});
993-
994-
const result = await dispatchPromise;
995-
activeOperation.complete();
996-
waitForIdleSpy.mockRestore();
997-
998-
expect(result.queuedFinal).toBe(true);
999-
expect(replyResolver).toHaveBeenCalledTimes(1);
1000-
expect(waitForIdleSpy).not.toHaveBeenCalled();
1001-
expect(firstFinalReplyPayload(dispatcher)?.text).toBe("visible after active run");
1002-
});
1003-
1004-
it("does not serialize custom resolvers before queue policy runs", async () => {
1005-
setNoAbort();
1006-
const sessionKey = "agent:main:telegram:group:-1003774691294:topic:3731";
1007-
const activeOperation = createReplyOperation({
1008-
sessionKey,
1009-
sessionId: "active-session",
1010-
resetTriggered: false,
1011-
});
1012-
activeOperation.setPhase("running");
1013-
let releaseFirstResolver!: () => void;
1014-
let releaseSecondResolver!: () => void;
1015-
const firstResolverReleased = new Promise<void>((resolve) => {
1016-
releaseFirstResolver = resolve;
1017-
});
1018-
const secondResolverReleased = new Promise<void>((resolve) => {
1019-
releaseSecondResolver = resolve;
1020-
});
1021-
const firstDispatcher = createDispatcher();
1022-
const secondDispatcher = createDispatcher();
1023-
const firstReplyResolver = vi.fn(async () => {
1024-
await firstResolverReleased;
1025-
return { text: "first visible reply" } satisfies ReplyPayload;
1026-
});
1027-
const secondReplyResolver = vi.fn(async () => {
1028-
await secondResolverReleased;
1029-
return { text: "second visible reply" } satisfies ReplyPayload;
1030-
});
1031-
const createTopicCtx = (messageSid: string, body: string) =>
1032-
buildTestCtx({
1033-
Provider: "telegram",
1034-
Surface: "telegram",
1035-
OriginatingChannel: "telegram",
1036-
SessionKey: sessionKey,
1037-
ChatType: "group",
1038-
IsForum: true,
1039-
MessageSid: messageSid,
1040-
MessageThreadId: 3731,
1041-
TransportThreadId: 3731,
1042-
To: "telegram:-1003774691294:topic:3731",
1043-
BodyForAgent: body,
1044-
});
1045-
1046-
const firstDispatch = dispatchReplyFromConfig({
1047-
ctx: createTopicCtx("27782", "first queued user message"),
1048-
cfg: automaticGroupReplyConfig,
1049-
dispatcher: firstDispatcher,
1050-
replyResolver: firstReplyResolver,
1051-
});
1052-
const secondDispatch = dispatchReplyFromConfig({
1053-
ctx: createTopicCtx("27783", "second queued user message"),
1054-
cfg: automaticGroupReplyConfig,
1055-
dispatcher: secondDispatcher,
1056-
replyResolver: secondReplyResolver,
1057-
});
1058-
1059-
await vi.waitFor(() => {
1060-
expect(firstReplyResolver.mock.calls.length + secondReplyResolver.mock.calls.length).toBe(2);
1061-
});
1062-
1063-
releaseFirstResolver();
1064-
releaseSecondResolver();
1065-
activeOperation.complete();
1066-
1067-
await expect(Promise.all([firstDispatch, secondDispatch])).resolves.toEqual([
1068-
expect.objectContaining({ queuedFinal: true }),
1069-
expect.objectContaining({ queuedFinal: true }),
1070-
]);
1071-
expect(firstFinalReplyPayload(firstDispatcher)?.text).toBe("first visible reply");
1072-
expect(firstFinalReplyPayload(secondDispatcher)?.text).toBe("second visible reply");
1073-
});
1074-
1075957
it("skips pre-dispatch admission when the caller already aborted", async () => {
1076958
setNoAbort();
1077959
const sessionKey = "agent:main:telegram:group:-1003774691294:topic:3731";

‎src/auto-reply/reply/dispatch-from-config.ts‎

Lines changed: 21 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -947,13 +947,24 @@ export async function dispatchReplyFromConfig(
947947
let dispatchAbortOperation: ReplyOperation | undefined;
948948
let preDispatchAbortOperation: ReplyOperation | undefined;
949949
type DispatchReplyOperationAcquisition = { status: "ready" } | { status: "busy" };
950-
const ensureDispatchReplyOperation = async (): Promise<DispatchReplyOperationAcquisition> => {
950+
const ensureDispatchReplyOperation = async (
951+
phase: "pre_dispatch" | "dispatch",
952+
): Promise<DispatchReplyOperationAcquisition> => {
951953
if (dispatchReplyOperation && !dispatchReplyOperation.result) {
952954
return { status: "ready" };
953955
}
954956
if (dispatchAbortOperation && !dispatchAbortOperation.result) {
955957
return dispatchReplyOperation ? { status: "ready" } : { status: "busy" };
956958
}
959+
if (
960+
phase === "dispatch" &&
961+
preDispatchAbortOperation?.result &&
962+
preDispatchAbortOperation.result.kind !== "completed" &&
963+
!dispatchReplyOperation
964+
) {
965+
dispatchAbortOperation = preDispatchAbortOperation;
966+
return { status: "busy" };
967+
}
957968
if (!dispatchOperationSessionKey) {
958969
return { status: "ready" };
959970
}
@@ -963,16 +974,17 @@ export async function dispatchReplyFromConfig(
963974
sessionStoreEntry.entry?.sessionId ??
964975
crypto.randomUUID();
965976
const replyTurnKind = resolveReplyTurnKind(params.replyOptions);
977+
const allowActivePreDispatch = phase === "pre_dispatch" && replyTurnKind === "visible";
966978
const admission = await admitReplyTurn({
967979
sessionKey: dispatchOperationSessionKey,
968980
sessionId: operationSessionId,
969981
kind: replyTurnKind,
970982
resetTriggered: false,
971983
upstreamAbortSignal: params.replyOptions?.abortSignal,
972-
waitForActive: false,
984+
waitForActive: !allowActivePreDispatch,
973985
});
974986
if (admission.status === "skipped") {
975-
if (replyTurnKind === "visible" && admission.reason === "active-run") {
987+
if (allowActivePreDispatch && admission.reason === "active-run") {
976988
preDispatchAbortOperation = admission.activeOperation;
977989
return { status: "ready" };
978990
}
@@ -1592,7 +1604,7 @@ export async function dispatchReplyFromConfig(
15921604
}
15931605
// Register the dispatch-owned operation before any plugin hook or model work
15941606
// so /stop can abort pre-run and in-run stalls through the same session lane.
1595-
if ((await ensureDispatchReplyOperation()).status === "busy") {
1607+
if ((await ensureDispatchReplyOperation("pre_dispatch")).status === "busy") {
15961608
return finishReplyOperationBusyDispatch();
15971609
}
15981610

@@ -1789,6 +1801,10 @@ export async function dispatchReplyFromConfig(
17891801
}
17901802
}
17911803

1804+
if ((await ensureDispatchReplyOperation("dispatch")).status === "busy") {
1805+
return finishReplyOperationBusyDispatch();
1806+
}
1807+
17921808
// When automatic source delivery is suppressed, still let the agent process
17931809
// the inbound message (context, memory, tool calls) but suppress automatic
17941810
// outbound source delivery.
@@ -2322,7 +2338,7 @@ export async function dispatchReplyFromConfig(
23222338
),
23232339
),
23242340
);
2325-
if ((await ensureDispatchReplyOperation()).status === "busy") {
2341+
if ((await ensureDispatchReplyOperation("dispatch")).status === "busy") {
23262342
return finishReplyOperationBusyDispatch({ recordAgentDispatchCompleted: true });
23272343
}
23282344

0 commit comments

Comments
 (0)