feat(web): grow real node halves in connection and hmr

connection binds the web transport: it injects httpServer + apiProxy and
registers toFetchHandler(ctx.apiProxy) under the /api prefix (the node:http
to fetch bridge moves in from the webserver, keeping the res-close disconnect
detection and drain/close backpressure waits). hmr owns dev reload: a
stat-poll watch per graph row driven by clientModuleHost.onGraphChanged,
rebuilt(id) on content change, and the /plugins/events SSE route (GET/HEAD
guarded); frame types are single-sourced in events.ts shared by both halves.

packages/client/hmr/package.json

@@ -28,15 +28,20 @@
     "immediately": true
   },
   "license": "BSD-3-Clause",
+  "dependencies": {
+    "schemastery": "^3.18.0"
+  },
   "peerDependencies": {
     "@cordisjs/plugin-loader": "^1.0.0-rc.5",
     "@deepseek-ai/dsh-client-modules": "^0.0.1",
+    "@deepseek-ai/dsh-host-webserver": "^0.0.1",
     "@deepseek-ai/dsh-invariants": "^0.0.1",
     "cordis": "^4.0.0-rc.7"
   },
   "devDependencies": {
     "@cordisjs/plugin-loader": "workspace:^",
     "@deepseek-ai/dsh-client-modules": "workspace:^",
+    "@deepseek-ai/dsh-host-webserver": "workspace:^",
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "cordis": "^4.0.0-rc.7"
   },

packages/client/hmr/src/client/index.ts

@@ -64,20 +64,11 @@
  */
 import type { Context } from 'cordis'
 import type { Entry, Loader } from '@cordisjs/plugin-loader'
-import type { WebBootGraph } from '@deepseek-ai/dsh-client-modules'
+import type { PluginsEventFrame } from '../events.ts'
+import { EVENTS_ENDPOINT } from '../events.ts'
 
-/**
- * Frames on the `GET /plugins/events` system SSE channel (owned host-side by
- * dsh-host-webserver's PluginEventFrame). Mirrored here because this is a
- * wire boundary: frames arrive as JSON text and are validated at the parse
- * point, not shared as a same-process typed seam.
- */
-export type PluginsEventFrame =
-  | { type: 'graph'; graph: WebBootGraph }
-  | { type: 'rebuilt'; id: string; rev: string }
-
-/** System SSE endpoint pushing graph/rebuilt frames (wire protocol constant). */
-export const EVENTS_ENDPOINT = '/plugins/events'
+export type { PluginsEventFrame } from '../events.ts'
+export { EVENTS_ENDPOINT } from '../events.ts'
 
 /** Cordis plugin name. */
 export const name = 'client-hmr'

packages/client/hmr/src/events.ts

@@ -0,0 +1,16 @@
+/**
+ * Wire protocol of the `/plugins/events` dev SSE channel — single source for
+ * both halves of this package. Frames still cross a wire boundary: the
+ * browser half validates them at its JSON parse point; sharing the type keeps
+ * the two ends from drifting, not from parsing.
+ */
+
+import type { WebBootGraph } from '@deepseek-ai/dsh-client-modules'
+
+/** One SSE frame: the full graph on connect, or one rebuilt bundle notice. */
+export type PluginsEventFrame =
+  | { type: 'graph'; graph: WebBootGraph }
+  | { type: 'rebuilt'; id: string; rev: string }
+
+/** System SSE endpoint pushing graph/rebuilt frames (wire protocol constant). */
+export const EVENTS_ENDPOINT = '/plugins/events'

packages/client/hmr/src/index.ts

@@ -1,9 +1,152 @@
 /**
- * HMR plugin, node half. The package IS a dshClient plugin (dev-only row in
- * the host graph): the reload driver lives in its client half in full
- * (src/client/); the empty apply exists so the plugin appears in the host
- * Loader (lifecycle governance + dshClient discovery).
+ * 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
+ * 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 type { ServerResponse } from 'node:http'
+import type { Context } from 'cordis'
+import z from 'schemastery'
+// Empty type imports carry the clientModuleHost/httpServer Context merges.
+import type {} from '@deepseek-ai/dsh-client-modules'
+import type {} from '@deepseek-ai/dsh-host-webserver'
+import type { PluginsEventFrame } from './events.ts'
+import { EVENTS_ENDPOINT } from './events.ts'
 
-/** Host plugin body — no host-side behavior for the HMR plugin. */
-export function apply(): void {}
+export type { PluginsEventFrame } from './events.ts'
+export { EVENTS_ENDPOINT } from './events.ts'
+
+/** Cordis plugin name. */
+export const name = 'client-hmr'
+
+/** Required services: the web plugin table and the route registry. */
+export const inject = ['clientModuleHost', 'httpServer']
+
+/** Plugin config, validated by the same-named schemastery schema. */
+export interface Config {
+  /** Bundle stat-poll interval in milliseconds (default 500, the build-side watcher's polling default). */
+  pollIntervalMs?: number
+}
+
+export const Config: z<Config> = z.object({
+  pollIntervalMs: z.number().step(1).min(1).default(500),
+})
+
+/** Serialize one frame as an SSE data line. */
+function sseData(frame: PluginsEventFrame): string {
+  return `data: ${JSON.stringify(frame)}\n\n`
+}
+
+/**
+ * Mount the dev chain: bundle watches, rebuilt reporting, and the SSE channel.
+ * @param ctx - host plugin context carrying clientModuleHost and httpServer.
+ * @param config - validated {@link Config}.
+ */
+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 }>()
+
+  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)
+      }
+    }
+    watchFile(path, { interval: pollIntervalMs, persistent: false }, listener)
+    watched.set(id, { path, listener })
+  }
+
+  // Diff the watch set against the current graph: drop watches for removed
+  // rows (or rows whose bundle path moved), add watches for new rows.
+  const syncWatches = (): void => {
+    const rows = new Map<string, string>()
+    for (const row of ctx.clientModuleHost.graph().entries) {
+      const path = ctx.clientModuleHost.clientPath(row.id)
+      if (path !== undefined) rows.set(row.id, path)
+    }
+    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) {
+      if (!watched.has(id)) watchRow(id, path)
+    }
+  }
+
+  ctx.effect(() => {
+    // Initial sync covers rows already in the graph; the subscription covers
+    // rows arriving later (boot-window activations, including this plugin's
+    // own row — no self-exemption, a modules/hmr rebuild rides the same chain).
+    syncWatches()
+    const unsubscribe = ctx.clientModuleHost.onGraphChanged(syncWatches)
+    return () => {
+      unsubscribe()
+      for (const { path, listener } of watched.values()) unwatchFile(path, listener)
+      watched.clear()
+    }
+  }, 'client-hmr: bundle watches')
+
+  // --- /plugins/events SSE channel ----------------------------------------
+  const connections = new Set<ServerResponse>()
+
+  const connect = (res: ServerResponse): void => {
+    res.writeHead(200, {
+      'content-type': 'text/event-stream',
+      'cache-control': 'no-cache',
+      'connection': 'keep-alive',
+    })
+    // Comment line on open so clients/proxies see a live channel even when
+    // no rebuild ever happens; EventSource frame parsing skips it naturally.
+    res.write(': connected\n\n')
+    res.write(sseData({ type: 'graph', graph: ctx.clientModuleHost.graph() }))
+    connections.add(res)
+    res.on('close', () => { connections.delete(res) })
+  }
+
+  ctx.effect(() => {
+    const disposeRoute = ctx.httpServer.register({
+      kind: 'exact',
+      path: EVENTS_ENDPOINT,
+      handler: (req, res) => {
+        // Named routes match ahead of the carrier's method gate; keep the old
+        // global 405 semantics for non-GET hits on this endpoint.
+        if (req.method !== 'GET' && req.method !== 'HEAD') {
+          res.writeHead(405)
+          res.end()
+          return
+        }
+        connect(res)
+      },
+    })
+    const unsubscribe = ctx.clientModuleHost.onRebuilt((id, rev) => {
+      const line = sseData({ type: 'rebuilt', id, rev })
+      for (const res of connections) res.write(line)
+    })
+    return () => {
+      unsubscribe()
+      disposeRoute()
+      for (const res of connections) res.destroy()
+      connections.clear()
+    }
+  }, 'client-hmr: /plugins/events channel')
+}

packages/client/hmr/src/invariant.ts

@@ -3,8 +3,7 @@
  * @module @deepseek-ai/dsh-client-hmr/invariant
  */
 
-/* jscpd:ignore-start */
-import type { Context } from 'cordis'
+import type { Context, Fiber } from 'cordis'
 import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
 
 const PACKAGE_NAME = '@deepseek-ai/dsh-client-hmr'
@@ -14,14 +13,43 @@ export const name = 'client-hmr-invariant'
 /** Service required before the companion can reserve package ownership. */
 export const inject = ['invariants']
 
+/** Live fs.watchFile pollers (this package is the composition's only stat-poll user). */
+function statWatchers(): number {
+  return process.getActiveResourcesInfo().filter(kind => kind === 'StatWatcher').length
+}
+
 /**
- * No runtime invariant: a dev-only reload driver — it consumes the loader
- * entry tree and module cache but owns no events and no cross-plugin mutable
- * state; reload correctness (dispose → style removal → re-execute ordering)
- * is observable only through the assembled browser runtime, not a host-side
- * event relation.
+ * Owned relation: every bundle stat watcher the node half starts must die
+ * with its fiber — a surviving poller would keep re-hashing bundles for a
+ * torn-down dev chain forever. Checked as a baseline delta: the StatWatcher
+ * count observed at fiber creation must be restored once disposal has drained
+ * the fiber's effects (`internal/plugin` fires at dispose start; the microtask
+ * hop lets the disposer queue its unload before `fiber.await()` joins it).
+ * SSE-connection and listener teardown live inside the same ctx.effect
+ * disposers, so the watcher count is the relation's observable proxy.
  */
-const install: InvariantInstaller = () => {}
+const install: InvariantInstaller = (ctx, fail) => {
+  const baselines = new WeakMap<Fiber, number>()
+  ctx.on('internal/plugin', (fiber) => {
+    if (fiber.name !== 'client-hmr') return
+    if (fiber.uid !== null) {
+      baselines.set(fiber, statWatchers())
+      return
+    }
+    const baseline = baselines.get(fiber)
+    if (baseline === undefined) return
+    // Returned to the emitter: emitPluginDisposed awaits-and-logs async
+    // listener failures, so a violation surfaces loudly instead of unhandled.
+    return (async () => {
+      await Promise.resolve()
+      await fiber.await()
+      const remaining = statWatchers()
+      if (remaining > baseline) {
+        fail(`client-hmr fiber disposed but ${remaining - baseline} bundle stat watcher(s) survived teardown`)
+      }
+    })()
+  }, { global: true })
+}
 
 /**
  * Register this package's invariant companion.
@@ -30,4 +58,3 @@ const install: InvariantInstaller = () => {}
  */
 export const apply = (ctx: Context): Promise<() => void> =>
   Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
-/* jscpd:ignore-end */

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

@@ -1,14 +1,116 @@
 /**
- * Node half of the HMR plugin: an empty apply placeholder (the reload driver
- * lives in the client half) whose only contract is mounting and disposing
- * cleanly in the host Loader.
+ * Node half of the HMR plugin: bundle watches follow the graph, stat changes
+ * report through clientModuleHost.rebuilt, and everything dies with the fiber.
  */
