fix(chat): recover anchored historical answers without replay

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -360,7 +360,19 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
             inputs = [(i, step) for i, step in indexed if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
             if not inputs:
                 return None
-            start, user = inputs[-1]
+            # Historical results are usable only through the original Core
+            # input anchor. Legacy records still require the latest input.
+            if input_anchor is not None:
+                anchored = input_anchor.get("start_step_index")
+                if type(anchored) is not int:
+                    return None
+                matching = [item for item in inputs if item[0] == anchored]
+                if len(matching) != 1:
+                    return None
+                start, user = matching[0]
+            else:
+                start, user = inputs[-1]
+            end = next((index for index, _ in inputs if index > start), total)
             metadata = user.get("metadata") or {}
             created = datetime.fromisoformat(metadata["createdAt"])
             if not datetime.fromisoformat(started_at) <= created <= datetime.fromisoformat(completed_at):
@@ -376,7 +388,7 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
                     or input_anchor.get("start_step_index") != start or input_anchor.get("input_sha256") != prompt_hash):
                 return None
             execution = metadata.get("executionId")
-            current = indexed[start:]
+            current = [(index, step) for index, step in indexed if start <= index < end]
             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"
@@ -399,7 +411,9 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
             return {"thread_id": cascade_id, "request_ref": request_ref, "execution_id": execution,
                 "input_created_at": metadata["createdAt"], "start_step_index": start, "total_steps": total,
                 "input_sha256": prompt_hash, "answer_sha256": hashlib.sha256(answer.encode()).hexdigest(),
-                "latest_input_verified": True, "normal_terminal_verified": True, "response": response,
+                "latest_input_verified": start == inputs[-1][0],
+                **({"end_step_index": end} if start != inputs[-1][0] else {}),
+                "normal_terminal_verified": True, "response": response,
                 **({"raw_response": answer} if include_raw_response else {})}
         except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
             return None

loopx/capabilities/native_chat/external_conversations.py

@@ -77,6 +77,12 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 or reservation.get("metadata_review"))
             and reservation.get("failed_turn_id") == row["turn_id"] else None)
         events = store.events_after(session["session_id"], turn["turn_id"], None)
+        anchors = [event["payload"]["native_input"] for event in events
+                   if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")]
+        # This only selects read-only result probes. Adoption still requires
+        # the exact original anchor, native final, current source and typed rule.
+        from ...chat import _review_openings
+        _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0)
         preparation = None
         later_equivalent = None
         if turn.get("error_code") == "manager_authorization_unavailable":
@@ -105,6 +111,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 "preparation_observation": preparation,
                 "later_equivalent_input_present": later_equivalent,
                 "preparation_scope_available": preparation_scope_available,
+                "native_input": anchors[0] if len(anchors) == 1 else None,
+                "response_literal_open": bool(open_literal),
                 "native_observation": observation, "source_verified": source_verified},
         })
 
@@ -174,7 +182,10 @@ def context_recovery_candidate(self, request_ref: str) -> bool:
             if not turn or turn.get("status") not in {"failed", "completed"}:
                 return False
             if turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work"):
-                return False
+                from ...chat import _review_openings
+                _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0)
+                if not open_literal:
+                    return False  # A lexical probe hint, never permission to adopt or send.
             return self._context_recovery_plan(row, discovery_only=True).get("recovery_candidate") is True
         except (KeyError, ValueError, RuntimeError, OSError):
             return False

loopx/control_plane/collaboration/conversation_binding.ts

@@ -484,9 +484,13 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && typeof recovered.message_text === "string" && recovered.message_text.length > 0
       && result != null && !["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].some(key => result[key])
       && Array.isArray(result.proposals) && result.proposals.length === 0;
+    const expandedResult = turn?.status === "completed" && response !== null && result != null
+      && recovered !== null && recovered.turn_id === row.turn_id && recovered.message_id === `native.${row.turn_id}.recovered`
+      && typeof recovered.message_text === "string" && recovered.message_text.length > 0
+      && extendsOriginalText(response, result);
     return {pending: !returned && row.command === null && row.status === "accepted"
       && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
-      && (metadataOwed || unfinished || turn?.status === "failed"
+      && (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)))
@@ -555,6 +559,22 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
   return {operation: "admit_turn", client_turn_id: `external-${request}`, session_id: session, turn_id: null};
 }
 
