fix(chat): retain settled file reads for interrupted answer reviews

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -434,7 +434,8 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
                     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_ERROR_MESSAGE": "errorMessage"}.get(kind)
+                    "CORTEX_STEP_TYPE_RUN_COMMAND": "runCommand", "CORTEX_STEP_TYPE_VIEW_FILE": "viewFile",
+                    "CORTEX_STEP_TYPE_ERROR_MESSAGE": "errorMessage"}.get(kind)
                 if payload is None or set(step) - {"type", "status", "metadata", payload}:
                     return None
                 if kind == "CORTEX_STEP_TYPE_RUN_COMMAND":
@@ -443,6 +444,18 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
                         return None
                     results.append({key: command[key] for key in ("commandLine", "cwd", "exitCode", "combinedOutput")
                                     if key in command})
+                elif kind == "CORTEX_STEP_TYPE_VIEW_FILE":
+                    # The original provider's visible file result is evidence,
+                    # never a request to open that URI or repeat an operation.
+                    view = step["viewFile"]
+                    uri, content = view.get("absolutePathUri"), view.get("content")
+                    if not isinstance(uri, str) or not isinstance(content, str):
+                        return None
+                    parsed = urlsplit(uri)
+                    if (parsed.scheme != "file" or parsed.netloc or not parsed.path.startswith("/")
+                            or parsed.query or parsed.fragment):
+                        return None
+                    results.append({"file_uri": uri, "content": content})
                 elif kind == "CORTEX_STEP_TYPE_PLANNER_RESPONSE":
                     if (step.get("plannerResponse") or {}).get("response") or (step.get("plannerResponse") or {}).get("modifiedResponse"):
                         return None  # A visible answer requires the existing result adoption owner.

loopx/control_plane/collaboration/conversation_binding.ts

@@ -1095,6 +1095,21 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
     && matchesWork(observed.unfinished_work)};
 }
 
