fix(chat): project Antigravity thinking summaries and harden turn recovery

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -878,14 +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.
+- The original private Lark owner sees the same bounded recent reasoning-summary
+  model as Codex (up to four summaries, rendered within the 1,200-character
+  projection) alongside the existing answer stream. Tool and command titles are
+  not repeated in the Lark tree view. Group and unqualified projections retain
+  coarse labels. Antigravity maps its redacted `plannerResponse.thinking`
+  summary through the shared activity contract; `rawThinking`, planner
+  completion and tool completion remain separate, and 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

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

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

loopx/capabilities/native_chat/antigravity_activity.py

@@ -8,12 +8,27 @@
 
 from ...presentation.codex_activity import CodexActivitySteps
 
+_BOILERPLATE_THINKING_PREFIXES = (
+    "i'm thinking through how to approach this.",
+    "thinking through how to approach this.",
+)
+
+
+def _clean_thinking_summary(value: Any) -> str:
+    if not isinstance(value, str):
+        return ""
+    lines = [line.strip() for line in value.splitlines() if line.strip()]
+    while lines and lines[0].lower() in _BOILERPLATE_THINKING_PREFIXES:
+        lines.pop(0)
+    return "\n\n".join(lines).strip()
+
 
 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.thinking_digests: dict[str, str] = {}
         self.output_digests: dict[str, str] = {}
 
     def observe(self, index: int, native: dict[str, Any]) -> list[dict[str, Any]]:
@@ -42,11 +57,27 @@ def finish(item: dict[str, Any]) -> None:
             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)
+            summary = _clean_thinking_summary(planner.get("thinking"))
+            if summary or planner.get("thinking"):
+                # Project only the native thinking summary (`thinking`), never
+                # `rawThinking`, through the shared path/secret redacting owner.
+                item_id = f"agy_think_{index}"
+                item: dict[str, Any] = {
+                    "id": item_id,
+                    "type": "reasoning",
+                    **({"summary": [summary]} if summary else {}),
+                }
+                digest = hashlib.sha256(summary.encode("utf-8")).hexdigest() if summary else ""
+                if item_id not in self.started:
+                    if digest:
+                        self.thinking_digests[item_id] = digest
+                    start(item)
+                elif digest and self.thinking_digests.get(item_id) != digest and item_id not in self.completed:
+                    self.thinking_digests[item_id] = digest
+                    emit(
+                        "item/reasoning/summaryTextDelta",
+                        self.steps._reasoning_step(item_id, "running", [summary]),
+                    )
                 if status == "CORTEX_STEP_STATUS_DONE":
                     finish(item)
             calls = planner.get("toolCalls")

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -40,6 +40,27 @@
     "gemini-4-argon-low": "MODEL_PLACEHOLDER_M419",
 }
 
