diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index 1ce68faade..ec4a0f02d5 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -342,7 +342,8 @@ class AntigravityAdapter: def completed_result_observation(*, cascade_id: str, request_ref: str, message: str, started_at: str, completed_at: str, work_dir: Path, input_anchor: dict[str, Any] | None = None, - include_raw_response: bool = False) -> dict[str, Any] | None: + include_raw_response: bool = False, + retained_result: dict[str, Any] | None = None) -> dict[str, Any] | None: """Read an existing result; never start, send to or cancel a cascade. Legacy owner requests require the exact Core-generated source context, @@ -350,27 +351,40 @@ def completed_result_observation(*, cascade_id: str, request_ref: str, message: New Turns additionally freeze their prompt hash before provider send. """ try: - found = discover_antigravity_language_server() - if found is None: - return None - def read(method: str, **payload: Any) -> dict[str, Any]: - return _rpc_post(found["url"], found["csrf_token"], method, - {"cascadeId": cascade_id, **payload}, timeout=5.0) - - before = read("GetCascadeTrajectory") - total = before.get("numTotalSteps") - if (before.get("status") != "CASCADE_RUN_STATUS_IDLE" or type(total) is not int or total < 2 - or (before.get("trajectory") or {}).get("cascadeId") != cascade_id): - return None - raw = read("GetCascadeTrajectorySteps", startIndex=0, endIndex=total).get("steps") - if not isinstance(raw, list): - return None - indexed, covered = _observed_turn_steps(raw, 0, total) - after = read("GetCascadeTrajectory") - if (covered != total or after.get("status") != before["status"] - or after.get("numTotalSteps") != total - or (after.get("trajectory") or {}).get("cascadeId") != cascade_id): - return None + if retained_result is not None: + # Core retained this exact invocation before parsing/delivery. + # It is not a fresh claim about the provider's latest input. + if (input_anchor is None or retained_result.get("thread_id") != cascade_id + or retained_result.get("start_step_index") != input_anchor.get("start_step_index")): + return None + total = retained_result.get("total_steps") + if type(total) is not int or total < 2: + return None + indexed, covered = _observed_turn_steps(retained_result["steps"], input_anchor["start_step_index"], total) + if covered != total: + return None + else: + found = discover_antigravity_language_server() + if found is None: + return None + def read(method: str, **payload: Any) -> dict[str, Any]: + return _rpc_post(found["url"], found["csrf_token"], method, + {"cascadeId": cascade_id, **payload}, timeout=5.0) + + before = read("GetCascadeTrajectory") + total = before.get("numTotalSteps") + if (before.get("status") != "CASCADE_RUN_STATUS_IDLE" or type(total) is not int or total < 2 + or (before.get("trajectory") or {}).get("cascadeId") != cascade_id): + return None + raw = read("GetCascadeTrajectorySteps", startIndex=0, endIndex=total).get("steps") + if not isinstance(raw, list): + return None + indexed, covered = _observed_turn_steps(raw, 0, total) + after = read("GetCascadeTrajectory") + if (covered != total or after.get("status") != before["status"] + or after.get("numTotalSteps") != total + or (after.get("trajectory") or {}).get("cascadeId") != cascade_id): + return None inputs = [(i, step) for i, step in indexed if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"] if not inputs: return None @@ -425,6 +439,7 @@ def read(method: str, **payload: Any) -> dict[str, Any]: "input_created_at": metadata["createdAt"], "start_step_index": start, "total_steps": total, "input_sha256": prompt_hash, "answer_sha256": hashlib.sha256(answer.encode()).hexdigest(), "latest_input_verified": start == inputs[-1][0], + **({"observation_source": "retained_turn_result"} if retained_result is not None else {}), **({"end_step_index": end} if start != inputs[-1][0] else {}), "normal_terminal_verified": True, "response": response, **({"raw_response": answer} if include_raw_response else {})} @@ -921,6 +936,26 @@ def _run_turn( break if event_sink is not None: + # Keep only completion facts and the original input/final text, + # never native tool output or raw reasoning. Core consumes this + # privately and durably before parsing can fail or lose text. + facts = [] + for index, step in current_steps: + metadata = step.get("metadata") or {} + fact = {"type": step.get("type"), "status": step.get("status"), "metadata": { + "createdAt": metadata.get("createdAt"), "executionId": metadata.get("executionId"), + "sourceTrajectoryStepInfo": {"stepIndex": index}}} + if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT": + fact["userInput"] = step.get("userInput") + if step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE": + fact["errorMessage"] = {"shouldShowUser": (step.get("errorMessage") or {}).get("shouldShowUser")} + if index == current_steps[-1][0]: + fact["plannerResponse"] = {key: planner[key] for key in + ("response", "modifiedResponse", "stopReason", "toolCalls") if key in planner} + facts.append(fact) + self._check_interrupted() + event_sink("turn.native_result", {"thread_id": self.cascade_id, "upstream_turn_id": turn_id, + "start_step_index": start_step_index, "total_steps": total_steps, "steps": facts}) tail = stream_filter.finish() if tail: emitted_visible_delta = True diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 78b8f120b5..29f39da982 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -351,6 +351,16 @@ def _original_native_result(self, row: dict[str, Any], session: dict[str, Any], if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")] if len(anchors) > 1: return None + # Retained facts can repair only a completed Turn's text. Failed-Turn + # recovery keeps its live/latest-input requirements; old capture does + # not authorize another invocation or change unknown-effect handling. + retained = self.controller.store.native_result(session["session_id"], turn["turn_id"]) + if (turn.get("status") == "completed" and len(anchors) == 1 and retained + and retained.get("upstream_turn_id") == turn.get("upstream_turn_id")): + return AntigravityAdapter.completed_result_observation( + cascade_id=session["upstream_thread_id"], request_ref=row["request_ref"], message=row["message"], + started_at=turn["started_at"], completed_at=turn["completed_at"], work_dir=Path(context["project"]), + input_anchor=anchors[0], include_raw_response=include_raw_response, retained_result=retained) return AntigravityAdapter.completed_result_observation( cascade_id=session["upstream_thread_id"], request_ref=row["request_ref"], message=row["message"], started_at=turn["started_at"], completed_at=turn["completed_at"], work_dir=Path(context["project"]), diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index d705a542c5..a6b2d59653 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -1722,6 +1722,16 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None: with self.lock: if (session_id, turn_id) in self.cancelled_turns: return + if kind == "turn.native_result": + if getattr(adapter, "adapter_kind", None) != "antigravity_ls": + raise ValueError("native result capture is unavailable for this adapter") + try: + self.store.retain_native_result(session_id, turn_id, payload) + except OSError as exc: + # Supplemental retention cannot discard a valid primary + # answer. Missing facts still cannot authorize recovery. + logging.getLogger(__name__).warning("Native completion capture unavailable: %s", type(exc).__name__) + return event_buffer.emit(kind, payload) if kind == "native_goal.status": self.store.update_session(session_id, native_goal=payload) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 84f1720a1b..ff660ba6a9 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -1390,6 +1390,28 @@ def load_turn(self, session_id: str, turn_id: str) -> dict[str, Any] | None: payload = _read_json(self._turn_path(session_id, turn_id)) return payload if payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION else None + def native_result(self, session_id: str, turn_id: str) -> dict[str, Any]: + """Private provider facts, outside public Turn/event projections.""" + token = _opaque_id(turn_id, field="turn_id") + return _read_json(self._session_dir(session_id) / "native-results" / f"{token}.json") + + def retain_native_result(self, session_id: str, turn_id: str, result: dict[str, Any]) -> None: + """Freeze once under the original Turn fence, before response handling.""" + with exclusive_file_lock(self._turn_path(session_id, turn_id), agent_id="loopx-chat", operation="retain_native_result"): + turn = self.load_turn(session_id, turn_id) + if turn is None or turn.get("status") != "running": + return + if not result.get("upstream_turn_id") or result["upstream_turn_id"] != turn.get("upstream_turn_id"): + raise ValueError("native result belongs to another invocation") + token = _opaque_id(turn_id, field="turn_id") + path = self._session_dir(session_id) / "native-results" / f"{token}.json" + if path.exists(): + if _read_json(path) != result: + raise ValueError("the original native result is immutable") + return + path.parent.mkdir(mode=0o700, exist_ok=True) + _atomic_write_json(path, result) + def update_turn( self, session_id: str, diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 1f6826e37b..5735e75d0f 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -350,7 +350,10 @@ def test_original_queue_recovers_historical_truncated_answer_without_execution(t start = AntigravityAdapter.start_turn def old_parser(adapter, message, event_sink): - response = start(adapter, message, event_sink) + # This legacy service had neither the fixed parser nor durable + # native result capture. Keep exercising its live-history path. + response = start(adapter, message, lambda kind, payload: + event_sink(kind, payload) if kind != "turn.native_result" else None) return {**response, "message": "Original `"} monkeypatch.setattr(AntigravityAdapter, "start_turn", old_parser) @@ -435,6 +438,147 @@ def crash(*args, **kwargs): fixture.close() +@pytest.mark.parametrize("case", ["normal", "provider-unavailable", "closed", "source-changed", "wrong-native-turn", "wrong-anchor", + "mixed-execution", "failed-core", "lost-ack"]) +def test_original_queue_retains_completed_answer_across_native_cleanup(tmp_path, language_server, monkeypatch, case): + from test_native_steward_private import steward + from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter + from loopx.chat_agent import CodexChatAgentError + from loopx.chat_store import _read_json + + monkeypatch.setenv("LOOPX_MANAGER_MODEL", "gemini-4-argon-high") + monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high") + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, *_ = next(fixture) + try: + binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="antigravity", context_kind="steward") + complete = "Original `` is literal.\n\n" + "完整原答复包含中文和🙂。\n" * 200 + "Original final paragraph." + language_server.update(plain_text=True, answers={0: '' + json.dumps({ + "message": complete, "proposals": [], "gate": None}) + ''}) + start = AntigravityAdapter.start_turn + + def damaged_response(adapter, message, event_sink): + response = start(adapter, message, event_sink) + if case == "failed-core": + raise CodexChatAgentError("Synthetic post-processing failure", gate=None, error_code="upstream_transport_failure") + return {**response, "message": "Original `"} + + monkeypatch.setattr(AntigravityAdapter, "start_turn", damaged_response) + event = provider.event("steward-app", "retained-original", "Complete original request") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + assert runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)["status"] == ( + "failed" if case == "failed-core" else "completed") + assert transport.reconcile_request(transport.root / f"{ref}.json") == 1 + retained = store.native_result(row["session_id"], row["turn_id"]) + result_path = store._session_dir(row["session_id"]) / "native-results" / f"{row['turn_id']}.json" + assert result_path.stat().st_mode & 0o777 == 0o600 + assert retained["steps"][-1]["plannerResponse"]["response"].endswith('') + assert all("thinking" not in step.get("plannerResponse", {}) and "runCommand" not in step for step in retained["steps"]) + assert not any(item["kind"] == "turn.native_result" for item in store.events_after(row["session_id"], row["turn_id"], None)) + monkeypatch.setattr(AntigravityAdapter, "start_turn", start) + later = provider.event("steward-app", "retained-later", "A separate later request") + transport.admit("steward-app", later) + later_row = transport.core.read_request(identity_ref(binding["provider_ref"], later["message_id"])) + assert runtime.wait_for_turn(session_id=later_row["session_id"], turn_id=later_row["turn_id"], timeout_sec=15)["status"] == "completed" + transport.reconcile_request(transport.root / f"{later_row['request_ref']}.json") + store.compact_completed_events(older_than_hours=0) + # The provider can no longer prove even a normal stop, and the original + # input is historical. Recovery must use the privately retained facts. + from copy import deepcopy + cleared = deepcopy(retained["steps"] + store.native_result(later_row["session_id"], later_row["turn_id"])["steps"]) + for step in cleared: + step["status"] = "CORTEX_STEP_STATUS_CLEARED" + step.get("plannerResponse", {}).pop("stopReason", None) + language_server.update(frames=[], frame={"at": 0, "steps": cleared, "status": "CASCADE_RUN_STATUS_IDLE"}) + if case in {"wrong-native-turn", "wrong-anchor", "mixed-execution"}: + if case == "wrong-native-turn": + retained["upstream_turn_id"] = "another-invocation" + elif case == "wrong-anchor": + retained["steps"][0]["userInput"]["items"][0]["text"] += "foreign" + else: + retained["steps"][-1]["metadata"]["executionId"] = "another-execution" + # Corrupt only the isolated fixture's private artifact. + result_path.write_text(json.dumps(retained)) + elif case == "provider-unavailable": + monkeypatch.setattr("loopx.capabilities.native_chat.antigravity_adapter.discover_antigravity_language_server", lambda: None) + elif case == "closed": + store.update_session(row["session_id"], status="closed") + elif case == "source-changed": + provider.messages[event["message_id"]]["content"] = "Edited source" + elif case == "lost-ack": + provider.lose_next_reply = True + turn_bytes = store._turn_path(row["session_id"], row["turn_id"]).read_bytes() + events_bytes = store._event_path(row["session_id"], row["turn_id"]).read_bytes() + retained_bytes = result_path.read_bytes() + writes = len(provider.writes) + calls = [item for item in language_server["calls"] if item[0] in {"StartCascade", "SendUserCascadeMessage", "CancelCascadeInvocation"}] + monkeypatch.setattr(runtime, "enqueue_turn", lambda **_: pytest.fail("Result recovery enqueued work")) + monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: pytest.fail("Result recovery resumed work")) + restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, runner=provider, cli_bin="lark-cli") + restarted.reconcile() + recovered = restarted.core.read_request(ref).get("native_result_recovery") + if case in {"normal", "provider-unavailable", "lost-ack"}: + assert recovered["native_observation"]["observation_source"] == "retained_turn_result" + assert recovered["native_observation"]["response"]["message"] == complete + assert recovered["delivery_verified"] is (case != "lost-ack") + assert len(provider.writes) == writes + 1 + if case != "lost-ack": + receipt = _read_json(transport.root / f"{ref}.json")["deliveries"][f"native_result.{row['turn_id']}"] + body = json.loads(provider.messages[receipt["attempt"]["message_ref"]]["body"]["content"]) + assert complete in "".join(item.get("text", "") for line in body["zh_cn"]["content"] for item in line) + restarted.reconcile() + assert len(provider.writes) == writes + 1 + else: + assert recovered is None and len(provider.writes) == writes + assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == turn_bytes + assert store._event_path(row["session_id"], row["turn_id"]).read_bytes() == events_bytes + assert result_path.read_bytes() == retained_bytes + assert [item for item in language_server["calls"] if item[0] in {"StartCascade", "SendUserCascadeMessage", "CancelCascadeInvocation"}] == calls + finally: + fixture.close() + + +def test_native_capture_io_failure_preserves_the_primary_answer(tmp_path, language_server, monkeypatch, caplog): + from test_native_steward_private import steward + + monkeypatch.setenv("LOOPX_MANAGER_MODEL", "gemini-4-argon-high") + monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high") + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, *_ = next(fixture) + try: + binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="antigravity", context_kind="steward") + retain = store.retain_native_result + + def blocked_capture(session_id, turn_id, result): + (store._session_dir(session_id) / "native-results").write_text("Synthetic occupied artifact directory") + return retain(session_id, turn_id, result) + + monkeypatch.setattr(store, "retain_native_result", blocked_capture) + language_server.update(plain_text=True, answers={0: "Verified original complete answer."}) + event = provider.event("steward-app", "original-primary", "Complete original request") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + turn = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15) + assert turn["status"] == "completed" + assert store.native_result(row["session_id"], row["turn_id"]) == {} + transport.reconcile_request(transport.root / f"{ref}.json") + receipt = json.loads((transport.root / f"{ref}.json").read_text())["deliveries"]["terminal"] + body = json.loads(provider.messages[receipt["attempt"]["message_ref"]]["body"]["content"]) + assert "Verified original complete answer." in "".join(item.get("text", "") for line in body["zh_cn"]["content"] for item in line) + assert sum(method == "SendUserCascadeMessage" for method, _ in language_server["calls"]) == 1 + assert not any(method == "CancelCascadeInvocation" for method, _ in language_server["calls"]) + warning = [record.getMessage() for record in caplog.records if record.name == "loopx.chat_runtime"] + assert any("Native completion capture unavailable: FileExistsError" in message for message in warning) + assert all(str(tmp_path) not in message for message in warning) + finally: + fixture.close() + + def test_active_original_reply_does_not_wait_for_older_native_result_probes(tmp_path, language_server, monkeypatch): # noqa: F811 from concurrent.futures import ThreadPoolExecutor from test_native_steward_private import steward