fix(chat): review cleared native answers without replaying work

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -414,8 +414,15 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
             if len(current) > 128 or raw != read("GetCascadeTrajectorySteps", startIndex=start, endIndex=total).get("steps") or before != read("GetCascadeTrajectory"):
                 return None
             user, terminal = current[0], current[-1]
+            # A cleared historical segment can retain visible answer text while
+            # losing its completion marker and tool payloads. It is not a
+            # verified final or evidence that effects did not occur. Only the
+            # existing restricted answer review may consume this text.
+            cleared_answer = (end < total
+                and terminal.get("type") == "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
+                and all(step.get("status") == "CORTEX_STEP_STATUS_CLEARED" for step in current))
             if (user.get("type") != "CORTEX_STEP_TYPE_USER_INPUT"
-                    or terminal.get("type") != "CORTEX_STEP_TYPE_ERROR_MESSAGE"):
+                    or not cleared_answer and terminal.get("type") != "CORTEX_STEP_TYPE_ERROR_MESSAGE"):
                 return None
             meta = user["metadata"]
             execution = meta.get("executionId")
@@ -429,15 +436,28 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
             results = []
             for step in current:
                 step_execution = (step.get("metadata") or {}).get("executionId")
-                if (step.get("status") != "CORTEX_STEP_STATUS_DONE"
-                        or step_execution != execution and not (step is terminal and step_execution is None)):
+                if (step.get("status") != ("CORTEX_STEP_STATUS_CLEARED" if cleared_answer else "CORTEX_STEP_STATUS_DONE")
+                        or step_execution != execution and not (not cleared_answer and step is terminal and step_execution is None)):
                     return None
                 kind = step.get("type")
                 payload = {"CORTEX_STEP_TYPE_USER_INPUT": "userInput", "CORTEX_STEP_TYPE_PLANNER_RESPONSE": "plannerResponse",
                     "CORTEX_STEP_TYPE_RUN_COMMAND": "runCommand", "CORTEX_STEP_TYPE_VIEW_FILE": "viewFile",
                     "CORTEX_STEP_TYPE_ERROR_MESSAGE": "errorMessage"}.get(kind)
+                if cleared_answer:
+                    payload = {"CORTEX_STEP_TYPE_USER_INPUT": "userInput",
+                        "CORTEX_STEP_TYPE_PLANNER_RESPONSE": "plannerResponse",
+                        "CORTEX_STEP_TYPE_GENERIC": "generic", "CORTEX_STEP_TYPE_CHECKPOINT": "checkpoint",
+                        "CORTEX_STEP_TYPE_SYSTEM_MESSAGE": "systemMessage"}.get(kind)
                 if payload is None or set(step) - {"type", "status", "metadata", payload}:
                     return None
+                if cleared_answer:
+                    if step is terminal:
+                        planner = step.get("plannerResponse") or {}
+                        text = planner.get("modifiedResponse") or planner.get("response")
+                        if planner.get("toolCalls") or not isinstance(text, str) or not text.strip():
+                            return None
+                        results.append({"visible_response": text})
+                    continue
                 if kind == "CORTEX_STEP_TYPE_RUN_COMMAND":
                     command = step["runCommand"]
                     if command.get("blocking") is not True or type(command.get("exitCode")) is not int:

loopx/control_plane/collaboration/conversation_binding.ts