+function pureResultText(response: JsonObject | null): boolean {
+  return response?.schema_version === "loopx_chat_agent_response_v0"
+    && typeof response.message === "string" && response.message.length > 0
+    && Array.isArray(response.proposals) && response.proposals.length === 0
+    && ["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].every(key => response[key] == null);
+}
+
+function extendsOriginalText(original: JsonObject, result: JsonObject): boolean {
+  const metadata = (value: JsonObject) => Object.fromEntries(Object.entries(value)
+    .filter(([key, value]) => key !== "message" && value != null));
+  return pureResultText(original) && pureResultText(result)
+    && (result.message as string).length > (original.message as string).length
+    && (result.message as string).startsWith(original.message as string)
+    && isDeepStrictEqual(metadata(original), metadata(result));
+}
+
 /** A new, correlated continuation after an observed pre-execution failure.
  * A delivered failure notice is not a completed business request. This bounded
  * migration path never reopens the failed Turn or retries unknown effects.
@@ -592,18 +612,30 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
       || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.some(value =>
         requireJsonObject(value, "prior preparation recovery").preparation_attempt != null))) return no;
   if (current.agent_id === "antigravity") {
+    const originalResponse = failed.response && typeof failed.response === "object" && !Array.isArray(failed.response)
+      ? failed.response as JsonObject : null;
+    const anchor = recovery.native_input && typeof recovery.native_input === "object" && !Array.isArray(recovery.native_input)
+      ? recovery.native_input as JsonObject : null;
+    // The lexical hint only bounds read-only probes. It confers no result,
+    // source or execution authority; an anchored native final is still needed.
+    const completedText = failed.status === "completed" && pureResultText(originalResponse)
+      && recovery.response_literal_open === true && recovery.queued_input_present === false
+      && 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);
     // Result adoption has no admission/dispatch authority. Keep the failed
     // Turn and original failure delivery, even when native tools ran.
     if (!steward || failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
-        || failed.client_turn_id !== `external-${row.request_ref}` || failed.status !== "failed"
-        || failed.response != null || failed.message !== row.message
-        || !["response_timeout", "antigravity_rpc_error", "antigravity_transport_error",
-          "upstream_transport_failure", "server_restarted"].includes(String(failed.error_code))
+        || failed.client_turn_id !== `external-${row.request_ref}` || failed.message !== row.message
+        || (!completedText && (failed.status !== "failed" || failed.response != null
+          || !["response_timeout", "antigravity_rpc_error", "antigravity_transport_error",
+            "upstream_transport_failure", "server_restarted"].includes(String(failed.error_code))))
         || recovery.attachments_present !== false || recovery.steering_observed !== false
         || recovery.stop_observed !== false) return no;
     const adopted = row.native_result_recovery == null ? null
       : requireJsonObject(row.native_result_recovery, "adopted native result");
     if (adopted?.delivery_verified === true) {
+      if (completedText) return no; // A repaired answer never starts a format review or business continuation.
       return planNativeMetadataReview(recovery, row, current, failed, adopted, version);
     }
     // A fresh native result requires an idle, stable trajectory. Do not spend
@@ -621,7 +653,15 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
     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
-      && proof.latest_input_verified === true && proof.normal_terminal_verified === true
+      && (completedText
+        ? proof.start_step_index === anchor!.start_step_index && proof.input_sha256 === anchor!.input_sha256
+          && (proof.latest_input_verified === true || 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))
+          && response !== null && extendsOriginalText(originalResponse!, response)
+        : proof.latest_input_verified === true)
+      && proof.normal_terminal_verified === true
       && Number.isFinite(created) && created >= start && created <= end
       && Number.isSafeInteger(proof.start_step_index) && Number(proof.start_step_index) >= 0
       && Number.isSafeInteger(proof.total_steps) && Number(proof.total_steps) > Number(proof.start_step_index) + 1

tests/control_plane_ts/conversation_binding.test.ts

@@ -92,6 +92,66 @@ test("Core preparation failures resume through one bounded original-source queue
   }
 });
 
+test("completed truncated text adopts only its anchored original final, never a later result or action", () => {
+  const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: "antigravity"};
+  const configured = planConversationBinding({...request, binding: selected}).state;
+  const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};
+  const context = resolveBoundConversation({current: configured, binding_id: row.binding_id,
+    observation, available_projects: [project], ...source}).context;
+  const ref = "f".repeat(24);
+  const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "antigravity",
+    upstream_mode: "default", status: "ready", upstream_thread_id: "native", active_turn_id: null,
+    channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`, steward_context: context};
+  const response = {schema_version: "loopx_chat_agent_response_v0", message: "Original `", proposals: [], gate: null};
+  const completed = {session_id: "original", turn_id: "truncated", client_turn_id: `external-${ref}`,
+    message: "Original request", status: "completed", response, error_code: null,
+    started_at: "2026-01-01T00:00:00Z", completed_at: "2026-01-01T00:00:10Z"};
+  const anchor = {thread_id: "native", start_step_index: 4, input_sha256: "1".repeat(64)};
+  const proof = {...anchor, request_ref: ref, execution_id: "original-execution",
+    input_created_at: "2026-01-01T00:00:05Z", total_steps: 12, end_step_index: 8,
+    answer_sha256: "2".repeat(64), latest_input_verified: false, normal_terminal_verified: true,
+    response: {...response, message: "Original `<loopx-review-json>` is a literal marker. Complete answer."}};
+  const original = {request_ref: ref, command: null, status: "accepted", delivery_verified: true,
+    message: completed.message, origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "truncated", source};
+  const recovery = {version: 21, failed_turn: completed, native_input: anchor, response_literal_open: true,
+    queued_input_present: false, steering_observed: false, stop_observed: false, attachments_present: false,
+    source_verified: true, native_observation: proof};
+  const packet = {request: original, binding: selected, current_session: session, context_recovery: recovery};
+  assert.equal(planBoundConversationRequest(packet).operation, "adopt_native_result");
+  assert.equal(planBoundConversationRequest(packet).recovery_eligible, true);
+  assert.equal(planBoundConversationRequest(packet).client_turn_id, undefined);
+  for (const change of [{native_input: null}, {response_literal_open: false}, {queued_input_present: true},
+    {source_verified: false}, {stop_observed: true}, {steering_observed: true}, {attachments_present: true},
+    {native_input: {...anchor, thread_id: "later"}}, {native_input: {...anchor, input_sha256: "3".repeat(64)}},
+    ...[{start_step_index: 5}, {end_step_index: 12}, {end_step_index: 5}, {end_step_index: null},
+      {normal_terminal_verified: false}, {input_created_at: "2026-01-02T00:00:00Z"},
+      {response: {...proof.response, message: response.message}},
+      {response: {...proof.response, message: "Later answer"}},
+      {response: {...proof.response, proposals: [{kind: "action"}]}},
+      {response: {...proof.response, gate: {kind: "response_metadata_invalid"}}},
+      {response: {...proof.response, extra_metadata: "changed"}}]
+      .map(value => ({native_observation: {...proof, ...value}})),
+    {failed_turn: {...completed, response: {...response, protected_action: {kind: "action"}}}}]) {
+    assert.equal(planBoundConversationRequest({...packet, context_recovery: {...recovery, ...change}}).recovery_eligible,
+      false, JSON.stringify(change));
+  }
+  for (const change of [{status: "closed"}, {status: "busy", active_turn_id: "newer"}, {active_turn_id: "newer"}]) {
+    assert.equal(planBoundConversationRequest({...packet, current_session: {...session, ...change}}).recovery_candidate, false);
+  }
+  const latest = {...proof, latest_input_verified: true};
+  delete (latest as Partial<typeof proof>).end_step_index;
+  assert.equal(planBoundConversationRequest({...packet, context_recovery: {...recovery, native_observation: latest}}).recovery_eligible, true);
+  const adopted = {turn_id: completed.turn_id, message_id: `native.${completed.turn_id}.recovered`,
+    message_text: "Frozen original answer", native_observation: proof, delivery_verified: false};
+  const pending = {...original, native_result_recovery: adopted};
+  assert.equal(planBoundConversationRequest({request: pending, current_session: session,
+    pending_observation: {turn: completed, stop_observed: false}}).pending, true);
+  const delivered = {...pending, native_result_recovery: {...adopted, delivery_verified: true}};
+  assert.equal(planBoundConversationRequest({request: delivered, current_session: session,
+    pending_observation: {turn: completed, stop_observed: false}}).pending, false);
+  assert.equal(planBoundConversationRequest({...packet, request: delivered}).recovery_candidate, false);
+});
+
 test("native terminal recovery adopts a result without Turn admission or lost failure provenance", () => {
   const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: "antigravity"};
   const configured = planConversationBinding({...request, binding: selected}).state;