fix(chat): expose observed command and tool progress

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

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

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 ""

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.

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:

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)

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.