一次 Harness 异步 Memory Flush PR 的排查与 Review 实录

本文记录的是审查当日的代码与结论；PR 状态和代码行号可能继续变化，请以文首 GitHub 链接为准。 PR：agentscope-ai/agentscope-java#2833
Issue：#2821
标题：fix(harness): support asynchronous memory flush
作者：guslegend0510
Review：xiaoyuuuuuupeng 的 Approve + 讨论评论
日期：2026-08-28

前言

这个 PR 动 8 个文件（约 +589 / -22），CI 全绿，测试覆盖也不错。但它修的是 Harness 用户很直观能感知的问题：主模型已经答完了，对话还要卡一两分钟才结束。

Issue #2821 的反馈很典型：开启长期记忆后，每次对话完成会同步卡在那里等待；用户用的是 HarnessAgent，不是旧版 ReActAgent。而 ReActAgent.longTermMemoryAsyncRecord(true) 虽然存在，但对 Harness 无效，且整段旧 API 已 @Deprecated。

本文记录：我们如何理解「两次 model 调用」、如何本地复现阻塞、PR 改了什么、Review 时发现了哪些设计 trade-off，以及与 qwen-code / gemini-cli 的差异。

1. 先搞清上下文：一次 call() 里其实有两次 LLM

用户只调了一次 agent.call()，但框架内部可能打 两次 模型：

第二次就是 Issue 里说的 flush LLM。它会把新事实追加到 memory/YYYY-MM-DD.md（日流水账），之后还有独立的 consolidation 流程合并进 MEMORY.md。

1.1 调用链（Harness）

HarnessAgent.call()
  └─ ReActAgent.buildAgentStream()     ← onAgent 中间件链
       ├─ 主 ReAct 循环（第一次 LLM）
       └─ MemoryFlushMiddleware        ← 默认 concatWith，等 flush 完成
            └─ MemoryFlushManager.flushMemories()
                 └─ model.stream()     ← 第二次 LLM（flush）

MemoryFlushMiddleware 注册在 HarnessAgent.build() 里，当 memoryModel != null && !disableMemoryHooks 时生效。

1.2 和旧版 ReActAgent 的对应关系

PR 本质上是把 ReAct 那条异步记录思路，补到 Harness 的 flush 管线上。

2. 旧代码有什么问题：文档写异步，实现是同步

2.1 实现：concatWith 阻塞响应流

修复前 MemoryFlushMiddleware.onAgent() 核心逻辑：

return next.apply(input)
        .concatWith(
                Mono.defer(() -> doFlush(agent, rc))
                        .subscribeOn(Schedulers.boundedElastic())
                        ...
                        .then(Mono.<AgentEvent>empty()));

concatWith 的含义：主响应事件发完之后，还要等 flush 这条 Mono 完成，整个 Flux 才算结束。call() 会等到 Flux 结束才返回。

2.2 文档：却说 flush 是 fire-and-forget

docs/v2/zh/docs/harness/memory.md 的 「Flush 的三个触发点」 一节写着：

Flush 和 offload 都是异步执行的：它们在响应流结束后通过 doOnComplete 以 fire-and-forget 方式启动，不会阻塞当前 call() 的返回。

类注释也写「Runs in onAgent’s doOnComplete」——但默认路径明明是 concatWith。

文档与代码不一致，这是排查时的第一个 red flag，也是 Review 时建议作者顺手改文档的原因。

2.3 Issue #2821 的用户场景

• 版本 2.0.0 / 2.0.1 均有
• 使用 HarnessAgent + 长期记忆
• 对话完成后有时要等 2–3 分钟
• 期望：像旧 ReAct 一样能异步写记忆

维护者确认：Harness 2.0.1 确实没有对应的异步配置，本 PR 就是补这个能力。

3. 如何排查：我们的 Review 步骤

3.1 第一步：读 Issue + PR 描述，对齐问题定义

从 #2821 和 #2833 确认：

• 问题不是「第三方 memory 中间件」，而是 Harness 自带的 per-call flush
• 修复是 opt-in：MemoryConfig.asyncFlush(true)，默认仍为同步
• 只改 路径 1（per-call flush）；compaction / consolidation / transcript 语义不变

3.2 第二步：读 diff 核心三处

异步路径要点：

if (asyncFlush) {
    return response.transformDeferredContextual(
            (events, contextView) ->
                    events.doOnComplete(() -> startAsyncFlush(agent, rc, contextView)));
}

