huangruiteng · Nov 2, 2026 · Nov 2, 2026 · Nov 2, 2026 · Nov 2, 2026
@@ -878,6 +878,14 @@ receipt → observed work or actionable failure → readable answer in the same
   upstream observations; never simulate phases, expose hidden reasoning, or
   infer progress from elapsed time. A period without new events has an explicit
   waiting caption, not an invented failure or continually changing animation.
+- The original private Lark owner can read the latest qualified public reasoning
+  summary (up to 1,200 characters) and four redacted activity titles alongside
+  the existing answer stream. Recent steps precede a potentially long summary.
+  Group and unqualified projections retain coarse
+  labels. Antigravity reports observed thinking spans and native command/tool
+  steps through the shared activity contract; its raw `thinking` field is not a
+  public summary, and planner completion is not tool completion. Final answers
+  still replace the draft on the original source-bound post.
 - Generate the visible answer once, then append a complete hidden review envelope
   with `message: ""`. The production parser reuses the preceding Markdown and
   applies the same redaction and typed metadata normalization. Legacy nonempty

@@ -598,6 +598,11 @@ CLI/SSH 导出使用 `goal-portfolio --manager-view todos --goal-id GOAL --todo-
   持续计算的证明。切换视图保留起始时间；恢复时使用已记录的回合时间，未知则不编造。
 - 工具和阶段记录可展开，只显示上游报告的信息，不模拟阶段或暴露隐藏推理。
   一段时间没有新事件时明确说明仍在等待，不推断失败，也不靠动画假装有进展。
+- 原飞书私聊受众可在答案流旁看到最新合格公开思考摘要（最多 1,200 字符）与四条
+  脱敏活动标题，最近步骤放在长摘要之前；群聊及未经限定的投影仍只显示粗粒度状态。
+  反重力复用共享活动契约
+  报告实际思考区段和命令、工具步骤；原始 `thinking` 字段不视作公开摘要，planner
+  结束不代表工具已完成。最终答案仍在绑定原请求的同一消息中替换草稿。
 - 可见答案只生成一次，随后附加完整隐藏 review envelope，`message: ""` 表示
   复用前面的 Markdown，仍经过相同脱敏与 typed 元数据归一化。旧的非空 message
   保持优先；缺失或 null 不启用该约定。Codex 采用完成的最终答案 item，排除之前

@@ -0,0 +1,92 @@
+"""Project observed native steps through the shared, redacted activity owner."""
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any
+from urllib.parse import unquote
+
+from ...presentation.codex_activity import CodexActivitySteps
+
+
+class AntigravityActivity:
+    def __init__(self, work_dir: Path) -> None:
+        self.steps = CodexActivitySteps(protected_paths=[work_dir])
+        self.started: set[str] = set()
+        self.completed: set[str] = set()
+        self.output_digests: dict[str, str] = {}
+
+    def observe(self, index: int, native: dict[str, Any]) -> list[dict[str, Any]]:
+        events: list[dict[str, Any]] = []
+
+        def emit(method: str, step: dict[str, Any] | None) -> None:
+            if step:
+                label = ({"reasoning": "Agent 正在思考", "command": "Agent 正在执行命令",
+                          "tool": "Agent 正在调用工具"}[step["kind"]]
+                         if method != "item/completed" else "Agent 返回了处理状态")
+                events.append({"method": method, "label": label, "step": step})
+
+        def start(item: dict[str, Any]) -> None:
+            if item["id"] not in self.started:
+                self.started.add(item["id"])
+                emit("item/started", self.steps.started(item))
+
+        def finish(item: dict[str, Any]) -> None:
+            if item["id"] not in self.completed:
+                self.completed.add(item["id"])
+                emit("item/completed", self.steps.completed(item))
+
+        kind = native.get("type")
+        status = native.get("status")
+        if kind == "CORTEX_STEP_TYPE_PLANNER_RESPONSE":
+            planner = native.get("plannerResponse")
+            if not isinstance(planner, dict):
+                return events
+            if planner.get("thinking"):
+                # The native field is not a qualified public summary. Report
+                # the observed thinking span without publishing its contents.
+                item: dict[str, Any] = {"id": f"agy_think_{index}", "type": "reasoning"}
+                start(item)
+                if status == "CORTEX_STEP_STATUS_DONE":
+                    finish(item)
+            calls = planner.get("toolCalls")
+            for offset, call in enumerate(calls if isinstance(calls, list) else []):
+                if isinstance(call, dict) and isinstance(call.get("name"), str):
+                    start({"id": f"agy_call_{index}_{offset}", "type": "dynamicToolCall",
+                           "tool": call["name"]})
+            # A finished planner has only requested the tool. It is not proof
+            # that the tool ran or completed; never fabricate that transition.
+        elif kind == "CORTEX_STEP_TYPE_VIEW_FILE":
+            view = native.get("viewFile")
+            if not isinstance(view, dict):
+                return events
+            uri = view.get("absolutePathUri")
+            name = unquote(uri[7:]) if isinstance(uri, str) and uri.startswith("file:///") else "[local-path]"
+            item = {"id": f"agy_read_{index}", "type": "commandExecution", "command": "",
+                    "commandActions": [{"type": "read", "name": name}]}
+            start(item)
+            if status in {"CORTEX_STEP_STATUS_DONE", "CORTEX_STEP_STATUS_ERROR"}:
+                item["status"] = "failed" if status == "CORTEX_STEP_STATUS_ERROR" else "completed"
+                finish(item)
+        elif kind == "CORTEX_STEP_TYPE_RUN_COMMAND":
+            command = native.get("runCommand")
+            if not isinstance(command, dict):
+                return events
+            item = {"id": f"agy_cmd_{index}", "type": "commandExecution",
+                    "command": command.get("commandLine") or command.get("command") or "command"}
+            start(item)
+            output = command.get("combinedOutput")
+            if isinstance(output, str) and output and item["id"] not in self.completed:
+                digest = hashlib.sha256(output.encode()).hexdigest()
+                if self.output_digests.get(item["id"]) != digest:
+                    self.output_digests[item["id"]] = digest
+                    # Only the observation travels, never stdout/stderr.
+                    emit("item/commandExecution/outputDelta", self.steps.activity_delta(
+                        item["id"], "observed", kind="command"))
+            if status in {"CORTEX_STEP_STATUS_DONE", "CORTEX_STEP_STATUS_ERROR"}:
+                exit_code = command.get("exitCode")
+                if type(exit_code) is int:
+                    item["exitCode"] = exit_code
+                item["status"] = "failed" if status == "CORTEX_STEP_STATUS_ERROR" else "completed"
+                finish(item)
+        return events
@@ -26,6 +26,7 @@
     parse_agent_response,
 )
 from ...chat_agent import CodexChatAgentError, _turn_prompt
+from .antigravity_activity import AntigravityActivity
 
 
 ANTIGRAVITY_DEFAULT_MODEL = "gemini-4-argon-high"
@@ -664,7 +665,7 @@ def _run_turn(
         stream_filter = VisibleResponseStreamFilter()
         emitted_visible_delta = False
         emitted_text_by_step: dict[int, str] = {}
-        announced_phases: set[tuple[int, str]] = set()
+        activity = AntigravityActivity(self.work_dir)
         latest_response_text = ""
         last_progress_at = time.monotonic()
         deadline = last_progress_at + self.hard_timeout_sec if self.hard_timeout_sec is not None else None
@@ -735,37 +736,19 @@ def _run_turn(
                                       "next_action": "Review permissions in Antigravity, then resend this request in LoopX."},
                                 error_code="host_tool_gate",
                             )
+                        planner = step.get("plannerResponse")
+                        if (step_type == "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
+                                and isinstance(planner, dict) and planner.get("toolCalls")
+                                and (self.runtime_profile == "restricted" or self._read_only_review)):
+                            raise CodexChatAgentError("Antigravity did not honor text-only Chat.",
+                                                      gate=None, error_code="unsupported_configuration")
+                        if event_sink is not None:
+                            for phase in activity.observe(step_idx, step):
+                                event_sink("agent.phase", phase)
                         if step_type == "CORTEX_STEP_TYPE_PLANNER_RESPONSE":
                             pr = step.get("plannerResponse") or {}
                             if not isinstance(pr, dict):
                                 continue
-                            thinking = str(pr.get("thinking") or "").strip()
-                            if thinking and (step_idx, "thinking") not in announced_phases:
-                                announced_phases.add((step_idx, "thinking"))
-                                if event_sink is not None:
-                                    event_sink(
-                                        "agent.phase",
-                                        {
-                                            "phase": "thinking",
-                                            "label": f"Gemini 4 Argon 思考中 ({self.model_id})",
-                                        },
-                                    )
-                            tool_calls = pr.get("toolCalls") or []
-                            if tool_calls and (self.runtime_profile == "restricted" or self._read_only_review):
-                                raise CodexChatAgentError("Antigravity did not honor text-only Chat.",
-                                                          gate=None, error_code="unsupported_configuration")
-                            if isinstance(tool_calls, list) and tool_calls and (step_idx, "tool") not in announced_phases:
-                                announced_phases.add((step_idx, "tool"))
-                                first_tool = tool_calls[0] if isinstance(tool_calls[0], dict) else {}
-                                tool_name = str(first_tool.get("name") or "tool")
-                                if event_sink is not None:
-                                    event_sink(
-                                        "agent.phase",
-                                        {
-                                            "phase": "tool_call",
-                                            "label": f"正在调用 {tool_name}",
-                                        },
-                                    )
                             text = str(pr.get("modifiedResponse") or pr.get("response") or "")
                             if text:
                                 prev_text = emitted_text_by_step.get(step_idx, "")

@@ -1213,7 +1213,9 @@ def _progress(self, path: Path, record: dict[str, Any], *, inbox: Callable[[], P
         stream = presentation_stream(record)
         events = self.core.controller.store.events_after(session_id or record["session_id"],
             turn_id or record["turn_id"], stream.get("cursor"))
-        text = project_progress(stream, events)
+        text = project_progress(stream, events, owner_private=(
+            record.get("source", {}).get("private_human_message") is True
+            and record.get("event", {}).get("chat_type") == "p2p"))
         if text:
             # Journal the exact presentation the sender freezes, not the raw
             # Core text. The same conversion also recovers older draft proofs.

@@ -1,7 +1,7 @@
 """Lark presentation of persisted Core events; never execution authority.
 
-Only the visible answer stream and coarse, observed activity kinds are shown.
-Reasoning text, tool arguments and command output stay out of the transport.
+The original private owner can also see qualified public summaries and redacted
+activity titles. Hidden reasoning, tool arguments and output never travel here.
 """
 from __future__ import annotations
 
@@ -71,8 +71,12 @@ def legacy_recovery_repairs(record: Mapping[str, Any]) -> list[dict[str, Any]]:
     return repairs
 
 
-def project_progress(state: dict[str, Any], events: list[dict[str, Any]]) -> str | None:
+def project_progress(state: dict[str, Any], events: list[dict[str, Any]], *,
+                     owner_private: bool = False) -> str | None:
     """Fold each exact Turn event once into a bounded, replayable view."""
+    if not owner_private:
+        state.pop("public_summary", None)
+        state.pop("activities", None)
     for event in events:
         state["cursor"] = event["event_id"]
         payload = event.get("payload") or {}
@@ -84,8 +88,10 @@ def project_progress(state: dict[str, Any], events: list[dict[str, Any]]) -> str
         elif event.get("kind") == "agent.phase":
             step = payload.get("step")
             kind = step.get("kind") if isinstance(step, Mapping) else None
-            # Summary deltas contain reasoning, not owner-facing progress.
-            if str(payload.get("method") or "").startswith("item/reasoning/"):
+            method = str(payload.get("method") or "")
+            if method.startswith("item/reasoning/") and not (
+                    owner_private and method == "item/reasoning/summaryTextDelta"
+                    and isinstance(step, Mapping) and step.get("reasoning_summary") is True):
                 continue
             # Core preparation and confirmed host start precede model output.
             # Render fixed activity text, never the context/provider label.
@@ -96,6 +102,18 @@ def project_progress(state: dict[str, Any], events: list[dict[str, Any]]) -> str
             label = {"reasoning": "正在思考", "command": "正在执行命令", "tool": "正在调用工具",
                      "search": "正在检索资料", "file_change": "正在修改文件"}.get(kind) if isinstance(kind, str) else None
             if label and isinstance(step, Mapping):
+                if owner_private:
+                    if kind == "reasoning" and step.get("reasoning_summary") is True:
+                        summary = step.get("detail") or step.get("title")
+                        if isinstance(summary, str):
+                            state["public_summary"] = summary[:1200]
+                    elif kind != "reasoning" and isinstance(step.get("title"), str):
+                        title = " ".join(step["title"].split())[:160].replace("`", "'")
+                        if title and isinstance(step.get("id"), str):
+                            activities = [row for row in state.get("activities", []) if row["id"] != step["id"]]
+                            activities.append({"id": step["id"], "title": title,
+                                               "state": step.get("state")})
+                            state["activities"] = activities[-4:]
                 if step.get("state") == "completed":
                     label = "已结束当前步骤，继续处理"
                 elif step.get("state") == "failed":