-import { describe, expect, it } from 'vitest'
-import { apply } from '@deepseek-ai/dsh-client-hmr'
+import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { Context } from 'cordis'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { WebBootGraph, ClientModuleHostService } from '@deepseek-ai/dsh-client-modules'
+import type { WebRoute, HttpServerService } from '@deepseek-ai/dsh-host-webserver'
+import { apply, Config, EVENTS_ENDPOINT, inject } from '../src/index.ts'
+
+const POLL_MS = 20
+
+let dir: string
+
+beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'dsh-hmr-')) })
+afterEach(() => { rmSync(dir, { recursive: true, force: true }) })
+
+/**
+ * Controllable clientModuleHost fake over a mutable id → bundle-path table.
+ * Structural (Pick+cast): the plugin only touches the read/notify surface;
+ * 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 {
+  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' })),
+    }),
+    clientPath: id => rows.get(id),
+    rebuilt: (id) => { rebuiltCalls.push(id); return 'r2' },
+    onRebuilt: () => () => {},
+    onGraphChanged: (listener) => {
+      graphListeners.add(listener)
+      return () => { graphListeners.delete(listener) }
+    },
+  }
+  return fake as FakeHost
+}
+
+// Structural fake: the plugin only touches register(); the service class
+// carries private state a literal cannot (and need not) reproduce.
+function fakeHttpServer(routes: WebRoute[]): HttpServerService {
+  const fake: Pick<HttpServerService, 'register' | 'tapIndex' | 'port'> = {
+    register(route) {
+      routes.push(route)
+      return () => { routes.splice(routes.indexOf(route), 1) }
+    },
+    tapIndex: () => () => {},
+    port: 0,
+  }
+  return fake as HttpServerService
+}
+
+async function mount(clientModuleHost: FakeHost, httpServer: HttpServerService) {
+  const ctx = new Context()
+  ctx.provide('clientModuleHost', clientModuleHost)
+  ctx.provide('httpServer', httpServer)
+  const fiber = ctx.plugin(
+    { inject: [...inject], Config, apply },
+    { pollIntervalMs: POLL_MS },
+  )
+  await fiber.await()
+  return fiber
+}
 
 describe('hmr node half', () => {
-  it('apply is a no-op host placeholder', () => {
-    apply()
-    expect(true).toBe(true) // reaching here without throw is the contract
+  it('watches graph bundles, reports stat changes, and unwatches on dispose', async () => {
+    const bundle = join(dir, 'a.js')
+    writeFileSync(bundle, 'v1')
+    const clientModuleHost = fakeClientModuleHost(new Map([['pkg-a', bundle]]))
+    const routes: WebRoute[] = []
+    const fiber = await mount(clientModuleHost, fakeHttpServer(routes))
+
+    expect(routes).toHaveLength(1)
+    expect(routes[0]).toMatchObject({ kind: 'exact', path: EVENTS_ENDPOINT })
+
+    // Nudge mtime past stat granularity so the poller sees a content signal.
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 2))
+    writeFileSync(bundle, 'v2-longer')
+    await vi.waitFor(() => { expect(clientModuleHost.rebuiltCalls).toContain('pkg-a') }, { timeout: 3_000 })
+
+    await fiber.dispose()
+    expect(routes).toHaveLength(0)
+    // Watcher gone: further file changes report nothing.
+    clientModuleHost.rebuiltCalls.length = 0
+    writeFileSync(bundle, 'v3-even-longer')
+    await new Promise(resolve => setTimeout(resolve, POLL_MS * 4))
+    expect(clientModuleHost.rebuiltCalls).toHaveLength(0)
+  })
+
+  it('follows graph changes: rows added after activation get watched', async () => {
+    const early = join(dir, 'early.js')
+    const late = join(dir, 'late.js')
+    writeFileSync(early, 'v1')
+    const rows = new Map([['pkg-early', early]])
+    const clientModuleHost = fakeClientModuleHost(rows)
+    const fiber = await mount(clientModuleHost, fakeHttpServer([]))
+
+    writeFileSync(late, 'v1')
+    rows.set('pkg-late', late)
+    clientModuleHost.fireGraphChanged()
+
+    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 })
+    await fiber.dispose()
   })
 })

packages/client/hmr/tsconfig.json

@@ -8,7 +8,7 @@
       "DOM",
       "DOM.Iterable"
     ],
-    "types": []
+    "types": ["node"]
   },
   "include": [
     "src"
@@ -23,6 +23,12 @@
     {
       "path": "../modules"
     },
+    {
+      "path": "../../host/webserver"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
     {
       "path": "../../support/invariants"
     }