Skip to content

Commit f8ed4de

Browse files
jalehmanjalehman
authored
refactor: add abort target session accessor (#96201)
* refactor: add abort target session accessor * refactor: centralize command abort session lookup * fix: keep abort runtime path best effort * fix: preserve abort target identity on persistence failure * fix: remember abort target when persistence is skipped * fix: abort runtime before metadata persistence * fix: preserve abort target fallback typing * fix: avoid stale abort memory fallback * fix: keep abort accessor ratchet narrow * fix: type abort persistence test mock * fix: align abort accessor ratchet test
1 parent b08d901 commit f8ed4de

10 files changed

Lines changed: 540 additions & 113 deletions

‎scripts/check-session-accessor-boundary.mjs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ export const migratedSessionAccessorFiles = new Set([
7979
"src/agents/embedded-agent-runner/tool-result-truncation.ts",
8080
"src/agents/embedded-agent-runner/transcript-rewrite.ts",
8181
"src/agents/embedded-agent-runner/transcript-runtime-state.ts",
82+
"src/auto-reply/reply/abort.ts",
8283
"src/auto-reply/reply/agent-runner-helpers.ts",
8384
"src/auto-reply/reply/agent-runner.ts",
8485
"src/auto-reply/reply/commands-subagents/action-info.ts",
@@ -127,6 +128,7 @@ export const migratedSessionAccessorWriteFiles = new Set([
127128
"src/agents/embedded-agent-runner/run.ts",
128129
"src/agents/embedded-agent-runner/run/attempt.ts",
129130
"src/agents/main-session-restart-recovery.ts",
131+
"src/auto-reply/reply/abort.ts",
130132
"src/auto-reply/reply/abort-cutoff.runtime.ts",
131133
"src/auto-reply/reply/agent-runner-cli-dispatch.ts",
132134
"src/auto-reply/reply/agent-runner-execution.ts",

‎src/auto-reply/reply/abort.test.ts‎

Lines changed: 235 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import path from "node:path";
55
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
66
import type { SubagentRunRecord } from "../../agents/subagent-registry.js";
77
import type { OpenClawConfig } from "../../config/config.js";
8+
import type { SessionAbortTargetResult } from "../../config/sessions/session-accessor.js";
89
import {
910
testing as abortTesting,
1011
getAbortMemory,
@@ -13,7 +14,6 @@ import {
1314
isAbortTrigger,
1415
resetAbortMemoryForTest,
1516
resolveAbortCutoffFromContext,
16-
resolveSessionEntryForKey,
1717
setAbortMemory,
1818
stopSubagentsForRequester,
1919
shouldSkipMessageByAbortCutoff,
@@ -382,32 +382,6 @@ describe("abort detection", () => {
382382
).toBe(false);
383383
});
384384

385-
it("resolves session entry when key exists in store", () => {
386-
const store = {
387-
"session-1": { sessionId: "abc", updatedAt: 0 },
388-
} as const;
389-
expect(resolveSessionEntryForKey(store, "session-1")).toEqual({
390-
entry: store["session-1"],
391-
key: "session-1",
392-
});
393-
expect(resolveSessionEntryForKey(store, "session-2")).toStrictEqual({});
394-
expect(resolveSessionEntryForKey(undefined, "session-1")).toStrictEqual({});
395-
});
396-
397-
it("resolves Telegram forum topic session when lookup key has different casing than store", () => {
398-
// Store normalizes keys to lowercase; caller may pass mixed-case. /stop in topic must find entry.
399-
const storeKey = "agent:main:telegram:group:-1001234567890:topic:99";
400-
const lookupKey = "Agent:Main:Telegram:Group:-1001234567890:Topic:99";
401-
const store = {
402-
[storeKey]: { sessionId: "agent-topic-99", updatedAt: 0 },
403-
} as Record<string, { sessionId: string; updatedAt: number }>;
404-
// Direct lookup fails (store uses lowercase keys); normalization fallback must succeed.
405-
expect(store[lookupKey]).toBeUndefined();
406-
const result = resolveSessionEntryForKey(store, lookupKey);
407-
expect(result.entry?.sessionId).toBe("agent-topic-99");
408-
expect(result.key).toBe(storeKey);
409-
});
410-
411385
it("fast-aborts even when text commands are disabled", async () => {
412386
const { cfg } = await createAbortConfig({ commandsTextEnabled: false });
413387

@@ -473,6 +447,240 @@ describe("abort detection", () => {
473447
expectSessionLaneCleared(sessionKey);
474448
});
475449

450+
it("fast-abort resolves canonical stored session identity before metadata persistence", async () => {
451+
const storeKey = "agent:main:telegram:group:-1001234567890:topic:99";
452+
const lookupKey = "Agent:Main:Telegram:Group:-1001234567890:Topic:99";
453+
const sessionId = "agent-topic-99";
454+
const { root, cfg } = await createAbortConfig({
455+
sessionIdsByKey: { [storeKey]: sessionId },
456+
});
457+
enqueueQueuedFollowupRun({ root, cfg, sessionId, sessionKey: storeKey });
458+
459+
const result = await runStopCommand({
460+
cfg,
461+
sessionKey: lookupKey,
462+
from: "telegram:123",
463+
to: "telegram:123",
464+
});
465+
466+
expect(result.handled).toBe(true);
467+
expect(runtimeAbortMocks.abortEmbeddedAgentRun).toHaveBeenCalledWith(sessionId);
468+
expect(getFollowupQueueDepth(storeKey)).toBe(0);
469+
expectSessionLaneCleared(storeKey);
470+
});
471+
472+
it("fast-abort still stops active runs when abort metadata persistence fails", async () => {
473+
const sessionKey = "telegram:persistence-failure";
474+
const sessionId = "session-persistence-failure";
475+
const activeSessionId = "active-persistence-failure";
476+
const { root, cfg } = await createAbortConfig({
477+
sessionIdsByKey: { [sessionKey]: sessionId },
478+
});
479+
runtimeAbortMocks.resolveActiveEmbeddedRunSessionId.mockReturnValue(activeSessionId);
480+
abortTesting.setDepsForTests({
481+
getAcpSessionManager: (() =>
482+
({
483+
resolveSession: acpManagerMocks.resolveSession,
484+
cancelSession: acpManagerMocks.cancelSession,
485+
}) as never) as never,
486+
abortEmbeddedAgentRun: runtimeAbortMocks.abortEmbeddedAgentRun,
487+
resolveActiveEmbeddedRunSessionId: runtimeAbortMocks.resolveActiveEmbeddedRunSessionId,
488+
markSessionAbortTarget: vi.fn(async () => {
489+
throw new Error("simulated persistence failure");
490+
}),
491+
getLatestSubagentRunByChildSessionKey:
492+
subagentRegistryMocks.getLatestSubagentRunByChildSessionKey,
493+
listSubagentRunsForController: subagentRegistryMocks.listSubagentRunsForRequester,
494+
markSubagentRunTerminated: subagentRegistryMocks.markSubagentRunTerminated,
495+
});
496+
enqueueQueuedFollowupRun({ root, cfg, sessionId, sessionKey });
497+
498+
const result = await runStopCommand({
499+
cfg,
500+
sessionKey,
501+
from: "telegram:123",
502+
to: "telegram:123",
503+
});
504+
505+
expect(result.handled).toBe(true);
506+
expect(runtimeAbortMocks.abortEmbeddedAgentRun).toHaveBeenCalledWith(activeSessionId);
507+
expect(getFollowupQueueDepth(sessionKey)).toBe(0);
508+
expectSessionLaneCleared(sessionKey);
509+
expect(getAbortMemory(sessionKey)).toBeUndefined();
510+
});
511+
512+
it("fast-abort uses resolved target identity when abort metadata save fails", async () => {
513+
const requestedKey = "Agent:Main:Telegram:Group:-1001234567890:Topic:99";
514+
const canonicalKey = "agent:main:telegram:group:-1001234567890:topic:99";
515+
const sessionId = "resolved-persistence-failure";
516+
const { root, cfg } = await createAbortConfig();
517+
abortTesting.setDepsForTests({
518+
getAcpSessionManager: (() =>
519+
({
520+
resolveSession: acpManagerMocks.resolveSession,
521+
cancelSession: acpManagerMocks.cancelSession,
522+
}) as never) as never,
523+
abortEmbeddedAgentRun: runtimeAbortMocks.abortEmbeddedAgentRun,
524+
resolveActiveEmbeddedRunSessionId: runtimeAbortMocks.resolveActiveEmbeddedRunSessionId,
525+
markSessionAbortTarget: vi.fn(async () => ({
526+
entry: {
527+
sessionId,
528+
updatedAt: 10,
529+
},
530+
persisted: false,
531+
persistenceError: "simulated persistence failure",
532+
sessionId,
533+
sessionKey: canonicalKey,
534+
})),
535+
resolveSessionAbortTarget: vi.fn(() => ({
536+
entry: {
537+
sessionId,
538+
updatedAt: 10,
539+
},
540+
sessionId,
541+
sessionKey: canonicalKey,
542+
})),
543+
getLatestSubagentRunByChildSessionKey:
544+
subagentRegistryMocks.getLatestSubagentRunByChildSessionKey,
545+
listSubagentRunsForController: subagentRegistryMocks.listSubagentRunsForRequester,
546+
markSubagentRunTerminated: subagentRegistryMocks.markSubagentRunTerminated,
547+
});
548+
enqueueQueuedFollowupRun({ root, cfg, sessionId, sessionKey: canonicalKey });
549+
550+
const result = await runStopCommand({
551+
cfg,
552+
sessionKey: requestedKey,
553+
from: "telegram:123",
554+
to: "telegram:123",
555+
});
556+
557+
expect(result.handled).toBe(true);
558+
expect(runtimeAbortMocks.abortEmbeddedAgentRun).toHaveBeenCalledWith(sessionId);
559+
expect(getFollowupQueueDepth(canonicalKey)).toBe(0);
560+
expectSessionLaneCleared(canonicalKey);
561+
expect(getAbortMemory(canonicalKey)).toBeUndefined();
562+
});
563+
564+
it("fast-abort uses abort memory when no persisted target entry exists", async () => {
565+
const sessionKey = "telegram:missing-persistence-target";
566+
const { cfg } = await createAbortConfig();
567+
abortTesting.setDepsForTests({
568+
getAcpSessionManager: (() =>
569+
({
570+
resolveSession: acpManagerMocks.resolveSession,
571+
cancelSession: acpManagerMocks.cancelSession,
572+
}) as never) as never,
573+
abortEmbeddedAgentRun: runtimeAbortMocks.abortEmbeddedAgentRun,
574+
resolveActiveEmbeddedRunSessionId: runtimeAbortMocks.resolveActiveEmbeddedRunSessionId,
575+
markSessionAbortTarget: vi.fn(async () => null),
576+
resolveSessionAbortTarget: vi.fn(() => null),
577+
getLatestSubagentRunByChildSessionKey:
578+
subagentRegistryMocks.getLatestSubagentRunByChildSessionKey,
579+
listSubagentRunsForController: subagentRegistryMocks.listSubagentRunsForRequester,
580+
markSubagentRunTerminated: subagentRegistryMocks.markSubagentRunTerminated,
581+
});
582+
583+
const result = await runStopCommand({
584+
cfg,
585+
sessionKey,
586+
from: "telegram:123",
587+
to: "telegram:123",
588+
});
589+
590+
expect(result.handled).toBe(true);
591+
expect(getAbortMemory(sessionKey)).toBe(true);
592+
});
593+
594+
it("fast-abort does not wait for abort metadata persistence before stopping runs", async () => {
595+
const sessionKey = "telegram:slow-persistence";
596+
const childKey = "agent:main:subagent:slow-persistence-child";
597+
const sessionId = "session-slow-persistence";
598+
const childSessionId = "session-slow-persistence-child";
599+
const { root, cfg } = await createAbortConfig({
600+
sessionIdsByKey: {
601+
[childKey]: childSessionId,
602+
[sessionKey]: sessionId,
603+
},
604+
});
605+
let finishPersistence: (() => void) | undefined;
606+
const persistenceStarted = new Promise<void>((resolveStarted) => {
607+
abortTesting.setDepsForTests({
608+
getAcpSessionManager: (() =>
609+
({
610+
resolveSession: acpManagerMocks.resolveSession,
611+
cancelSession: acpManagerMocks.cancelSession,
612+
}) as never) as never,
613+
abortEmbeddedAgentRun: runtimeAbortMocks.abortEmbeddedAgentRun,
614+
resolveActiveEmbeddedRunSessionId: runtimeAbortMocks.resolveActiveEmbeddedRunSessionId,
615+
markSessionAbortTarget: vi.fn(
616+
() =>
617+
new Promise<SessionAbortTargetResult | null>((resolvePersistence) => {
618+
resolveStarted();
619+
finishPersistence = () => {
620+
resolvePersistence({
621+
entry: {
622+
sessionId,
623+
updatedAt: 10,
624+
},
625+
persisted: true,
626+
sessionId,
627+
sessionKey,
628+
});
629+
};
630+
}),
631+
),
632+
resolveSessionAbortTarget: vi.fn(() => ({
633+
entry: {
634+
sessionId,
635+
updatedAt: 10,
636+
},
637+
sessionId,
638+
sessionKey,
639+
})),
640+
getLatestSubagentRunByChildSessionKey:
641+
subagentRegistryMocks.getLatestSubagentRunByChildSessionKey,
642+
listSubagentRunsForController: subagentRegistryMocks.listSubagentRunsForRequester,
643+
markSubagentRunTerminated: subagentRegistryMocks.markSubagentRunTerminated,
644+
});
645+
});
646+
enqueueQueuedFollowupRun({ root, cfg, sessionId, sessionKey });
647+
subagentRegistryMocks.listSubagentRunsForRequester.mockReturnValueOnce([
648+
{
649+
runId: "slow-child-run",
650+
childSessionKey: childKey,
651+
requesterSessionKey: sessionKey,
652+
requesterDisplayKey: sessionKey,
653+
task: "slow child",
654+
cleanup: "keep",
655+
createdAt: Date.now(),
656+
},
657+
]);
658+
659+
const resultPromise = runStopCommand({
660+
cfg,
661+
sessionKey,
662+
from: "telegram:123",
663+
to: "telegram:123",
664+
});
665+
await persistenceStarted;
666+
667+
expect(runtimeAbortMocks.abortEmbeddedAgentRun).toHaveBeenCalledWith(sessionId);
668+
expect(runtimeAbortMocks.abortEmbeddedAgentRun).toHaveBeenCalledWith(childSessionId);
669+
expect(subagentRegistryMocks.markSubagentRunTerminated).toHaveBeenCalledWith({
670+
childSessionKey: childKey,
671+
reason: "killed",
672+
runId: "slow-child-run",
673+
});
674+
expect(getFollowupQueueDepth(sessionKey)).toBe(0);
675+
expectSessionLaneCleared(sessionKey);
676+
677+
finishPersistence?.();
678+
await expect(resultPromise).resolves.toMatchObject({
679+
aborted: true,
680+
handled: true,
681+
});
682+
});
683+
476684
it("plain-language stop on ACP-bound session triggers ACP cancel", async () => {
477685
const sessionKey = "agent:codex:acp:test-1";
478686
const sessionId = "session-123";

0 commit comments

Comments
 (0)