Merge pull request #621 from deepseek-harness/fix-webplugins-watch-flake

fix(client-hmr): close construction-time bundle watch blind window

.agents/notes/implemented/architecture/2026-07-23-client-plugin-loading-model.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write
-2026-07-23-client-plugin-loading-model.md: 9f8b69739213b9bdc52e4b4de4d663419e596c66
-2026-07-23-client-plugin-loading-model.zh.md: 05a78fbba9859378178720f012af462382b3ab0f
+2026-07-23-client-plugin-loading-model.md: 3513e026785fc366455bb32bf788a3a098275bb1
+2026-07-23-client-plugin-loading-model.zh.md: f31a1b076a5d93db44c67463730b38283d44ff7f

.agents/notes/implemented/architecture/2026-07-23-client-plugin-loading-model.md

@@ -76,7 +76,7 @@ Why is the roster yml rows and not a scan? Because which plugins compose into a
 
 Whether hot reload is active is a composition decision: dev compositions mount the `client-hmr` row (a normal plugin package, appended by `--dev`) whose node half brings the bundle watch and the SSE channel; prod compositions mount nothing and have neither.
 
-How does a rebuilt bundle become a reload signal? The hmr node half observes it itself — no builder tells it. It reads the graph's bundle paths from `ctx.clientModuleHost.clientPath(id)` and stat-polls each with `fs.watchFile`, following graph membership through `onGraphChanged` (rows added late in the boot window get watches; vanished rows drop them; all lifecycles ride `ctx.effect`). Polling is by design: inotify does not fire on the weka network mount, the same reason the build-side watcher needs `--poll`. On a mtime/size change it calls `clientModuleHost.rebuilt(id)` — the single re-hash entry point — and when the `rev` actually changed, broadcasts a `rebuilt` frame on `GET /plugins/events` — a system SSE channel that sends the full graph on connect and `rebuilt` frames on change, presentation-only wire that never enters the session log. The poll interval is a validated config field (default 500ms), not a constant. Rebuilding the bundles is any tsdown watch process's business — `scripts/dev-web.ts` remains as the watch-build entry point, its package list dshClient-discovered by scanning `packages/*/*/package.json` at startup — and builder and host share zero protocol. A torn read of a half-written bundle self-heals: the stats keep changing while the write completes, so the next poll tick re-hashes again and broadcasts the final rev.
+How does a rebuilt bundle become a reload signal? The hmr node half observes it itself — no builder tells it. It reads bundle paths from `ctx.clientModuleHost.clientPath(id)`, and one HMR-owned interval stat-polls every current graph row. Adding a row is ordered as synchronous stat baseline, then immediate `clientModuleHost.rebuilt(id)`: a write after the module host's graph hash but before that baseline is caught by the immediate re-hash, while a write after the baseline leaves a stat delta for the next poll. This avoids `fs.watchFile`, whose asynchronous first baseline can silently absorb a construction-time rebuild. Watch membership follows `onGraphChanged`; vanished rows drop out, and a bundle missing at poll time keeps its row dirty so reappearance forces a re-hash even with identical metadata. On a mtime/size delta or dirty row, `clientModuleHost.rebuilt(id)` is the single re-hash entry point; when the `rev` actually changed, the node half broadcasts a `rebuilt` frame on `GET /plugins/events` — a system SSE channel that sends the full graph on connect and `rebuilt` frames on change, presentation-only wire that never enters the session log. Polling is deliberate because inotify does not fire on the weka network mount, the same reason the build-side watcher needs `--poll`; the interval is a validated config field (default 500ms), and disposal clears the one timer. Rebuilding bundles is any tsdown watch process's business — `scripts/dev-web.ts` remains the watch-build entry point, its package list dshClient-discovered by scanning `packages/*/*/package.json` at startup — and builder and host share zero protocol. A torn read self-heals: stats keep changing while the write completes, so the next poll re-hashes and broadcasts the final rev.
 
 On the browser side, the driver reloads one plugin per frame, serialized:
 

.agents/notes/implemented/architecture/2026-07-23-client-plugin-loading-model.zh.md

@@ -76,7 +76,7 @@ vendored Loader 经其 `internal` seam 消费模块系统——唯一调用点
 
 热重载是否启用是一项组合决策：dev 组合挂载 `client-hmr` 行（一个常规的插件包，由 `--dev` 追加），其 node 半带来 bundle 监视与 SSE 通道；prod 组合不挂载，两者皆无。
 
