emrg: TUI queue-injection client support (P3 of #655)

Agent.md

@@ -112,7 +112,7 @@ Community needs voiced in HN agent-UI discussions map directly to EMRG's design:
 pkill -f "emrg.server"; rm -f ~/.emrg/emrgd.port; python -m emrg
 ```
 
-Python: `uv run pytest tests/ -v` (703) — import check: `uv run python -c "from emrg.client.app import run_client"`
+Python: `uv run pytest tests/ -v` (705) — import check: `uv run python -c "from emrg.client.app import run_client"`
 GUI: `cd emrg/gui && npm test` (212: 43 daemon_client + 19 conn-manager + 22 app-commands + 91 renderer smoke + 15 i18n + 7 integration + 3 commands + 5 build-config + 7 gui-state) — 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 路径不受影响)

emrg/client/app.py

@@ -237,6 +237,11 @@ def _status_left(title: str, sid: str, model: str = "") -> str:
  _welcomed = False # show welcome message once on first connect
  _request_start: float = 0.0 # timestamp when current request started
  _elapsed_task: asyncio.Task | None = None # background timer task
+ # P1 queue-injection client side (daemon #655): messages sent while the
+ # session is busy are queued daemon-side (task_queued). Track them here so
+ # `queued_requeue` can re-send with the same request id (without re-adding
+ # chat rows) and `queued_cancelled` clears on abort/disconnect.
+ _queued_sends: list[dict] = [] # {"id", "prompt", "images"}
 
  def _short_path(p: str) -> str:
  home = os.path.expanduser("~")
@@ -303,6 +308,7 @@ async def read_server():
  nonlocal stream_buffer, status, history, chat, busy, server_id, need_new_assistant, session_id, session_title, msg_count, tool_args, _welcomed
  nonlocal current_model
  nonlocal _last_center, _elapsed_task, conn
+ nonlocal _request_start
 
  async def _reconnect():
  """Attempt reconnection — blocks until successful."""
@@ -311,6 +317,7 @@ async def _reconnect():
  if _elapsed_task is not None:
  _elapsed_task.cancel(); _elapsed_task = None
  busy = False # pending request is lost
+ _queued_sends.clear() # daemon drops the queue on disconnect (queued_cancelled)
  chat.add("system", "⏸ server connection lost — reconnecting...")
  status.update(center="reconnecting...")
  term.render()
@@ -374,6 +381,65 @@ async def _reconnect():
  term.set_title(f"{session_title or session_id} @ {project_name}")
  term.render(); continue
 
+ # P1 queue-injection client side (daemon #655): messages sent
+ # while the session is busy are queued daemon-side and injected
+ # at the next round boundary. The TUI tracks them so a
+ # queued_requeue re-sends with the same request id.
+ if data.get("type") == "task_queued":
+ # Daemon queued our task (session busy). The user row is
+ # already shown; confirm the queue position.
+ pos = data.get("position", 0)
+ chat.add("system", f"⏳ Queued (position {pos}) — will run after the current turn.")
+ chat.dirty = True; term.render()
+ continue
+
+ if data.get("type") == "steer_committed":
+ # Injected into the running turn — no longer needs requeue.
+ rid = data.get("request_id", "")
+ if rid:
+ _queued_sends[:] = [q for q in _queued_sends if q.get("id") != rid]
+ continue
+
+ if data.get("type") == "queued_requeue":
+ # Turn ended with queued messages never injected — re-send
+ # them through the normal path (the daemon lock is released
+ # now). Rows were already added at the original submit, so
+ # do NOT re-add them or double-count msg_count.
+ ids = set(data.get("request_ids", []) or [])
+ to_resend = [q for q in _queued_sends if q.get("id") in ids]
+ _queued_sends.clear()
+ if to_resend:
+ was_busy = busy
+ busy = True; need_new_assistant = True; stream_buffer = ""
+ _request_start = time.time()
+ if _elapsed_task is None:
+ _elapsed_task = asyncio.create_task(_run_elapsed_timer())
+ for i, q in enumerate(to_resend):
+ rid = await conn.send_task(
+ session_id=session_id, cwd=cwd, prompt=q["prompt"],
+ images=q.get("images"), id=q["id"],
+ )
+ # Track every re-sent message the daemon will queue:
+ # re-send #1 starts a new turn (busy=True above), so
+ # re-sends #2+ arrive while busy and get queued
+ # daemon-side (task_queued) — untracked they would be
+ # silently lost at the next queued_requeue. Also
+ # track all re-sends when a turn was already running
+ # (multi-client). steer_committed removes ids that
+ # get injected mid-turn, so the loop converges.
+ if was_busy or i > 0:
+ _queued_sends.append({"id": rid, "prompt": q["prompt"], "images": q.get("images")})
+ chat.add("system", f"→ Re-sending {len(to_resend)} queued message(s).")
+ chat.dirty = True; term.render()
+ continue
+
+ if data.get("type") == "queued_cancelled":
+ if _queued_sends:
+ _queued_sends.clear()
+ chat.add("system", "⏹ Queued message(s) cancelled.")
+ chat.dirty = True; term.render()
+ continue
+
  # Tool lifecycle: create a ToolCard on start, update on end.
  if data.get("type") == "tool_start":
  ts = ToolStart.from_dict(data)
@@ -1834,9 +1900,11 @@ def _is_image_token(s, i):
  inp.text = ""; inp.cursor = 0; inp.dirty = True; term.render()
  return True
 
- if busy:
- logger.debug("ENTER blocked by busy")
- term.render(); return True
+ # P1 queue-injection (daemon #655): sending while busy no longer
+ # blocks — the daemon queues the task (task_queued) and injects
+ # it at the next round boundary, or re-sends via queued_requeue
+ # when the turn ends. Track the send for requeue.
+ was_busy = busy
 
  busy = True; need_new_assistant = True # rant #32: force new StreamingMarkdown per response
  _request_start = time.time()
@@ -1860,8 +1928,10 @@ def _is_image_token(s, i):
  _pending_images[:] = [img for img in _pending_images if img.get("label") in inp.text]
  images = _pending_images or None
  _pending_images = []
- await conn.send_task(session_id=session_id, cwd=cwd, prompt=text,
- images=images)
+ rid = await conn.send_task(session_id=session_id, cwd=cwd, prompt=text,
+ images=images)
+ if was_busy:
+ _queued_sends.append({"id": rid, "prompt": text, "images": images})
  logger.info("task sent, prompt_len=%d chars", len(text))
  inp.text = ""; inp.cursor = 0; inp.dirty = True; term.render(); return True
  if b == 0x1B and len(data) >= 2 and data[1] in (0x0D, 0x0A):

emrg/client/daemon_manager.py

@@ -228,15 +228,20 @@ def __init__(self, ws):
  self._ws = ws
 
  async def send_task(self, session_id: str, cwd: str, prompt: str,
- images: list | None = None) -> None:
+ images: list | None = None, id: str | None = None) -> str:
  """聊天发送：TaskRequest(type="task")。images 支持 /image 粘贴图。
 
  内部 json.dumps(req.to_dict(), ensure_ascii=False) 以 str 发送（不 .encode()）。
+ `id` 显式指定请求 id（P1 queue requeue 复用原 id 以匹配 queued_requeue）；
+ 返回最终请求 id（未指定时为内部生成的 uuid）。
  """
  req = TaskRequest(session_id=session_id, cwd=cwd, prompt=prompt)
+ if id:
+ req.id = id
  if images:
  req.images = images
  await self._ws.send(json.dumps(req.to_dict(), ensure_ascii=False))
+ return req.id
 
  async def send_command(self, type_: str, **params) -> None:
  """通用命令：ping/list_*/set_*/rant/compact/... 只发不读。

emrg/server/evolution_prompt.md

@@ -426,6 +426,7 @@ When reading rants, follow these rules:
 > - Packaging gen-assets doc (#690 并行周期 doc-only：Agent.md + DEVELOPMENT.md 新增 Packaging 段——图标产物 gitignore（仅 icon.svg 提交，#688），本地安装包构建需先 `bash packaging/gen-assets.sh`（幂等；渲染优先级 rsvg-convert → Chrome headless → sips；.icns 需 macOS iconutil 否则跳过带提示）；#467/#468 宿主对称原则（CI 构建时生成 + 宿主本地自检文档化）；合并 6b8fff3) ✅
 > - stop-git.ps1 EMRG-tree snapshot kill (#692 宿主 rant 2026-08-11T19:47:44 修正 #689：`Get-CimInstance Win32_Process` 祖先回溯对**已死 daemon 的孤儿 git 进程**解析失败（查父返回 $null → 不杀 → Inno DeleteFile code 5 回归）——改**向下 BFS 快照**：step 0 杀任何进程前从 EMRG 根（emrgd.pid daemon + EMRG.exe + `python.exe -m emrg` TUI 排除 emrg.server）BFS 整棵树写 `%TEMP%\emrg-stop-pids.txt`；step 4 只杀快照集内仍占 `install\git\` 的 PID；:verify 只查快照集存活；宿主 Git Bash sh/vim 永不被碰（不在快照集）且孤儿进程生前已入快照仍被抓；R125 同族（Inno 不干涉宿主工具）；另 README 吸引力文案（rant 19:50:37）；合并 5d57d60) ✅
 > - TUI cursor-left CLEAR_TO_EOL fix + status bar reorg (#693 宿主 rants 2026-08-11T19:59:09/20:02:43：①光标左移右侧字符消失——write_frame 尾部 `row_dirty_end`+`CLEAR_TO_EOL`(\x1b[0K) 行尾清理把光标右侧未变字符整行清掉（diff 只含光标附近 2 格）；修复=**整块删除** row_dirty_end 声明/dirty_end 计算/尾部 CUP+EL；SPACER_TAIL 残影由 WIDE 字 2 列天然覆盖 + 行内收缩（prev 字符→curr 空）走正常分支写空格；review ❌ 纠偏=显式 spacer 空格写入有 off-by-one（WIDE 后光标在 x+2 非 x+1，空格落偏右移字符）→ 彻底删除 elif 块（7296269）；②状态栏重组=左段 bold magenta `title (sid[:8]) [model] [1:23] · 3 msgs · ~/proj`（模型独立 `current_model` 跟踪、耗时纯文本 [m:ss] 去 ⏱、消息数+目录走 `left_extra`）、中段 dim 仅 `id @ host`（服务端 ID + 主机名）、右段移除（`_update_right()`→`_update_left_extra()`）；/model 切换刷新左段；+8 测试 695→703（test_output +3 / test_buffer +1 / 新 test_status_line +4），GUI 212；合并 2d12ad8) ✅
+> - TUI queue-injection client side (#655 P3 follow-up，自发现：daemon P1 排队注入已 e2e 验证但 TUI 不可达——busy 时 ENTER 被静默吞掉（app.py:1837），4 个广播帧无人处理；修复=①`send_task` 增可选 `id` 参数并返回最终请求 id（重发复用原 id）；②ENTER busy 不再拦截（was_busy 捕获），发送后若当时 busy 记入 `_queued_sends`；③read_server 新增 4 帧处理——task_queued 显示 '⏳ Queued (position N)'、steer_committed 从队列移除、queued_requeue 以原 id 静默重发（**不重加 user 行/不重复 msg_count**，busy 置 True + need_new_assistant 保证响应进新 md 行 + 重启耗时计时）、queued_cancelled 清队列+提示；④断线重连清 `_queued_sends`（daemon 断连即 drop 队列）；+2 测试 703→705（send_task 显式 id 透传 / 返回生成 id），GUI 212 不变) ✅
 
 #### 2.2 Latest GitHub code changes
 

tests/test_daemon_manager.py

@@ -319,6 +319,21 @@ def test_send_task_no_images(self):
  sent = json.loads(conn._ws.sent[0])
  assert "images" not in sent
 
+ def test_send_task_explicit_id(self):
+ conn = self._conn()
+ rid = asyncio.run(conn.send_task(
+ session_id="s1", cwd="/tmp/x", prompt="hi", id="abc-123"))
+ sent = json.loads(conn._ws.sent[0])
+ assert sent["id"] == "abc-123"
+ assert rid == "abc-123"
+
+ def test_send_task_returns_generated_id(self):
+ conn = self._conn()
+ rid = asyncio.run(conn.send_task(session_id="s1", cwd="/tmp/x", prompt="hi"))
+ sent = json.loads(conn._ws.sent[0])
+ assert sent["id"] == rid
+ assert rid # non-empty uuid
+
  def test_send_command_payload(self):
  conn = self._conn()
  asyncio.run(conn.send_command("set_model", model="gpt-4o"))