emrg: DaemonClient per-connection delta batching (GUI multi-session rant P2 slice 3)

Agent.md

@@ -66,7 +66,7 @@ EMRG is a self-evolving AI agent architecture experiment. Python implementation,
   - Streaming chat with delta rendering (16ms batching), markdown on done (marked + DOMPurify + local highlight.js subset), tool call status cards (2000-char truncation + expand)
   - Session list/switch/new/delete + right-click rename (context menu, #423) synced with daemon; own-stream busy lock (G65); broadcast streams from other clients tagged "来自其他客户端"
   - Disconnect/reconnect: red status dot, auto daemon respawn (stale-port detection), session resume, input bar restored on disconnect (no 30s fake-timeout)
-  - Unit tests `npm test` (118: 31 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config); RESPONSE_TYPES mirror daemon protocol verified against `daemon.py`
+  - Unit tests `npm test` (124: 37 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config); RESPONSE_TYPES mirror daemon protocol verified against `daemon.py`
 - **Auto project tracking** — Automatically detects and records working directories; project-scoped sessions
 - **Rant-driven evolution** — User feedback via `/rant` drives automatic self-improvement cycles
 - **Headless GitHub auth** — Non-interactive evolution auto-extracts `GH_TOKEN` from git credential store (osxkeychain / credential helper); PR comment/LGTM queries fall back to REST API (GraphQL needs `read:org` scope)
@@ -94,7 +94,7 @@ pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.port; python -m emrg
 ```
 
 Python: `uv run pytest tests/ -v` (680) — import check: `uv run python -c "from emrg.client.app import run_client"`
-GUI: `cd emrg/gui && npm test` (118: 31 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js`
+GUI: `cd emrg/gui && npm test` (124: 37 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config) — syntax: `node --check main.js preload.js daemon_client.js renderer/js/*.js`
 CI: `uv run pytest` + GUI tests + **actionlint workflow lint** (`rhysd/actionlint@v1.7.12` gate, #444 — workflow 解析错误在 PR CI 即失败，如 `if:` secrets 上下文)
 Re-trigger: `scripts/re-trigger-ci.sh [branch]` (workflow_dispatch, #527 — 替代空 commit 重触发：Actions outage 会整段丢弃 push 事件，dispatch 走 API 路径不受影响)
 

README.cn.md

@@ -282,7 +282,7 @@ uv run python -m emrg     # 启动 TUI
 cd emrg/gui
 npm ci               # 安装依赖（生产模式可 --omit=dev）
 npm start            # 启动 GUI（自动拉起 daemon）
-npm test             # 运行 Node 测试（118 项：31 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config；集成测试在 CI 跑，本地可 npm run test:integration）
+npm test             # 运行 Node 测试（124 项：37 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config；集成测试在 CI 跑，本地可 npm run test:integration）
 ```
 
 CI 通过 GitHub Actions 自动运行测试并检查冲突标记（`.github/workflows/test.yml`）。

README.md

@@ -281,7 +281,7 @@ uv run python -m emrg     # launch TUI
 cd emrg/gui
 npm ci               # install deps (production: --omit=dev)
 npm start            # launch GUI (auto-starts daemon)
-npm test             # run Node tests (118: 31 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config; integration runs in CI, local: npm run test:integration)
+npm test             # run Node tests (124: 37 daemon_client + 5 conn-manager + 22 app-commands + 32 renderer smoke + 15 i18n + 7 integration + 3 commands + 3 build-config; integration runs in CI, local: npm run test:integration)
 ```
 
 CI runs tests and checks for conflict markers automatically via GitHub Actions (`.github/workflows/test.yml`).

emrg/gui/daemon_client.js

@@ -70,7 +70,7 @@ const RESPONSE_TYPES = {
 };
 
 class DaemonClient {
-  constructor({ projectDir = os.homedir(), logger = console, authTimeoutMs = AUTH_TIMEOUT_MS, isPackaged = false } = {}) {
+  constructor({ projectDir = os.homedir(), logger = console, authTimeoutMs = AUTH_TIMEOUT_MS, isPackaged = false, deltaBatchMs = 0 } = {}) {
     this.projectDir = projectDir;
     this.logger = logger;
     this._authTimeoutMs = authTimeoutMs; // G142 测试可注入短超时（默认 10s）
@@ -87,6 +87,12 @@ class DaemonClient {
     this._reconnectTimer = null;
     this._stopReconnect = false;
     this._spawnAttempts = 0; // 连接生命周期内 spawn 计数（成功 auth 后归零）
+    // P2 connManager（rant 2026-08-10T15:07:19）：deltaBuf 批量（G122 16ms）每连接一份。
+    // deltaBatchMs > 0 时本实例自行批量 message_delta，终态（done/error/cancelled）前
+    // 强制冲刷保序（rant 14:11 孤儿节点教训）；默认 0 = 每帧即时发（既有行为不变）。
+    this._deltaBatchMs = deltaBatchMs;
+    this._deltaBuf = [];
+    this._deltaTimer = null;
   }
 
   // ── 生命周期 ────────────────────────────────────────────
@@ -431,13 +437,28 @@ class DaemonClient {
   }
 
   close() {
+    this._flushDeltaBuf(); // 断连前冲刷残留 delta（防丢失）
     if (this.ws) {
       try { this.ws.close(); } catch { /* ignore */ }
       this.ws = null;
     }
     this.connected = false;
   }
 
+  // P2（rant 2026-08-10T15:07:19 + 14:11）：批量冲刷 delta 缓冲。
+  // 有定时器则清；有残留则按 {chunks} 形状一次性发出（与 main.js G122 同形）。
+  _flushDeltaBuf() {
+    if (this._deltaTimer) {
+      clearTimeout(this._deltaTimer);
+      this._deltaTimer = null;
+    }
+    if (this._deltaBuf.length) {
+      const chunks = this._deltaBuf;
+      this._deltaBuf = [];
+      this._emit("message_delta", { chunks });
+    }
+  }
+
   // ── 事件 ────────────────────────────────────────────────
 
   onEvent(callback) {
@@ -557,6 +578,7 @@ class DaemonClient {
       return;
     }
     if (frame.type === "cancelled") {
+      this._flushDeltaBuf(); // 终态前冲刷（rant 14:11 同源：delta 不晚于终态）
       this._emit("cancelled", frame);
       return;
     }
@@ -569,16 +591,25 @@ class DaemonClient {
       return;
     }
     if (frame.done) {
+      this._flushDeltaBuf(); // 终态前冲刷 delta：保证 delta 不晚于终态（rant 14:11）
       this._onDone(frame);
       this._emit("done", frame);
       return;
     }
     if (frame.delta) {
       this._onDelta(frame);
+      if (this._deltaBatchMs > 0) {
+        this._deltaBuf.push(frame);
+        if (!this._deltaTimer) {
+          this._deltaTimer = setTimeout(() => this._flushDeltaBuf(), this._deltaBatchMs);
+        }
+        return; // 批量模式：不即时发单帧
+      }
       this._emit("message_delta", frame);
       return;
     }
     if (frame.error) {
+      this._flushDeltaBuf(); // 终态前冲刷（rant 14:11 同源）
       this._emit("error", frame);
       return;
     }

emrg/gui/test/daemon_client.test.js

@@ -169,6 +169,96 @@ test("ensureConnected: port 文件缺失 → 拉起 daemon（spawn 参数正确
   assert.strictEqual(spawnCalls.projectDir, tmpHome);
 });
 
+test("P2 deltaBatchMs: 批量合并 message_delta，终态前冲刷保序（rant 14:11）", async () => {
+  const client = new DaemonClient({ projectDir: tmpHome, deltaBatchMs: 16 });
+  await connectClient(client);
+  const seen = [];
+  client.onEvent((type, data) => seen.push([type, data]));
+  const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
+  const rid = "req-b";
+  // 多条 delta → 不即时发（批量模式），16ms 后合并为一次 {chunks}
+  send({ request_id: rid, content: "a", done: false, delta: true });
+  send({ request_id: rid, content: "b", done: false, delta: true });
+  assert.deepStrictEqual(seen.filter(([t]) => t === "message_delta"), [],
+    "delta must not emit immediately in batch mode");
+  await new Promise((r) => setTimeout(r, 40));
+  const deltas = seen.filter(([t]) => t === "message_delta");
+  assert.strictEqual(deltas.length, 1, "deltas batched into one message_delta");
+  assert.ok(Array.isArray(deltas[0][1].chunks), "batched payload has chunks array");
+  assert.strictEqual(deltas[0][1].chunks.length, 2);
+});
+
+test("P2 deltaBatchMs: done 终态到达 → 先冲刷残留 delta 再发 done（顺序保证）", async () => {
+  const client = new DaemonClient({ projectDir: tmpHome, deltaBatchMs: 1000 }); // 定时器远未到期
+  await connectClient(client);
+  const seen = [];
+  client.onEvent((type, data) => seen.push([type, data]));
+  const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
+  const rid = "req-c";
+  send({ request_id: rid, content: "a", done: false, delta: true });
+  send({ request_id: rid, content: "a", done: true, delta: false });
+  // delta 必须出现在 done 之前（顺序保证：delta 不晚于终态）
+  const idxDelta = seen.findIndex(([t]) => t === "message_delta");
+  const idxDone = seen.findIndex(([t]) => t === "done");
+  assert.ok(idxDelta >= 0, "delta flushed before done");
+  assert.ok(idxDone > idxDelta, "done must come after flushed delta");
+  assert.strictEqual(seen[idxDelta][1].chunks.length, 1);
+});
+
+test("P2 deltaBatchMs: cancelled 终态 → 冲刷残留 delta 再发 cancelled", async () => {
+  const client = new DaemonClient({ projectDir: tmpHome, deltaBatchMs: 1000 });
+  await connectClient(client);
+  const seen = [];
+  client.onEvent((type, data) => seen.push([type, data]));
+  const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
+  send({ request_id: "req-d", content: "x", done: false, delta: true });
+  send({ type: "cancelled" });
+  const types = seen.map(([t]) => t);
+  assert.ok(types.indexOf("message_delta") < types.indexOf("cancelled"),
+    "delta must be flushed before cancelled");
+  assert.strictEqual(seen.find(([t]) => t === "message_delta")[1].chunks.length, 1);
+});
+
+test("P2 deltaBatchMs: 默认 0 = 每帧即时发（既有行为回归）", async () => {
+  const client = new DaemonClient({ projectDir: tmpHome });
+  await connectClient(client);
+  const seen = [];
+  client.onEvent((type, data) => seen.push([type, data]));
+  const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
+  send({ request_id: "req-e", content: "a", done: false, delta: true });
+  send({ request_id: "req-e", content: "b", done: false, delta: true });
+  const deltas = seen.filter(([t]) => t === "message_delta");
+  assert.strictEqual(deltas.length, 2, "default mode emits per frame");
+  assert.ok(!Array.isArray(deltas[0][1].chunks), "default payload is the frame, not chunks");
+});
+
+test("P2 deltaBatchMs: close 冲刷残留 delta", async () => {
+  const client = new DaemonClient({ projectDir: tmpHome, deltaBatchMs: 1000 });
+  await connectClient(client);
+  const seen = [];
+  client.onEvent((type, data) => seen.push([type, data]));
+  const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
+  send({ request_id: "req-f", content: "y", done: false, delta: true });
+  client.close();
+  const deltas = seen.filter(([t]) => t === "message_delta");
+  assert.strictEqual(deltas.length, 1, "close must flush pending delta");
+  assert.strictEqual(deltas[0][1].chunks.length, 1);
+});
+
+test("P2 deltaBatchMs: error 终态 → 冲刷残留 delta 再发 error", async () => {
+  const client = new DaemonClient({ projectDir: tmpHome, deltaBatchMs: 1000 });
+  await connectClient(client);
+  const seen = [];
+  client.onEvent((type, data) => seen.push([type, data]));
+  const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
+  send({ request_id: "req-g", content: "z", done: false, delta: true });
+  send({ error: "boom" });
+  const types = seen.map(([t]) => t);
+  assert.ok(types.indexOf("message_delta") < types.indexOf("error"),
+    "delta must be flushed before error");
+  assert.strictEqual(seen.find(([t]) => t === "message_delta")[1].chunks.length, 1);
+});
+
 test("P2 skipStart: port 文件缺失 → 抛错不拉起 daemon（connManager 独占 daemon 生命周期）", async () => {
   fs.rmSync(PORT_FILE(tmpHome), { force: true });
   const client = new DaemonClient({ projectDir: tmpHome });