From 86ba27db35c4668266055fc5ab2e592c78982021 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 06:54:38 +0800 Subject: [PATCH] fix(chat): recover original image preflight failures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 33 ++++++- .../collaboration/conversation_binding.ts | 47 +++++++-- .../extensions/lark/private_conversations.py | 19 ++-- .../conversation_binding.test.ts | 61 ++++++++++++ tests/test_lark_private_conversations.py | 98 +++++++++++++++++++ 5 files changed, 239 insertions(+), 19 deletions(-) diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 1870aca0ad..28bbfde376 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -9,6 +9,7 @@ from collections.abc import Callable, Mapping, Sequence from copy import deepcopy from pathlib import Path +import hashlib import json import re from datetime import datetime, timedelta, timezone @@ -31,6 +32,20 @@ _QUOTA_BACKLOG_BATCH_BYTES = 1024 * 1024 +def _preparation_image_identity(value: Any) -> dict[str, Any] | None: + """Fingerprint already normalized Core images without exporting their bytes.""" + if not isinstance(value, list) or not value: + return None + try: + images = normalize_chat_image_attachments(value) + if images != value: + return None + encoded = json.dumps(images, sort_keys=True, separators=(",", ":")).encode() + return {"count": len(images), "sha256": hashlib.sha256(encoded).hexdigest()} + except (ValueError, TypeError, OverflowError): + return None + + def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]: identity: dict[str, str] = {} for field in ("goal_instance_id", "creation_operation_id"): @@ -123,10 +138,12 @@ def _context_recovery_plan(self, row: dict[str, Any], *, _, 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": + image_failure = turn.get("error_code") == "runtime_error" and bool(row.get("attachments")) + if turn.get("error_code") == "manager_authorization_unavailable" or image_failure: # Collect canonical facts only. The typed planner owns the exact # pre-dispatch sequence, retry budget and authority decision. - preparation = {"kind": "core_context_preparation_failure", "session_id": session["session_id"], + preparation = {"kind": ("core_attachment_preparation_failure" if image_failure + else "core_context_preparation_failure"), "session_id": session["session_id"], "turn_id": turn["turn_id"], "events": [ {"event_id": item["event_id"], "kind": item["kind"], **{key: item.get("payload", {})[key] for key in ("phase", "schema_version", "error_code") @@ -141,9 +158,17 @@ def _context_recovery_plan(self, row: dict[str, Any], *, return effect_runtime_result("collaboration.conversation.request", { "request": {key: value for key, value in row.items() if key != "attachments"}, "current_session": session, "binding": selected_binding, - "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn, + "context_recovery": {"version": MANAGER_CONTEXT_VERSION, + "failed_turn": ({key: value for key, value in turn.items() if key != "attachments"} + if image_failure else turn), + **({"image_attachments": {"request": _preparation_image_identity(row.get("attachments")), + "failed": _preparation_image_identity(turn.get("attachments")), + "admitted": _preparation_image_identity((admitted or {}).get("attachments"))}, + "native_result_present": bool(store.native_result(session["session_id"], turn["turn_id"]))} + if image_failure else {}), "observed_at": datetime.now(timezone.utc).isoformat(), - "admitted_turn": admitted, + "admitted_turn": ({key: value for key, value in admitted.items() if key != "attachments"} + if image_failure and admitted is not None else admitted), "queued_input_present": (discovery_reads.queued(session["session_id"]) if discovery_reads is not None else bool(store.queued_turns(session["session_id"]))), "attachments_present": bool(row.get("attachments")), diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index ccc98d5dcb..eb5f98b300 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -1192,15 +1192,38 @@ function planResultReview(recovery: JsonObject, row: JsonObject, current: JsonOb /** A local manager-context failure precedes native dispatch. Only the Core's * exact preparation event sequence can qualify it; absence alone cannot. */ +function samePreparationImages(left: unknown, right: unknown): boolean { + if (!left || typeof left !== "object" || Array.isArray(left)) return false; + const image = left as JsonObject; + return Number.isSafeInteger(image.count) && Number(image.count) > 0 + && typeof image.sha256 === "string" && BARE_SHA256_PATTERN.test(image.sha256) + && isDeepStrictEqual(left, right); +} + +function legacyImagePreparation(recovery: JsonObject, failed: JsonObject): boolean { + const images = recovery.image_attachments as JsonObject | undefined; + // A bounded migration for the old adapter's exact pre-dispatch rejection. + // Neither generic runtime errors nor absent upstream IDs establish safety. + return failed.error_code === "runtime_error" + && failed.error === "attachments currently require the Codex Agent endpoint" + && recovery.attachments_present === true && recovery.native_result_present === false + && images != null && samePreparationImages(images.request, images.failed); +} + function planPreparationRecovery(recovery: JsonObject, row: JsonObject, current: JsonObject, failed: JsonObject, version: number): JsonObject | null { - if (failed.error_code !== "manager_authorization_unavailable") return null; + const images = legacyImagePreparation(recovery, failed); + if (failed.error_code !== "manager_authorization_unavailable" && !images) return null; const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false}; if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id || failed.status !== "failed" || failed.response !== null || failed.upstream_turn_id !== null - || failed.delta_count !== 0 || failed.message !== row.message || recovery.attachments_present !== false + || failed.delta_count !== 0 || failed.message !== row.message || (!images && recovery.attachments_present !== false) || recovery.steering_observed !== false || recovery.native_observation != null) return no; const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "preparation recovery"); + // Only one migration of the original rejected image input. An admitted + // continuation that later fails cannot become another attachment retry. + if (images && previous !== null && (previous.failed_turn_id !== failed.turn_id + || previous.preparation_attempt !== 1)) return no; const otherKinds = ["routing_attempt", "quota_attempt", "read_only_review", "metadata_review", "work_attempt"]; if (previous !== null && (!Number.isSafeInteger(previous.preparation_attempt) || Number(previous.preparation_attempt) < 1 || Number(previous.preparation_attempt) > 3 @@ -1221,7 +1244,9 @@ function planPreparationRecovery(recovery: JsonObject, row: JsonObject, current: if (previous?.client_turn_id !== client || admitted.client_turn_id !== client || admitted.session_id !== current.session_id || admitted.turn_id === failed.turn_id || admitted.message !== row.message || admitted.origin !== row.origin - || !Array.isArray(admitted.attachments ?? []) || (admitted.attachments as unknown[] | undefined)?.length) return no; + || (images ? !samePreparationImages((recovery.image_attachments as JsonObject).request, + (recovery.image_attachments as JsonObject).admitted) + : !Array.isArray(admitted.attachments ?? []) || (admitted.attachments as unknown[] | undefined)?.length)) return no; return {...no, recovery_candidate: true, recovery_eligible: recovery.source_verified === true, client_turn_id: client, admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted preparation Turn"), preparation_attempt: previous?.preparation_attempt, @@ -1241,7 +1266,8 @@ function planPreparationRecovery(recovery: JsonObject, row: JsonObject, current: function planRoutingRecovery(recovery: JsonObject, row: JsonObject, current: JsonObject, failed: JsonObject, previous: JsonObject | null, version: number): JsonObject | null { const quota = failed.error_code === "usage_limit_exceeded"; - const preparation = failed.error_code === "manager_authorization_unavailable"; + const images = legacyImagePreparation(recovery, failed); + const preparation = failed.error_code === "manager_authorization_unavailable" || images; const kind = preparation ? "preparation" : quota ? "quota" : "routing"; const attemptKey = `${kind}_attempt`; if (!quota && !preparation && !["host_gate", "upstream_transport_failure"].includes(String(failed.error_code))) return null; @@ -1279,12 +1305,15 @@ function planRoutingRecovery(recovery: JsonObject, row: JsonObject, current: Jso if (!observed) return candidate; if (preparation) { const events = observed.events; - if (observed.kind !== "core_context_preparation_failure" || observed.session_id !== failed.session_id - || observed.turn_id !== failed.turn_id || !Array.isArray(events) || events.length !== 4) return candidate; - const [queued, phase, context, terminal] = events.map(value => requireJsonObject(value, "Core preparation event")); + if (observed.kind !== (images ? "core_attachment_preparation_failure" : "core_context_preparation_failure") + || observed.session_id !== failed.session_id || observed.turn_id !== failed.turn_id + || !Array.isArray(events) || (events.length !== 4 && !(images && events.length === 3))) return candidate; + const rows = events.map(value => requireJsonObject(value, "Core preparation event")); + const queued = rows[0], terminal = rows.at(-1)!, context = rows.at(-2)!; + const phase = rows.length === 4 ? rows[1] : null; const ids = events.map(value => (value as JsonObject).event_id); - const verified = new Set(ids).size === 4 && ids.every(value => typeof value === "string" && value.length > 0) - && queued.kind === "turn.queued" && phase.kind === "agent.phase" && phase.phase === "manager_context" + const verified = new Set(ids).size === rows.length && ids.every(value => typeof value === "string" && value.length > 0) + && queued.kind === "turn.queued" && (phase === null || phase.kind === "agent.phase" && phase.phase === "manager_context") && context.kind === "manager.context" && context.schema_version === "manager_turn_context_v1" && terminal.kind === "turn.failed" && terminal.error_code === failed.error_code; return {...candidate, recovery_eligible: verified && recovery.source_verified === true diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index a8d0c736f8..051377abe4 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -1622,14 +1622,21 @@ def _recovery_source_observation(self, record: dict[str, Any], selected: dict[st if source.get("msg_type", source.get("message_type")) != message_type: return None if message_type == "post": - # A text-only post already froze its provider rendering at intake. - # Require that exact source and the same stripped model input; do - # not download resources or treat a partial media caption as text. + # Posts froze their provider rendering at intake. Text-only input + # must also equal its admitted text; image input retains its + # original model text and frozen attachment payload below. frozen = record.get("source_content") - if (record.get("attachments") != [] or native.get("attachments") - or record.get("attachment_notice") or not isinstance(frozen, str) - or frozen.strip() != expected): + images = record.get("attachments") + frozen_images = (isinstance(images, list) and bool(images) + and images == native.get("attachments") and record.get("message") == native.get("message")) + if (record.get("attachment_notice") or not isinstance(frozen, str) + or (not frozen_images and (images != [] or native.get("attachments") + or frozen.strip() != expected))): return None + # The original post's image keys/caption must still render exactly + # as frozen at intake. Core separately verifies the saved image + # bytes against the original pre-dispatch Turn; never redownload or + # silently execute a text-only replacement of an image request. expected = frozen elif message_type != "text": return None diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index faf42aea47..9c1f39757d 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/tests/control_plane_ts/conversation_binding.test.ts @@ -239,6 +239,67 @@ test("Core preparation failures resume through one bounded original-source queue } }); +test("legacy image preflight failures need exact Core and image proof before one original continuation", () => { + for (const endpoint of ["codex", "antigravity"]) { + const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: endpoint}; + 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), client = `external-${ref}`; + const session = {session_id: "original", goal_id: "loopx-manager", agent_id: endpoint, + upstream_mode: endpoint === "codex" ? "chat" : "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 failed = {session_id: "original", turn_id: "failed", client_turn_id: client, message: "Explain this image", + status: "failed", delta_count: 0, response: null, upstream_turn_id: null, error_code: "runtime_error", + error: "attachments currently require the Codex Agent endpoint", completed_at: "2026-01-01T00:00:00Z"}; + const images = {count: 1, sha256: "a".repeat(64)}; + const proof = {kind: "core_attachment_preparation_failure", session_id: "original", turn_id: "failed", events: [ + {event_id: "one", kind: "turn.queued"}, + {event_id: "three", kind: "manager.context", schema_version: "manager_turn_context_v1"}, + {event_id: "four", kind: "turn.failed", error_code: "runtime_error"}]}; + const packet = {request: {request_ref: ref, command: null, status: "accepted", delivery_verified: true, + message: failed.message, origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "failed", source}, + binding: selected, current_session: session, context_recovery: {version: 22, failed_turn: failed, + observed_at: "2026-01-01T00:10:00Z", steering_observed: false, stop_observed: false, attachments_present: true, + image_attachments: {request: images, failed: images}, native_result_present: false, + source_verified: true, queued_input_present: false, later_equivalent_input_present: false, + preparation_scope_available: true, preparation_observation: proof}}; + const plan = planBoundConversationRequest(packet); + assert.equal(plan.recovery_eligible, true); + assert.equal(plan.preparation_attempt, 1); + assert.deepEqual(plan.native_thread_ids, []); + const withPhase = {...proof, events: [proof.events[0], + {event_id: "two", kind: "agent.phase", phase: "manager_context"}, ...proof.events.slice(1)]}; + assert.equal(planBoundConversationRequest({...packet, context_recovery: { + ...packet.context_recovery, preparation_observation: withPhase}}).recovery_eligible, true); + for (const change of [{source_verified: false}, {stop_observed: true}, {queued_input_present: true}, + {later_equivalent_input_present: true}, {preparation_scope_available: false}, {attachments_present: false}, + {native_result_present: true}, {native_result_present: undefined}, {native_observation: {effects_observed: true}}, + {image_attachments: {request: images, failed: {...images, sha256: "b".repeat(64)}}}, + {image_attachments: {request: {count: 0, sha256: images.sha256}, failed: {count: 0, sha256: images.sha256}}}, + {preparation_observation: {...proof, events: proof.events.slice(1)}}, + {preparation_observation: {...proof, events: [...proof.events, {event_id: "native", kind: "turn.started"}]}}, + ...[{upstream_turn_id: "started"}, {delta_count: 1}, {response: {}}, {client_turn_id: "foreign"}, + {error: "The worker exited; execution may have occurred"}].map(value => ({failed_turn: {...failed, ...value}}))]) { + assert.notEqual(planBoundConversationRequest({...packet, context_recovery: { + ...packet.context_recovery, ...change}}).recovery_eligible, true); + } + const reservation = {version: 22, failed_turn_id: "failed", preparation_attempt: 1, client_turn_id: plan.client_turn_id}; + const admitted = {session_id: "original", turn_id: "continued", client_turn_id: plan.client_turn_id, + message: failed.message, origin: "lark"}; + const crash = {...packet, request: {...packet.request, context_recovery: reservation}, context_recovery: { + ...packet.context_recovery, admitted_turn: admitted, image_attachments: {request: images, failed: images, admitted: images}}}; + assert.equal(planBoundConversationRequest(crash).admitted_turn_id, "continued"); + assert.notEqual(planBoundConversationRequest({...crash, context_recovery: {...crash.context_recovery, + image_attachments: {...crash.context_recovery.image_attachments, admitted: {...images, sha256: "b".repeat(64)}}}}).recovery_eligible, true); + assert.equal(planBoundConversationRequest({...packet, request: {...packet.request, turn_id: "continued", + context_recovery: reservation}, context_recovery: {...packet.context_recovery, + failed_turn: {...failed, turn_id: "continued", client_turn_id: plan.client_turn_id}, + preparation_observation: {...proof, turn_id: "continued"}}}).recovery_candidate, false); + } +}); + 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; diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index ff26f237a8..c778b8cf5d 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -19,6 +19,104 @@ from loopx.extensions.lark.goal_topic_runtime import poll_lark_goal_topic_profile_once +@pytest.mark.parametrize("fault", [None, "admission-crash", "source-image", "source-caption", "saved-image", "extra-native-event", "revoked", "stop"]) +def test_original_post_image_preflight_recovery_keeps_images_source_and_failed_turn(tmp_path, monkeypatch, fault): + from test_native_steward_private import steward, finish + from test_lark_private_images import image_runner + from test_chat_image_attachments import PNG_DATA_URL + from test_manager_preparation_recovery import advance_recovery_clock + from loopx.chat_store import _read_json, _atomic_write_json + from loopx.chat_manager import MANAGER_CONTEXT_VERSION + import loopx.chat_runtime as runtime_module + + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, capture, _ = next(fixture) + transport.runner = image_runner(provider) + original_start = runtime_module.start_attachment_turn + + def old_preflight(*_args, **_kwargs): + raise ValueError("attachments currently require the Codex Agent endpoint") + + monkeypatch.setattr(runtime_module, "start_attachment_turn", old_preflight) + try: + event = provider.event("steward-app", "original-image-preflight", "Explain this image\n![Image](img_example)", kind="post") + assert transport.admit("steward-app", event)["status"] == "durably_accepted" + row = transport.core.pending()[0] + failed = finish(runtime, row) + assert failed["status"] == "failed" and failed["upstream_turn_id"] is None and failed["delta_count"] == 0 + assert not any(json.loads(line)["method"] == "turn/start" for line in capture.read_text().splitlines()) + transport.reconcile() + path = transport.root / f"{row['request_ref']}.json" + record = _read_json(path) + assert record["deliveries"]["terminal"]["verified"] + original_failure_bytes = store._turn_path(row["session_id"], row["turn_id"]).read_bytes() + monkeypatch.setattr(runtime_module, "start_attachment_turn", original_start) + advance_recovery_clock(monkeypatch) + if fault in {"source-image", "source-caption"}: + provider.messages[event["message_id"]]["content"] = ("Explain this image\n![Image](img_replaced)" + if fault == "source-image" else "Different intent\n![Image](img_example)") + elif fault == "saved-image": + original = _read_json(transport.core.root / f"{row['request_ref']}.json") + original["attachments"][0]["name"] = "a different frozen image" + _atomic_write_json(transport.core.root / f"{row['request_ref']}.json", original) + elif fault == "extra-native-event": + events_after = store.events_after + monkeypatch.setattr(store, "events_after", lambda sid, tid, cursor, **kw: events_after(sid, tid, cursor, **kw) + + ([{"event_id": "unknown-effect", "kind": "turn.started", "payload": {}}] if tid == row["turn_id"] else [])) + elif fault == "revoked": + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + elif fault == "stop": + transport.admit("steward-app", provider.event("steward-app", "stop-original-image", "/stop")) + if fault in {None, "admission-crash"}: + assert transport.core.context_recovery_candidate(row["request_ref"]) + if fault == "admission-crash": + enqueue = runtime.enqueue_turn + + def committed_then_disconnected(**kwargs): + enqueue(**kwargs) + monkeypatch.setattr(runtime, "enqueue_turn", enqueue) + raise OSError("synthetic crash after original queue admission") + + monkeypatch.setattr(runtime, "enqueue_turn", committed_then_disconnected) + with pytest.raises(OSError, match="after original queue admission"): + transport.reconcile_request(path) + # Exercise the existing reply worker after its next scheduled visit; + # discovery's monotonic backoff has separate real-pool coverage below. + if fault == "revoked": + with pytest.raises(ValueError, match="no longer authorized"): + transport.reconcile_request(path) + else: + transport.reconcile_request(path) + current = transport.core.read_request(row["request_ref"]) + if fault not in {None, "admission-crash"}: + assert current["turn_id"] == row["turn_id"] and not current.get("context_recovery") + assert not any(json.loads(line)["method"] == "turn/start" for line in capture.read_text().splitlines()) + return + assert current["session_id"] == row["session_id"] and current["turn_id"] != row["turn_id"] + assert current["context_recovery"]["preparation_attempt"] == 1 + continued = runtime.wait_for_turn(session_id=row["session_id"], turn_id=current["turn_id"], timeout_sec=15) + assert continued["status"] == "completed" and continued["attachments"] == failed["attachments"] + transport.reconcile() + delivered = _read_json(path) + terminal = delivered["deliveries"][f"terminal.recovery.{MANAGER_CONTEXT_VERSION}.preparation.1"] + assert terminal["verified"] and terminal["text"] == continued["response"]["message"] + actual = provider.messages[terminal["attempt"]["message_ref"]] + body = json.loads(actual["body"]["content"]) + assert body["zh_cn"]["content"][0][0]["text"] == terminal["text"] + assert actual["chat_id"] == event["chat_id"] and actual["sender"]["id"] == provider.profile_apps["steward-app"] + assert delivered["deliveries"]["terminal"] == record["deliveries"]["terminal"] + assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original_failure_bytes + wire = [json.loads(line) for line in capture.read_text().splitlines()] + model_calls = [call for call in wire if call["method"] == "turn/start"] + assert len(model_calls) == 1 + assert any(item.get("type") == "image" and item["url"] == PNG_DATA_URL for item in model_calls[0]["params"]["input"]) + transport.reconcile() + assert sum(json.loads(line)["method"] == "turn/start" for line in capture.read_text().splitlines()) == 1 + assert len([call for call in provider.calls if "+messages-resources-download" in call]) == 1 + finally: + fixture.close() + + @pytest.mark.parametrize("fault", [None, "stop", "source-changed", "later-source", "no-capacity", "native-tool", "stop-after-discovery", "source-changed-after-discovery"]) def test_original_worker_continues_proven_unstarted_argon_quota_on_codex(tmp_path, language_server, monkeypatch, fault): # noqa: F811