@@ -639,6 +639,12 @@ 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") {
+    if (failed.error_code === "runtime_error") {
+      // Parsing/presentation can fail after native work. This route admits
+      // only a restricted review of independently read original evidence;
+      // it never retries the failed business Turn or adopts an unverified final.
+      return planInterruptedStewardAnswer(recovery, row, current, failed, version);
+    }
     if (failed.error_code === "antigravity_step_error") {
       const quota = planNativeQuotaRecovery(recovery, row, current, failed, version);
       const observed = recovery.native_observation as JsonObject | null | undefined;
@@ -1083,6 +1089,9 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
 function interruptedVisibleResult(value: unknown): boolean {
   if (!value || typeof value !== "object" || Array.isArray(value)) return false;
   const result = value as JsonObject;
+  if (Object.keys(result).join(",") === "visible_response") {
+    return typeof result.visible_response === "string" && result.visible_response.trim().length > 0;
+  }
   if (Object.keys(result).sort().join(",") === "content,file_uri") {
     if (typeof result.file_uri !== "string" || typeof result.content !== "string") return false;
     try {

tests/control_plane_ts/conversation_binding.test.ts

@@ -58,6 +58,25 @@ test("settled historical native work permits only one restricted steward answer
   assert.equal(plan.read_only_review, true);
   assert.equal(plan.quota_attempt, undefined);
   assert.equal(plan.client_turn_id, `external-${ref}-recovery-22-review-1`);
+  const runtimeFailure = {...packet, context_recovery: {...packet.context_recovery,
+    failed_turn: {...failed, error_code: "runtime_error"},
+    native_observation: {...proof, visible_results: [{visible_response: "Existing visible draft"}]}}};
+  const draftReview = planBoundConversationRequest(runtimeFailure);
+  assert.equal(draftReview.recovery_eligible, true);
+  assert.equal(draftReview.answer_review, true);
+  assert.equal(draftReview.read_only_review, true);
+  assert.equal(draftReview.quota_attempt, undefined);
+  for (const result of [{visible_response: ""}, {visible_response: null},
+    {visible_response: "draft", newEffect: true}, {visible_response: "x".repeat(256 * 1024)}]) {
+    assert.notEqual(planBoundConversationRequest({...runtimeFailure,
+      context_recovery: {...runtimeFailure.context_recovery,
+        native_observation: {...proof, visible_results: [result]}}}).recovery_eligible, true);
+  }
+  for (const change of [{source_verified: false}, {stop_observed: true}, {steering_observed: true},
+    {attachments_present: true}, {queued_input_present: true}, {metadata_capacity: null}]) {
+    assert.notEqual(planBoundConversationRequest({...runtimeFailure,
+      context_recovery: {...runtimeFailure.context_recovery, ...change}}).recovery_eligible, true);
+  }
   const fileResult = {file_uri: "file:///authorized/existing-result.txt", content: "Original visible file result"};
   assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery,
     native_observation: {...proof, visible_results: [...proof.visible_results, fileResult]}}}).recovery_eligible, true);

tests/test_steward_interrupted_answer.py

@@ -101,7 +101,62 @@ def read(_url, _token, method, _body, **_):
     assert not set(calls) - {"GetCascadeTrajectory", "GetCascadeTrajectorySteps"}
 
 
-@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "no-capacity", "review-action", "review-gate", "queue-change", "already-codex", "post-enqueue-crash", "lost-ack", "new-question-during-answer", "restricted-review", "primary-quota-unreadable", "native-tail", "authenticated-host", "file-read", "file-running"])
+@pytest.mark.parametrize("fault", [None, "open", "busy", "mixed", "foreign", "changed", "wrong-hash",
+    "no-visible-answer", "tool-call", "unknown", "oversize"])
+def test_cleared_visible_answer_is_only_stable_historical_review_evidence(monkeypatch, fault):
+    prompt = "Original synthetic source"
+    meta = {"createdAt": "2026-01-01T00:00:05Z", "executionId": "original"}
+    steps = [
+        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_CLEARED", "metadata": meta,
+         "userInput": {"items": [{"text": prompt}]}},
+        {"type": "CORTEX_STEP_TYPE_GENERIC", "status": "CORTEX_STEP_STATUS_CLEARED", "metadata": meta,
+         "generic": {}},
+        {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "status": "CORTEX_STEP_STATUS_CLEARED", "metadata": meta,
+         "plannerResponse": {"response": "Existing visible draft", "thinking": "private reasoning"}},
+        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE", "metadata": meta,
+         "userInput": {"items": [{"text": "Later question"}]}}]
+    if fault == "open":
+        steps.pop()
+    elif fault == "mixed":
+        steps[1]["status"] = "CORTEX_STEP_STATUS_RUNNING"
+    elif fault == "foreign":
+        steps[1]["metadata"] = {**meta, "executionId": "other"}
+    elif fault == "no-visible-answer":
+        steps[2]["plannerResponse"].pop("response")
+    elif fault == "tool-call":
+        steps[2]["plannerResponse"]["toolCalls"] = [{"name": "unsettled"}]
+    elif fault == "unknown":
+        steps[1]["type"] = "FUTURE_NATIVE_OPERATION"
+    elif fault == "oversize":
+        steps[2]["plannerResponse"]["response"] = "x" * (256 * 1024)
+    trajectory = {"trajectory": {"cascadeId": "native"}, "numTotalSteps": len(steps),
+        "status": "RUNNING" if fault == "busy" else "CASCADE_RUN_STATUS_IDLE"}
+    calls = []
+    def read(_url, _token, method, _body, **_):
+        calls.append(method)
+        if method == "GetCascadeTrajectory":
+            return deepcopy(trajectory)
+        current = deepcopy(steps)
+        if fault == "changed" and calls.count("GetCascadeTrajectorySteps") > 1:
+            current[2]["plannerResponse"]["response"] = "Changed draft"
+        return {"steps": current}
+    monkeypatch.setattr(native, "discover_antigravity_language_server", lambda: {"url": "local", "csrf_token": "test"})
+    monkeypatch.setattr(native, "_rpc_post", read)
+    monkeypatch.setattr(native, "manager_input_matches", lambda *_args, **_kwargs: True)
+    proof = native.AntigravityAdapter.interrupted_answer_observation(cascade_id="native", request_ref="f" * 24,
+        message="Original question", started_at="2026-01-01T00:00:00Z", completed_at="2026-01-01T00:00:10Z",
+        input_anchor={"thread_id": "native", "start_step_index": 0,
+            "input_sha256": "0" * 64 if fault == "wrong-hash" else hashlib.sha256(prompt.encode()).hexdigest()})
+    if fault is None:
+        assert proof["visible_results"] == [{"visible_response": "Existing visible draft"}]
+        assert proof["read_only_review"] is True and proof["no_final_answer"] is True
+        assert "private reasoning" not in json.dumps(proof)
+    else:
+        assert proof is None
+    assert not set(calls) - {"GetCascadeTrajectory", "GetCascadeTrajectorySteps"}
+
+
+@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "no-capacity", "review-action", "review-gate", "queue-change", "already-codex", "post-enqueue-crash", "lost-ack", "new-question-during-answer", "restricted-review", "primary-quota-unreadable", "native-tail", "authenticated-host", "file-read", "file-running", "cleared-answer"])
 def test_original_worker_answers_interrupted_steward_question_once(tmp_path, language_server, monkeypatch, fault):  # noqa: F811
     import loopx.chat_manager as manager_module
     import loopx.chat_runtime as runtime_module
