diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index b49dc120b8..e040e6faa1 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/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. diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index eb5f98b300..f21e5bd9cd 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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"}; diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 9c1f39757d..e07b91e29b 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/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"); diff --git a/tests/test_steward_interrupted_answer.py b/tests/test_steward_interrupted_answer.py index b1c0e12772..5b6e236da5 100644 --- a/tests/test_steward_interrupted_answer.py +++ b/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]