• doOnComplete 时 快照 AgentState.getContext()（defensive copy）
• 后台 Mono.defer(() -> doFlush(rc, messages)).subscribeOn(ASYNC_FLUSH_SCHEDULER).subscribe()
• 失败只记 warn，不影响已完成的响应

3.3 第三步：重点检查清单

3.4 第四步：本地复现（独立 worktree，不动当前分支）

在 upstream/main 上建 worktree ../asj-pr2833-repro，写最小复现测试 MemoryFlushBlockingReproTest：

cd ../asj-pr2833-repro
mvn -pl agentscope-harness spotless:apply
mvn -pl agentscope-harness -am test -Dtest=MemoryFlushBlockingReproTest

两个用例均 通过，证实 main 上默认行为：

1. Middleware 层：下游事件已发出，流在 memory model 完成前不 complete
2. HarnessAgent.call() 层：主模型 instant 返回，flush 挂着时 call() pending；release flush 后才完成

这直接对应 Issue #2821 的用户体感。

4. PR 如何修复：asyncFlush 语义

4.1 配置方式

HarnessAgent.builder()
    ...
    .memory(MemoryConfig.builder()
        .asyncFlush(true)
        .build())
    .build();

4.2 同步 vs 异步

4.3 bounded scheduler 与丢弃

private static final Scheduler ASYNC_FLUSH_SCHEDULER =
        Schedulers.newBoundedElastic(1, 3, "memory-flush");

这与 StaticLongTermMemoryHook 的 long-term-memory-record scheduler 完全同一套路。

4.4 和 flushTrigger.throttled(...) 的区别

Review 讨论中还澄清了一个易混点：

两者可同时使用：降频 + 后台跑，减轻队列压力。

5. Review 结论：已解决 vs 待讨论

已解决（值得 Approve）

待讨论（不挡 merge）

A. 进程级 shared scheduler

static scheduler 意味着 全 JVM 共用 一条 flush 队列，不是按 Agent / workspace / user 隔离。

不同 Agent 通常 不共用 同一个 MEMORY.md（workspace + IsolationScope 已隔离文件路径），因此「串行写文件防冲突」不足以解释为什么要跨 Agent 共享队列。

更实际的风险：Agent A 的慢 flush 占用 worker，Agent B 的 flush 被 reject。

B. 异步 + 慢模型 → 可能 silently 丢记忆

开了 asyncFlush 且记忆模型慢、或短时间多轮结束时：

• 队列满 → reject → 对应轮 memory/YYYY-MM-DD.md 没写入
• 主回复已返回，用户无感
• 不是整个长期记忆系统失效，是 漏掉被拒绝的那几轮

单用户、正常节奏一般 OK；多 Agent 同进程、高频连发需警惕。

C. 文档不一致

memory.md 「When flush fires」仍写 flush 默认 fire-and-forget，与代码不符。PR 加了 Example 7，但没改旧句。

6. 与 qwen-code、gemini-cli 对照

排查过程中还对比了另外两个 coding agent 项目的长期记忆策略，帮助理解 AgentScope 这套设计的「行业位置」。

6.1 总览

6.2 qwen-code：extract 是什么、怎么调度

qwen-code 没有 AgentScope 那种「同进程再调一次 flush LLM」的路径，而是 Managed Auto-Memory：turn 结束后 fork 记忆抽取子 agent，读对话和现有 memory 文件，用工具写 topic 文件，再重建 MEMORY.md。

用户一轮结束 (client.ts)
  → runManagedAutoMemoryBackgroundTasks()     // fire-and-forget，不阻塞主回复
  → MemoryManager.scheduleExtract()
  → runExtract() → runAutoMemoryExtract()
  → runAutoMemoryExtractionByAgent()          // fork 子 agent
  → 写 ~/.qwen/projects/<project>/memory/...
  → rebuildManagedAutoMemoryIndex()           // 重建 MEMORY.md
  → scheduleDream()                           // 可选 consolidation（另有一套门槛）

核心文件：

6.2.1 调度策略（不是 Reactor 1+3 reject）

MemoryManager.scheduleExtract() 按 projectRoot 维度的状态机，而不是全局 static scheduler：

和 AgentScope 的关键差别：

• AgentScope 异步 flush：1 运行 + 3 排队，第 5 个 RejectedExecutionException → 该轮 flush 完全不执行
• qwen extract：1 运行 + 1 个可覆盖的 trailing → 高频连聊时中间几轮 不各自开 extract，但 不会无声 reject 掉「再跑一次」的机会

