fix(chat): restore historical answers and retain unanswered originals

loopx/capabilities/native_chat/external_conversations.py

@@ -6,7 +6,7 @@
 """
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from copy import deepcopy
 from pathlib import Path
 import json
@@ -27,6 +27,8 @@
 # Request journals are named by their 24-hex request_ref; lock metadata and
 # other siblings in this directory are not requests.
 _REQUEST_JOURNAL_STEM = re.compile(r"[a-f0-9]{24}")
+QUOTA_BACKLOG_BATCH_SIZE = 64
+_QUOTA_BACKLOG_BATCH_BYTES = 1024 * 1024
 
 
 def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]:
@@ -186,8 +188,8 @@ def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, Any],
                 and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"])
                 for item in (source_controls or self.pending)())))
 
-    def quota_backlog_observer(self) -> Callable[[dict[str, Any]], bool]:
-        """Share lazy journal IO within one read-only observation, never admission."""
+    def quota_backlog_observer(self) -> Callable[[Sequence[dict[str, Any]]], list[bool]]:
+        """Batch the existing typed projection; share controls only within this read."""
         controls: list[dict[str, Any]] | None = None
 
         def read_controls() -> list[dict[str, Any]]:
@@ -196,29 +198,87 @@ def read_controls() -> list[dict[str, Any]]:
                 controls = self.pending()
             return controls
 
-        return lambda row: self.quota_backlog_pending(row, source_controls=read_controls)
+        def observe(rows: Sequence[dict[str, Any]]) -> list[bool]:
+            if len(rows) > QUOTA_BACKLOG_BATCH_SIZE:
+                raise ValueError("select at most 64 pending requests per observation")
+            params = [self._quota_pending_inputs(row) for row in rows]
+            flags = self._quota_pending_flags(params)
+            stopped = []
+            for index, (row, values, pending) in enumerate(zip(rows, params, flags, strict=True)):
+                if pending:
+                    events = self.controller.store.events_after(row["session_id"], row["turn_id"], None)
+                    if self._source_stop_observed(row, values["pending_observation"]["turn"], events,
+                                                 source_controls=read_controls):
+                        values["pending_observation"]["stop_observed"] = True
+                        stopped.append(index)
+            for index, pending in zip(stopped, self._quota_pending_flags([params[i] for i in stopped]), strict=True):
+                flags[index] = pending
+            return flags
+
+        return observe
 
-    def quota_backlog_pending(self, row: dict[str, Any], *,
-                              source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
-        """Retain quota and declared unfinished work without admitting recovery."""
+    @staticmethod
+    def _quota_pending_flags(params: Sequence[dict[str, Any]]) -> list[bool]:
         from ...control_plane.effect_runtime import effect_runtime_result
 
+        flags: list[bool] = []
+        batch: list[dict[str, Any]] = []
+        size = 0
+
+        def flush() -> None:
+            if not batch:
+                return
+            result = effect_runtime_result("collaboration.conversation.request", {"pending_observations": batch})
+            values = result.get("pending_flags") if isinstance(result, dict) else None
+            if (not isinstance(values, list) or len(values) != len(batch)
+                    or any(type(value) is not bool for value in values)):
+                raise ValueError("pending observation results are unavailable")
+            flags.extend(values)
+            batch.clear()
+
+        for values in params:
+            encoded = len(json.dumps(values, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
+            if batch and (len(batch) >= QUOTA_BACKLOG_BATCH_SIZE or size + encoded > _QUOTA_BACKLOG_BATCH_BYTES):
+                flush()
+                size = 0
+            if encoded > _QUOTA_BACKLOG_BATCH_BYTES:
+                # Preserve the original single-request transport and its byte
+                # refusal; batching must not introduce a new content limit.
+                result = effect_runtime_result("collaboration.conversation.request", values)
+                pending = result.get("pending") if isinstance(result, dict) else None
+                if type(pending) is not bool:
+                    raise ValueError("pending observation result is unavailable")
+                flags.append(pending)
+                continue
+            batch.append(values)
+            size += encoded
+        flush()
+        return flags
+
+    def _quota_pending_inputs(self, row: dict[str, Any]) -> dict[str, Any]:
         store = self.controller.store
-        session = store.load_session(row["session_id"])
-        turn = store.load_turn(row["session_id"], row["turn_id"])
-        observation = {"turn": turn, "stop_observed": False}
-        params = {
+        return {
             "request": {key: value for key, value in row.items() if key != "attachments"},
-            "current_session": session,
-            "pending_observation": observation,
+            "current_session": store.load_session(row["session_id"]),
+            "pending_observation": {"turn": store.load_turn(row["session_id"], row["turn_id"]),
+                                    "stop_observed": False},
         }
+
+    def quota_backlog_pending(self, row: dict[str, Any], *,
+                              source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
+        """Retain quota and declared unfinished work without admitting recovery."""
+        from ...control_plane.effect_runtime import effect_runtime_result
+
+        params = self._quota_pending_inputs(row)
+        observation = params["pending_observation"]
+        turn = observation["turn"]
         # The typed owner first excludes work that has no retained obligation even
         # without a stop. Stop evidence can only remove a pending request;
         # never expose a positive result before reading those native facts.
         if effect_runtime_result("collaboration.conversation.request", params).get("pending") is not True:
             return False
         assert turn is not None  # Required by the typed pending projection.
-        events = store.events_after(row["session_id"], row["turn_id"], None)
+        events = self.controller.store.events_after(row["session_id"], row["turn_id"], None)
         if self._source_stop_observed(row, turn, events, source_controls=source_controls):
             observation["stop_observed"] = True
             return effect_runtime_result("collaboration.conversation.request", params).get("pending") is True
@@ -439,6 +499,9 @@ def _preparation_scope_available(self, session: dict[str, Any]) -> bool:
     def _native_result_text(recovered: dict[str, Any]) -> str:
         response = recovered["native_observation"]["response"]
         text = "已恢复本次请求的原答复：\n\n" + response["message"]
+        if recovered["native_observation"].get("latest_input_verified") is False:
+            text = ("已补回本次请求在原生会话中生成的历史答复。本次只补回正文，没有重新执行业务。"
+                    "文中提到其他消息“已处理”的说法仍需逐条核实，未答的问题继续保留。\n\n" + response["message"])
         if any(response.get(key) for key in ("context_handoff", "proposals", "protected_action", "goal_draft")):
             text += "\n\n操作状态：本次仅恢复原答复，结构化操作或委派尚未提交；未重新执行原请求。"
         if (response.get("gate") or {}).get("summary") and response["gate"]["summary"] not in text:
@@ -461,8 +524,8 @@ def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
         if len(anchors) > 1:
             return None
         # Retained facts can repair only a completed Turn's text. Failed-Turn
-        # recovery keeps its live/latest-input requirements; old capture does
-        # not authorize another invocation or change unknown-effect handling.
+        # recovery keeps its live native read and exact input requirements; old
+        # capture does not authorize another invocation or unknown-effect replay.
         retained = self.controller.store.native_result(session["session_id"], turn["turn_id"])
         if (turn.get("status") == "completed" and len(anchors) == 1 and retained
                 and retained.get("upstream_turn_id") == turn.get("upstream_turn_id")):

loopx/chat_manager.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import hashlib
+import time
 from pathlib import Path
 from typing import Any, Callable, Mapping
 
@@ -991,7 +992,7 @@ def manager_session_quota_route(controller: Any, session: Mapping[str, Any]) ->
         allocation = manager_executor_allocation(controller, machine_defaults=defaults, environ=environment,
             endpoint_capabilities=[{**row, "available": row.get("agent_id") == endpoint} for row in capabilities])
         return {"allocation": allocation, "blocked_primary": route.get("blocked_primary"),
-                "observed_at": capacity["observed_at"] if capacity else None}
+                "observed_at": capacity["observed_at"] if capacity else time.time()}
     except Exception as exc:
         import logging
         logging.getLogger(__name__).warning("Steward quota observation held (%s)", type(exc).__name__)

loopx/control_plane/collaboration/conversation_binding.ts

@@ -446,6 +446,20 @@ export function listLocalBoundSessions(params: JsonObject): JsonObject {
  * command; Core binds its exact target before IO. Replays retain that target.
  */
 export function planBoundConversationRequest(params: JsonObject): JsonObject {
+  if (params.pending_observations !== undefined) {
+    const observations = params.pending_observations;
+    if (Object.keys(params).length !== 1 || !Array.isArray(observations) || observations.length > 64) {
+      throw new EffectRuntimeRequestError("select at most 64 read-only pending observations");
+    }
+    return {pending_flags: observations.map(value => {
+      const item = requireJsonObject(value, "pending observation input");
+      if (Object.keys(item).some(key => !["request", "current_session", "pending_observation"].includes(key))
+          || item.pending_observation === undefined) {
+        throw new EffectRuntimeRequestError("pending batches cannot admit or recover work");
+      }
+      return planBoundConversationRequest(item).pending;
+    })};
+  }
   const row = requireJsonObject(params.request, "external request");
   const request = ref(row.request_ref, "external request identity");
   const command = row.command;
@@ -509,7 +523,7 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && (expandedResult || metadataOwed || unfinished || turn?.status === "failed"
         && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure",
           "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error",
-          "manager_authorization_unavailable"].includes(String(turn.error_code)))
+          "manager_authorization_unavailable", "runtime_error", "server_restarted"].includes(String(turn.error_code)))
       && !(current?.session_id === row.session_id && current?.status === "closed")
       && observation.stop_observed !== true};
   }
@@ -687,6 +701,16 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
     const created = Date.parse(String(proof.input_created_at));
     const start = Date.parse(String(failed.started_at)), end = Date.parse(String(failed.completed_at));
     const response = proof.response == null ? null : requireJsonObject(proof.response, "native result response");
+    // Returning an already generated historical answer is not replay. Bind its
+    // exact original input and closed native segment; later ordinary questions
+    // must not strand that answer. Legacy unanchored results stay latest-only.
+    const anchoredHistorical = anchor !== null && anchor.thread_id === current.upstream_thread_id
+      && Number.isSafeInteger(anchor.start_step_index) && Number(anchor.start_step_index) >= 0
+      && typeof anchor.input_sha256 === "string" && BARE_SHA256_PATTERN.test(anchor.input_sha256)
+      && proof.start_step_index === anchor.start_step_index && proof.input_sha256 === anchor.input_sha256
+      && proof.latest_input_verified === false && Number.isSafeInteger(proof.end_step_index)
+      && Number(proof.end_step_index) > Number(proof.start_step_index) + 1
+      && Number(proof.end_step_index) < Number(proof.total_steps);
     return {...candidate, recovery_eligible: recovery.source_verified === true
       && proof.thread_id === current.upstream_thread_id && proof.request_ref === row.request_ref
       && typeof proof.execution_id === "string" && proof.execution_id.length > 0
@@ -697,7 +721,7 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
             && Number(proof.end_step_index) > Number(proof.start_step_index) + 1
             && Number(proof.end_step_index) < Number(proof.total_steps))
           && response !== null && extendsOriginalText(originalResponse!, response)
-        : proof.latest_input_verified === true)
+        : proof.latest_input_verified === true || anchoredHistorical)
       && proof.normal_terminal_verified === true
       && Number.isFinite(created) && created >= start && created <= end
       && Number.isSafeInteger(proof.start_step_index) && Number(proof.start_step_index) >= 0

loopx/extensions/lark/private_conversations.py

@@ -15,7 +15,7 @@
 from pathlib import Path
 from typing import Any
 
-from ...capabilities.native_chat.external_conversations import ChatExternalConversations
+from ...capabilities.native_chat.external_conversations import ChatExternalConversations, QUOTA_BACKLOG_BATCH_SIZE
 from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json
 from ...control_plane.collaboration import conversation_reply_context
 from ...control_plane.effect_runtime import effect_runtime_request_scope
@@ -225,6 +225,17 @@ def profiles(self) -> dict[str, dict[str, str]]:
     def health(self) -> dict[str, dict[str, int]]:
         health: dict[str, dict[str, int]] = {}
         quota_pending = self.core.quota_backlog_observer()
+        completed: list[tuple[dict[str, Any], dict[str, Any]]] = []
+
+        def observe_completed() -> None:
+            flags = quota_pending([native for _, native in completed])
+            for (row, _), pending in zip(completed, flags, strict=True):
+                if pending:
+                    health[row["binding_id"]]["pending_count"] += 1
+                    health[row["binding_id"]]["recovery_count"] += int(any(
+                        phase.get("started") and not phase.get("verified") for phase in row["deliveries"].values()))
+            completed.clear()
+
         for path in [*self.root.glob("*.json"), *self.root.glob("web-projections/*.json")]:
             row = _read_json(path)
             entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0})
@@ -239,11 +250,14 @@ def health(self) -> dict[str, dict[str, int]]:
                 if (native.get("status") == "accepted" and native.get("command") is None
                         and all(native.get(key) == row.get(key) for key in
                             ("binding_id", "source", "session_id", "turn_id"))):
-                    pending = quota_pending(native)
+                    completed.append((row, native))
+                    if len(completed) >= QUOTA_BACKLOG_BATCH_SIZE:
+                        observe_completed()
             if pending:
                 entry["pending_count"] += 1
                 entry["recovery_count"] += int(any(phase.get("started") and not phase.get("verified")
                                                    for phase in row["deliveries"].values()))
+        observe_completed()
         return health
 
     def _binding(self, profile: str) -> dict[str, Any]: