From 8b235e6d4e0e4fdf7f3c4d38725fd9134c5b53e7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 10 Oct 2026 23:46:57 +0800 Subject: [PATCH] fix(chat): project observed command and tool progress Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- ...lligent-review-presentation-surfaces-v0.md | 15 +++ .../rfcs/loopx-overall-roadmap-v0.md | 6 ++ loopx/chat_agent.py | 8 ++ loopx/extensions/lark/private_progress.py | 11 ++- loopx/presentation/codex_activity.py | 26 ++++- tests/test_chat_activity.py | 96 +++++++++++++++++++ tests/test_lark_private_progress.py | 50 ++++++++++ 7 files changed, 209 insertions(+), 3 deletions(-) diff --git a/docs/architecture/rfcs/intelligent-review-presentation-surfaces-v0.md b/docs/architecture/rfcs/intelligent-review-presentation-surfaces-v0.md index e5fe5394e6..22b1bb6367 100644 --- a/docs/architecture/rfcs/intelligent-review-presentation-surfaces-v0.md +++ b/docs/architecture/rfcs/intelligent-review-presentation-surfaces-v0.md @@ -733,6 +733,21 @@ final-result identities separately. Reconnect and replay must neither start a second model Turn nor duplicate events or drop the final result. No adapter may claim tool activity it did not emit. +The native Codex adapter now projects observed command-output and MCP-progress +notifications into the same running step, retaining only its already-redacted +identity/title and observed elapsed duration. It coalesces bursts over five +seconds, drops unknown/terminal/wrong-Turn items and never stores output or tool +progress text in the presentation. Chat's existing step duration updates in +place; Lark's existing bounded post editor shows a new-output/progress cue with +the observed duration. A silent tool produces no synthetic update, and output +is not a completed check. This source slice does not qualify packaged Lark +timing, stop/recovery or the complete R2/R3 journey. + +原生 Codex 的命令输出与 MCP 进度通知现在更新原运行步骤，只保留已脱敏名称与实际 +经过时长；五秒合并突发事件，不存输出正文，不把旧回合、未知或已结束步骤重新激活。 +Chat 沿用原步骤时长显示，飞书沿用原消息编辑器呈现新输出/进度提示。工具静默时不 +制造进展，收到输出不代表检查通过；安装版时序、停止恢复与 R2/R3 整体验收仍开放。 + Stop and correction bind the current session and Turn. An old control cannot affect a later Turn. Stop readback distinguishes actual interruption, already terminal and unsupported/refused. A busy correction is accepted as native diff --git a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md index 1793561446..7e4f71367b 100644 --- a/docs/architecture/rfcs/loopx-overall-roadmap-v0.md +++ b/docs/architecture/rfcs/loopx-overall-roadmap-v0.md @@ -418,6 +418,12 @@ Chat, session, collaboration and presentation owners as R2/R3 integration work. Do not call a candidate catalog, spinner or queued inbox receipt a completed worker handoff. This checkpoint does not lower R1–R3 or G1 gates. +Observed native command-output and MCP-progress notifications now refresh the +existing Chat work step and bounded Lark post, without persisting their text or +inventing activity during silence. See the same presentation checkpoint for +coalescing, source isolation and the remaining packaged timing/stop/recovery +qualification; this source repair does not close R2/R3. + CLI and MCP receiver inbox guidance now composes the existing shared conversation answer instruction, making the verbatim requester-facing use of report text explicit. It preserves requested depth, evidence and material failures; it adds no diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index 4f67cb33cc..3600204376 100644 --- a/loopx/chat_agent.py +++ b/loopx/chat_agent.py @@ -1376,6 +1376,14 @@ def send( ) if step: phase = "Agent 正在思考" + elif method in {"item/commandExecution/outputDelta", "item/mcpToolCall/progress"} and isinstance(params, dict): + command_output = method == "item/commandExecution/outputDelta" + step = steps.activity_delta( + params.get("itemId"), params.get("delta" if command_output else "message"), + kind="command" if command_output else "tool", + ) + if step: + phase = "Agent 收到命令新输出" if command_output else "Agent 收到工具进展" if method == "item/started": item_type = ( str(item.get("type") or "") if isinstance(item, dict) else "" diff --git a/loopx/extensions/lark/private_progress.py b/loopx/extensions/lark/private_progress.py index a146b17c51..052e5860af 100644 --- a/loopx/extensions/lark/private_progress.py +++ b/loopx/extensions/lark/private_progress.py @@ -36,14 +36,21 @@ def project_progress(state: dict[str, Any], events: list[dict[str, Any]]) -> str elif payload.get("method") == "turn/started": state["label"] = "已开始处理" label = {"reasoning": "正在思考", "command": "正在执行命令", "tool": "正在调用工具", - "search": "正在检索资料", "file_change": "正在修改文件"}.get(kind) - if label: + "search": "正在检索资料", "file_change": "正在修改文件"}.get(kind) if isinstance(kind, str) else None + if label and isinstance(step, Mapping): if step.get("state") == "completed": label = "已结束当前步骤，继续处理" elif step.get("state") == "failed": label = "当前步骤失败，Agent 正在处理" elif kind == "command" and step.get("verb") == "read": label = "正在读取内容" + if step.get("state") == "running" and payload.get("method") in { + "item/commandExecution/outputDelta", "item/mcpToolCall/progress", + }: + duration = step.get("duration_ms") + if isinstance(duration, int) and not isinstance(duration, bool) and duration >= 0: + observed = "收到新输出" if kind == "command" else "收到工具进展" + label = f"{label} · {observed}（已运行 {duration // 1000} 秒）" state["label"] = label elif payload.get("method") == "item/started": # Older adapters emit only this fixed vocabulary, without steps. diff --git a/loopx/presentation/codex_activity.py b/loopx/presentation/codex_activity.py index a1156c47b1..ad9e1890d8 100644 --- a/loopx/presentation/codex_activity.py +++ b/loopx/presentation/codex_activity.py @@ -30,6 +30,7 @@ TITLE_LIMIT = 160 DETAIL_LIMIT = 4000 REASONING_UPDATE_INTERVAL_SEC = 1.5 +ACTIVITY_UPDATE_INTERVAL_SEC = 5.0 _ITEM_KINDS = { "reasoning": "reasoning", @@ -110,6 +111,8 @@ def __init__( self._started: dict[str, float] = {} self._reasoning: dict[str, dict[str, list[str]]] = {} self._reasoning_emitted: dict[str, float] = {} + self._active: dict[str, dict[str, Any]] = {} + self._activity_emitted: dict[str, float] = {} def _clean(self, value: Any) -> str: def clean_path(match: re.Match[str]) -> str: @@ -148,7 +151,12 @@ def started(self, item: Any) -> dict[str, Any] | None: if item.get("type") == "reasoning": # The first update waits one interval so it carries a phrase, not a token. self._reasoning_emitted.setdefault(step_id, now) - return self._step(item, step_id, "running") + step = self._step(item, step_id, "running") + if step["kind"] in {"command", "tool"}: + # Keep only the redacted presentation, never the host payload. + self._active[step_id] = step + self._activity_emitted.setdefault(step_id, now) + return step def completed(self, item: Any) -> dict[str, Any] | None: if not isinstance(item, dict) or item.get("type") not in _ITEM_KINDS: @@ -168,8 +176,24 @@ def completed(self, item: Any) -> dict[str, Any] | None: step["duration_ms"] = max(0, int((self._clock() - started) * 1000)) self._reasoning.pop(step_id, None) self._reasoning_emitted.pop(step_id, None) + self._active.pop(step_id, None) + self._activity_emitted.pop(step_id, None) return step + def activity_delta(self, item_id: Any, delta: Any, *, kind: str) -> dict[str, Any] | None: + """Observe real output/progress without retaining or displaying its text.""" + if not isinstance(delta, str) or not delta: + return None + step_id = _step_id(item_id) + step = self._active.get(step_id) + if step is None or step["kind"] != kind: + return None # Unknown, wrong-kind and already-terminal items stay absent. + now = self._clock() + if now - self._activity_emitted[step_id] < ACTIVITY_UPDATE_INTERVAL_SEC: + return None + self._activity_emitted[step_id] = now + return {**step, "duration_ms": max(0, int((now - self._started[step_id]) * 1000))} + def reasoning_delta( self, item_id: Any, delta: Any, *, summary: bool, index: Any = 0 ) -> dict[str, Any] | None: diff --git a/tests/test_chat_activity.py b/tests/test_chat_activity.py index 6a4d105537..89b983243e 100644 --- a/tests/test_chat_activity.py +++ b/tests/test_chat_activity.py @@ -8,6 +8,7 @@ from pathlib import Path from loopx.presentation.codex_activity import ( + ACTIVITY_UPDATE_INTERVAL_SEC, COMMAND_VERBS, REASONING_UPDATE_INTERVAL_SEC, STEP_KINDS, @@ -94,6 +95,101 @@ def test_user_and_answer_items_are_not_steps(): assert steps.started(item) is None and steps.completed(item) is None +def test_observed_output_is_throttled_and_never_retained_or_promoted_to_completion(): + clock = _Clock() + steps = CodexActivitySteps(clock=clock) + command = _command("python -u worker.py") + steps.started(command) + assert steps.activity_delta("exec-1", "private output", kind="command") is None + clock.now += ACTIVITY_UPDATE_INTERVAL_SEC + for bad in ("", None, {"text": "private output"}): + assert steps.activity_delta("exec-1", bad, kind="command") is None + assert steps.activity_delta("unknown", "private output", kind="command") is None + assert steps.activity_delta("exec-1", "private output", kind="tool") is None + update = steps.activity_delta("exec-1", "private output", kind="command") + assert update == {"id": "exec-1", "kind": "command", "state": "running", "verb": "run", + "title": "python -u worker.py", "duration_ms": 5000} + assert update["state"] == "running" and "exit_code" not in update + assert steps.activity_delta("exec-1", "more private output", kind="command") is None + clock.now += ACTIVITY_UPDATE_INTERVAL_SEC + assert steps.activity_delta("exec-1", "more output", kind="command")["duration_ms"] == 10000 + assert "private output" not in repr(vars(steps)) + assert steps.completed({**command, "exitCode": 1})["state"] == "failed" + assert steps.activity_delta("exec-1", "late output", kind="command") is None + assert not steps._active and not steps._activity_emitted + steps.started({"type": "mcpToolCall", "id": "tool", "server": "fixture", "tool": "inspect"}) + clock.now += ACTIVITY_UPDATE_INTERVAL_SEC + assert steps.activity_delta("tool", "private progress", kind="tool")["title"] == "fixture · inspect" + assert steps.activity_delta("tool", "wrong method", kind="command") is None + + +def test_native_output_progress_is_source_bound_and_clean_before_persisted_replay(monkeypatch, tmp_path): + from loopx import chat_agent + from loopx.chat_store import ChatSessionStore + from loopx.presentation import codex_activity + + # Timing/coalescing is covered above; this exercises actual stdio routing, + # Turn filtering, terminal cleanup and durable replay without slow sleeps. + monkeypatch.setattr(codex_activity, "ACTIVITY_UPDATE_INTERVAL_SEC", 0) + peer = tmp_path / "peer.py" + peer.write_text(r''' +import json, sys +def emit(value): + print(json.dumps(value), flush=True) +for raw in sys.stdin: + request = json.loads(raw) + if "id" not in request: + continue + if request["method"] != "turn/start": + emit({"id": request["id"], "result": {"thread": {"id": "thread"}}}) + continue + turn = "current" + emit({"id": request["id"], "result": {"turn": {"id": turn}}}) + for item, method, field in [ + ({"id": "command", "type": "commandExecution", "command": "python -u worker.py"}, "item/commandExecution/outputDelta", "delta"), + ({"id": "tool", "type": "mcpToolCall", "server": "fixture", "tool": "inspect"}, "item/mcpToolCall/progress", "message"), + ]: + emit({"method": "item/started", "params": {"threadId": "thread", "turnId": turn, "item": item}}) + for thread, tid, iid, text in [("other", turn, item["id"], "outside-thread"), + ("thread", "stale", item["id"], "outside-turn"), + ("thread", turn, "unknown", "unknown-item"), + ("thread", turn, item["id"], ""), + ("thread", turn, item["id"], "private-host-output")]: + emit({"method": method, "params": {"threadId": thread, "turnId": tid, "itemId": iid, field: text}}) + emit({"method": "item/completed", "params": {"threadId": "thread", "turnId": turn, "item": {**item, "status": "completed"}}}) + emit({"method": method, "params": {"threadId": "thread", "turnId": turn, "itemId": item["id"], field: "late-private-output"}}) + emit({"method": "item/agentMessage/delta", "params": {"threadId": "thread", "turnId": turn, "delta": "Ready."}}) + emit({"method": "turn/completed", "params": {"threadId": "thread", "turn": {"id": turn, "status": "completed"}}}) +''', encoding="utf-8") + popen = chat_agent.subprocess.Popen + monkeypatch.setattr(chat_agent.shutil, "which", lambda _: sys.executable) + monkeypatch.setattr(chat_agent.subprocess, "Popen", lambda command, **kwargs: + popen([sys.executable, str(peer), *command[1:]], **kwargs)) + session = chat_agent.CodexChatAgentSession.start( + codex_bin="fixture-host", work_dir=tmp_path, goal_id="fixture", + objective="Inspect synthetic activity.", codex_home=tmp_path / "host-home", + idle_timeout_sec=5, hard_timeout_sec=10, + ) + root = tmp_path / "store" + store = ChatSessionStore(root) + try: + session.send("Inspect activity.", on_event=lambda kind, payload: + store.append_event("session", "turn", kind=kind, payload=payload)) + events = ChatSessionStore(root).events_after("session", "turn", None) + updates = [event for event in events if event["payload"].get("method") in { + "item/commandExecution/outputDelta", "item/mcpToolCall/progress"}] + assert len(updates) == 2 + assert {event["payload"]["step"]["kind"] for event in updates} == {"command", "tool"} + assert all(event["payload"]["step"]["state"] == "running" for event in updates) + saved = "\n".join(p.read_text() for p in store.root.rglob("*.events.jsonl")) + for secret in ("outside-thread", "outside-turn", "unknown-item", "private-host-output", "late-private-output"): + assert secret not in saved + assert [event["payload"]["step"]["state"] for event in events if "step" in event["payload"]] == [ + "running", "running", "completed", "running", "running", "completed"] + finally: + session.close() + + def test_reasoning_streams_throttled_and_prefers_the_model_summary(): clock = _Clock() steps = CodexActivitySteps(clock=clock) diff --git a/tests/test_lark_private_progress.py b/tests/test_lark_private_progress.py index 2141c2bcdd..68c55b07c7 100644 --- a/tests/test_lark_private_progress.py +++ b/tests/test_lark_private_progress.py @@ -117,6 +117,56 @@ def test_native_preparation_is_visible_without_answer_or_private_labels(): "phase": "unrecognized", "label": "private output"}}]) is None +def test_observed_tool_progress_changes_the_view_without_exposing_output_or_inventing_activity(): + state = {} + for kind, method, observed in [ + ("command", "item/commandExecution/outputDelta", "收到新输出"), + ("tool", "item/mcpToolCall/progress", "收到工具进展"), + ]: + def update(seconds): + return {"event_id": f"{kind}-{seconds}", "kind": "agent.phase", "payload": { + "method": method, "label": "private host label", "step": { + "id": kind, "kind": kind, "state": "running", "title": "private title", + "detail": "private details", "duration_ms": seconds * 1000}}} + first = project_progress(state, [update(10)]) + second = project_progress(state, [update(20)]) + assert observed in first and "已运行 10 秒" in first + assert "已运行 20 秒" in second and second != first + assert project_progress(state, []) == second # No timer-synthesized progress. + assert "private" not in second and "answer" not in state + done = {"event_id": "done", "kind": "agent.phase", "payload": { + "method": "item/completed", "step": {"id": "tool", "kind": "tool", "state": "failed"}}} + assert project_progress(state, [done]) == "⏳ **当前步骤失败，Agent 正在处理**" + + +def test_output_activity_updates_the_original_post_then_preserves_terminal_answer(ordinary): # noqa: F811 + store, runtime, provider, transport, row = start(ordinary) + try: + sid, tid = row["session_id"], row["turn_id"] + for duration in (10000, 20000): + store.append_event(sid, tid, kind="agent.phase", payload={ + "method": "item/commandExecution/outputDelta", "step": { + "id": "read", "kind": "command", "state": "running", "verb": "read", + "title": "private command", "duration_ms": duration}}) + allow_update(transport, row) + transport.reconcile() + assert f"已运行 {duration // 1000} 秒" in provider.edits[-1][2] + assert "第一段回答" in provider.edits[-1][2] and "private" not in provider.edits[-1][2] + count = len(provider.edits) + allow_update(transport, row) + transport.reconcile() + assert len(provider.edits) == count + runtime.adapters[sid].steer_turn("finish", store.load_turn(sid, tid)["upstream_turn_id"]) + runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10) + transport.reconcile() + assert provider.edits[-1][2] == "Steered response." + assert len(provider.writes) == 1 and {ref for _, ref, _ in provider.edits} == {"om_out_0"} + transport.reconcile() + assert len(provider.edits) == count + 1 + finally: + runtime.close() + + def test_native_preparation_updates_original_post_before_model_output(ordinary): # noqa: F811 store, runtime, provider, transport = streaming(ordinary) # This host accepts the Turn, then waits without any answer/tool/reasoning event.