diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index e040e6faa1..f8fdf7670f 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/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: diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 2df7baba14..b499a438c5 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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 { diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index e07b91e29b..ab3cb45d51 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/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); diff --git a/tests/test_steward_interrupted_answer.py b/tests/test_steward_interrupted_answer.py index 5b6e236da5..71ee09dc39 100644 --- a/tests/test_steward_interrupted_answer.py +++ b/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":