Skip to content

emrg: remove non-stream dead code — _run_chat_once / TaskRequest.stream (rant 2026-08-10T21:20:38)#654

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
argszero merged 1 commit into from
Aug 10, 2026
Merged

emrg: remove non-stream dead code — _run_chat_once / TaskRequest.stream (rant 2026-08-10T21:20:38) #654

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion emrg/client/app.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1854,7 +1854,7 @@ def _is_image_token(s, i):
images = _pending_images or None
_pending_images = []
await conn.send_task(session_id=session_id, cwd=cwd, prompt=text,
stream=True, images=images)
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):
Expand Down
4 changes: 2 additions & 2 deletions emrg/client/daemon_manager.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -228,12 +228,12 @@ def __init__(self, ws):
self._ws = ws

async def send_task(self, session_id: str, cwd: str, prompt: str,
stream: bool = True, images: list | None = None) -> None:
images: list | None = None) -> None:
"""聊天发送：TaskRequest(type="task")。images 支持 /image 粘贴图。

内部 json.dumps(req.to_dict(), ensure_ascii=False) 以 str 发送（不 .encode()）。
"""
req = TaskRequest(session_id=session_id, cwd=cwd, prompt=prompt, stream=stream)
req = TaskRequest(session_id=session_id, cwd=cwd, prompt=prompt)
if images:
req.images = images
await self._ws.send(json.dumps(req.to_dict(), ensure_ascii=False))
Expand Down
11 changes: 4 additions & 7 deletions emrg/gui/daemon_client.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -486,11 +486,11 @@ class DaemonClient {

// ── 消息发送 ────────────────────────────────────────────

sendTask({ sessionId, cwd, prompt, stream = true, images = null, requestId = null, mode = "auto" }) {
sendTask({ sessionId, cwd, prompt, images = null, requestId = null, mode = "auto" }) {
// G32：request_id 必须作为 id 字段发出（daemon 只回显不自生成）
// G96：stream 必须显式 true（daemon 读 stream 默认 False）
// G143：外部预生成 requestId 优先（renderer send 前标记自有流，消除 IPC 往返竞态窗口）
// WorkBuddy P2：mode="ask" → daemon 不启用工具（纯对话）
// rant 21:20:38：非 stream 路径已删除——所有 task 恒走 tool_loop（流式）
const rid = requestId || crypto.randomUUID();
const payload = {
type: "task",
Expand All@@ -499,17 +499,14 @@ class DaemonClient {
cwd,
prompt,
timestamp: new Date().toISOString(),
stream,
images,
};
if (mode && mode !== "auto") payload.mode = mode;
this._setCurrentStream(rid);
// G65：自有流锁——本连接发出流式 task 即标记，done/error/cancelled/断连释放
// （多会话各自独立；main.js emrg:sendMessage 的 G65 切会话检查读本字段）
if (stream) {
this.ownStream = true;
this.ownStreamRequestId = rid;
}
this.ownStream = true;
this.ownStreamRequestId = rid;
this.ws.send(JSON.stringify(payload));
return rid;
}
Expand Down
2 changes: 1 addition & 1 deletion emrg/gui/main.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -287,7 +287,7 @@ vision = false
// G143：renderer 预生成 requestId（send 前标记自有流，消除 IPC 往返竞态窗口）
// WorkBuddy P2：mode="ask" → 纯对话（daemon 不启用工具）
// G65：conn.sendTask 内部标记 ownStream（每连接独立锁）
rid = conn.sendTask({ sessionId, cwd: sessionCwd, prompt: text, stream: true, requestId, mode });
rid = conn.sendTask({ sessionId, cwd: sessionCwd, prompt: text, requestId, mode });
} catch (e) {
conn._releaseOwnStream(); // sendTask 抛异常（ws.send 失败）→ 释放锁，防 G65 锁泄漏
throw e;
Expand Down
27 changes: 13 additions & 14 deletions emrg/gui/test/daemon_client.test.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -7,7 +7,7 @@
* - ensureConnected：port 文件读取 + auth 首帧 + auth_ok 处理
* - 坏 JSON 帧 → 忽略不崩
* - ws close → 触发 disconnected 事件（重连回调由 main 层调度）
* - sendTask：payload（type=task + session_id + prompt + images + id + stream:true）
* - sendTask：payload（type=task + session_id + prompt + images + id，无 stream 字段——非 stream 路径已删）
* - sendCommand：payload（type + params）；cancel 无多余字段（G24）
* - 帧分类（G21+G58）：tool_start/tool_end/delta/done/cancelled/error/pong/
* list_result/command_result 各帧正确分类
Expand DownExpand Up@@ -524,7 +524,7 @@ test("断连清 _currentStream + G94 timer（#338 回归：不弹虚假超时）
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
// 发起任务 → _currentStream 建立
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: tmpHome, prompt: "hi", stream: true });
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: tmpHome, prompt: "hi" });
assert.ok(client._currentStream, "_currentStream 已建立");
assert.ok(client._currentStream.requestId === rid);
// 收到 delta → _resetStreamTimer 挂起 G94 30s timer
Expand All@@ -540,7 +540,7 @@ test("断连清 _currentStream + G94 timer（#338 回归：不弹虚假超时）
assert.deepStrictEqual(doneEvents, [], "断连后无虚假 done 事件");
});

test("sendTask payload（G32/G96）", async () => {
test("sendTask payload（G32，无 stream 字段——rant 21:20:38）", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hello", images: null });
Expand All@@ -550,16 +550,15 @@ test("sendTask payload（G32/G96）", async () => {
assert.strictEqual(frame.session_id, "s_260803_1730_abcd1234");
assert.strictEqual(frame.cwd, "/proj");
assert.strictEqual(frame.prompt, "hello");
assert.strictEqual(frame.stream, true, "stream 显式 true（G96）");
assert.strictEqual(frame.images, null);
assert.strictEqual(frame.images, null);
assert.ok(frame.timestamp);
});

test("sendTask 外部预生成 requestId（G143）", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
const outer = "s_260803_1730_outer1234";
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: outer });
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: outer });
assert.strictEqual(rid, outer, "返回外部预生成 id");
const frame = JSON.parse(currentMockWs.sent.at(-1));
assert.strictEqual(frame.id, outer, "payload id 用外部预生成 id");
Expand DownExpand Up@@ -829,19 +828,19 @@ test("18:47:37: stale projectDir port + spawn 节流失败 → probe 复用 cano

// ── P2 自有流锁（G65 每连接独立；rant 15:07:19）──────────────────────────

test("P2 ownStream: sendTask(stream:true) 标记 ownStream + requestId", async () => {
test("P2 ownStream: sendTask 恒标记 ownStream + requestId（非 stream 路径已删）", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: "req-own-1" });
assert.strictEqual(client.ownStream, true, "stream:true must set ownStream");
const rid = client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: "req-own-1" });
assert.strictEqual(client.ownStream, true, "ownStream set unconditionally");
assert.strictEqual(client.ownStreamRequestId, "req-own-1");
assert.strictEqual(rid, "req-own-1");
});