+_SETTLED_STEP_STATUSES = {
+    "CORTEX_STEP_STATUS_DONE",
+    "CORTEX_STEP_STATUS_ERROR",
+    "CORTEX_STEP_STATUS_CANCELED",
+    "CORTEX_STEP_STATUS_CLEARED",
+}
+
+
+def _step_allows_completed_turn(step: dict[str, Any], *, final: dict[str, Any]) -> bool:
+    """Only a settled intermediate step may precede a final planner answer."""
+    error = step.get("errorMessage") or {}
+    return (
+        step.get("status") in _SETTLED_STEP_STATUSES
+        and not (step is final and step.get("status") != "CORTEX_STEP_STATUS_DONE")
+        and not (
+            step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE"
+            and (step is final or (isinstance(error, dict) and error.get("shouldShowUser") is True))
+        )
+    )
+
+
 ANTIGRAVITY_MODEL_OPTIONS: tuple[dict[str, Any], ...] = tuple(
     {
         "id": f"gemini-4-argon-{effort}",
@@ -369,7 +390,7 @@ def completed_result_observation(*, cascade_id: str, request_ref: str, message:
                     return None
                 def read(method: str, **payload: Any) -> dict[str, Any]:
                     return _rpc_post(found["url"], found["csrf_token"], method,
-                                     {"cascadeId": cascade_id, **payload}, timeout=5.0)
+                                     {"cascadeId": cascade_id, **payload}, timeout=10.0)
 
                 before = read("GetCascadeTrajectory")
                 total = before.get("numTotalSteps")
@@ -416,16 +437,14 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
                 return None
             execution = metadata.get("executionId")
             current = [(index, step) for index, step in indexed if start <= index < end]
+            if not current:
+                return None
+            final = current[-1][1]
             if not isinstance(execution, str) or not execution or any(
-                    (step.get("status") != "CORTEX_STEP_STATUS_DONE"
-                     and not (step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE"
-                              and step.get("status") == "CORTEX_STEP_STATUS_CLEARED"))
-                    or (step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE"
-                        and (step.get("errorMessage") or {}).get("shouldShowUser") is True)
+                    not _step_allows_completed_turn(step, final=final)
                     or (step.get("metadata") or {}).get("executionId") != execution
                     for _, step in current):
                 return None
-            final = current[-1][1]
             planner = final.get("plannerResponse") or {}
             answer = planner.get("modifiedResponse") or planner.get("response")
             if (final.get("type") != "CORTEX_STEP_TYPE_PLANNER_RESPONSE" or planner.get("toolCalls")
@@ -649,14 +668,34 @@ def capabilities(self) -> dict[str, Any]:
         return {"adapter_kind": self.adapter_kind, "streaming": True,
                 "resume": True, "interrupt": True, "steering": False}
 
+    def start_turn_with_attachments(
+        self,
+        message: str,
+        event_sink: Callable[[str, dict[str, Any]], None] | None = None,
+        attachments: list[dict[str, Any]] | None = None,
+    ) -> dict[str, Any]:
+        from ...chat_attachments import normalize_chat_image_attachments
+
+        normalized = normalize_chat_image_attachments(attachments or [])
+        media = [
+            {
+                "mimeType": row["mime_type"],
+                "inlineData": row["data_url"].split(",", 1)[1],
+                "description": row["name"],
+            }
+            for row in normalized
+        ]
+        return self.start_turn(message, event_sink, media=media)
+
     def start_turn(self, message: str,
                    event_sink: Callable[[str, dict[str, Any]], None] | None = None,
-                   *, read_only_review: bool = False) -> dict[str, Any]:
+                   *, read_only_review: bool = False,
+                   media: list[dict[str, Any]] | None = None) -> dict[str, Any]:
         self._read_only_review = read_only_review
         self._interrupted.clear()
         self._cancel_acknowledged = False
         try:
-            return self._run_turn(message, event_sink)
+            return self._run_turn(message, event_sink, media=media)
         except BaseException as failure:
             # A timeout/transport or permission failure is not a successful turn.
             # Retain an unconfirmed cancellation as an error for the caller fence.
@@ -676,6 +715,8 @@ def _run_turn(
         self,
         message: str,
         event_sink: Callable[[str, dict[str, Any]], None] | None = None,
+        *,
+        media: list[dict[str, Any]] | None = None,
     ) -> dict[str, Any]:
         if self._closed:
             raise CodexChatAgentError(
@@ -690,7 +731,7 @@ def _run_turn(
             self.csrf_token,
             "GetCascadeTrajectory",
             {"cascadeId": self.cascade_id},
-            timeout=5.0,
+            timeout=10.0,
         )
         if before_traj.get("status") != "CASCADE_RUN_STATUS_IDLE":
             raise CodexChatAgentError(
@@ -705,6 +746,7 @@ def _run_turn(
             "cascadeId": self.cascade_id,
             "items": [{"text": full_prompt}],
             "cascadeConfig": user_config,
+            **({"media": media} if media else {}),
         }
         with self._send_lock:
             self._check_interrupted()
@@ -741,7 +783,7 @@ def _run_turn(
                     self.csrf_token,
                     "GetCascadeTrajectory",
                     {"cascadeId": self.cascade_id},
-                    timeout=2.0,
+                    timeout=10.0,
                 )
             except CodexChatAgentError as exc:
                 # Losing an idempotent observation does not mean that the
@@ -770,7 +812,7 @@ def _run_turn(
                             "startIndex": start_step_index,
                             "endIndex": total_steps,
                         },
-                        timeout=2.0,
+                        timeout=10.0,
                     )
                 except CodexChatAgentError as exc:
                     if exc.error_code != "antigravity_transport_error":
@@ -928,7 +970,10 @@ def _run_turn(
                                                   gate=None, error_code="upstream_transport_failure")
                 if (final.get("type") != "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
                         or final.get("status") != "CORTEX_STEP_STATUS_DONE" or planner.get("toolCalls")
-                        or any(step.get("status") != "CORTEX_STEP_STATUS_DONE" for _, step in current_steps)
+                        or any(
+                            not _step_allows_completed_turn(step, final=final)
+                            for _, step in current_steps
+                        )
                         or not str(planner.get("modifiedResponse") or planner.get("response") or "").strip()
                         or planner.get("stopReason") != "STOP_REASON_STOP_PATTERN"):
                     raise CodexChatAgentError("Antigravity ended without a verified final response; execution may have occurred.",

loopx/capabilities/native_chat/codex_adapter.py

@@ -425,7 +425,8 @@ def start_attachment_turn(adapter: Any, session: dict[str, Any], turn: dict[str,
                           event_sink: Callable[[str, dict[str, Any]], None], attachments: list[dict[str, Any]]) -> dict[str, Any]:
     """Keep attachment support and recovered source-route checks at the adapter."""
     from ...chat_attachments import require_text_document_route
+    from .antigravity_adapter import AntigravityAdapter
     require_text_document_route(session, attachments, origin=str(turn.get("origin") or "web"))
-    if not isinstance(adapter, CodexAppServerAdapter):
-        raise ValueError("attachments currently require the Codex Agent endpoint")
+    if not isinstance(adapter, (CodexAppServerAdapter, AntigravityAdapter)):
+        raise ValueError("attachments currently require the Codex or Antigravity Agent endpoint")
     return adapter.start_turn_with_attachments(message, event_sink, attachments)

loopx/extensions/lark/README.md

@@ -492,8 +492,10 @@ skip a control-plane receipt.
 
 Native private project Chat and steward conversations present persisted Core
 Turn events as a mutable, source-bound Markdown reply. Answer chunks are coalesced at a two-second interval; changes only to fixed
-activity labels are coalesced at ten seconds to avoid message churn. Reasoning,
-tool arguments and command output are excluded. The transport does not run models.
+activity labels are coalesced at ten seconds to avoid message churn. The private
+owner sees the same bounded reasoning-summary strategy as Codex; unqualified
+reasoning, tool/command titles, arguments and output are excluded. The transport
+does not run models.
 
 The conservative edit budget applies to each draft, rather than the whole Turn.
 When it is exhausted, the transport verifies a short continuation notice on the

loopx/extensions/lark/private_progress.py

@@ -1,7 +1,8 @@
 """Lark presentation of persisted Core events; never execution authority.
 
-The original private owner can also see qualified public summaries and redacted
-activity titles. Hidden reasoning, tool arguments and output never travel here.
+The original private owner sees bounded reasoning summaries in the same shape
+as Codex, without repeating tool/command titles. Hidden reasoning, tool
+arguments and output never travel here.
 """
 from __future__ import annotations
 
@@ -74,9 +75,10 @@ def legacy_recovery_repairs(record: Mapping[str, Any]) -> list[dict[str, Any]]:
 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."""
+    state.pop("activities", None)
     if not owner_private:
         state.pop("public_summary", None)
-        state.pop("activities", None)
+        state.pop("thinking_summaries", None)
     for event in events:
         state["cursor"] = event["event_id"]
         payload = event.get("payload") or {}
@@ -102,18 +104,25 @@ def project_progress(state: dict[str, Any], events: list[dict[str, Any]], *,
             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 owner_private and kind == "reasoning" and step.get("reasoning_summary") is True:
+                    summary = step.get("detail") or step.get("title")
+                    if isinstance(summary, str) and summary.strip():
+                        cleaned = " ".join(summary.split())[:280]
+                        step_id = str(step.get("id") or "reasoning")
+                        summaries = [
+                            row
+                            for row in state.get("thinking_summaries", [])
+                            if isinstance(row, Mapping) and row.get("id") != step_id
+                        ]
+                        summaries.append({"id": step_id, "text": cleaned})
+                        summaries = summaries[-4:]
+                        state["thinking_summaries"] = summaries
+                        if len(summaries) == 1:
+                            state["public_summary"] = summary.strip()[:1200]
+                        else:
+                            state["public_summary"] = "\n".join(
+                                f"• {row['text']}" for row in summaries
+                            )[:1200]
                 if step.get("state") == "completed":
                     label = "已结束当前步骤，继续处理"
                 elif step.get("state") == "failed":
@@ -142,10 +151,6 @@ def project_progress(state: dict[str, Any], events: list[dict[str, Any]], *,
     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 "")