6.2.2 为什么说 trailing 是「合并」而不是「丢弃」

extract 带 cursor（extract-cursor.json），记录：

• sessionId
• processedOffset：上次成功处理到 history 的第几条

每次 extract 检查 history.slice(startOffset) 里是否有 新的 user 消息；子 agent 读的是 当前完整对话上下文（session cache-safe params），不是某一截固定快照。

因此 trailing 跑起来时，params 里的 history 已是 最新全量。中间轮虽然没各自启动 extract，但内容会进 trailing 那一次。

时间线示例（extract 很慢，用户 10 秒内连聊 4 轮）：

结果：没有 4 次独立 extract，也没有第 4 次被 reject；而是 #1 抽一轮 + trailing 用最新 history 补抽。配合 cursor，设计目标是 最终一致，不是每轮即时落盘。

6.2.3 与 AgentScope 异步 flush 对照

这也是为什么 Review 里说 qwen 策略 比 AgentScope 的 hard reject 温和——不是 queue 满就抛异常丢掉唯一任务，而是 「少跑几次，最后一次用最新对话补抽」。

6.3 gemini-cli：无 turn-end flush

• 长期记忆靠 agent 回合内 write_file 改 MEMORY.md / GEMINI.md（同步）
• Auto Memory 在 CLI 启动时 扫 idle session 做抽取，产出 inbox patch，需用户审核
• 没有 AgentScope 式「turn 结束再调一次模型写记忆」

6.4 对我们 Review 的启示

AgentScope 的 asyncFlush / ReAct 的 longTermMemoryAsyncRecord 属于 「turn 边界 + 同进程 LLM + 硬限流队列」 一派；qwen 是 「turn 边界 + fork 子 agent + 按 project 单飞/trailing 合并」；gemini 则 根本不在 turn 边界做自动抽取。

若业务不能接受 silently 丢记忆：

• AgentScope：要么别开 asyncFlush，要么改进 scheduler（按 isolation key 拆分、失败重试、更大队列）
• qwen：高频连聊接受「合并成 trailing 一次 extract」的延迟；极端情况下仍可能因 skip（memory_pressure、memory_tool 等）漏抽
• gemini：依赖 turn 内工具写 MD，或启动时 Auto Memory + 用户审核 inbox

7. 对话里澄清的几个概念

Review 过程中和作者/读者对齐的几个问题，一并记入：

Q：是「两次 call()」吗？

不是。 用户只调一次 call()；框架内部串了两次 model.stream()（主回复 + flush）。

Q：asyncFlush 只作用于第二次吗？

是。 只改 per-call flush 的完成语义；主 ReAct 仍同步走完。

Q：opt-in 是什么意思？

默认 asyncFlush=false，行为和现在完全一样（仍同步等 flush）。只有显式 .asyncFlush(true) 才后台跑。

Q：flush LLM 是什么？

对话结束后，单独再问一次记忆模型「这轮有什么该长期记住的」，写入日流水账；不是主回复本身。

8. FAQ

Q：不开 asyncFlush merge 这个 PR 会变快吗？

不会。 默认行为不变，仍同步等 flush。

Q：开了 asyncFlush 就一定丢记忆吗？

不会必然丢。 只有队列满（1 运行 + 3 排队，第 5 个起 reject）或进程立刻退出（close() 不 await）时才可能漏写。

Q：和 longTermMemoryAsyncRecord 完全一样吗？

思路对齐（bounded 异步 + fire-and-forget），但 存储与调度不同：

• ReAct：LongTermMemory.record() + 进程级 newBoundedElastic(1,3)
• Harness flush：MemoryFlushManager → 日流水账 + 同样 1+3 reject
• qwen extract：fork 子 agent + 每 project 单飞/trailing（见 §6.2）

Q：qwen 的 extract 为什么「不太会丢」？

不是不会丢，而是 丢法不同：中间轮合并进 trailing，用最新 history 补抽；AgentScope 异步则是队列满 hard reject，该轮 flush 完全不跑。见 §6.2.2 时间线。

Q：为什么文档写异步实现却是同步？

历史文档/注释滞后；concatWith 是 2.x harness 的实际行为。本 PR 的 Example 7 开始纠正，但旧段落仍待改。

9. 小结

这是一个 「用户痛点清晰、默认保守、opt-in 加速」 的 PR。修 #2821 的方向正确，测试到位；进程级 scheduler 与文档一致性适合作为 follow-up 继续讨论。

附录：关键文件索引