test("P2 ownStream: 自有 done（request 匹配）→ 释放锁；广播 done（不匹配）→ 保持", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: "req-own-2" });
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: "req-own-2" });
const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
// 广播 done（其他客户端/其他流）→ 锁保持
send({ request_id: "req-other", done: true, delta: false });
Expand All@@ -855,7 +854,7 @@ test("P2 ownStream: 自有 done（request 匹配）→ 释放锁；广播 done
test("P2 ownStream: timeout 兜底 done（无匹配 request）→ 释放锁", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: "req-own-3" });
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: "req-own-3" });
const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
send({ request_id: "req-stale", done: true, delta: false, timeout: true });
assert.strictEqual(client.ownStream, false, "timeout done must release own lock");
Expand All@@ -864,7 +863,7 @@ test("P2 ownStream: timeout 兜底 done（无匹配 request）→ 释放锁", as
test("P2 ownStream: session busy 即发 error → 释放锁（防 G65 锁泄漏）", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: "req-own-4" });
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: "req-own-4" });
const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
send({ error: "session busy: another stream running" });
assert.strictEqual(client.ownStream, false, "session busy error must release lock");
Expand All@@ -873,7 +872,7 @@ test("P2 ownStream: session busy 即发 error → 释放锁（防 G65 锁泄漏
test("P2 ownStream: cancelled（request 匹配）→ 释放锁", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: "req-own-5" });
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: "req-own-5" });
const send = (obj) => currentMockWs.emit("message", Buffer.from(JSON.stringify(obj)));
send({ type: "cancelled", request_id: "req-own-5" });
assert.strictEqual(client.ownStream, false, "own cancelled must release lock");
Expand All@@ -882,7 +881,7 @@ test("P2 ownStream: cancelled（request 匹配）→ 释放锁", async () => {
test("P2 ownStream: 断连 → 释放锁", async () => {
const client = new DaemonClient({ projectDir: tmpHome });
await connectClient(client);
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", stream: true, requestId: "req-own-6" });
client.sendTask({ sessionId: "s_260803_1730_abcd1234", cwd: "/proj", prompt: "hi", requestId: "req-own-6" });
assert.strictEqual(client.ownStream, true);
client.close();
assert.strictEqual(client.ownStream, false, "disconnect must release lock");
Expand Down
2 changes: 0 additions & 2 deletions emrg/protocol.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -29,7 +29,6 @@ class TaskRequest:
timestamp: str = field(
default_factory=lambda: datetime.now().isoformat()
)
stream: bool = False
images: Optional[list[dict]] = None

def to_dict(self) -> dict:
Expand All@@ -40,7 +39,6 @@ def to_dict(self) -> dict:
"cwd": self.cwd,
"prompt": self.prompt,
"timestamp": self.timestamp,
"stream": self.stream,
}
if self.images:
d["images"] = self.images
Expand Down
88 changes: 7 additions & 81 deletions emrg/server/daemon.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -554,7 +554,6 @@ async def _handle_client(self, ws) -> None:
cwd=cwd,
prompt=data.get("prompt", ""),
timestamp=data.get("timestamp", ""),
stream=data.get("stream", False),
images=data.get("images"),
)
except Exception as e:
Expand All@@ -570,19 +569,14 @@ async def _handle_client(self, ws) -> None:
_tool_task.cancel()
session = self._get_or_create_session(session_id, Path(cwd))
logger.info(
'task received: session=%s prompt="%s" → routing via LLM (stream=%s)',
session_id, _redact_string(req.prompt[:60]), req.stream,
'task received: session=%s prompt="%s" → routing via LLM',
session_id, _redact_string(req.prompt[:60]),
)
_cancel_event = asyncio.Event()
self._session_busy[session_id] = True # lock (released in *locked wrapper)
if req.stream:
_tool_task = asyncio.create_task(
self._run_tool_loop_locked(req, ws, session, _cancel_event, allow_tools=allow_tools)
)
else:
_tool_task = asyncio.create_task(
self._run_chat_once_locked(req, ws, session)
)
_tool_task = asyncio.create_task(
self._run_tool_loop_locked(req, ws, session, _cancel_event, allow_tools=allow_tools)
)
continue

await self._process_message(data, ws)
Expand DownExpand Up@@ -1642,64 +1636,6 @@ def _build_user_content(text: str, images: list[dict] | None, vision: bool = Fal
content.insert(0, {"type": "text", "text": "请分析这张图片"})
return content

async def _run_chat_once(
self, req: TaskRequest, ws, session: Session
) -> None:
"""Non-streaming single-turn chat (no tool loop)."""
system_prompt = self._build_system_prompt(session)
history_messages = session.get_messages_for_llm()
user_content = self._build_user_content(req.prompt, req.images, self.llm.config.vision)
messages: list[dict] = [
{"role": "system", "content": system_prompt},
*history_messages,
{"role": "user", "content": user_content},
]
tools = self.tools.to_openai_tools()

# Persist user message (with image references if present)
user_record: dict = {
"type": "message",
"role": "user",
"content": req.prompt,
}
if req.images:
user_record["images"] = req.images
session.append_message(user_record)

try:
msg = await self.llm.chat(messages, tools=tools)
content = msg.get("content", "")

# Log LLM request/response
self._log_llm_exchange(
session, messages, tools, content,
finish_reason=msg.get("finish_reason", "stop"),
tool_calls=msg.get("tool_calls"),
)

# Persist assistant message
session.append_message({
"type": "message",
"role": "assistant",
"content": content or "",
})

await self._broadcast(session.session_id, {
"request_id": req.id,
"content": content or "",
"done": True,
"delta": False,
"session_id": session.session_id,
})

# Fire-and-forget: reflect on whether to save memories
self._maybe_reflect_memory(session, req.prompt, content or "")
except Exception as e:
logger.exception("LLM error")
await self._broadcast(session.session_id, {
"error": f"LLM error: {e}. Check config at ~/.emrg/config.toml",
})

# ── Phase 2 session-lock wrappers (protocol-contract §2.6.5) ──
# The caller fire-and-forgets with asyncio.create_task() and never awaits,
# so the lock MUST be released inside the task (wrapper finally) — not in
Expand All@@ -1719,16 +1655,6 @@ async def _run_tool_loop_locked(
finally:
self._session_busy[session_id] = False

async def _run_chat_once_locked(
self, req: TaskRequest, ws, session: Session,
) -> None:
"""Run _run_chat_once and release the session busy lock on exit."""
session_id = session.session_id
try:
await self._run_chat_once(req, ws, session)
finally:
self._session_busy[session_id] = False

async def _run_tool_loop(
self, req: TaskRequest, ws, session: Session,
cancel_event: asyncio.Event | None = None,
Expand DownExpand Up@@ -2110,8 +2036,8 @@ def _log_llm_exchange(
) -> None:
"""Log a complete LLM request/response exchange to the session.

Centralizes the 4 identical append_llm patterns from _run_chat_once
and _run_tool_loop, ensuring consistent logging format.
Centralizes the identical append_llm patterns from _run_tool_loop,
ensuring consistent logging format.
"""
session.append_llm({
"type": "request",
Expand Down
4 changes: 2 additions & 2 deletions tests/test_daemon_manager.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -298,15 +298,15 @@ def _conn(self, frames=None):
def test_send_task_payload(self):
conn = self._conn()
asyncio.run(conn.send_task(
session_id="s1", cwd="/tmp/x", prompt="hello", stream=True,
session_id="s1", cwd="/tmp/x", prompt="hello",
images=[{"path": "/tmp/a.png", "label": "[image1]"}],
))
sent = json.loads(conn._ws.sent[0])
assert sent["type"] == "task"
assert sent["session_id"] == "s1"
assert sent["cwd"] == "/tmp/x"
assert sent["prompt"] == "hello"
assert sent["stream"] is True
assert "stream" not in sent # non-stream path removed (rant 21:20:38)
assert sent["images"] == [{"path": "/tmp/a.png", "label": "[image1]"}]

def test_send_task_no_images(self):
Expand Down
2 changes: 1 addition & 1 deletion tests/test_daemon_manager_e2e.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -80,7 +80,7 @@ async def _test():

# streaming task → delta frames → done
await conn.send_task(session_id="e2e-session",
cwd=str(tmp), prompt="你好", stream=True)
cwd=str(tmp), prompt="你好")
frames = []
while True:
frame = await conn.recv(timeout=5)
Expand Down
4 changes: 1 addition & 3 deletions tests/test_protocol.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,15 +17,13 @@ def test_task_request_to_dict():
session_id="s_260718_1200_a3f9",
cwd="/home/user/project",
prompt="hello",
stream=True,
)
d = req.to_dict()
assert d["type"] == "task"
assert d["id"] == "abc-123"
assert d["session_id"] == "s_260718_1200_a3f9"
assert d["cwd"] == "/home/user/project"
assert d["prompt"] == "hello"
assert d["stream"] is True
assert "timestamp" in d


Expand All@@ -35,7 +33,7 @@ def test_task_request_defaults():
assert d["type"] == "task"
assert d["id"] # auto-generated UUID
assert d["session_id"] == ""
assert d["stream"] is False
assert "stream" not in d # non-stream path removed (rant 21:20:38)


def test_task_response_from_dict():
Expand Down
Loading