Skip to content

fix(chat): recover original image preflight failures #8073

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
loopx-agent merged 1 commit into from
Nov 4, 2026
Merged

fix(chat): recover original image preflight failures #8073

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 29 additions & 4 deletions loopx/capabilities/native_chat/external_conversations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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"):
Expand Down Expand Up @@ -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")
Expand All @@ -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")),
Expand Down
47 changes: 38 additions & 9 deletions loopx/control_plane/collaboration/conversation_binding.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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;
Expand Down Expand Up @@ -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
Expand Down
19 changes: 13 additions & 6 deletions loopx/extensions/lark/private_conversations.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
61 changes: 61 additions & 0 deletions tests/control_plane_ts/conversation_binding.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading
Loading