fix(dws): isolate direct-message ingestion across conversations - #10691
Conversation
github-actions
Bot
added
the
review/self-reported
|
Thanks for the PR! Template looks good ✓ Problem: an observed bug, not theoretical hardening. The linked open issue #10688 ( Direction: aligned. With more than one active conversation this is a real correctness problem for daemon users, and fixing it at the channel's admission boundary — while ChannelBase keeps serializing turns within a session — is the right layer. No direction concerns. Size: 257 production lines (dws-channel.ts 225, dws-client.ts 30, index.ts 2) + 319 test lines, all contained in Approach: the scope feels right — durable admission before turn completion, detached per-message dispatch, the same boundary for history pages, and backpressure instead of oldest-eviction on a full journal. Every piece is needed for the stated goal; no drive-by changes spotted. One deliberate asymmetry worth knowing: document-mention notifications still block their history window ( Risk: no elevated risk signals — none of the changed files match the high-risk paths from the revert-history analysis. Moving on to code review. 🔍 中文说明感谢贡献！ 模板完整 ✓ 问题：已观测到的 bug，不是理论性加固。关联的 open issue #10688（ 方向：对齐。对有多会话并发的 daemon 用户这是真实的正确性问题，在 channel 的接收入口修复（ChannelBase 继续保证会话内串行）是正确的层次。无方向性问题。 规模：257 行生产代码（dws-channel.ts 225、dws-client.ts 30、index.ts 2）+ 319 行测试，全部限于 方案：范围合理——先持久接收再等任务完成、按消息独立调度、历史页使用相同接收边界、持久化队列满时背压而非淘汰最旧条目。每一部分都是目标所必需，未发现顺手改动。一个有意为之的不对称：文档提及通知仍然阻塞其历史窗口（等待 风险：无升级风险信号——改动文件均未命中 revert 历史分析中的高风险路径。 进入代码审查。🔍 — Qwen Code · qwen3.8-max Reviewed at |
Code reviewReviewed the full diff at the commit in the footer; it is contained to
One suggestion, non-blocking: a live direct-message turn that fails under budget gets logged twice — once by the catch attached in One thing I verified explicitly rather than assumed: the new backpressure in sequenceDiagram
participant P1 as DWS event reader
participant P2 as line wrapper
participant P3 as admission
participant P4 as pending journal
participant P5 as scheduled task
participant P6 as session queue
P1->>P2: event line
P2->>P3: onMessage, returns dispatch handle
P3->>P4: persist entry, saveCursor
P3-->>P2: admitted resolves, reader moves on
P5->>P6: dispatch turn, in admission order
P5->>P4: remove entry on success
P5-->>P2: completion failure reported via onError
TestingUnattended CI run — the review is static and I do not build or execute PR code. The evidence below is the PR's own CI on the reviewed commit, fetched through the API at review time. The author's local results (257 DWS tests, builds, lint, typecheck on macOS) are their claim, not independently re-run.
No failures at fetch time (bot orchestration checks omitted from the table). The main unit suite — Sandboxed verification would settle what static reading cannot: Not verified: live DWS endpoint E2E — no such environment exists in CI, and the author marked it out of scope. 中文说明代码审查完整审查了页脚所指提交的全部 diff，改动限于
一个非阻断建议：预算内的私聊任务失败会被记录两次—— 有一处我特意做了验证而非假设： 测试无人值守 CI 运行——审查为静态审查，不构建或执行 PR 代码。以上证据来自 PR 自身在受审提交上的 CI，通过 API 在审查时获取。作者自报的本地结果（257 个 DWS 测试、构建、lint、typecheck，macOS）是其声明，未经独立复跑。 获取时无失败（表格省略了机器人编排类检查）。主单元测试套件 沙盒验证可以补足静态审查无法确定的部分： 未验证：真实 DWS 端点 E2E——CI 中没有该环境，作者也标记为范围外。 — Qwen Code · qwen3.8-max Reviewed at |
|
Confidence: 4/5 — solid, focused fix whose tests map onto the issue's acceptance criteria; one non-blocking log-duplication nit, and CI is the last gate. Stepping back: this PR does exactly what the issue asked for, and nothing more. I proposed the same shape independently before reading the diff — complete the reader callback after durable admission, dispatch journaled messages independently across conversations, backpressure on a full journal — and the implementation matches it. Where it deviates from the naive version, the details are right: document-mention notifications deliberately keep blocking history-window semantics (the pinned-watermark abort contract), the redelivery in-flight contract and the stale-message pull-back are preserved, and capacity waiters are resolved on disconnect with a lifecycle-generation guard. The backpressure I worried about is provably bounded — parked entries are re-driven every poll and dropped after their retry budget, so a full journal stalls new admission for at most a few poll cycles. The 257-line production delta is minimal for the goal, the existing invariant comments are respected, and the new tests map one-to-one onto the issue's acceptance criteria. The author has a high volume of open PRs this week, but judged on its own merit this one stands up. What remains is evidence uncertainty, not code uncertainty: the unit suite on the reviewed commit is still running, and "the new tests fail on base" is my static inference (the Approval deferred until CI lands green on 中文说明Confidence: 4/5 —— 扎实、聚焦的修复，测试与 issue 的验收标准一一对应；仅一个非阻断的日志重复小问题，CI 是最后的关卡。 整体来看：这个 PR 恰好做了 issue 要求的事，没有多余内容。读 diff 之前我独立提出了同样的方案——读取器回调在持久接收后即完成、按会话独立调度日志中的消息、日志满时背压——实现与之吻合。它偏离朴素版本的地方，正是细节处理正确之处：文档提及通知刻意保留阻塞的历史窗口语义（水位回拉/中止契约）、重投递在飞契约与过期消息回拉都得到保留、容量等待者在断开连接时被唤醒且有生命周期代数守卫。我担心的背压被证明有界——停泊条目每次轮询都会被重新驱动，重试预算耗尽后被丢弃，所以日志满时新接收最多停滞几个轮询周期。 257 行生产改动对目标而言是最小的，既有不变量注释得到尊重，新测试与 issue 验收标准一一对应。作者本周有大量 open PR，但按自身价值评判，这个 PR 站得住。剩下的是证据层面的不确定性，而非代码层面：受审提交上的单元测试套件仍在运行，"新测试在基线上会失败"是我的静态推断（上面指名的 批准推迟到 — Qwen Code · qwen3.8-max Reviewed at |
qwen-code-ci-bot
left a comment
qwen-code-ci-bot
left a comment
There was a problem hiding this comment.
Not explored to full depth (tool budget reached): "agent reverse-audit (round 5)": full trace of whether a steered turn's ChannelLoopSkippedError propagates through ChannelBase's sessionQueues chain to reject handleInbound (re-park of the canc….
Deferred under the convergence posture (round 2, not a blocker) — recorded, not requested in this round:
packages/channels/dws/src/dws-channel.ts:1183 — [probe] History/live admission dispatch is uncapped (probe: 60 of 60 concurrent turns from one history poll)packages/channels/dws/src/dws-channel.ts:1958 — [test] Unwitnessed capacity-waiter clear in removePendingMessage (surviving deletion mutant)packages/channels/dws/src/dws-channel.ts:1997 — [test] Unwitnessed processingMessages.delete in the document re-drive path (surviving deletion mutant)
中文说明
未探索到全部深度（达到工具调用预算）："agent reverse-audit (round 5)"：full trace of whether a steered turn's ChannelLoopSkippedError propagates through ChannelBase's sessionQueues chain to reject handleInbound (re-park of the canc…。
收敛姿态下延后（第 2 轮，非阻断）——已记录，本轮不要求修改：共 3 条（原文未翻译，列表见上方英文部分）。
— qwen3.8-max via Qwen Code /review (v0.22.3)
|
Qwen Code review did not complete successfully. The review pipeline failed before a review could be posted. A transient error is retried automatically; if you are seeing this, retry with |
The replay cap measured queuedDirectMessages.size, which in followup mode also counts the live per-conversation backlog: entries are added at schedule time and removed only when their own turn finishes, so N chained messages hold N entries while one turn runs. A single conversation's backlog of 16 could therefore keep every poll from replaying any parked failed direct message of any other conversation, and parked entries have no other redelivery surface. Count only replay-started dispatches against the cap, tracked in a map cleared on disconnect alongside queuedDirectMessages so replay-alone concurrency still stays at or below the cap across polls. Also add the missing test witnesses for the disconnect and tail cleanup introduced earlier in this branch: the conversation-tail reset on disconnect, the identity-guarded tail release in scheduleDirectMessage, and failed ambient parking when a capacity wait is released by disconnect.
qwen-code-dev-bot
left a comment
qwen-code-dev-bot
left a comment
There was a problem hiding this comment.
Re-reviewed at head 47852c6e (the round-2 CHANGES_REQUESTED was on the superseded cf2084fe).
- Both of that round's Criticals are answered by the two new commits and I read them line by line. The conversation tail is now a
{started, completed}pair so a live steer turn only serializes behind the previous turn's start (FIFO is already guaranteed at the ChannelBase layer) instead of queueing on its completion — the wedge path R2-1 described is gone, and dispose resolves both sides; the replay cap counts a dedicatedreplayDirectDispatchesmap with identity-guarded cleanup, so long-lived live entries can no longer starve parked replay (R1-1's ordering concern is covered by new regression tests for both followup and default-steer modes). - All 20 review threads are resolved. The round's three deferred items are two unwatched-cleanup test mutants and one uncapped-concurrency observation, explicitly non-blocking.
- CI: the head was pushed minutes ago and its lanes are still queued; the head the review ran on had
Test (ubuntu)green, and the increment is additive fixes with pinned tests. Per the channel convention CI is outside my gate and this call is on the review itself.
Maintainer verification — real-environment A/B on a live daemonI built a live rig for this PR rather than relying on the unit tests alone, and ran the same scenarios against two bundles produced from one tree: Rig — a stand-in 1. The reported bug reproduces on
|
| Observable | main |
this PR |
|---|---|---|
| B answered while A hangs | no | yes |
| B's reply latency after its message arrived | 30 227 ms | 120 ms |
chat message list-all polls between t=5 s and t=32 s |
1 | 6 |
The third row is a second symptom I did not expect and that is worth recording: on main the hanging turn also wedges pollOnce. The history poll at t≈6.5 s returned both msg-a1 and msg-b1, then the loop awaited the turn and no further history or mention poll ran for the next 27 s — so the history path is not a fallback for this bug, it has the same head-of-line block. This PR keeps the loop polling on its normal 5 s cadence throughout.
2. In-conversation ordering holds, with a behaviour change worth knowing about
A1 (hanging) then A2 in the same conversation, 1 s apart. On both arms A2 is never answered before A1 finishes. The mechanism differs:
main: A2 is not even read until A1 completes, then runs as a second turn — two replies (t=24138 ms, t=24218 ms).- this PR: A2 reaches
ChannelBaseimmediately and steers the active turn (steer: cancelled active turn … turn msg-a1 threw after cancellation). The restarted turn carries both messages and produces one reply covering both. Afterwards the cursor is clean:processedMessages = [conv-A\0msg-a1, conv-A\0msg-a2],pendingMessages = [],inboundFailures = []— no duplicate, no leak, nothing parked.
That is the same steering other channels already get, and DWS was the outlier only because the block made it unreachable. Still, it is a user-visible change: a second DM during a long turn now interrupts it instead of queueing behind it.
3. Durable admission does what the description claims
While both turns are still in flight, the on-disk journal already holds both messages on this PR and is empty on main. To find out what that actually buys, I SIGKILLed the whole daemon process tree mid-turn and restarted:
- with a healthy history window, both arms recover (main via
chat message list-all, this PR via its journal). So the journal is a second path, not the only one. - with history unavailable after the crash,
mainloses both messages permanently and this PR recovers both in ~2.4 s.
4. The cost side: admission is no longer rate-limited by the reader
20 direct messages across 20 conversations, all hanging upstream:
main |
this PR | |
|---|---|---|
| peak concurrent agent turns | 1 | 20 |
| daemon sessions spawned | 1 | 20 |
This is the intended consequence of the change, but it is worth stating in the risk section: DWS direct-message concurrency is now bounded only by the 5 000-entry journal, and qwen serve --max-sessions is the only operator backstop.
I probed that backstop (--max-sessions 3, 10 DMs across 10 conversations) and it behaves identically on both arms: 7 of the 10 fail with POST /session: Session limit reached (3), burn all 5 retry attempts within ~20 s, and are dropped permanently (dropping a DWS message after 5 failed turns); 3 of 10 are answered. Same counts before and after, so this is pre-existing and not caused by this PR — a transient capacity refusal should not consume the inbound retry budget, but that belongs in its own issue.
5. Suites, types, lint, mutation
packages/channels/dws: 286/286 pass.packages/channels/base: 1173/1173 pass.npm run typecheck(whole repo) andeslint packages/channels/dws/src --max-warnings=0: clean.- Merged locally with
origin/main(no conflicts,packages/channels/dwsuntouched by the merge):npm ci, dws suite 286/286, repo typecheck clean. - 9/9 targeted mutants killed against the PR's own suite, including every site an earlier review round called unwitnessed:
| Mutant | Verdict |
|---|---|
admitted := completed (revert the isolation entirely) |
killed (5 tests) |
drop queuedDirectMessages.clear() in disconnect() |
killed |
drop directConversationTails.clear() in disconnect() |
killed |
drop the capacity-waiter drain in disconnect() |
killed |
force the === task identity guard to always delete |
killed |
report a detached failure even when admission failed (dws-client.ts) |
killed (2 tests) |
| skip durable admission before scheduling | killed (9 tests) |
| drop the pending-capacity backpressure throw | killed (4 tests) |
| drop the per-conversation tail chain for live dispatch | killed (2 tests) |
The red Test (ubuntu-latest, Node 22.x) job is unrelated to this PR: hooks/hook-runner.process.test.ts, utils/shellAstParser.test.ts ("classifies adversarial rule inputs within the CPU budget" — a shared-runner CPU-budget assertion) and vscode-ide-companion's imageSupport.bundle.test.ts. None of them touch packages/channels.
Residual notes (non-blocking)
rememberFailedMessagestill awaitspendingMessageCapacityWaiterswith no timeout. It is reachable only on the failed-turn parking path forgroup/group-allsources, which are awaited by their event reader — so at 5 000 parked messages that reader can stall. It self-heals (the 5-attempt budget removes entries, anddisconnect()drains the waiters), andrememberPendingMessagecorrectly throws instead of waiting, so the earlierpollOncewedge concern is addressed. Worth a comment saying why the two parking paths differ.- At journal capacity, a live direct message is refused rather than admitted, so it falls back to history replay. Given §1 showed history polling shares the block on
main— and is healthy here — that is fine, but it is the one path where a message can still be delayed indefinitely by a full journal.
Verdict: the fix does what it says on a real daemon with real subprocesses, the regression risk the earlier review rounds chased (in-conversation ordering, cursor/journal leaks, unwitnessed guards) does not reproduce, and the suites are load-bearing. LGTM from my side.
中文说明
维护者验证 —— 真实 daemon 上的 A/B 实测
我没有只跑单测，而是为这个 PR 搭了一套真实链路验证台，用同一棵树产出的两个打包产物跑同一组场景：main（merge base cf6c9bbec2）和本 PR head（47852c6ef6）。packages/channels/dws 在 cf6c9bbec2 与今天的 origin/main 之间逐字节相同，所以"修改前"这一臂就是 main 的行为。回退 base 后再把 PR 源码恢复重新构建，产物与第一次 PR 构建逐字节一致（diff -rq 无差异），因此两臂只差那三个 dws 源文件。
验证台：PATH 上放一个假的 dws 可执行文件，实现 dws-client.ts 真正会调用的命令面（version、profile list、auth status、event consume <topic> --format compact、chat message send|reply|add-emoji|remove-emoji，以及忠实实现的 chat message list-all 历史窗口）；用 dist/cli.js 起真实的 qwen serve --port 0 --channel dwsprobe；隔离的 QWEN_HOME；一个可以把某个 turn 挂住直到放行的 OpenAI 兼容上游。每次 dws 调用都带时间戳，所以下面的时延来自出站 DWS 调用流水账，而不是往 channel 里插桩。macOS 26.6.2，Node v24.18.1。
1. 问题在 main 上确实复现，本 PR 修好了
会话 A 的模型调用挂住；1 秒后会话 B 的私聊消息到达。
| 观测项 | main |
本 PR |
|---|---|---|
| A 挂住期间 B 是否被回复 | 否 | 是 |
| B 消息到达后的回复时延 | 30 227 ms | 120 ms |
t=5s 到 t=32s 之间的 chat message list-all 次数 |
1 | 6 |
第三行是我事先没预料到、但值得记录的第二个症状：在 main 上，挂住的 turn 同时把 pollOnce 也楔死了。t≈6.5s 那次历史轮询其实已经把 msg-a1 和 msg-b1 都取回来了，但随后循环去 await 那个 turn，接下来 27 秒内再没有任何一次历史或 mention 轮询——也就是说历史补偿路径并不是这个 bug 的兜底，它有同样的队头阻塞。本 PR 下轮询全程保持 5 秒节奏。
2. 会话内顺序成立，但有一个值得知道的行为变化
同一会话内先 A1（挂住）后 A2，间隔 1 秒。两臂都不会在 A1 完成前回复 A2，但机制不同：
main：A1 完成前根本读不到 A2，之后作为第二个 turn 跑，产生两条回复（t=24138 ms、t=24218 ms）。- 本 PR：A2 立刻抵达
ChannelBase并 steer 掉当前 turn（steer: cancelled active turn … turn msg-a1 threw after cancellation），重启的 turn 同时带上两条消息，产生一条覆盖两者的回复。事后游标是干净的：processedMessages = [conv-A\0msg-a1, conv-A\0msg-a2]，pendingMessages = []，inboundFailures = []——没有重复、没有泄漏、没有残留 parked 条目。
这就是其它 channel 本来就有的 steering 行为，DWS 之前只是因为阻塞而够不到它。不过这仍是用户可见的变化：长任务进行中再发一条私聊，现在会打断它，而不是排队等它结束。
3. 持久化接收确实兑现了描述里的说法
两个 turn 都还在飞的时候，本 PR 的磁盘日志里已经有这两条消息，而 main 上是空的。为了弄清这到底买到了什么，我在 turn 进行中对整个 daemon 进程树发 SIGKILL 再重启：
- 历史窗口正常时，两臂都能恢复（main 靠
chat message list-all，本 PR 靠自己的日志）。所以日志是第二条路径，不是唯一路径。 - 崩溃后历史不可用时，
main永久丢失两条消息，本 PR 约 2.4 秒内两条全部恢复。
4. 代价一侧：接收不再受读取器限流
20 条私聊分布在 20 个会话，上游全部挂住：
main |
本 PR | |
|---|---|---|
| 峰值并发 agent turn | 1 | 20 |
| 拉起的 daemon session | 1 | 20 |
这是这次改动的预期后果，但值得写进风险一节：DWS 私聊并发现在只受 5000 条日志上限约束，qwen serve --max-sessions 是运维侧唯一的兜底。
我顺手探了这个兜底（--max-sessions 3，10 条私聊 / 10 个会话），结果两臂完全一致：10 条里有 7 条以 POST /session: Session limit reached (3) 失败，在约 20 秒内耗尽全部 5 次重试并被永久丢弃（dropping a DWS message after 5 failed turns），10 条里只答了 3 条。改动前后计数相同，所以这是既有问题，不是本 PR 引入的——临时性的容量拒绝不该消耗入站重试预算，但那应该单开 issue。
5. 测试、类型、lint、变异
packages/channels/dws：286/286 通过。packages/channels/base：1173/1173 通过。- 全仓
npm run typecheck和eslint packages/channels/dws/src --max-warnings=0：干净。 - 本地与
origin/main合并（无冲突，packages/channels/dws未被合并影响）后：npm ci、dws 套件 286/286、全仓 typecheck 干净。 - 9/9 定向变异体全部被杀，包含此前评审轮次点名"无测试见证"的每一处：
admitted := completed（整体回退隔离）、disconnect()里三处清理、=== task身份守卫、dws-client.ts的 admission 失败仍上报、跳过持久化接收、去掉容量背压抛出、去掉活跃调度的按会话尾链。
红掉的 Test (ubuntu-latest, Node 22.x) 与本 PR 无关：hooks/hook-runner.process.test.ts、utils/shellAstParser.test.ts（"classifies adversarial rule inputs within the CPU budget"，共享 runner 上的 CPU 预算断言）和 vscode-ide-companion 的 imageSupport.bundle.test.ts，都不碰 packages/channels。
遗留说明（不阻塞）
rememberFailedMessage仍然无超时地 awaitpendingMessageCapacityWaiters。它只在group/group-all源的失败 parking 路径上可达，而这些源是被各自事件读取器 await 的——所以在 5000 条 parked 的极端下，那个读取器可能停住。它能自愈（5 次预算会移除条目，disconnect()也会放行 waiter），而且rememberPendingMessage已经改成抛错而非等待，先前pollOnce楔死的顾虑已经解决。建议补一句注释说明两条 parking 路径为何不同。- 日志满时活跃私聊会被拒绝而非接收，从而回落到历史补偿。鉴于 §1 已证明 main 上历史轮询同样被阻塞（这里则是健康的），这没问题，但这是日志满时消息仍可能被无限期延后的唯一路径。
结论：在带真实子进程的真实 daemon 上，这个修复确实兑现了它的说法；此前几轮评审追查的回归风险（会话内顺序、游标/日志泄漏、无见证的守卫）都没有复现；测试也确实承重。我这边 LGTM。
Addendum — the red
|
| Failure | Status |
|---|---|
core/src/utils/shellAstParser.test.ts → expected 1192.671 to be less than 1000 |
Issue #10734, fixed on main by #10735 (ec15ba6c46, merged today 06:19 UTC) — it rescales maxClassificationCpuMs from 1000 to 4000. Not an ancestor of 47852c6ef6; this branch still has 1000. |
vscode-ide-companion/src/utils/imageSupport.bundle.test.ts → Test timed out in 5000ms |
Fixed on main by #10672 (bde667f8f8) — it raises testTimeout to 60 s on ecs-qwen-* runners for exactly this esbuild-driven bundle guard. Not an ancestor of 47852c6ef6. |
core/src/hooks/hook-runner.process.test.ts → expected undefined to be defined |
Not fixed on main; the file is byte-identical between the merge base and main. It is the wait-for-condition family catalogued in the open issue #10490 ("a different test set each run" on the shared runners). |
Locally at 47852c6ef6 on an idle host, all three pass: shellAstParser + hook-runner.process 568/568, imageSupport.bundle 2/2.
So: merging origin/main into this branch should clear two of the three reds outright, and the third is a tracked shared-runner flake that a re-run clears. A merge rather than a rebase, so the existing review threads stay anchored.
中文说明
补充说明 —— 红掉的 Test (ubuntu-latest, Node 22.x) 归因
上面的报告只说了"与本 PR 无关"，这里补上归因：三处失败里有两处在 main 上已经修好了，而这个分支落后 main 65 个提交。
run 33616236954，runner ecs-qwen-hk4-11，跑了 1 小时 37 分（该 job 正常跑完是 25–28 分钟）。三处失败都不在 packages/channels：
| 失败 | 状态 |
|---|---|
core/src/utils/shellAstParser.test.ts → expected 1192.671 to be less than 1000 |
即 issue #10734，main 上已由 #10735 修复（ec15ba6c46，今天 06:19 UTC 合入），把 maxClassificationCpuMs 从 1000 调到 4000。它不是 47852c6ef6 的祖先，本分支上仍然是 1000。 |
vscode-ide-companion/src/utils/imageSupport.bundle.test.ts → Test timed out in 5000ms |
main 上已由 #10672 修复（bde667f8f8），针对的正是这个 esbuild 打包守卫，在 ecs-qwen-* runner 上把 testTimeout 提到 60 秒。同样不是 47852c6ef6 的祖先。 |
core/src/hooks/hook-runner.process.test.ts → expected undefined to be defined |
main 上未修；该文件在 merge base 与 main 之间逐字节相同。它属于 open issue #10490 记录的"等待条件超时"那一类（共享 runner 上"每次红的测试集都不一样"）。 |
在空闲机器上以 47852c6ef6 本地跑，三个全过：shellAstParser + hook-runner.process 568/568，imageSupport.bundle 2/2。
结论：把 origin/main merge 进这个分支，三处红里有两处会直接消失，剩下那处是有记录的共享 runner 抖动，重跑即可。用 merge 而不是 rebase，以免现有的评审线程失效。
yiliang114
left a comment
yiliang114
left a comment
There was a problem hiding this comment.
Review findings on 47852c6 (local dws suite passes 286/286 at this head — the three findings below are pressure-path gaps the suite does not exercise; not approving until addressed):
P2 — rememberFailedMessage's unbounded capacity wait can wedge the shared at/group subscription pump (dws-channel.ts ~1941). A live non-direct message whose turn fails while the pending journal is full blocks in the waiter loop with no timeout; for non-direct sources the subscription handler returns the full dispatch promise to the line pump, so one failed group/at turn during journal pressure stops live event processing for every conversation on that subscription until the journal drains — the exact cross-conversation coupling this PR set out to remove. Pre-change rememberPendingMessage evicted the oldest entry and never blocked. Suggest a timeout/attempt cap on the wait, then park-and-continue (or drop under the existing budget accounting) instead of blocking the shared reader indefinitely.
P2 — a full journal aborts the entire direct-history window every poll (~1198, ~1913). rememberPendingMessage now throws on capacity exhaustion, so receiveDirectMessage(...).admitted rejects mid-window-loop and the window aborts without advancing notificationWatermark (the old per-message try/catch that let the page keep moving was removed). Live direct admission also rejects, so the live subscription error-restarts in a churn. Direct history ingestion for all conversations stalls while the journal is full; recovery is bounded only by replay draining or MAX_INBOUND_ATTEMPTS burn-off — self-healing, but the direct channel is dark in between. Suggest skip-and-continue on capacity exhaustion in the history window (the message stays unmarked, so the un-advanced watermark refetches it), logging once per poll.
P2 — the replay-cap deferral breaks the per-conversation FIFO this PR introduces (~2124). When replayDirectDispatches.size >= 16, a parked older message is skipped without creating a tail entry; a newer live message for the same conversation then dispatches first, and the skipped message re-queues behind it next poll — observable out-of-order replies whenever >=16 replay dispatches are in flight (typical after an outage with many parked messages), contradicting the stated per-session ordering guarantee. Suggest recording a lightweight tail/sequence for a cap-skipped conversation (or blocking live admission behind the oldest pending entry for that conversation).
Residual notes (not blocking): admission persists before dispatch, so there is no dispatched-but-uncheckpointed loss; re-admission dedups via hasPendingMessage/queuedDirectMessages; a tiny pre-existing crash window remains in processImMessage's success path (remove → mark → saveCursor as one write). 16 hung turns would saturate the replay cap until reconnect (requires a turn hang outside this diff).
chiga0
left a comment
chiga0
left a comment
There was a problem hiding this comment.
Deep-tier review at head 47852c6e, run locally on linux / node v24.20.0.
Verification
- DWS suite at head: 286/286 pass; package
tsc --buildclean. - Non-vacuity probes: reverting the client wrapper to await
result.completedfails 4 tests; removing the journal push fromrememberPendingMessagefails 5 — the new tests genuinely constrain the decoupling and persist-before-admit. - A default-steer probe (second DM into the same conversation while the first turn runs) shows
cancelSessioninvoked on the running turn. Not filed as a finding: the author settled this in the round-1 thread ("Steer mode intentionally remains concurrent so a later message can steer or cancel the active turn"). Note only: the Reviewer Test Plan sentence "A1 still completes before A2 starts" holds only in followup/replay mode, not default steer.
Cross-check of @yiliang114's open P2s at this head
- Y1 (unbounded capacity wait in
rememberFailedMessagewedges the shared pump) — confirmed, with scope correction. Probe: journal seeded to 5,000, a failing group-all turn parks in the capacity-waiter loop (dws-channel.ts:1935); its subscription promise stays pending — the line pump is paused for that source — until a slot is freed, and the wait has no timeout. Scope:atmessages are never parked (source.kind !== 'at',dws-channel.ts:1820), so only group/group-all subscriptions wedge; direct is decoupled. A bounded wait that then parks-and-continues (or drops under the existing budget accounting) looks like the right shape. - Y2 (full journal aborts the direct-history window) — confirmed mechanism, refuted restart churn. Admission throws at capacity (
dws-channel.ts:1913), the window loop aborts andnotificationWatermarkis held — codified by the newdoes not block or advance direct history at pending capacitytest, self-healing once replay drains, and consistent with the backpressure tradeoff #10688 asked for. But "live subscription error-restarts in a churn" does not reproduce:startImSource'sonErrorrecordslastErroronly forDwsEventProcessErrorand merely logs other errors (dws-channel.ts:1393); the subscription is never stopped or restarted, and the event-streamlineQueuecatches and resumes. Cost under pressure is stalled direct history plus stderr noise, not restart churn. - Y3 (replay-cap skip breaks per-conversation FIFO) — confirmed. Probe: 17 parked direct messages across 17 conversations; one poll dispatches 16 (cap,
dws-channel.ts:2125) and skips the 17th without creating a conversation tail; a newer live message into the skipped conversation then starts immediately while the parked one is still pending, and runs only on a later poll. That violates #10688's "Messages A1 and A2 remain ordered" under cap saturation — the post-outage backlog case this PR targets.
Verdict: not approving. Y1 and Y3 are independently confirmed at this head and remain unaddressed; Y2 is confirmed in part. Everything else I checked was clean: persist-before-advance on live/history/replay, dedup via queuedDirectMessages/hasPendingMessage, disconnect cleanup of tails/queues/waiters, identity guards on tail cleanup, watermark pullback handling, and the DwsClientLike widening (implementers returning Promise<void> remain assignable).
Scope: all 5 changed files at 47852c6e plus the ChannelBase dispatch/steer/session-queue path and the event-stream line pump. Not covered: live DWS-endpoint E2E (no environment), Windows/macOS behaviour (linux only).
Reviewed with AI assistance.
|
Released in v0.23.0. |
6 participants
What this PR does
This change separates direct-message admission from Agent-turn completion. Each accepted direct message is persisted before the shared DWS reader continues, then dispatched independently while the existing session queue preserves order within one conversation. History recovery follows the same admission boundary, and a full persistence journal now applies backpressure instead of evicting an older admitted message.
Why it's needed
Direct messages currently share one ordered event reader, and its callback waits for the complete Agent turn. A long-running request in Conversation A therefore prevents a later request in Conversation B from even entering the channel. History recovery has the same head-of-line behavior because it waits for each turn before admitting the next message and advancing the checkpoint.
Reviewer Test Plan
How to verify
Run the DWS event-stream, client, and channel unit tests. Confirm that a blocked turn in Conversation A does not prevent Conversation B from starting, A1 still completes before A2 starts, live ingestion persists both messages before either blocked turn is released, history recovery advances after durable admission, detached failures remain observable, and a full journal blocks new admission until capacity becomes available without evicting an older entry.
Evidence (Before & After)
N/A — this is a non-UI channel scheduling change covered by deterministic unit tests.
Tested on
Environment (optional)
Node.js 22; 261 focused DWS tests, the DWS package build, the full repository build, focused ESLint, and the full repository typecheck all passed locally.
Risk & Scope
Linked Issues
Fixes #10688
中文说明
本 PR 做了什么
此变更将私聊消息的接收确认与 Agent 任务完成解耦。每条通过检查的私聊消息都会先持久化，再让共享的 DWS 读取器继续读取；随后消息可独立调度，而现有会话队列继续保证同一会话内的顺序。历史补偿使用相同的接收边界；当持久化队列已满时会施加背压，不再淘汰更早已接收的消息。
为什么需要
私聊消息目前共用一个顺序事件读取器，其回调会等待整个 Agent 任务完成。因此，会话 A 中的长任务会阻止会话 B 的后续请求进入 channel。历史补偿也存在相同的队头阻塞，因为它会等待每个任务完成后才接收下一条消息并推进检查点。
Reviewer 测试计划
如何验证
运行 DWS event-stream、client 和 channel 单元测试。确认会话 A 的阻塞任务不会阻止会话 B 启动；A1 仍会在 A2 启动前完成；在释放任一阻塞任务前，实时入口已持久化两条消息；历史补偿可在持久接收后推进；异步任务失败仍可观察；持久化队列满时，新消息会等待容量释放，且不会淘汰旧条目。
证据（修改前后）
N/A —— 这是非 UI 的 channel 调度变更，由确定性单元测试覆盖。
测试平台
环境（可选）
Node.js 22；本地已通过 261 个 DWS 定向测试、DWS 包构建、全仓构建、定向 ESLint 和全仓类型检查。
风险与范围
关联 Issue
Fixes #10688