@@ -10,6 +10,7 @@ import {
1010type MemoryFlushPlanResolver,
1111} from "../../plugins/memory-state.js";
1212import type { TemplateContext } from "../templating.js";
13+import type { ReplyPayload } from "../types.js";
1314import {
1415runMemoryFlushIfNeeded,
1516runPreflightCompactionIfNeeded,
@@ -25,6 +26,8 @@ const refreshQueuedFollowupSessionMock = vi.fn();
2526const incrementCompactionCountMock = vi.fn();
2627const ensureSelectedAgentHarnessPluginMock = vi.fn();
2728const ensureMemoryFlushTargetFileMock = vi.fn();
29+const emitAgentEventMock = vi.fn();
30+const TEST_MAX_FLUSH_FAILURES = 3;
28312932function registerMemoryFlushPlanResolverForTest(resolver: MemoryFlushPlanResolver): void {
3033registerMemoryCapability("memory-core", { flushPlanResolver: resolver });
@@ -171,6 +174,7 @@ describe("runMemoryFlushIfNeeded", () => {
171174refreshQueuedFollowupSessionMock.mockReset();
172175ensureMemoryFlushTargetFileMock.mockReset().mockResolvedValue(undefined);
173176ensureSelectedAgentHarnessPluginMock.mockReset().mockResolvedValue(undefined);
177+emitAgentEventMock.mockReset();
174178incrementCompactionCountMock.mockReset().mockImplementation(async (params) => {
175179const sessionKey = String(params.sessionKey ?? "");
176180if (!sessionKey || !params.sessionStore?.[sessionKey]) {
@@ -208,6 +212,7 @@ describe("runMemoryFlushIfNeeded", () => {
208212incrementCompactionCount: incrementCompactionCountMock as never,
209213ensureSelectedAgentHarnessPlugin: ensureSelectedAgentHarnessPluginMock as never,
210214registerAgentRunContext: vi.fn() as never,
215+emitAgentEvent: emitAgentEventMock as never,
211216randomUUID: () => "00000000-0000-0000-0000-000000000001",
212217now: () => 1_700_000_000_000,
213218});
@@ -489,6 +494,236 @@ describe("runMemoryFlushIfNeeded", () => {
489494expect(visibleErrorPayloads).toEqual([]);
490495});
491496497+it("increments memoryFlushFailureCount on non-abort flush failure", async () => {
498+const storePath = path.join(rootDir, "sessions.json");
499+const sessionEntry: SessionEntry = {
500+sessionId: "session",
501+updatedAt: Date.now(),
502+totalTokens: 80_000,
503+compactionCount: 1,
504+};
505+await writeTestSessionStore(storePath, "main", sessionEntry);
506+runWithModelFallbackMock.mockRejectedValueOnce(new Error("provider crashed during flush"));
507+508+await runMemoryFlushIfNeeded({
509+cfg: { agents: { defaults: { compaction: { memoryFlush: {} } } } },
510+followupRun: createTestFollowupRun(),
511+sessionCtx: { Provider: "whatsapp" } as unknown as TemplateContext,
512+defaultModel: "anthropic/claude-opus-4-7",
513+agentCfgContextTokens: 100_000,
514+resolvedVerboseLevel: "off",
515+ sessionEntry,
516+sessionStore: { main: sessionEntry },
517+sessionKey: "main",
518+ storePath,
519+isHeartbeat: false,
520+replyOperation: createReplyOperation(),
521+});
522+523+const persisted = JSON.parse(await fs.readFile(storePath, "utf8")) as { main: SessionEntry };
524+expect(persisted.main.memoryFlushFailureCount).toBe(1);
525+expect(persisted.main.memoryFlushLastFailedAt).toBe(1_700_000_000_000);
526+expect(persisted.main.memoryFlushLastFailureError).toContain("provider crashed during flush");
527+expect(emitAgentEventMock).toHaveBeenCalledWith(
528+expect.objectContaining({
529+stream: "lifecycle",
530+data: expect.objectContaining({
531+phase: "memory_flush_failed",
532+attempt: 1,
533+maxAttempts: TEST_MAX_FLUSH_FAILURES,
534+}),
535+}),
536+);
537+});
538+539+it("does not track failure on abort error", async () => {
540+const storePath = path.join(rootDir, "sessions.json");
541+const sessionEntry: SessionEntry = {
542+sessionId: "session",
543+updatedAt: Date.now(),
544+totalTokens: 80_000,
545+compactionCount: 1,
546+memoryFlushFailureCount: 0,
547+};
548+await writeTestSessionStore(storePath, "main", sessionEntry);
549+const abortErr = new Error("operation aborted by user");
550+abortErr.name = "AbortError";
551+runWithModelFallbackMock.mockRejectedValueOnce(abortErr);
552+553+await runMemoryFlushIfNeeded({
554+cfg: { agents: { defaults: { compaction: { memoryFlush: {} } } } },
555+followupRun: createTestFollowupRun(),
556+sessionCtx: { Provider: "whatsapp" } as unknown as TemplateContext,
557+defaultModel: "anthropic/claude-opus-4-7",
558+agentCfgContextTokens: 100_000,
559+resolvedVerboseLevel: "off",
560+ sessionEntry,
561+sessionStore: { main: sessionEntry },
562+sessionKey: "main",
563+ storePath,
564+isHeartbeat: false,
565+replyOperation: createReplyOperation(),
566+});
567+568+const persisted = JSON.parse(await fs.readFile(storePath, "utf8")) as { main: SessionEntry };
569+expect(persisted.main.memoryFlushFailureCount).toBe(0);
570+expect(persisted.main.memoryFlushLastFailedAt).toBeUndefined();
571+expect(persisted.main.memoryFlushLastFailureError).toBeUndefined();
572+});
573+574+it("clears failure counters on successful flush", async () => {
575+const storePath = path.join(rootDir, "sessions.json");
576+const sessionEntry: SessionEntry = {
577+sessionId: "session",
578+updatedAt: Date.now(),
579+totalTokens: 80_000,
580+compactionCount: 1,
581+memoryFlushFailureCount: 2,
582+memoryFlushLastFailedAt: 1_699_999_999_000,
583+memoryFlushLastFailureError: "provider crashed during flush",
584+};
585+await writeTestSessionStore(storePath, "main", sessionEntry);
586+587+await runMemoryFlushIfNeeded({
588+cfg: { agents: { defaults: { compaction: { memoryFlush: {} } } } },
589+followupRun: createTestFollowupRun(),
590+sessionCtx: { Provider: "whatsapp" } as unknown as TemplateContext,
591+defaultModel: "anthropic/claude-opus-4-7",
592+agentCfgContextTokens: 100_000,
593+resolvedVerboseLevel: "off",
594+ sessionEntry,
595+sessionStore: { main: sessionEntry },
596+sessionKey: "main",
597+ storePath,
598+isHeartbeat: false,
599+replyOperation: createReplyOperation(),
600+});
601+602+const persisted = JSON.parse(await fs.readFile(storePath, "utf8")) as { main: SessionEntry };
603+expect(persisted.main.memoryFlushFailureCount).toBe(0);
604+expect(persisted.main.memoryFlushLastFailedAt).toBeUndefined();
605+expect(persisted.main.memoryFlushLastFailureError).toBeUndefined();
606+});
607+608+it("marks flush as completed after MAX_FLUSH_FAILURES to break retry loop", async () => {
609+const storePath = path.join(rootDir, "sessions.json");
610+const sessionEntry: SessionEntry = {
611+sessionId: "session",
612+updatedAt: Date.now(),
613+totalTokens: 80_000,
614+compactionCount: 1,
615+memoryFlushFailureCount: TEST_MAX_FLUSH_FAILURES - 1,
616+};
617+await writeTestSessionStore(storePath, "main", sessionEntry);
618+runWithModelFallbackMock.mockRejectedValueOnce(new Error("provider crashed during flush"));
619+620+const visibleErrorPayloads: ReplyPayload[] = [];
621+await runMemoryFlushIfNeeded({
622+cfg: { agents: { defaults: { compaction: { memoryFlush: {} } } } },
623+followupRun: createTestFollowupRun(),
624+sessionCtx: { Provider: "whatsapp" } as unknown as TemplateContext,
625+defaultModel: "anthropic/claude-opus-4-7",
626+agentCfgContextTokens: 100_000,
627+resolvedVerboseLevel: "off",
628+ sessionEntry,
629+sessionStore: { main: sessionEntry },
630+sessionKey: "main",
631+ storePath,
632+isHeartbeat: false,
633+replyOperation: createReplyOperation(),
634+onVisibleErrorPayloads: (payloads) => {
635+visibleErrorPayloads.push(...payloads);
636+},
637+});
638+639+const persisted = JSON.parse(await fs.readFile(storePath, "utf8")) as { main: SessionEntry };
640+expect(persisted.main.memoryFlushCompactionCount).toBe(1);
641+expect(persisted.main.memoryFlushFailureCount).toBe(TEST_MAX_FLUSH_FAILURES);
642+expect(emitAgentEventMock).toHaveBeenCalledWith(
643+expect.objectContaining({
644+stream: "lifecycle",
645+data: expect.objectContaining({
646+phase: "memory_flush_exhausted",
647+attempt: TEST_MAX_FLUSH_FAILURES,
648+maxAttempts: TEST_MAX_FLUSH_FAILURES,
649+}),
650+}),
651+);
652+expect(visibleErrorPayloads[0]).toEqual(
653+expect.objectContaining({
654+text: expect.stringContaining("skipping for this cycle"),
655+isError: true,
656+}),
657+);
658+});
659+660+it("retries flush on subsequent messages until MAX_FLUSH_FAILURES", async () => {
661+const storePath = path.join(rootDir, "sessions.json");
662+const sessionEntry: SessionEntry = {
663+sessionId: "session",
664+updatedAt: Date.now(),
665+totalTokens: 80_000,
666+compactionCount: 1,
667+};
668+await writeTestSessionStore(storePath, "main", sessionEntry);
669+runWithModelFallbackMock.mockRejectedValue(new Error("provider crashed during flush"));
670+671+const params = {
672+cfg: { agents: { defaults: { compaction: { memoryFlush: {} } } } },
673+followupRun: createTestFollowupRun(),
674+sessionCtx: { Provider: "whatsapp" } as unknown as TemplateContext,
675+defaultModel: "anthropic/claude-opus-4-7",
676+agentCfgContextTokens: 100_000,
677+resolvedVerboseLevel: "off" as const,
678+ sessionEntry,
679+sessionStore: { main: sessionEntry },
680+sessionKey: "main",
681+ storePath,
682+isHeartbeat: false,
683+replyOperation: createReplyOperation(),
684+};
685+686+await runMemoryFlushIfNeeded(params);
687+await runMemoryFlushIfNeeded({ ...params, replyOperation: createReplyOperation() });
688+689+expect(runWithModelFallbackMock).toHaveBeenCalledTimes(2);
690+691+const persisted = JSON.parse(await fs.readFile(storePath, "utf8")) as { main: SessionEntry };
692+expect(persisted.main.memoryFlushFailureCount).toBe(2);
693+});
694+695+it("next message retries flush after failure", async () => {
696+const storePath = path.join(rootDir, "sessions.json");
697+const sessionEntry: SessionEntry = {
698+sessionId: "session",
699+updatedAt: Date.now(),
700+totalTokens: 80_000,
701+compactionCount: 1,
702+};
703+await writeTestSessionStore(storePath, "main", sessionEntry);
704+runWithModelFallbackMock.mockRejectedValueOnce(new Error("provider crashed during flush"));
705+706+const params = {
707+cfg: { agents: { defaults: { compaction: { memoryFlush: {} } } } },
708+followupRun: createTestFollowupRun(),
709+sessionCtx: { Provider: "whatsapp" } as unknown as TemplateContext,
710+defaultModel: "anthropic/claude-opus-4-7",
711+agentCfgContextTokens: 100_000,
712+resolvedVerboseLevel: "off" as const,
713+ sessionEntry,
714+sessionStore: { main: sessionEntry },
715+sessionKey: "main",
716+ storePath,
717+isHeartbeat: false,
718+replyOperation: createReplyOperation(),
719+};
720+721+await runMemoryFlushIfNeeded(params);
722+await runMemoryFlushIfNeeded({ ...params, replyOperation: createReplyOperation() });
723+724+expect(runWithModelFallbackMock).toHaveBeenCalledTimes(2);
725+});
726+492727it("runs memory flush on the configured maintenance model without active fallbacks", async () => {
493728registerMemoryFlushPlanResolverForTest(() => ({
494729softThresholdTokens: 4_000,
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。