diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index 6cd16d030e..395c71f6a7 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -360,7 +360,19 @@ def read(method: str, **payload: Any) -> dict[str, Any]: inputs = [(i, step) for i, step in indexed if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"] if not inputs: return None - start, user = inputs[-1] + # Historical results are usable only through the original Core + # input anchor. Legacy records still require the latest input. + if input_anchor is not None: + anchored = input_anchor.get("start_step_index") + if type(anchored) is not int: + return None + matching = [item for item in inputs if item[0] == anchored] + if len(matching) != 1: + return None + start, user = matching[0] + else: + start, user = inputs[-1] + end = next((index for index, _ in inputs if index > start), total) metadata = user.get("metadata") or {} created = datetime.fromisoformat(metadata["createdAt"]) if not datetime.fromisoformat(started_at) <= created <= datetime.fromisoformat(completed_at): @@ -376,7 +388,7 @@ def read(method: str, **payload: Any) -> dict[str, Any]: or input_anchor.get("start_step_index") != start or input_anchor.get("input_sha256") != prompt_hash): return None execution = metadata.get("executionId") - current = indexed[start:] + current = [(index, step) for index, step in indexed if start <= index < end] if not isinstance(execution, str) or not execution or any( (step.get("status") != "CORTEX_STEP_STATUS_DONE" and not (step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE" @@ -399,7 +411,9 @@ def read(method: str, **payload: Any) -> dict[str, Any]: return {"thread_id": cascade_id, "request_ref": request_ref, "execution_id": execution, "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": True, "normal_terminal_verified": True, "response": response, + "latest_input_verified": start == inputs[-1][0], + **({"end_step_index": end} if start != inputs[-1][0] else {}), + "normal_terminal_verified": True, "response": response, **({"raw_response": answer} if include_raw_response else {})} except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError): return None diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index ec4e0ab6c3..78b8f120b5 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -77,6 +77,12 @@ def _context_recovery_plan(self, row: dict[str, Any], *, or reservation.get("metadata_review")) and reservation.get("failed_turn_id") == row["turn_id"] else None) events = store.events_after(session["session_id"], turn["turn_id"], None) + anchors = [event["payload"]["native_input"] for event in events + if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")] + # This only selects read-only result probes. Adoption still requires + # the exact original anchor, native final, current source and typed rule. + from ...chat import _review_openings + _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0) preparation = None later_equivalent = None if turn.get("error_code") == "manager_authorization_unavailable": @@ -105,6 +111,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *, "preparation_observation": preparation, "later_equivalent_input_present": later_equivalent, "preparation_scope_available": preparation_scope_available, + "native_input": anchors[0] if len(anchors) == 1 else None, + "response_literal_open": bool(open_literal), "native_observation": observation, "source_verified": source_verified}, }) @@ -174,7 +182,10 @@ def context_recovery_candidate(self, request_ref: str) -> bool: if not turn or turn.get("status") not in {"failed", "completed"}: return False if turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work"): - return False + from ...chat import _review_openings + _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0) + if not open_literal: + return False # A lexical probe hint, never permission to adopt or send. return self._context_recovery_plan(row, discovery_only=True).get("recovery_candidate") is True except (KeyError, ValueError, RuntimeError, OSError): return False diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 928407ae5e..ceb3bbe64d 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -484,9 +484,13 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject { && typeof recovered.message_text === "string" && recovered.message_text.length > 0 && result != null && !["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].some(key => result[key]) && Array.isArray(result.proposals) && result.proposals.length === 0; + const expandedResult = turn?.status === "completed" && response !== null && result != null + && recovered !== null && recovered.turn_id === row.turn_id && recovered.message_id === `native.${row.turn_id}.recovered` + && typeof recovered.message_text === "string" && recovered.message_text.length > 0 + && extendsOriginalText(response, result); return {pending: !returned && row.command === null && row.status === "accepted" && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id - && (metadataOwed || unfinished || turn?.status === "failed" + && (expandedResult || metadataOwed || unfinished || turn?.status === "failed" && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure", "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error", "manager_authorization_unavailable"].includes(String(turn.error_code))) @@ -555,6 +559,22 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject { return {operation: "admit_turn", client_turn_id: `external-${request}`, session_id: session, turn_id: null}; } +function pureResultText(response: JsonObject | null): boolean { + return response?.schema_version === "loopx_chat_agent_response_v0" + && typeof response.message === "string" && response.message.length > 0 + && Array.isArray(response.proposals) && response.proposals.length === 0 + && ["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].every(key => response[key] == null); +} + +function extendsOriginalText(original: JsonObject, result: JsonObject): boolean { + const metadata = (value: JsonObject) => Object.fromEntries(Object.entries(value) + .filter(([key, value]) => key !== "message" && value != null)); + return pureResultText(original) && pureResultText(result) + && (result.message as string).length > (original.message as string).length + && (result.message as string).startsWith(original.message as string) + && isDeepStrictEqual(metadata(original), metadata(result)); +} + /** A new, correlated continuation after an observed pre-execution failure. * A delivered failure notice is not a completed business request. This bounded * migration path never reopens the failed Turn or retries unknown effects. @@ -592,18 +612,30 @@ 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") { + const originalResponse = failed.response && typeof failed.response === "object" && !Array.isArray(failed.response) + ? failed.response as JsonObject : null; + const anchor = recovery.native_input && typeof recovery.native_input === "object" && !Array.isArray(recovery.native_input) + ? recovery.native_input as JsonObject : null; + // The lexical hint only bounds read-only probes. It confers no result, + // source or execution authority; an anchored native final is still needed. + const completedText = failed.status === "completed" && pureResultText(originalResponse) + && recovery.response_literal_open === true && recovery.queued_input_present === false + && anchor !== null && anchor.thread_id === current.upstream_thread_id + && Number.isSafeInteger(anchor.start_step_index) && Number(anchor.start_step_index) >= 0 + && typeof anchor.input_sha256 === "string" && BARE_SHA256_PATTERN.test(anchor.input_sha256); // Result adoption has no admission/dispatch authority. Keep the failed // Turn and original failure delivery, even when native tools ran. if (!steward || failed.session_id !== current.session_id || failed.turn_id !== row.turn_id - || failed.client_turn_id !== `external-${row.request_ref}` || failed.status !== "failed" - || failed.response != null || failed.message !== row.message - || !["response_timeout", "antigravity_rpc_error", "antigravity_transport_error", - "upstream_transport_failure", "server_restarted"].includes(String(failed.error_code)) + || failed.client_turn_id !== `external-${row.request_ref}` || failed.message !== row.message + || (!completedText && (failed.status !== "failed" || failed.response != null + || !["response_timeout", "antigravity_rpc_error", "antigravity_transport_error", + "upstream_transport_failure", "server_restarted"].includes(String(failed.error_code)))) || recovery.attachments_present !== false || recovery.steering_observed !== false || recovery.stop_observed !== false) return no; const adopted = row.native_result_recovery == null ? null : requireJsonObject(row.native_result_recovery, "adopted native result"); if (adopted?.delivery_verified === true) { + if (completedText) return no; // A repaired answer never starts a format review or business continuation. return planNativeMetadataReview(recovery, row, current, failed, adopted, version); } // A fresh native result requires an idle, stable trajectory. Do not spend @@ -621,7 +653,15 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO return {...candidate, recovery_eligible: recovery.source_verified === true && proof.thread_id === current.upstream_thread_id && proof.request_ref === row.request_ref && typeof proof.execution_id === "string" && proof.execution_id.length > 0 - && proof.latest_input_verified === true && proof.normal_terminal_verified === true + && (completedText + ? proof.start_step_index === anchor!.start_step_index && proof.input_sha256 === anchor!.input_sha256 + && (proof.latest_input_verified === true || proof.latest_input_verified === false + && Number.isSafeInteger(proof.end_step_index) + && Number(proof.end_step_index) > Number(proof.start_step_index) + 1 + && Number(proof.end_step_index) < Number(proof.total_steps)) + && response !== null && extendsOriginalText(originalResponse!, response) + : proof.latest_input_verified === true) + && proof.normal_terminal_verified === true && Number.isFinite(created) && created >= start && created <= end && Number.isSafeInteger(proof.start_step_index) && Number(proof.start_step_index) >= 0 && Number.isSafeInteger(proof.total_steps) && Number(proof.total_steps) > Number(proof.start_step_index) + 1 diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 167ef8fa13..d746824414 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -92,6 +92,66 @@ test("Core preparation failures resume through one bounded original-source queue } }); +test("completed truncated text adopts only its anchored original final, never a later result or action", () => { + const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: "antigravity"}; + const configured = planConversationBinding({...request, binding: selected}).state; + const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true}; + const context = resolveBoundConversation({current: configured, binding_id: row.binding_id, + observation, available_projects: [project], ...source}).context; + const ref = "f".repeat(24); + const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "antigravity", + upstream_mode: "default", status: "ready", upstream_thread_id: "native", active_turn_id: null, + channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`, steward_context: context}; + const response = {schema_version: "loopx_chat_agent_response_v0", message: "Original `", proposals: [], gate: null}; + const completed = {session_id: "original", turn_id: "truncated", client_turn_id: `external-${ref}`, + message: "Original request", status: "completed", response, error_code: null, + started_at: "2026-01-01T00:00:00Z", completed_at: "2026-01-01T00:00:10Z"}; + const anchor = {thread_id: "native", start_step_index: 4, input_sha256: "1".repeat(64)}; + const proof = {...anchor, request_ref: ref, execution_id: "original-execution", + input_created_at: "2026-01-01T00:00:05Z", total_steps: 12, end_step_index: 8, + answer_sha256: "2".repeat(64), latest_input_verified: false, normal_terminal_verified: true, + response: {...response, message: "Original `` is a literal marker. Complete answer."}}; + const original = {request_ref: ref, command: null, status: "accepted", delivery_verified: true, + message: completed.message, origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "truncated", source}; + const recovery = {version: 21, failed_turn: completed, native_input: anchor, response_literal_open: true, + queued_input_present: false, steering_observed: false, stop_observed: false, attachments_present: false, + source_verified: true, native_observation: proof}; + const packet = {request: original, binding: selected, current_session: session, context_recovery: recovery}; + assert.equal(planBoundConversationRequest(packet).operation, "adopt_native_result"); + assert.equal(planBoundConversationRequest(packet).recovery_eligible, true); + assert.equal(planBoundConversationRequest(packet).client_turn_id, undefined); + for (const change of [{native_input: null}, {response_literal_open: false}, {queued_input_present: true}, + {source_verified: false}, {stop_observed: true}, {steering_observed: true}, {attachments_present: true}, + {native_input: {...anchor, thread_id: "later"}}, {native_input: {...anchor, input_sha256: "3".repeat(64)}}, + ...[{start_step_index: 5}, {end_step_index: 12}, {end_step_index: 5}, {end_step_index: null}, + {normal_terminal_verified: false}, {input_created_at: "2026-01-02T00:00:00Z"}, + {response: {...proof.response, message: response.message}}, + {response: {...proof.response, message: "Later answer"}}, + {response: {...proof.response, proposals: [{kind: "action"}]}}, + {response: {...proof.response, gate: {kind: "response_metadata_invalid"}}}, + {response: {...proof.response, extra_metadata: "changed"}}] + .map(value => ({native_observation: {...proof, ...value}})), + {failed_turn: {...completed, response: {...response, protected_action: {kind: "action"}}}}]) { + assert.equal(planBoundConversationRequest({...packet, context_recovery: {...recovery, ...change}}).recovery_eligible, + false, JSON.stringify(change)); + } + for (const change of [{status: "closed"}, {status: "busy", active_turn_id: "newer"}, {active_turn_id: "newer"}]) { + assert.equal(planBoundConversationRequest({...packet, current_session: {...session, ...change}}).recovery_candidate, false); + } + const latest = {...proof, latest_input_verified: true}; + delete (latest as Partial).end_step_index; + assert.equal(planBoundConversationRequest({...packet, context_recovery: {...recovery, native_observation: latest}}).recovery_eligible, true); + const adopted = {turn_id: completed.turn_id, message_id: `native.${completed.turn_id}.recovered`, + message_text: "Frozen original answer", native_observation: proof, delivery_verified: false}; + const pending = {...original, native_result_recovery: adopted}; + assert.equal(planBoundConversationRequest({request: pending, current_session: session, + pending_observation: {turn: completed, stop_observed: false}}).pending, true); + const delivered = {...pending, native_result_recovery: {...adopted, delivery_verified: true}}; + assert.equal(planBoundConversationRequest({request: delivered, current_session: session, + pending_observation: {turn: completed, stop_observed: false}}).pending, false); + assert.equal(planBoundConversationRequest({...packet, request: delivered}).recovery_candidate, false); +}); + test("native terminal recovery adopts a result without Turn admission or lost failure provenance", () => { const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: "antigravity"}; const configured = planConversationBinding({...request, binding: selected}).state; diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py index 28464ee707..47d3fed25c 100644 --- a/tests/test_antigravity_steward_adapter.py +++ b/tests/test_antigravity_steward_adapter.py @@ -161,7 +161,7 @@ def do_POST(self): response = {"steps": state["frame"]["steps"] if state.get("frames") is not None else []} start = 0 if state.get("ignore_step_range") else body["startIndex"] // 2 for index in (() if state.get("frames") is not None else range(start, state["sent_count"])): - answer = f"Argon answer {index + 1}" + answer = (state.get("answers") or {}).get(index, f"Argon answer {index + 1}") if not state.get("plain_text"): answer = '' + json.dumps({"message": answer, "proposals": [], "context_handoff": (state.get("original_handoffs") or [{"goal_id": "work", "agent_id": "worker", "brief": "invalid"}] @@ -473,6 +473,51 @@ def test_completed_result_readback_is_exact_and_never_executes(tmp_path, languag for method, _ in language_server["calls"]) +@pytest.mark.parametrize("fault", [None, "no-anchor", "wrong-anchor", "mixed-execution", "unfinished", "stream-error"]) +def test_historical_completed_result_stays_inside_its_original_anchored_input(tmp_path, language_server, fault): + import hashlib + + ref = "f" * 24 + context = {"schema_version": "manager_turn_context_v1", "context_delegation": {"source_id": ref}} + 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_RUN_COMMAND", "runCommand": {"exitCode": 0}}, + {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "plannerResponse": { + "response": "Original complete answer", "stopReason": "STOP_REASON_STOP_PATTERN"}}, + {"type": "CORTEX_STEP_TYPE_USER_INPUT", "userInput": {"items": [{"text": "Later input"}], "userResponse": "Later input"}}, + {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "plannerResponse": { + "response": "Different later answer", "stopReason": "STOP_REASON_STOP_PATTERN"}}, + ] + for index, step in enumerate(steps): + step.update(status="CORTEX_STEP_STATUS_DONE", metadata={ + "createdAt": "2026-01-01T00:00:05Z" if index < 3 else "2026-01-01T00:01:00Z", + "executionId": "original" if index < 3 else "later", + "sourceTrajectoryStepInfo": {"stepIndex": index}}) + anchor = {"thread_id": "synthetic-cascade", "start_step_index": 0, + "input_sha256": hashlib.sha256(prompt.encode()).hexdigest()} + if fault == "wrong-anchor": + anchor["input_sha256"] = "0" * 64 + elif fault == "mixed-execution": + steps[1]["metadata"]["executionId"] = "later" + elif fault == "unfinished": + steps[1]["status"] = "CORTEX_STEP_STATUS_RUNNING" + elif fault == "stream-error": + steps[2]["plannerResponse"]["stopReason"] = "STOP_REASON_CLIENT_STREAM_ERROR" + language_server.update(active=True, frames=[], frame={"at": 0, "steps": steps, "status": "CASCADE_RUN_STATUS_IDLE"}) + 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=None if fault == "no-anchor" else anchor) + assert (proof is not None) == (fault is None) + if proof: + assert proof["response"]["message"] == "Original complete answer" + assert proof["execution_id"] == "original" and proof["end_step_index"] == 3 + assert proof["latest_input_verified"] is False and proof["total_steps"] == 5 + assert {method for method, _ in language_server["calls"]} <= { + "GetCascadeModelConfigData", "GetCascadeTrajectory", "GetCascadeTrajectorySteps"} + + @pytest.mark.parametrize("change", ["none", "wrong-message", "wrong-source", "wrong-schema", "suffix", "truncated"]) def test_manager_readback_does_not_confuse_quoted_frames_with_transport(change): from loopx.capabilities.native_chat.manager_input import manager_input_matches diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index d64cb2bb50..f395eae93f 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -328,6 +328,113 @@ def crash(*args, **kwargs): fixture.close() +@pytest.mark.parametrize("case", ["normal", "source-changed", "closed", "lost-ack", "message-crash", "non-prefix", "action-change"]) +def test_original_queue_recovers_historical_truncated_answer_without_execution(tmp_path, language_server, monkeypatch, case): # noqa: F811 + from test_native_steward_private import steward + from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter + 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 a literal marker. The complete original answer." + native_response = {"message": "Different original answer" if case == "non-prefix" else complete, + "proposals": [], "gate": None} + if case == "action-change": + native_response["unfinished_work"] = {"remaining": "More work", "dependency": "Worker", "effects_settled": True} + language_server.update(plain_text=True, answers={0: '' + json.dumps(native_response) + ''}) + start = AntigravityAdapter.start_turn + + def old_parser(adapter, message, event_sink): + response = start(adapter, message, event_sink) + return {**response, "message": "Original `"} + + monkeypatch.setattr(AntigravityAdapter, "start_turn", old_parser) + event = provider.event("steward-app", "historical-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"] == "completed" + assert transport.reconcile_request(transport.root / f"{ref}.json") == 1 + monkeypatch.setattr(AntigravityAdapter, "start_turn", start) + later = provider.event("steward-app", "historical-later", "A separate later request") + transport.admit("steward-app", later) + later_ref = identity_ref(binding["provider_ref"], later["message_id"]) + later_row = transport.core.read_request(later_ref) + assert runtime.wait_for_turn(session_id=later_row["session_id"], turn_id=later_row["turn_id"], timeout_sec=15)["status"] == "completed" + assert later_row["session_id"] == row["session_id"] + assert transport.reconcile_request(transport.root / f"{later_ref}.json") == 1 + path = transport.root / f"{ref}.json" + old = _read_json(path) + turn_bytes = store._turn_path(row["session_id"], row["turn_id"]).read_bytes() + event_bytes = store._event_path(row["session_id"], row["turn_id"]).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")) + monkeypatch.setattr(AntigravityAdapter, "start_turn", lambda *_, **__: pytest.fail("Result recovery started model work")) + if case == "source-changed": + provider.messages[event["message_id"]]["content"] = "Edited source" + elif case == "closed": + store.update_session(row["session_id"], status="closed") + elif case == "lost-ack": + provider.lose_next_reply = True + elif case == "message-crash": + append = store.append_message + + def crash(*args, **kwargs): + if str(kwargs.get("message_id", "")).startswith("native."): + raise OSError("Synthetic recovery message crash") + return append(*args, **kwargs) + + monkeypatch.setattr(store, "append_message", crash) + with pytest.raises(OSError, match="recovery message crash"): + transport.reconcile() + monkeypatch.setattr(store, "append_message", append) + monkeypatch.setattr(ChatExternalConversations, "_native_result_text", + staticmethod(lambda _: pytest.fail("Frozen message was regenerated"))) + if case == "lost-ack": + try: + transport.reconcile() + except OSError: + pass + restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, + runner=provider, cli_bin="lark-cli") + restarted.reconcile() # The ordinary original queue discovers the historical result. + recovered = restarted.core.read_request(ref).get("native_result_recovery") + if case in {"source-changed", "closed", "non-prefix", "action-change"}: + assert recovered is None and len(provider.writes) == writes + else: + assert recovered["native_observation"]["latest_input_verified"] is False + assert recovered["native_observation"]["end_step_index"] == 2 + assert recovered["native_observation"]["total_steps"] == 4 + assert recovered["native_observation"]["response"]["message"] == complete + assert recovered["delivery_verified"] is (case != "lost-ack") + assert len(provider.writes) == writes + 1 + receipt = _read_json(path)["deliveries"][f"native_result.{row['turn_id']}"] + if case == "lost-ack": + assert receipt["blocker"] == "provider_delivery_receipt_unknown" + assert restarted.health()[row["binding_id"]]["pending_count"] == 1 + else: + message = next(item for item in store.messages(row["session_id"]) if item["message_id"] == recovered["message_id"]) + body = json.loads(provider.messages[receipt["attempt"]["message_ref"]]["body"]["content"]) + assert body["zh_cn"]["content"][0][0]["text"] == message["text"] + assert complete in message["text"] and "Argon answer 2" not in message["text"] + restarted.reconcile() + assert len(provider.writes) == writes + 1 + assert _read_json(path)["deliveries"]["terminal"] == old["deliveries"]["terminal"] + 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() == event_bytes + assert len(list(store._turn_path(row["session_id"], row["turn_id"]).parent.glob("*.json"))) == 2 + assert [item for item in language_server["calls"] if item[0] in {"StartCascade", "SendUserCascadeMessage", "CancelCascadeInvocation"}] == calls + 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