@@ -122,6 +177,8 @@ def test_original_worker_answers_interrupted_steward_question_once(tmp_path, lan
         original = transport.core.read_request(ref)
         failed = runtime.wait_for_turn(session_id=original["session_id"], turn_id=original["turn_id"], timeout_sec=15)
         assert failed["status"] == "failed" and failed["delta_count"] == 0
+        if fault == "cleared-answer":
+            store.update_turn(original["session_id"], original["turn_id"], error_code="runtime_error")
         original_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes()
         path = transport.root / f"{ref}.json"
         transport.reconcile_request(path)
@@ -161,6 +218,22 @@ def visible_tools(*args, **kwargs):
                         (5, "CORTEX_STEP_TYPE_ERROR_MESSAGE", {"errorMessage": {"error": {"errorCode": 429}}})]:
                     steps.append({"type": kind, "status": "CORTEX_STEP_STATUS_DONE", "metadata": {
                         **meta, "executionId": "other-execution", "sourceTrajectoryStepInfo": {"stepIndex": index}}, **payload})
+            if fault == "cleared-answer":
+                # Completion metadata/tool payloads were cleared, but the
+                # original input and visible response survive. A later input
+                # closes this segment; no completed-result adoption is proved.
+                steps = [user,
+                    {"type": "CORTEX_STEP_TYPE_GENERIC", "metadata": {**meta,
+                        "sourceTrajectoryStepInfo": {"stepIndex": 1}}, "generic": {}},
+                    {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "metadata": {**meta,
+                        "sourceTrajectoryStepInfo": {"stepIndex": 2}},
+                     "plannerResponse": {"response": "Original verified output, not a new execution",
+                        "thinking": "never forward this"}}]
+                for step in steps:
+                    step["status"] = "CORTEX_STEP_STATUS_CLEARED"
+                steps.append({"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
+                    "metadata": {**meta, "executionId": "later", "sourceTrajectoryStepInfo": {"stepIndex": 3}},
+                    "userInput": {"items": [{"text": "Later independent question"}]}})
             return {"steps": steps}
         monkeypatch.setattr(native, "_rpc_post", visible_tools)
         if fault == "native-tail":