-重建好的 bundle 怎么变成重载信号？hmr 的 node 半自己观察——没有构建器来通知它。它从 `ctx.clientModuleHost.clientPath(id)` 读取图上各行的 bundle 路径并用 `fs.watchFile` 逐一 stat 轮询，监视集合的成员随 `onGraphChanged` 走（boot 窗口内晚到的行补上监视、消失的行撤下监视，生命周期全部收 `ctx.effect`）。轮询是刻意选择：inotify 在 weka 网络挂载上不触发，构建侧监视器需要 `--poll` 也是同一原因。mtime/size 一变，它调用 `clientModuleHost.rebuilt(id)`——重哈希的唯一入口；当 `rev` 真的变了，才在 `GET /plugins/events` 上广播 `rebuilt` 帧——这是一条系统级 SSE（Server-Sent Events）通道，连接即发全量图，变更时发 `rebuilt` 帧，仅供呈现的 wire，永不进会话日志。轮询间隔是一个经校验的配置字段（默认 500ms），不是常量。重建 bundle 则是任意一个 tsdown watch 进程的事——`scripts/dev-web.ts` 仍作为 watch 构建入口保留，其包清单在启动时扫描 `packages/*/*/package.json` 按 dshClient 发现——构建器与 host 共享零协议。写一半的 bundle 被撕裂读取会自愈：写入完成期间 stat 持续变化，下一个轮询节拍会再次重哈希并广播最终的 rev。
+重建好的 bundle 怎么变成重载信号？hmr 的 node 半自己观察——没有构建器来通知它。它从 `ctx.clientModuleHost.clientPath(id)` 读取图上各行的 bundle 路径，由 HMR 自持的单个定时器对当前图上的每一行做 stat 轮询。新增图行时，顺序固定为先同步取得 stat 基线，再立即调用 `clientModuleHost.rebuilt(id)`：在模块 host 算出图哈希之后、取得基线之前发生的写入会被这次立即重哈希捕获；取得基线之后发生的写入则会留下 stat 差异，供下一次轮询捕获。这避开了 `fs.watchFile`：它以异步首次 stat 建立基线，可能把构造期间的重建静默吸收进基线。监视集合的成员随 `onGraphChanged` 更新；消失的行撤下监视，轮询时缺失的 bundle 则让对应行保持标脏状态，文件重现时即使元数据相同也强制重哈希。mtime/size 变化或行处于标脏状态时，`clientModuleHost.rebuilt(id)` 是重哈希的唯一入口；当 `rev` 真的变了，node 半才在 `GET /plugins/events` 上广播 `rebuilt` 帧——这是一条系统级 SSE（Server-Sent Events）通道，连接即发全量图，变更时发 `rebuilt` 帧，仅供呈现的 wire，永不进会话日志。轮询是刻意选择：inotify 在 weka 网络挂载上不触发，构建侧监视器需要 `--poll` 也是同一原因；轮询间隔是一个经校验的配置字段（默认 500ms），dispose（资源释放）会清掉那一个定时器。重建 bundle 则是任意一个 tsdown watch 进程的事——`scripts/dev-web.ts` 仍作为 watch 构建入口保留，其包清单在启动时扫描 `packages/*/*/package.json` 按 dshClient 发现——构建器与 host 共享零协议。写一半的 bundle 被撕裂读取会自愈：写入完成期间 stat 持续变化，下一个轮询节拍会再次重哈希并广播最终的 rev。
 
 浏览器侧，驱动插件每帧重载一个插件，串行执行：
 

docs/config-catalog.md