@@ -124,6 +142,12 @@ def project_progress(state: dict[str, Any], events: list[dict[str, Any]]) -> str
     if not state.get("label") and not state.get("answer"):
         return None
     header = f"⏳ **{state.get('label') or '正在处理'}**"
+    if owner_private and state.get("activities"):
+        header += "\n\n**最近步骤**\n" + "\n".join(
+            f"{'✓' if row['state'] == 'completed' else '!' if row['state'] == 'failed' else '·'} `{row['title']}`"
+            for row in state["activities"])
+    if owner_private and state.get("public_summary"):
+        header += "\n\n**思考摘要**\n" + state["public_summary"]
     answer = str(state.get("answer") or "")
     if answer:
         if state.get("clipped"):

@@ -0,0 +1,48 @@
+"""Native observations become bounded owner activity, never a transcript."""
+import json
+
+from loopx.capabilities.native_chat.antigravity_activity import AntigravityActivity
+
+
+def test_repeated_native_observations_do_not_invent_tool_completion_or_publish_reasoning(tmp_path):
+    activity = AntigravityActivity(tmp_path)
+    planner = {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "status": "CORTEX_STEP_STATUS_DONE",
+               "plannerResponse": {"thinking": "hidden thought content", "toolCalls": [
+                   {"name": "read_file", "arguments": {"secret": "private arguments"}}]}}
+    events = activity.observe(1, planner)
+    assert [(row["method"], row["step"]["kind"]) for row in events] == [
+        ("item/started", "reasoning"), ("item/completed", "reasoning"), ("item/started", "tool")]
+    assert activity.observe(1, planner) == []
+    assert events[-1]["step"]["title"] == "read_file"
+    assert "hidden thought" not in json.dumps(events) and "private arguments" not in json.dumps(events)
+    assert not any(row["step"]["kind"] == "tool" and row["method"] == "item/completed" for row in events)
+
+
+def test_command_redaction_failure_and_output_observation_use_shared_contract(tmp_path):
+    activity = AntigravityActivity(tmp_path)
+    native = {"type": "CORTEX_STEP_TYPE_RUN_COMMAND", "status": "CORTEX_STEP_STATUS_GENERATING",
+              "runCommand": {"commandLine": f"cat {tmp_path}/source.txt --token secret-value /private/other.txt",
+                             "combinedOutput": "private command output"}}
+    rows = activity.observe(2, native)
+    assert len(rows) == 1
+    assert "source.txt" in rows[0]["step"]["title"]
+    assert "secret-value" not in json.dumps(rows) and "private command output" not in json.dumps(rows)
+    assert str(tmp_path) not in json.dumps(rows) and "/private/other.txt" not in json.dumps(rows)
+    assert activity.observe(2, native) == []
+    native["status"] = "CORTEX_STEP_STATUS_DONE"
+    native["runCommand"]["exitCode"] = 3
+    final = activity.observe(2, native)
+    assert len(final) == 1 and final[0]["step"]["state"] == "failed"
+    assert final[0]["step"]["exit_code"] == 3
+    assert activity.observe(2, native) == []
+
+
+def test_observed_native_file_view_redacts_uri_and_never_emits_contents(tmp_path):
+    activity = AntigravityActivity(tmp_path)
+    native = {"type": "CORTEX_STEP_TYPE_VIEW_FILE", "status": "CORTEX_STEP_STATUS_DONE",
+              "viewFile": {"absolutePathUri": (tmp_path / "file name.txt").as_uri(), "content": "private file body"}}
+    rows = activity.observe(3, native)
+    assert [row["step"]["state"] for row in rows] == ["running", "completed"]
+    assert all(row["step"]["verb"] == "read" and row["step"]["title"] == "file name.txt" for row in rows)
+    assert str(tmp_path) not in json.dumps(rows) and "private file body" not in json.dumps(rows)
+    assert activity.observe(3, native) == []