+function interruptedVisibleResult(value: unknown): boolean {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return false;
+  const result = value as JsonObject;
+  if (Object.keys(result).sort().join(",") === "content,file_uri") {
+    if (typeof result.file_uri !== "string" || typeof result.content !== "string") return false;
+    try {
+      const uri = new URL(result.file_uri);
+      return uri.protocol === "file:" && uri.host === "" && uri.pathname.startsWith("/")
+        && uri.search === "" && uri.hash === "";
+    } catch { return false; }
+  }
+  return typeof result.commandLine === "string" && result.commandLine.length > 0
+    && Number.isSafeInteger(result.exitCode);
+}
+
 /** Interrupted work may have effects. Never retry it. The original worker may
  * reserve one read-only review of existing results, with its own presentation.
  * The adapter retains only the original runtime's authorized reads, with
@@ -1143,9 +1158,7 @@ function planInterruptedStewardAnswer(recovery: JsonObject, row: JsonObject, cur
       || proof.read_only_review !== true || proof.no_final_answer !== true || proof.effects_observed !== true
       || !Array.isArray(proof.visible_results) || !proof.visible_results.length || proof.visible_results.length > 128
       || Buffer.byteLength(JSON.stringify(proof.visible_results), "utf8") > 256 * 1024
-      || proof.visible_results.some(value => !value || typeof value !== "object" || Array.isArray(value)
-        || typeof (value as JsonObject).commandLine !== "string" || !(value as JsonObject).commandLine
-        || !Number.isSafeInteger((value as JsonObject).exitCode))) return candidate;
+      || !proof.visible_results.every(interruptedVisibleResult)) return candidate;
   const capacity = nativeRecoveryCapacity(recovery, row, current, candidate);
   // This is a new restricted answer review, never native business replay.
   return {...capacity, recovery_eligible: capacity.recovery_eligible === true && capacity.answer_executor === "codex"};

tests/control_plane_ts/conversation_binding.test.ts

@@ -58,6 +58,18 @@ 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 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);
+  assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery,
+    native_observation: {...proof, visible_results: [fileResult]}}}).recovery_eligible, true);
+  for (const result of [{...fileResult, file_uri: "https://untrusted.invalid/result.txt"},
+    {...fileResult, file_uri: "file://remote-host/result.txt"}, {...fileResult, file_uri: "file:///result.txt?next=run"},
+    {...fileResult, content: null}, {file_uri: fileResult.file_uri}, {...fileResult, newEffect: true},
+    {...fileResult, content: "x".repeat(256 * 1024)}]) {
+    assert.notEqual(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery,
+      native_observation: {...proof, visible_results: [result]}}}).recovery_eligible, true);
+  }
   assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery,
     metadata_capacity: {alternate_executor: capacity.alternate_executor}}}).recovery_eligible, true,
     "a verified alternate route does not depend on an unrelated primary quota read");

tests/test_steward_interrupted_answer.py

@@ -15,7 +15,9 @@
 from loopx.extensions.lark.conversation_identity import identity_ref
 
 
-@pytest.mark.parametrize("fault", [None, "terminal-without-execution", "foreign-terminal", "busy", "changed", "truncated", "wrong-hash", "async", "unknown", "final", "reasoning"])
+@pytest.mark.parametrize("fault", [None, "terminal-without-execution", "foreign-terminal", "busy", "changed", "truncated", "wrong-hash", "async", "unknown", "final", "reasoning",
+    "file-read", "file-only", "file-running", "file-foreign", "file-changed", "file-remote", "file-missing-content",
+    "file-missing-uri", "file-unknown-field", "file-oversize"])
 def test_interrupted_native_inventory_is_exact_stable_and_contains_no_reasoning(monkeypatch, fault):
     # The real matcher is exercised in the original-worker integration below;
     # this bounded native inventory fixture isolates index/effect/read stability.
@@ -33,6 +35,26 @@ def test_interrupted_native_inventory_is_exact_stable_and_contains_no_reasoning(
          "errorMessage": {"error": {"errorCode": 429}}},
         {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE", "metadata": meta,
          "userInput": {"items": [{"text": "later independent question"}]}}]
+    if isinstance(fault, str) and fault.startswith("file-"):
+        view = {"type": "CORTEX_STEP_TYPE_VIEW_FILE", "status": "CORTEX_STEP_STATUS_DONE", "metadata": meta,
+            "viewFile": {"absolutePathUri": "file:///authorized/existing-result.txt", "content": "Original visible file result"}}
+        steps.insert(3, view)
+        if fault == "file-only":
+            steps.pop(2)
+        elif fault == "file-running":
+            view["status"] = "CORTEX_STEP_STATUS_RUNNING"
+        elif fault == "file-foreign":
+            view["metadata"] = {**meta, "executionId": "another-execution"}
+        elif fault == "file-remote":
+            view["viewFile"]["absolutePathUri"] = "https://untrusted.invalid/result.txt"
+        elif fault == "file-missing-content":
+            view["viewFile"].pop("content")
+        elif fault == "file-missing-uri":
+            view["viewFile"].pop("absolutePathUri")
+        elif fault == "file-unknown-field":
+            view["newEffect"] = {}
+        elif fault == "file-oversize":
+            view["viewFile"]["content"] = "x" * (256 * 1024)
     if fault == "async":
         steps[2]["runCommand"]["blocking"] = False
     elif fault == "unknown":
@@ -56,6 +78,8 @@ def read(_url, _token, method, _body, **_):
         result = deepcopy(steps[:-1] if fault == "truncated" else steps)
         if fault == "changed" and calls.count(method) > 1:
             result[2]["runCommand"]["combinedOutput"] = "changed"
+        if fault == "file-changed" and calls.count(method) > 1:
+            result[3]["viewFile"]["content"] = "changed"
         return {"steps": result}
     monkeypatch.setattr(native, "discover_antigravity_language_server", lambda: {"url": "synthetic", "csrf_token": "synthetic"})
     monkeypatch.setattr(native, "_rpc_post", read)
@@ -64,16 +88,20 @@ def read(_url, _token, method, _body, **_):
         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 in {None, "terminal-without-execution"}:
-        assert proof["end_step_index"] == 4 and proof["total_steps"] == 5
-        assert proof["visible_results"][0]["combinedOutput"] == "already observed"
+    if fault in {None, "terminal-without-execution", "file-read", "file-only"}:
+        assert proof["end_step_index"] == len(steps) - 1 and proof["total_steps"] == len(steps)
+        if fault != "file-only":
+            assert proof["visible_results"][0]["combinedOutput"] == "already observed"
+        if fault in {"file-read", "file-only"}:
+            assert proof["visible_results"][-1] == {"file_uri": "file:///authorized/existing-result.txt",
+                "content": "Original visible file result"}
         assert "thinking" not in json.dumps(proof) and "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"])
+@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"])
 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
@@ -104,7 +132,7 @@ def visible_tools(*args, **kwargs):
             method = args[2]
             if method == "GetCascadeTrajectory":
                 packet = post(*args, **kwargs)
-                packet["numTotalSteps"] += 4 if native_tail else 2
+                packet["numTotalSteps"] += 4 if native_tail else 3 if fault in {"file-read", "file-running"} else 2
                 return packet
             if method != "GetCascadeTrajectorySteps":
                 return post(*args, **kwargs)
@@ -121,6 +149,13 @@ def visible_tools(*args, **kwargs):
                     "blocking": True, "exitCode": 0, "combinedOutput": "Original verified output, not a new execution"}}
             error["metadata"]["sourceTrajectoryStepInfo"] = {"stepIndex": 3}
             steps = [user, planner, tool, error]
+            if fault in {"file-read", "file-running"}:
+                error["metadata"]["sourceTrajectoryStepInfo"] = {"stepIndex": 4}
+                steps.insert(3, {"type": "CORTEX_STEP_TYPE_VIEW_FILE",
+                    "status": "CORTEX_STEP_STATUS_RUNNING" if fault == "file-running" else "CORTEX_STEP_STATUS_DONE",
+                    "metadata": {**meta, "sourceTrajectoryStepInfo": {"stepIndex": 3}},
+                    "viewFile": {"absolutePathUri": (workspace / "original-result.txt").as_uri(),
+                        "content": "Original visible file result"}})
             if native_tail:
                 for index, kind, payload in [(4, "CORTEX_STEP_TYPE_USER_INPUT", {"userInput": {"items": [{"text": "Another native question"}]}}),
                         (5, "CORTEX_STEP_TYPE_ERROR_MESSAGE", {"errorMessage": {"error": {"errorCode": 429}}})]:
@@ -236,7 +271,7 @@ def crash(**kwargs):
             provider.lose_next_reply = True
         transport.reconcile_request(path)
         recovered = transport.core.read_request(ref)
-        if fault in {"stop", "source-changed", "no-capacity", "queue-change"}:
+        if fault in {"stop", "source-changed", "no-capacity", "queue-change", "file-running"}:
             assert recovered["turn_id"] == original["turn_id"] and not recovered.get("context_recovery")
         else:
             assert recovered["context_recovery"]["answer_review"] is True
@@ -257,6 +292,8 @@ def crash(**kwargs):
             profile = "loopx_steward_host_read" if host_reads else "loopx_workspace_only_read"
             assert reviews[0]["permissions"] == profile
             assert "Original verified output" in reviews[0]["input"][0]["text"]
+            if fault == "file-read":
+                assert "Original visible file result" in reviews[0]["input"][0]["text"]
             assert "never forward this" not in reviews[0]["input"][0]["text"]
             opens = [p["params"] for p in packets if p.get("method") == "thread/start"]
             review_opens = [p for p in opens if p.get("permissions") == profile]