@@ -285,7 +285,7 @@ export interface Config {
 }
 ```
 
-Source: [`packages/client/hmr/src/index.ts:30`](../packages/client/hmr/src/index.ts)
+Source: [`packages/client/hmr/src/index.ts:29`](../packages/client/hmr/src/index.ts)
 
 ## `@deepseek-ai/dsh-code-runtime-worker`
 

packages/client/hmr/README.md

@@ -2,7 +2,7 @@
 
 Hot reload for fetch-arrival client plugins. A static-arrival entry composed only into `--dev` graphs (`dsh web --dev`); production graphs omit the row, so the shell-bundled code stays inert.
 
-The plugin subscribes to the webserver's system SSE channel (`GET /plugins/events`) and reloads one plugin per `rebuilt` frame, serialized through a queue (the bundle handoff slot is single). The sequence per frame — `prefetch` (fetch the new bundle before touching anything), `invalidate`, `registry.delete` (before the fiber: a bare fiber dispose trips the vendored Loader's self-dispose branch, which would mark the entry disabled), drain the old fiber, delete `entry.fiber`, remove owned `<style data-plugin>` tags, `entry.refresh()` re-imports and remounts, `fiber.await()` rethrows startup failures loud. Dependents reload through cordis itself: a fiber's activation epoch strings its service providers' uids, so replacing a provider's fiber cascades every dependent with zero client-side graph analysis. Rebuild detection lives on the webserver: in dev mode it stat-polls each plugin's built `lib/client.js` (`fs.watchFile`) and broadcasts the `rebuilt` frame when the bundle's rev changes, so any tsdown watch process producing the bundle triggers HMR with no builder→host channel.
+The browser half subscribes to the system SSE channel (`GET /plugins/events`) and reloads one plugin per `rebuilt` frame, serialized through a queue (the bundle handoff slot is single). The sequence per frame — `prefetch` (fetch the new bundle before touching anything), `invalidate`, `registry.delete` (before the fiber: a bare fiber dispose trips the vendored Loader's self-dispose branch, which would mark the entry disabled), drain the old fiber, delete `entry.fiber`, remove owned `<style data-plugin>` tags, `entry.refresh()` re-imports and remounts, `fiber.await()` rethrows startup failures loud. Dependents reload through cordis itself: a fiber's activation epoch strings its service providers' uids, so replacing a provider's fiber cascades every dependent with zero client-side graph analysis. The node half detects rebuilds with one interval that stat-polls each graph bundle from a synchronous baseline, immediately re-hashes after adding a row, retains missing rows as dirty, and broadcasts only real rev changes; any tsdown watch process producing the bundle therefore triggers HMR with no builder→host channel.
 
 ## Model Experience
 

packages/client/hmr/src/index.ts

@@ -1,13 +1,12 @@
 /**
- * HMR plugin, node half: the host end of the dev reload chain. Stat-polls
- * every graph row's client bundle (fs.watchFile — polling by design: network
+ * HMR plugin, node half: the host end of the dev reload chain. One interval
+ * stat-polls every graph row's client bundle (polling by design: network
  * mounts deliver no inotify events), reports content changes through
  * `clientModuleHost.rebuilt(id)`, and serves the `/plugins/events` SSE channel
  * broadcasting graph/rebuilt frames to the browser half (src/client/).
  * Dev-only row: prod compositions never mount this plugin.
  */
-import type { Stats } from 'node:fs'
-import { unwatchFile, watchFile } from 'node:fs'
+import { statSync } from 'node:fs'
 import type { ServerResponse } from 'node:http'
 import type { Context } from 'cordis'
 import z from 'schemastery'
@@ -41,6 +40,13 @@ function sseData(frame: PluginsEventFrame): string {
   return `data: ${JSON.stringify(frame)}\n\n`
 }
 
+interface WatchedBundle {
+  path: string
+  mtimeMs: number
+  size: number
+  dirty: boolean
+}
+
 /**
  * Mount the dev chain: bundle watches, rebuilt reporting, and the SSE channel.
  * @param ctx - host plugin context carrying clientModuleHost and httpServer.
@@ -50,29 +56,59 @@ export function apply(ctx: Context, config: Config): void {
   // schemastery's .default() guarantees the field is set after validation.
   const pollIntervalMs = config.pollIntervalMs as number
 
-  // --- bundle watch: one fs.watchFile stat poll per graph row -------------
-  const watched = new Map<string, { path: string; listener: (curr: Stats, prev: Stats) => void }>()
+  // --- bundle watch: one HMR-owned stat poll ------------------------------
+  const watched = new Map<string, WatchedBundle>()
+
+  const rehash = (id: string, watch: WatchedBundle, current: { mtimeMs: number; size: number }): void => {
+    try {
+      // rebuilt() re-hashes; an unchanged hash stays silent (clientModuleHost
+      // fires onRebuilt only on a real rev change).
+      ctx.clientModuleHost.rebuilt(id)
+    } catch (error) {
+      const code = (error as NodeJS.ErrnoException).code
+      if (code === 'ENOENT') {
+        watch.dirty = true
+        return
+      }
+      ctx.logger.warn(error)
+    }
+    watch.mtimeMs = current.mtimeMs
+    watch.size = current.size
+    watch.dirty = false
+  }
 
   const watchRow = (id: string, path: string): void => {
-    const listener = (curr: Stats, prev: Stats): void => {
-      // fs.watchFile fires on any stat delta (atime included); only content
-      // signals count. An all-zero curr means the file vanished mid-rebuild
-      // — the completing write fires the next tick, so skipping is safe.
-      if (curr.mtimeMs === prev.mtimeMs && curr.size === prev.size) return
-      if (curr.mtimeMs === 0) return
-      try {
-        // rebuilt() re-hashes; an unchanged hash stays silent (clientModuleHost
-        // fires onRebuilt only on a real rev change). A torn read of a
-        // half-written bundle self-heals on the next poll tick.
-        ctx.clientModuleHost.rebuilt(id)
-      } catch (error) {
-        const code = (error as NodeJS.ErrnoException).code
-        if (code === 'ENOENT') return // mid-rename window; the completed write fires the next poll tick
-        ctx.logger.warn(error)
-      }
+    let baseline: { mtimeMs: number; size: number }
+    try {
+      baseline = statSync(path)
+    } catch (error) {
+      watched.set(id, { path, mtimeMs: 0, size: 0, dirty: true })
+      if ((error as NodeJS.ErrnoException).code !== 'ENOENT') ctx.logger.warn(error)
+      return
+    }
+    const watch = { path, mtimeMs: baseline.mtimeMs, size: baseline.size, dirty: false }
+    watched.set(id, watch)
+    // The module host hashed before publishing the graph. Re-hash immediately
+    // after capturing this baseline so a write in between cannot become an
+    // already-current baseline paired with a stale graph rev.
+    rehash(id, watch, baseline)
+  }
+
+  const pollWatches = (): void => {
+    for (const [id, watch] of watched) {
+      let current: { mtimeMs: number; size: number }
+      try {
+        current = statSync(watch.path)
+      } catch (error) {
+        watch.dirty = true
+        if ((error as NodeJS.ErrnoException).code !== 'ENOENT') ctx.logger.warn(error)
+        continue
+      }
+      if (!watch.dirty && current.mtimeMs === watch.mtimeMs && current.size === watch.size) continue
+      // Stat-before-hash preserves a detectable older baseline for writes that
+      // land during hashing. Repeated stat changes heal a torn read.
+      rehash(id, watch, current)
     }
-    watchFile(path, { interval: pollIntervalMs, persistent: false }, listener)
-    watched.set(id, { path, listener })
   }
 
   // Diff the watch set against the current graph: drop watches for removed
@@ -85,7 +121,6 @@ export function apply(ctx: Context, config: Config): void {
     }
     for (const [id, watch] of watched) {
       if (rows.get(id) === watch.path) continue
-      unwatchFile(watch.path, watch.listener)
       watched.delete(id)
     }
     for (const [id, path] of rows) {
@@ -99,9 +134,11 @@ export function apply(ctx: Context, config: Config): void {
     // own row — no self-exemption, a modules/hmr rebuild rides the same chain).
     syncWatches()
     const unsubscribe = ctx.clientModuleHost.onGraphChanged(syncWatches)
+    const timer = setInterval(pollWatches, pollIntervalMs)
+    timer.unref()
     return () => {
       unsubscribe()
-      for (const { path, listener } of watched.values()) unwatchFile(path, listener)
+      clearInterval(timer)
       watched.clear()
     }
   }, 'client-hmr: bundle watches')

packages/client/hmr/tests/node-half.spec.ts

@@ -2,7 +2,7 @@
  * Node half of the HMR plugin: bundle watches follow the graph, stat changes
  * report through clientModuleHost.rebuilt, and everything dies with the fiber.
  */
-import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
+import { mkdtempSync, rmSync, statSync, unlinkSync, utimesSync, writeFileSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { Context } from 'cordis'
@@ -24,18 +24,29 @@ afterEach(() => { rmSync(dir, { recursive: true, force: true }) })
  * the service class carries private scan state a literal need not reproduce.
  */
 type FakeHost = ClientModuleHostService & { rebuiltCalls: string[]; fireGraphChanged(): void }
-function fakeClientModuleHost(rows: Map<string, string>): FakeHost {
+interface FakeHostOptions {
+  beforeGraphRead?: () => void
+  rebuilt?: (id: string) => string | undefined
+}
+
+function fakeClientModuleHost(rows: Map<string, string>, options: FakeHostOptions = {}): FakeHost {
   const graphListeners = new Set<() => void>()
   const rebuiltCalls: string[] = []
   const fake: Pick<FakeHost, 'graph' | 'clientPath' | 'rebuilt' | 'onRebuilt' | 'onGraphChanged' | 'rebuiltCalls' | 'fireGraphChanged'> = {
     rebuiltCalls,
     fireGraphChanged: () => { for (const l of graphListeners) l() },
-    graph: (): WebBootGraph => ({
-      rev: 'r',
-      entries: [...rows.keys()].map(id => ({ id, url: `/plugins/${id}/client.js?rev=r`, rev: 'r' })),
-    }),
+    graph: (): WebBootGraph => {
+      options.beforeGraphRead?.()
+      return {
+        rev: 'r',
+        entries: [...rows.keys()].map(id => ({ id, url: `/plugins/${id}/client.js?rev=r`, rev: 'r' })),
+      }
+    },
     clientPath: id => rows.get(id),
-    rebuilt: (id) => { rebuiltCalls.push(id); return 'r2' },
+    rebuilt: (id) => {
+      rebuiltCalls.push(id)
+      return options.rebuilt?.(id) ?? 'r2'
+    },
     onRebuilt: () => () => {},
     onGraphChanged: (listener) => {
       graphListeners.add(listener)
@@ -81,6 +92,8 @@ describe('hmr node half', () => {
 
     expect(routes).toHaveLength(1)
     expect(routes[0]).toMatchObject({ kind: 'exact', path: EVENTS_ENDPOINT })
+    expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a'])
+    clientModuleHost.rebuiltCalls.length = 0
 
     // Nudge mtime past stat granularity so the poller sees a content signal.
     await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
@@ -103,14 +116,89 @@ describe('hmr node half', () => {
     const rows = new Map([['pkg-early', early]])
     const clientModuleHost = fakeClientModuleHost(rows)
     const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+    clientModuleHost.rebuiltCalls.length = 0
 
     writeFileSync(late, 'v1')
     rows.set('pkg-late', late)
     clientModuleHost.fireGraphChanged()
+    expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-late'])
+    clientModuleHost.rebuiltCalls.length = 0
 
     await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
     writeFileSync(late, 'v2-longer')
     await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toContain('pkg-late') }, { timeout: 3_000 })
+
+    rows.delete('pkg-late')
+    clientModuleHost.fireGraphChanged()
+    clientModuleHost.rebuiltCalls.length = 0
+    writeFileSync(late, 'v3-even-longer')
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 3))
+    expect(clientModuleHost.rebuiltCalls).toHaveLength(0)
+    await fiber.dispose()
+  })
+
+  it('rehashes after baseline capture so a construction-window write cannot become the baseline', async () => {
+    const bundle = join(dir, 'construction.js')
+    writeFileSync(bundle, 'v1')
+    let rewrite = true
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]), {
+      beforeGraphRead: () => {
+        if (!rewrite) return
+        rewrite = false
+        // The graph carries the hash from before this write. The old
+        // fs.watchFile registration asynchronously captured the new file as
+        // its first baseline and never requested a re-hash.
+        writeFileSync(bundle, 'v2-written-during-watch-construction')
+      },
+    })
+
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+
+    expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a'])
+    clientModuleHost.rebuiltCalls.length = 0
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 3))
+    expect(clientModuleHost.rebuiltCalls).toHaveLength(0)
+    await fiber.dispose()
+  })
+
+  it('marks a vanished bundle dirty so identical metadata still re-hashes after it reappears', async () => {
+    const bundle = join(dir, 'replace.js')
+    writeFileSync(bundle, 'seed')
+    const fixedTime = new Date(1_600_000_000_000)
+    utimesSync(bundle, fixedTime, fixedTime)
+    const baseline = statSync(bundle)
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]))
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+    clientModuleHost.rebuiltCalls.length = 0
+
+    unlinkSync(bundle)
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
+    writeFileSync(bundle, 'x'.repeat(baseline.size))
+    utimesSync(bundle, fixedTime, fixedTime)
+    const restored = statSync(bundle)
+    expect({ mtimeMs: restored.mtimeMs, size: restored.size }).toEqual({
+      mtimeMs: baseline.mtimeMs,
+      size: baseline.size,
+    })
+    await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a']) }, { timeout: 3_000 })
+    await fiber.dispose()
+  })
+
+  it('retains a dirty baseline when the immediate re-hash races a rename', async () => {
+    const bundle = join(dir, 'rename.js')
+    writeFileSync(bundle, 'v1')
+    let first = true
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]), {
+      rebuilt: () => {
+        if (!first) return 'r2'
+        first = false
+        throw Object.assign(new Error('bundle renamed'), { code: 'ENOENT' })
+      },
+    })
+
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+
+    await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toEqual(['pkg-a', 'pkg-a']) }, { timeout: 3_000 })
     await fiber.dispose()
   })
 })