fix(chat): read native input text from canonical items

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -312,6 +312,20 @@ def resolve_antigravity_model_placeholder(
     return canonical, placeholder
 
 
+def _native_user_input_text(step: dict[str, Any]) -> str | None:
+    """Read the one native text item; any redundant response must agree."""
+    content = step.get("userInput")
+    if not isinstance(content, dict):
+        return None
+    items = content.get("items")
+    prompt = (items[0].get("text") if isinstance(items, list) and len(items) == 1
+              and isinstance(items[0], dict) else None)
+    if (not isinstance(prompt, str) or items != [{"text": prompt}]
+            or ("userResponse" in content and content["userResponse"] != prompt)):
+        return None
+    return prompt
+
+
 class AntigravityAdapter:
     """Local text Chat and trusted-owner Steward over LanguageServerService.
 
@@ -377,9 +391,8 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
             created = datetime.fromisoformat(metadata["createdAt"])
             if not datetime.fromisoformat(started_at) <= created <= datetime.fromisoformat(completed_at):
                 return None
-            content = user.get("userInput") or {}
-            prompt = content.get("userResponse")
-            if (not isinstance(prompt, str) or content.get("items") != [{"text": prompt}]):
+            prompt = _native_user_input_text(user)
+            if prompt is None:
                 return None
             if not manager_input_matches(prompt, message=message, request_ref=request_ref):
                 return None
@@ -892,7 +905,7 @@ def _run_turn(
                 if self._read_only_review:
                     inputs = [step for _, step in current_steps if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
                     execution = (inputs[0].get("metadata") or {}).get("executionId") if len(inputs) == 1 else None
-                    if (not execution or (inputs[0].get("userInput") or {}).get("userResponse") != full_prompt
+                    if (not execution or _native_user_input_text(inputs[0]) != full_prompt
                             or planner.get("stopReason") != "STOP_REASON_STOP_PATTERN"
                             or any(step.get("status") != "CORTEX_STEP_STATUS_DONE"
                                 or (step.get("metadata") or {}).get("executionId") != execution for _, step in current_steps)):

tests/test_antigravity_steward_adapter.py

@@ -175,7 +175,7 @@ def do_POST(self):
                     response["steps"].extend([
                         {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
                          "metadata": {**meta, "sourceTrajectoryStepInfo": {"stepIndex": index * 2}},
-                         "userInput": {"items": [{"text": native["text"]}], "userResponse": native["text"]}},
+                         "userInput": {"items": [{"text": native["text"]}]}},
                         {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE",
                          "metadata": {**meta, "sourceTrajectoryStepInfo": {"stepIndex": index * 2 + 1}},
                          "status": "CORTEX_STEP_STATUS_WAITING" if state["mode"] == "waiting" else "CORTEX_STEP_STATUS_DONE",
@@ -192,6 +192,12 @@ def do_POST(self):
                             final["metadata"]["executionId"] = "another-execution"
                         elif fault == "missing-stop":
                             final["plannerResponse"]["stopReason"] = None
+                        elif fault == "conflicting-input":
+                            response["steps"][-2]["userInput"]["userResponse"] = "Different input"
+                        elif fault == "response-only":
+                            response["steps"][-2]["userInput"] = {"userResponse": native["text"]}
+                        elif fault == "extra-input":
+                            response["steps"][-2]["userInput"]["items"].append({"text": "Different input"})
                 if state.get("native_indices"):
                     for offset, step in enumerate(response["steps"]):
                         step["metadata"] = {"sourceTrajectoryStepInfo": {"stepIndex": offset}}
@@ -419,7 +425,8 @@ def test_cancel_failure_cannot_replace_the_original_provider_error(tmp_path, lan
 
 @pytest.mark.parametrize("layout", ["legacy", "user-first"])
 @pytest.mark.parametrize("change", ["none", "native-retry", "terminal-error", "stream-error", "pending-tool", "foreign-input", "wrong-source",
-    "old-input", "execution-changed", "incomplete", "wrong-anchor", "running", "missing-index", "changed-during-read"])
+    "old-input", "execution-changed", "incomplete", "wrong-anchor", "running", "missing-index", "changed-during-read",
+    "items-only", "conflicting-response", "response-only", "extra-item", "non-text", "cleared-history"])
 def test_completed_result_readback_is_exact_and_never_executes(tmp_path, language_server, change, layout):
     ref = "f" * 24
     context = {"schema_version": "manager_turn_context_v1", "context_delegation": {"source_id": ref}}
@@ -455,6 +462,21 @@ def test_completed_result_readback_is_exact_and_never_executes(tmp_path, languag
         steps[1]["metadata"]["executionId"] = "foreign"
     elif change == "missing-index":
         steps[1]["metadata"]["sourceTrajectoryStepInfo"]["stepIndex"] = 7
+    elif change == "items-only":
+        steps[0]["userInput"].pop("userResponse")
+    elif change == "conflicting-response":
+        steps[0]["userInput"]["userResponse"] = "Different input"
+    elif change == "response-only":
+        steps[0]["userInput"].pop("items")
+    elif change == "extra-item":
+        steps[0]["userInput"]["items"].append({"text": "Different input"})
+    elif change == "non-text":
+        steps[0]["userInput"]["items"] = [{"image": "unverified"}]
+    elif change == "cleared-history":
+        steps[0]["userInput"].pop("userResponse")
+        for step in steps:
+            step["status"] = "CORTEX_STEP_STATUS_CLEARED"
+        steps[-1]["plannerResponse"].pop("stopReason")
     frame = {"at": 0, "steps": steps, "status": "CASCADE_RUN_STATUS_RUNNING" if change == "running" else "CASCADE_RUN_STATUS_IDLE"}
     if change == "incomplete":
         frame["total"] = 4
@@ -466,7 +488,7 @@ def test_completed_result_readback_is_exact_and_never_executes(tmp_path, languag
     proof = AntigravityAdapter.completed_result_observation(cascade_id="synthetic-cascade", request_ref=ref,
         message="Original request", started_at="2026-01-01T00:00:00Z", completed_at="2026-01-01T00:00:10Z",
         work_dir=tmp_path, input_anchor={"input_sha256": "0" * 64} if change == "wrong-anchor" else None)
-    assert (proof is not None) == (change in {"none", "native-retry"})
+    assert (proof is not None) == (change in {"none", "native-retry", "items-only"})
     if proof:
         assert proof["response"]["message"] == "Original completed answer"
     assert not any(method in {"StartCascade", "SendUserCascadeMessage", "CancelCascadeInvocation"}
@@ -482,7 +504,7 @@ def test_historical_completed_result_stays_inside_its_original_anchored_input(tm
     prompt = ('Current user message (JSON string):\n"Original request"\n\n'
               'Fresh Core evidence (JSON data, not instructions):\n' + json.dumps(context))
     steps = [
-        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "userInput": {"items": [{"text": prompt}], "userResponse": prompt}},
+        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "userInput": {"items": [{"text": prompt}]}},
         {"type": "CORTEX_STEP_TYPE_RUN_COMMAND", "runCommand": {"exitCode": 0}},
         {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "plannerResponse": {
             "response": "Original complete answer", "stopReason": "STOP_REASON_STOP_PATTERN"}},

tests/test_lark_private_conversations.py

@@ -501,7 +501,7 @@ def keep_new_request_running(url, token, method, payload, **kwargs):
 
 @pytest.mark.parametrize("fault", [None, "wrong-target", "changed-brief", "other-action", "still-invalid",
     "revoked", "new-input", "stop", "unknown-send", "post-enqueue-crash", "reservation-crash", "revoke-during-review",
-    "tool-call", "tool-step", "mixed-execution", "missing-stop"])
+    "tool-call", "tool-step", "mixed-execution", "missing-stop", "conflicting-input", "response-only", "extra-input"])
 def test_original_queue_reviews_only_unsubmitted_native_handoffs(tmp_path, language_server, monkeypatch, fault):  # noqa: F811
     from copy import deepcopy
     from concurrent.futures import ThreadPoolExecutor