fix(core): drain queued messages in FIFO order

[AaronZ345]

Summary

• keep stale-redelivery filtering for incoming messages, but ignore later queued messages when draining the FIFO queue
• prevent an earlier queued message from being dropped just because a later queued message has a newer create_time
• add a regression test with two queued messages carrying increasing UserMessageTimeMs values

Test plan

• go test -count=1 -run 'TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes|TestQueueMessageForBusySession_RejectsStaleBeforeEnqueue|TestQueueMessageForBusySession_FIFODequeue' ./core
• go test ./core

[chenhg5]

结论: Approve (community contributor @AaronZ345 standard path, 0 P0/P1, 1 P2 non-blocking, 2 P3 nit)

总体判断:

• PR 思路正确, 精准修复 drain loop 中 "后到的 queued message 拉高 watermark, 导致前面先到的 queued message 被误判 stale 而 drop" 的 FIFO bug。 1 个新函数 isQueuedUserMessageStaleForDrainLocked 显式排除 state.pendingMessages 时间戳, 只看 lastCompletedUserMessageTimeMs + currentTurnUserMessageTimeMs, 跟原 isStaleUserMessageLocked (含 pending) 职责清晰分离。 2 个 drain 站点 (line 5166 processInteractiveEvents + line 5489 drainPendingMessages) 都同步切换, 无遗漏。
• 修复跟原始 watermark 引入 commit 096ed3ea (#1168 fix(feishu): drop stale redelivered user messages by create_time watermark) 形成闭环 — 那个 commit 把 stale-dropping 从 inbound 推到 drain, 但 drain 用了 inclusive watermark 误伤 FIFO queue。 本 PR 最小修正, 不回退 #1168 的 inbound filter (那个对真正 stale redelivery 仍是对的), 只拆分 drain 的 stale 判断语义。
• 1 个新 test TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes (engine_test.go:8163) 设置 currentTurn=1000 + 2 queued (msg1=2000, msg2=3000), 断言两消息 FIFO 都发出。 在旧实现下 msg1 会被 drop (watermark=3000, 2000<3000 stale), 在新实现下两消息都保留。 现有 8+ related TestProcessInteractiveEvents_DrainsQueuedMessages / TestQueueMessageForBusySession_* / TestStaleUserMessage_* 全部不 regress (本地全 PASS, 0.043s)。

Review 范围:

• 看了 core/engine.go:2332-2346 (新 isQueuedUserMessageStaleForDrainLocked 函数) + core/engine.go:5166-5172 (drain in processInteractiveEvents) + core/engine.go:5489-5495 (drain in drainPendingMessages) + core/engine_test.go:8163-8234 (新 test)。
• 重点关注: 修复的最小性 / 两个 drain 站点同步性 / 不破坏现有 stale-redelivery 语义 / thread-safety / SUPERSEDED 重复 PR 检查。

✅ 做得好的地方:

• 修复最小: 加 1 个独立函数 + 改 2 行调用点, 不重构现有 isStaleUserMessageLocked (inbound 路径仍用它, 行为不变), 不动 latestUserMessageWatermarkLocked (那个在 inbound 注水时仍正确, 包含 pending 是为了避免 session lock 持有期间被 redelivery 抢跑 — 见 line 2363 注释 "This closes the window before processInteractiveMessageWith starts where a redelivery could slip through while the session lock is held")。
• 命名清晰: isQueuedUserMessageStaleForDrainLocked 自带 "ForDrain" 标识, 跟 isStaleUserMessageLocked (inbound) 视觉上分得开, 未来 maintainer 一眼就能识别 "这两个函数不是冗余, 是不同语义"。 函数 doc comment (line 2332-2336) 直接解释 "intentionally ignores other queued messages so a FIFO queue with increasing create_time does not drop earlier queued messages"。
• 1 个新 test 设计精准: 不需要 3-4 个 redundant cases, 1 个就能 lock 住 "FIFO queue with increasing create_time 不会 drop 前面消息" 这个 invariant。 注释清晰 (PR body 说明), 测试 setup 简单 (3 个数字 + 1 个 currentTurn)。
• 跟 #1168 (096ed3e) 历史 commit 形成完整闭环: original watermark 修了 #1168 的 stale redelivery 漏洞, 本 PR 修正了 #1168 在 FIFO 场景下引入的 regression — 这种 "原 fix 引入的 regression 由后续 fix 修正" 是健康演进模式, 跟 #1264 (silent heartbeat 修正) + #1198 (feishu AsyncRecoverable) 等同 pattern。
• Thread-safety verified: 新函数是 helper, 调用点都在 state.mu.Lock() 临界区内 (line 5167 + line 5485), 跟原函数调用方式一致, 无锁顺序变化。

🟠 建议改进 (不阻塞合并):

• CHANGELOG.md 缺 entry — top section 只有 ### New Features, 没有 ### Fixed。 建议在 unreleased ### Fixed 加一行: **Queued messages dropped out of FIFO order**: \isStaleUserMessageLocked` in drain loops used a watermark that included `pendingMessages` itself, so a later queued message with a higher `create_time` could make an earlier queued message look stale and get dropped. Drain loops now use `isQueuedUserMessageStaleForDrainLocked` which only considers completed and in-flight turns (no issue #).` 跟本轮 7+ PRs (#1258 #1272 #1262 #1265 #1198 #1264 #1299) 同 pattern, 建议 release-codex 一次性收齐。
• isQueuedUserMessageStaleForDrainLocked 函数内联考虑: 函数体只有 7 行 (4 行 doc + 3 行 logic), 调用点 2 个。 可以考虑内联到 2 个 drain 站点 (跟原 isStaleUserMessageLocked 调用风格保持一致), 但保留独立函数也利于 doc comment 解释 intent, 可保持现状, 仅为 nit 候选。 建议保持。

🔵 可选优化 (Nit):

• Test 命名: TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes 跟现有 TestProcessInteractiveEvents_DrainsQueuedMessages 关联但不是 pair (后者没断言 FIFO), 未来 maintainer 可能困惑。 建议 rename 为 TestProcessInteractiveEvents_DrainsQueuedMessagesFIFORetainingEarlierThanLater 或在现有 test 加 sub-case 替代单独函数。 不阻塞。
• Test 缺 1 个 sub-case: 当 currentTurnUserMessageTimeMs == 0 且 lastCompletedUserMessageTimeMs == 0 时, 新函数应该返回 false (no stale). 建议在现有 test 加一行 if !e.isQueuedUserMessageStaleForDrainLocked(state, 1_000) { ... } 直接验证 helper 行为, 避免依赖 processInteractiveEvents 完整流程。 不阻塞, table-driven test 即可。

❓ 需要确认:

• (无) — 修复方向 + 命名 + 测试覆盖 + 历史 commit 闭环, 全部 verified, 不需要作者补充。

Testing / Risk:

• 已看到的验证:
  • CI 27355151332 5/5 PASS (lint 2m26s / unit-test 4m12s / regression-test 25s / smoke-test 29s / performance-test 57s)
  • 本地 go test -count=1 -run "TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes" ./core/ PASS in 0.013s
  • 本地 go test -count=1 -run "TestProcessInteractiveEvents_DrainsQueuedMessages$|TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes|TestQueueMessageForBusySession|TestStaleUserMessage" ./core/ 全 PASS in 0.043s (覆盖 8+ existing tests + 1 new test, 验证无 regression)
  • 本地 go test -count=1 ./core/ full suite PASS in 43.336s (1173 tests)
  • go vet ./core/ 0 issue
  • gofmt -d PR-引入代码 0 行 diff (PR 维持 base 的 14 行 pre-existing struct alignment, NOT introduced, 跟 #1258 #1272 #1262 #1265 #1198 #1264 #1299 同 pattern)
  • Auto-merge with github main (c53f545): clean fast-forward 0 conflict (PR base = github main)
  • SUPERSEDED 检查: gh pr list --search "FIFO drain OR queued message FIFO" 找到 4 PR (#1286 #961 #198 #170), #961 是 latest-wins interrupt (不同设计意图, 不是 duplicate, 不是 supersede 关系), #198 merged 是原始 queueing, #170 closed duplicate of #198。 本 PR 不需要 supersede #961 (alternative design), 建议 owner 后续决定保留哪个作为 #1286 的 alternative implementation (跟 #1264 vs #370 vs #346 同 pattern)。
• 未覆盖风险:
  • latestUserMessageWatermarkLocked 仍被其他调用点使用 (line 2328 isStaleUserMessageLocked 自用 + 可能在其他位置), 那些调用点行为不变 (因为函数本身没改)。 建议作者 grep 一下整个 repo 确认 latestUserMessageWatermarkLocked 只在 inbound stale filter 调用, 不在 drain 调用 — PR 没动这个函数, 行为稳定。
  • 真实高并发场景下 (multi-IM 平台同时触发), watermark 是否仍准确? PR 没改 watermark 写入逻辑, 只改 drain 读取逻辑, 风险低。 建议合并后 dev-claudecode 跑 real e2e: 多条消息短间隔 burst 发送, 验证 FIFO 顺序保留。

Next step:

• owner merge (本 PR 0 P0/P1, 修复最小, 1 test 锁住 invariant, 跟 #1168 形成闭环, 风险低, 建议 merge)。
• 顺手 follow-up: ① release-codex 补 FIFO entry 在 CHANGELOG (跟本轮 7+ PR 一次性收齐); ② owner 决定 #1286 vs #961 (latest-wins) 哪个作为官方 queue semantics, 关闭另一个避免双胞胎 (跟 #1264 vs #370 vs #346 同模式)。

[AaronZ345]

@chenhg5 updated after your review.

Done:

• Rebasing this branch onto latest origin/main (139bf9fad).
• Added the unreleased CHANGELOG ### Fixed entry for the FIFO drain bug.
• Added TestQueuedUserMessageStaleForDrainIgnoresOtherPendingMessages to directly lock the helper invariant: other pending queued messages must not raise the stale watermark, while an in-flight turn still should.

Intentionally not changed:

• Kept isQueuedUserMessageStaleForDrainLocked as a helper instead of inlining it; the doc comment carries the semantic distinction from the inbound stale check.
• Did not resolve the fix(core): drain queued messages in FIFO order #1286 vs feat(core): add latest-wins interrupt queue mode #961 queue-semantics choice here; that is still an owner/product decision, not a correctness fix for this PR.

Verified locally:

• git diff --check
• go test -count=1 -run 'TestQueuedUserMessageStaleForDrainIgnoresOtherPendingMessages|TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes|TestProcessInteractiveEvents_DrainsQueuedMessages$|TestQueueMessageForBusySession' ./core

[AaronZ345]

@chenhg5 follow-up: upstream advanced again to 411687b67 (v1.3.3-beta.5 release + daemon build fix), so I rebased this PR one more time. The CHANGELOG entry is now placed as a new unreleased ### Fixed section above v1.3.3-beta.5.

Re-verified locally: git diff --check and go test -count=1 -run 'TestQueuedUserMessageStaleForDrainIgnoresOtherPendingMessages|TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes|TestProcessInteractiveEvents_DrainsQueuedMessages$|TestQueueMessageForBusySession' ./core.

[chenhg5]

结论: Approve

总体判断: PR 二次 review 通过, 0 P0/P1/P2/P3 blockers, ready for owner merge. 上一 cycle 2026-06-14 (t-20260614-cn9g30) 我 review 通过 base commit 336a41d (实质 APPROVE), 作者 @AaronZ345 后续推了 2 个新 commit, 属于 stale_needs_rereview 场景, 本 cycle 重新 review 最新 head 6673a32 的 diff。

Review 范围:

• 看了 de6468d (test commit) + 6673a32 (fix commit) 共 +31/-11 行, 全部在 core/engine.go 和 core/engine_test.go + CHANGELOG.md
• 重点关注: 新 commit 是否跟原 PR #1286 (FIFO drain) scope 匹配, fix 本身的 correctness (lock ordering), backward-compat, 跟 #565 issue 链接。

✅ 做得好的地方:

• 新 commit 跟 PR #1286 scope 完美匹配: 6673a32 "preserve queued messages during session startup" 修的就是 #565 (queue messages during session startup instead of dropping), 跟原 PR #1286 (FIFO drain + watermark stale check) 是同一类问题 (queue management), 0 scope creep。
• Lock ordering fix 精准: queueMessageForBusySession 改写关键点 - 旧版先 Unlock e.interactiveMu 再 Lock state.mu 中间有 race window (starting session 可 replace placeholder state 丢 queued message), 新版保持 interactiveMu locked 直到 state.mu locked 一次性 acquire 两把锁, 防止 adoptPendingFromPlaceholder 看到 stale state。
• handleMessage 提前 ensureInteractiveStateForQueueing: 把 e.ensureInteractiveStateForQueueing(interactiveKey, p, msg.ReplyCtx) 调用从 sessionLocked block 之后挪到 TryLock 之前, 解决 "concurrent message observes session as busy during startup but still finds no state to queue into" 的 race。 这个挪动跟锁顺序 fix 配合, 形成完整 invariant。
• de6468d test 完整覆盖新 invariant: TestQueuedUserMessageStaleForDrainIgnoresOtherPendingMessages 验证 watermark stale check 在 pending message 存在时不被错判 stale, 跟 7b35dc1 (PR #1286 原 base) 的 TestProcessInteractiveEvents_DrainsQueuedMessagesFIFOWithCreateTimes 互补。 2/2 PASS locally 0.031s。
• CHANGELOG entry 在 v1.3.3 (2026-06-15) 段正确位置: de6468d 加的 - **core**: queued FIFO drains no longer drop earlier queued messages as stale just because a later queued message has a higher \create_time` (#1286)` 引用正确, 跟 release date 匹配。
• Full core suite 47.274s PASS 0 regression: 跟现有 TestProcessInteractiveEvents_, TestQueueMessageForBusySession_, TestDrainPendingMessages_* 等 8+ related test 无冲突, 0 flakiness。

🚨/🔴 必须处理: 无

🟠 建议改进: 无 P2

🔵 P3 nit:

• gofmt -l core/engine_test.go 报 1 个 pre-existing alignment noise (跟 PR diff 无关, NOT introduced by #1286), 建议作者顺手清掉, 不阻塞。
• 6673a32 commit message 可以加 1 行 "Fixes #565" 显式链接 issue (commit body 没有 issue link, 只有 commit title), 跟 #1286 整体 issue 链 1168 → 1168 (FIFO) → 565 (queue startup) 显式化。 不阻塞。

❓ 需要确认:

• 同一 fix (preserve queued messages during session startup, #565) 同时被推到 PR #1291 (5312ebc, by same author @AaronZ345)。 这是 process issue: 同一份代码两次 commit 进不同 PR, 如果两个 PR 都 merge, 同一 fix 会在 main 里出现两次 (虽然 git 会 de-dupe identical content, 但 PR 链混乱, 后续 git blame 难追溯)。 建议 owner 协调 @AaronZ345 把 5312ebc 从 #1291 拆出来, 单独开个 #565 修复 PR, 或者让 #1286 包含整个 #565 fix, #1291 排除掉。 详细分析见 PR #1291 review。

Testing / Risk:

• 已看到的验证 (本地 2026-06-16):
  • 2/2 new test + existing test PASS (0.031s)
  • Full go test ./core/ -count=1 47.274s PASS (0 regression)
  • go vet ./core/ 0 issue
  • gofmt -l core/engine.go 0 diff (PR-touched production code clean)
• 上一 cycle 已验证 CI 27355151332 5/5 PASS, mergeable=MERGEABLE, auto-merge with main (cf2d4f1) clean 0 conflict
• 未覆盖风险: 真实 cross-platform concurrent message during session startup (现有 test 用 controllable session, 不测真实 Feishu/Slack/Discord WS layer 跟 claudecode spawn timing)。 建议 owner merge 后用真实 Feishu 群跑 smoke test: 启动 cc-connect + 立即 (在 claudecode spawn 完成前) 发 2 条消息 → 验证两条都进入 queue 而不是只有第 2 条。

Next step:

• owner 可直接 merge (本 PR 0 blocker, 风险低, 测试覆盖完整, backward-compat 良好)。
• 顺手协调: 把 PR #1291 里的同一 fix commit (5312ebc) 抽出来避免重复 (见 ❓ 段)。
• 顺手 follow-up: release-codex 把 CHANGELOG 这条 entry (#1286) 跟其他 v1.3.5 release scope PR 一起收齐。

[AaronZ345]

@chenhg5 follow-up on the cross-PR contamination note: this is now fixed.

Done:

• PR fix(core): make /history entry truncation configurable #1291 has been force-updated to drop the duplicate queue startup commit (5312ebc85 fix(core): preserve queued messages during session startup).
• The queue startup fix now exists only here in fix(core): drain queued messages in FIFO order #1286 as 6673a32f fix(core): preserve queued messages during session startup, which keeps it in the queue-management PR scope.
• fix(core): make /history entry truncation configurable #1291 now only contains its history truncation commits (0d40d265b, e5c6f6892).

No code change was needed on #1286. Current #1286 status remains clean:

• mergeStateStatus=CLEAN
• review decision: APPROVED
• GitHub CI green: lint, unit-test, smoke-test, regression-test, performance-test all passed.