From 868e88722d057cd99c772ac70eea5836ae056ab7 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 21:42:03 +0800
Subject: [PATCH] fix(chat): retain completed handoffs until safe metadata
recovery
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
loopx/canary/module_metric_baseline.json | 2 +-
.../native_chat/antigravity_adapter.py | 110 ++++++++++--
.../native_chat/external_conversations.py | 41 ++++-
loopx/chat_store.py | 14 +-
.../collaboration/conversation_binding.ts | 121 ++++++++++++--
.../conversation_binding.test.ts | 66 +++++++-
tests/test_antigravity_steward_adapter.py | 20 ++-
tests/test_chat_store_input_validation.py | 38 +++++
tests/test_lark_private_conversations.py | 158 +++++++++++++++++-
9 files changed, 523 insertions(+), 47 deletions(-)
diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json
index 3041c487a6..f99cf6f984 100644
--- a/loopx/canary/module_metric_baseline.json
+++ b/loopx/canary/module_metric_baseline.json
@@ -44,7 +44,7 @@
"lines": 2035
},
"loopx/chat_store.py": {
- "lines": 2066
+ "lines": 2074
},
"loopx/codex_cli_probe.py": {
"any_count": 37,
diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py
index ec4a0f02d5..8bd320436e 100644
--- a/loopx/capabilities/native_chat/antigravity_adapter.py
+++ b/loopx/capabilities/native_chat/antigravity_adapter.py
@@ -84,6 +84,27 @@ def _check_hard_timeout(value: float | None) -> None:
raise ValueError("Antigravity hard timeout must be finite and positive")
+def _native_completion_facts(cascade_id: str, current: list[tuple[int, dict[str, Any]]]) -> dict[str, Any]:
+ """One private completion projection, without tool outputs or reasoning."""
+ facts = []
+ for index, step in current:
+ 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[-1][0]:
+ planner = step.get("plannerResponse") or {}
+ fact["plannerResponse"] = {key: planner[key] for key in
+ ("response", "modifiedResponse", "stopReason", "toolCalls") if key in planner}
+ facts.append(fact)
+ return {"thread_id": cascade_id, "start_step_index": current[0][0],
+ "total_steps": current[-1][0] + 1, "steps": facts}
+
+
class _LocalRedirectHandler(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req: Any, fp: Any, code: int, msg: str,
headers: Any, newurl: str) -> None:
@@ -338,12 +359,80 @@ class AntigravityAdapter:
adapter_kind = "antigravity_ls"
+ @staticmethod
+ def metadata_review_capacity(*, cascade_id: str, model_id: str) -> dict[str, Any] | None:
+ """Observe the existing model's quota; never probe it with generation.
+
+ A positive fraction alone is insufficient: an invocation can have
+ reached its limit with a small advertised remainder. Preserve that
+ native failure alongside the exact model's quota epoch.
+ """
+ try:
+ found = discover_antigravity_language_server()
+ if found is None:
+ return None
+ def read(method: str, payload: dict[str, Any]) -> dict[str, Any]:
+ return _rpc_post(found["url"], found["csrf_token"], method, payload, timeout=5.0)
+ before = read("GetCascadeTrajectory", {"cascadeId": cascade_id})
+ 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
+ start = max(0, total - 128)
+ steps, covered = _observed_turn_steps(read("GetCascadeTrajectorySteps",
+ {"cascadeId": cascade_id, "startIndex": start, "endIndex": total})["steps"], start, total)
+ inputs = [i for i, step in steps if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
+ if not inputs and start:
+ steps, covered = _observed_turn_steps(read("GetCascadeTrajectorySteps",
+ {"cascadeId": cascade_id, "startIndex": 0, "endIndex": total})["steps"], 0, total)
+ inputs = [i for i, step in steps if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
+ if not inputs or covered != total:
+ return None
+ latest = [step for i, step in steps if i >= inputs[-1]]
+ exhausted = any(str(((step.get("errorMessage") or {}).get("error") or {}).get(
+ "userErrorMessage") or "").startswith("Individual quota reached.")
+ for step in latest)
+ planner = latest[-1].get("plannerResponse") or {}
+ normal = (latest[-1].get("type") == "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
+ and planner.get("stopReason") == "STOP_REASON_STOP_PATTERN" and not planner.get("toolCalls")
+ and all(step.get("status") == "CORTEX_STEP_STATUS_DONE" for step in latest)
+ and isinstance((latest[0].get("metadata") or {}).get("executionId"), str)
+ and bool((latest[0].get("metadata") or {}).get("executionId"))
+ and len({(step.get("metadata") or {}).get("executionId") for step in latest}) == 1)
+ if not normal and not exhausted:
+ return None # An unknown native failure is not capacity evidence.
+ status = read("GetUserStatus", {})
+ user = status["userStatus"]
+ account = user.get("email")
+ if not isinstance(account, str) or not account.strip():
+ return None
+ rows = user["cascadeModelConfigData"]["clientModelConfigs"]
+ matching = [row for row in rows if row.get("modelId") == model_id]
+ if len(matching) != 1:
+ return None
+ quota = matching[0]["quotaInfo"]
+ fraction, reset = quota.get("remainingFraction", 0), quota.get("resetTime")
+ if (type(fraction) not in (int, float) or not math.isfinite(fraction)
+ or not 0 <= fraction <= 1 or not isinstance(reset, str)
+ or datetime.fromisoformat(reset).tzinfo is None):
+ return None
+ after = read("GetCascadeTrajectory", {"cascadeId": cascade_id})
+ if (after.get("status") != before["status"] or after.get("numTotalSteps") != total
+ or (after.get("trajectory") or {}).get("cascadeId") != cascade_id):
+ return None
+ return {"model_id": model_id, "account_ref": hashlib.sha256(account.strip().casefold().encode()).hexdigest(),
+ "remaining_fraction": fraction, "reset_time": reset,
+ "observed_at": time.time(), "quota_exhausted": exhausted, "normal_completion": normal}
+ except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
+ return None
+
@staticmethod
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,
- retained_result: dict[str, Any] | None = None) -> dict[str, Any] | None:
+ retained_result: dict[str, Any] | None = None,
+ include_retained_result: bool = False) -> 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,
@@ -442,6 +531,7 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
**({"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,
+ **({"retained_result": _native_completion_facts(cascade_id, current)} if include_retained_result else {}),
**({"raw_response": answer} if include_raw_response else {})}
except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
return None
@@ -939,23 +1029,9 @@ def _run_turn(
# 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})
+ event_sink("turn.native_result", {**_native_completion_facts(self.cascade_id, current_steps),
+ "upstream_turn_id": turn_id})
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 29f39da982..3295bcce61 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -51,7 +51,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
source_verified: bool = False,
selected_binding: dict[str, Any] | None = None,
discovery_only: bool = False,
- preparation_scope_available: bool = False) -> dict[str, Any]:
+ preparation_scope_available: bool = False,
+ metadata_capacity: dict[str, Any] | None = None) -> dict[str, Any]:
from ...chat_manager import MANAGER_CONTEXT_VERSION
from ...control_plane.effect_runtime import effect_runtime_result
@@ -113,6 +114,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
"preparation_scope_available": preparation_scope_available,
"native_input": anchors[0] if len(anchors) == 1 else None,
"response_literal_open": bool(open_literal),
+ "metadata_capacity": metadata_capacity,
"native_observation": observation, "source_verified": source_verified},
})
@@ -184,7 +186,8 @@ def context_recovery_candidate(self, request_ref: str) -> bool:
if turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work"):
from ...chat import _review_openings
_, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0)
- if not open_literal:
+ invalid_metadata = ((turn.get("response") or {}).get("gate") or {}).get("kind") == "response_metadata_invalid"
+ if not open_literal and not invalid_metadata:
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):
@@ -230,8 +233,15 @@ def source_requests() -> list[dict[str, Any]]:
# matches after reacquiring the route fence; fresh authority/source
# checks and the typed eligibility decision remain below that fence.
proof = None
+ capacity = None
if not plan.get("admitted_turn_id") and metadata_review:
- proof = self._original_native_result(row, session, turn, context, include_raw_response=True)
+ proof = self._original_native_result(row, session, turn, context, include_raw_response=True,
+ include_retained_result=turn.get("status") == "completed")
+ if proof is not None:
+ from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL
+ capacity = AntigravityAdapter.metadata_review_capacity(
+ cascade_id=session["upstream_thread_id"],
+ model_id=session.get("agent_model_id") or ANTIGRAVITY_DEFAULT_MODEL)
elif not plan.get("admitted_turn_id") and not preparation_retry:
proof = CodexAppServerAdapter.failed_request_observation(
codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
@@ -239,6 +249,7 @@ def source_requests() -> list[dict[str, Any]]:
**({"read_only_review": True} if plan.get("read_only_review") else {}),
**({"completed_work": True, "work_dir": Path(context["project"])} if plan.get("work_attempt") else {}),
thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
+ facts = proof.pop("retained_result", None) if proof is not None and metadata_review else None
selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
source = source_observation(selected)
scope_available = (self._preparation_scope_available(session)
@@ -265,8 +276,23 @@ def source_requests() -> list[dict[str, Any]]:
or source_requests() != requests):
return row
plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
- selected_binding=selected["binding"], preparation_scope_available=scope_available)
+ selected_binding=selected["binding"], preparation_scope_available=scope_available,
+ metadata_capacity=capacity)
+ if plan.get("retain_metadata_result") and facts and not self.controller.store.native_result(row["session_id"], turn["turn_id"]):
+ try:
+ self.controller.store.retain_native_result(row["session_id"], turn["turn_id"],
+ {**facts, "upstream_turn_id": turn["upstream_turn_id"]}, completed_turn=turn)
+ except (OSError, ValueError) as exc:
+ import logging
+ logging.getLogger(__name__).warning("Metadata proof retention held (%s)", type(exc).__name__)
+ return row # Preserve the delivered answer and unspent review.
+ proof = self._original_native_result(row, session, turn, context, include_raw_response=True)
+ plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
+ selected_binding=selected["binding"], metadata_capacity=capacity)
if not plan.get("recovery_eligible"):
+ if plan.get("metadata_quota_wait") is not None and row.get("metadata_quota_wait") != plan["metadata_quota_wait"]:
+ row["metadata_quota_wait"] = plan["metadata_quota_wait"]
+ _atomic_write_json(path, row)
return row
recovery = row.get("context_recovery")
if recovery is None or recovery.get("client_turn_id") != plan["client_turn_id"]:
@@ -343,7 +369,8 @@ def _native_result_text(recovered: dict[str, Any]) -> str:
def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
turn: dict[str, Any], context: dict[str, Any], *,
- include_raw_response: bool = False) -> dict[str, Any] | None:
+ include_raw_response: bool = False,
+ include_retained_result: bool = False) -> dict[str, Any] | None:
from .antigravity_adapter import AntigravityAdapter
events = self.controller.store.events_after(session["session_id"], turn["turn_id"], None)
@@ -360,11 +387,13 @@ def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
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)
+ input_anchor=anchors[0], include_raw_response=include_raw_response, retained_result=retained,
+ **({"include_retained_result": True} if include_retained_result else {}))
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] if anchors else None,
+ **({"include_retained_result": True} if include_retained_result else {}),
**({"include_raw_response": True} if include_raw_response else {}))
def _recover_native_result(self, path: Path, row: dict[str, Any], session: dict[str, Any],
diff --git a/loopx/chat_store.py b/loopx/chat_store.py
index ff660ba6a9..9096f6953b 100644
--- a/loopx/chat_store.py
+++ b/loopx/chat_store.py
@@ -1395,11 +1395,19 @@ def native_result(self, session_id: str, turn_id: str) -> dict[str, Any]:
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."""
+ def retain_native_result(self, session_id: str, turn_id: str, result: dict[str, Any], *,
+ completed_turn: dict[str, Any] | None = None) -> None:
+ """Freeze once; a completed backfill requires the exact unchanged Turn.
+
+ The automatic recovery owner first verifies the original native proof
+ and current source. This fence changes neither its Turn nor its events.
+ """
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":
+ if completed_turn is not None:
+ if turn != completed_turn or turn is None or turn.get("status") != "completed":
+ raise ValueError("the original completed Turn changed")
+ elif 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")
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index ceb3bbe64d..2f368778e1 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -566,6 +566,15 @@ function pureResultText(response: JsonObject | null): boolean {
&& ["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].every(key => response[key] == null);
}
+function unsubmittedMetadata(response: JsonObject | null): boolean {
+ const gate = response?.gate as JsonObject | undefined;
+ return response?.schema_version === "loopx_chat_agent_response_v0"
+ && typeof response.message === "string" && response.message.length > 0
+ && gate?.kind === "response_metadata_invalid" && typeof gate.summary === "string"
+ && Array.isArray(response.proposals) && response.proposals.length === 0
+ && ["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));
@@ -623,17 +632,26 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
&& 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);
+ const completedMetadata = failed.status === "completed" && unsubmittedMetadata(originalResponse)
+ && 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.message !== row.message
- || (!completedText && (failed.status !== "failed" || failed.response != null
+ || (!completedText && !completedMetadata && (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 (completedMetadata) {
+ // The original prose was already delivered. Review only its unsubmitted
+ // metadata: no duplicate prose adoption, Turn rewrite or business replay.
+ return planNativeMetadataReview(recovery, row, current, failed, null, version);
+ }
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);
@@ -758,22 +776,48 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
/** A text-only format review of already returned native prose. This is never
* another business attempt, and any failed/unknown review exhausts the route.
*/
+function textualBrief(value: unknown, depth = 0): boolean {
+ if (depth > 8) return false;
+ if (typeof value === "string") return true;
+ if (Array.isArray(value)) return value.every(item => textualBrief(item, depth + 1));
+ return value !== null && typeof value === "object"
+ && Object.values(value).every(item => textualBrief(item, depth + 1));
+}
+
function planNativeMetadataReview(recovery: JsonObject, row: JsonObject, current: JsonObject,
- failed: JsonObject, adopted: JsonObject, version: number): JsonObject {
+ failed: JsonObject, adopted: JsonObject | null, version: number): JsonObject {
const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
- const frozen = requireJsonObject(adopted.native_observation, "adopted native observation");
- const response = requireJsonObject(frozen.response, "adopted native response");
- if (adopted.turn_id !== failed.turn_id || adopted.message_id !== `native.${failed.turn_id}.recovered`
- || typeof adopted.message_text !== "string" || !adopted.message_text
- || (response.gate as JsonObject | null)?.kind !== "response_metadata_invalid"
- || frozen.thread_id !== current.upstream_thread_id || frozen.request_ref !== row.request_ref
- || frozen.normal_terminal_verified !== true || frozen.latest_input_verified !== true) return no;
const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-metadata-1`;
const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "metadata review reservation");
if (previous !== null && (previous.metadata_review !== true || previous.version !== version
|| previous.failed_turn_id !== failed.turn_id || previous.client_turn_id !== client)) return no;
const candidate = {...no, recovery_candidate: true, metadata_review: true, client_turn_id: client,
session_id: current.session_id, failed_turn_id: failed.turn_id};
+ if (recovery.admitted_turn == null && (current.status !== "ready" || current.active_turn_id != null
+ || recovery.queued_input_present !== false)) return no;
+ const supplied = adopted?.native_observation ?? previous?.native_observation ?? recovery.native_observation;
+ if (supplied == null) return adopted === null ? candidate : no;
+ const {raw_response: _raw, ...frozen} = requireJsonObject(supplied, "original native observation");
+ const response = requireJsonObject(frozen.response, "original native response");
+ if (!unsubmittedMetadata(response) || frozen.thread_id !== current.upstream_thread_id
+ || frozen.request_ref !== row.request_ref || frozen.normal_terminal_verified !== true) return no;
+ if (adopted !== null) {
+ if (adopted.turn_id !== failed.turn_id || adopted.message_id !== `native.${failed.turn_id}.recovered`
+ || typeof adopted.message_text !== "string" || !adopted.message_text
+ || frozen.latest_input_verified !== true) return no;
+ } else {
+ const anchor = requireJsonObject(recovery.native_input, "original completed input");
+ const created = Date.parse(String(frozen.input_created_at));
+ if (!isDeepStrictEqual(failed.response, response) || frozen.start_step_index !== anchor.start_step_index
+ || typeof frozen.execution_id !== "string" || !frozen.execution_id
+ || frozen.input_sha256 !== anchor.input_sha256 || !Number.isFinite(created)
+ || created < Date.parse(String(failed.started_at)) || created > Date.parse(String(failed.completed_at))
+ || !Number.isSafeInteger(frozen.total_steps) || Number(frozen.total_steps) <= Number(frozen.start_step_index) + 1
+ || ![frozen.input_sha256, frozen.answer_sha256].every(v => typeof v === "string" && BARE_SHA256_PATTERN.test(v))
+ || !(frozen.latest_input_verified === true || frozen.latest_input_verified === false
+ && Number.isSafeInteger(frozen.end_step_index) && Number(frozen.end_step_index) > Number(frozen.start_step_index) + 1
+ && Number(frozen.end_step_index) < Number(frozen.total_steps))) return no;
+ }
if (previous !== null && recovery.admitted_turn != null) {
const admitted = requireJsonObject(recovery.admitted_turn, "admitted metadata review");
if (admitted.client_turn_id !== client || admitted.session_id !== current.session_id
@@ -791,7 +835,9 @@ function planNativeMetadataReview(recovery: JsonObject, row: JsonObject, current
if (!isDeepStrictEqual(observed, frozen) || typeof raw !== "string" || raw.length > 100000
|| createHash("sha256").update(raw).digest("hex") !== frozen.answer_sha256) return candidate;
// The parser retained prose but admitted none of the malformed action batch.
- // Only string-brief handoffs qualify; other bad/unknown actions stay held.
+ // Transcribe bounded JSON briefs as exact evidence strings; this does not
+ // interpret their fields or grant actions. The normal owner must still
+ // validate the formatter's canonical brief and each current recipient.
let payload: JsonObject;
try {
const start = raw.lastIndexOf(""), end = raw.lastIndexOf("");
@@ -802,21 +848,62 @@ function planNativeMetadataReview(recovery: JsonObject, row: JsonObject, current
"goal_draft", "protected_action", "unfinished_work"].includes(key))
|| [payload.goal_draft, payload.protected_action, payload.gate, payload.unfinished_work].some(v => v != null)
|| !Array.isArray(payload.proposals ?? []) || (payload.proposals as unknown[] | undefined)?.length
- || !Array.isArray(payload.context_handoff) || !payload.context_handoff.length
- || payload.context_handoff.length > 32) return candidate;
+ ) return candidate;
+ const handoffs = Array.isArray(payload.context_handoff) ? payload.context_handoff : [payload.context_handoff];
+ if (!handoffs.length || handoffs.length > 32) return candidate;
const targets = new Set();
- for (const value of payload.context_handoff) {
+ const transcribed: JsonObject[] = [];
+ for (const value of handoffs) {
if (!value || typeof value !== "object" || Array.isArray(value)) return candidate;
const item = value as JsonObject;
- if (Object.keys(item).sort().join(",") !== "agent_id,brief,goal_id"
+ const keys = Object.keys(item).filter(key => key !== "schema_version").sort().join(",");
+ if (keys !== "agent_id,brief,goal_id" || item.schema_version != null && item.schema_version !== "context_handoff_v0"
|| [item.goal_id, item.agent_id].some(v => typeof v !== "string" || !/^[A-Za-z0-9._-]{1,160}$/.test(v))
- || typeof item.brief !== "string" || !item.brief.trim() || item.brief.length > 6000) return candidate;
+ || !(typeof item.brief === "string" || item.brief && typeof item.brief === "object" && !Array.isArray(item.brief))
+ || !textualBrief(item.brief)) return candidate;
+ const brief = typeof item.brief === "string" ? item.brief : JSON.stringify(item.brief);
+ if (!brief.trim() || brief.length > 6000) return candidate;
const target = JSON.stringify([item.goal_id, item.agent_id]);
if (targets.has(target)) return candidate;
targets.add(target);
+ transcribed.push({goal_id: item.goal_id, agent_id: item.agent_id, brief});
+ }
+ return metadataReviewCapacity(recovery, row, current, {...candidate, metadata_handoffs: transcribed,
+ ...(adopted === null && recovery.source_verified === true ? {retain_metadata_result: true} : {})});
+}
+
+/** A model list/positive remainder is not proof a known quota failure cleared.
+ * Keep the existing request pending without reserving its one format attempt.
+ */
+function metadataReviewCapacity(recovery: JsonObject, row: JsonObject, current: JsonObject,
+ candidate: JsonObject): JsonObject {
+ if (recovery.source_verified !== true) return candidate;
+ const capacity = recovery.metadata_capacity as JsonObject | null | undefined;
+ const now = Date.parse(String(recovery.observed_at)) / 1000;
+ if (!capacity || capacity.model_id !== (current.agent_model_id ?? "gemini-4-argon-high")
+ || typeof capacity.account_ref !== "string" || !BARE_SHA256_PATTERN.test(capacity.account_ref)
+ || typeof capacity.remaining_fraction !== "number" || !Number.isFinite(capacity.remaining_fraction)
+ || capacity.remaining_fraction < 0 || capacity.remaining_fraction > 1
+ || typeof capacity.observed_at !== "number" || !Number.isFinite(capacity.observed_at)
+ || !Number.isFinite(now) || now < capacity.observed_at || now - capacity.observed_at > 120
+ || typeof capacity.quota_exhausted !== "boolean" || typeof capacity.normal_completion !== "boolean"
+ || typeof capacity.reset_time !== "string"
+ || !Number.isFinite(Date.parse(String(capacity.reset_time)))) return candidate;
+ const waiting = row.metadata_quota_wait as JsonObject | null | undefined;
+ if (waiting != null && (typeof waiting.model_id !== "string" || typeof waiting.remaining_fraction !== "number"
+ || typeof waiting.account_ref !== "string" || !BARE_SHA256_PATTERN.test(waiting.account_ref)
+ || !Number.isFinite(waiting.remaining_fraction) || waiting.remaining_fraction < 0 || waiting.remaining_fraction > 1
+ || typeof waiting.reset_time !== "string" || !Number.isFinite(Date.parse(waiting.reset_time)))) return candidate;
+ const sameEpoch = waiting?.model_id === capacity.model_id && waiting.account_ref === capacity.account_ref
+ && Date.parse(String(waiting.reset_time)) === Date.parse(capacity.reset_time);
+ // Fractions can fluctuate across refreshes. Only a quota/account epoch
+ // change or the known reset boundary can clear this recorded hard failure.
+ const restored = waiting != null && (!sameEpoch || now >= Date.parse(String(waiting.reset_time)) / 1000);
+ if (capacity.remaining_fraction <= 0 || waiting != null && !restored
+ || waiting == null && capacity.quota_exhausted === true) {
+ return {...candidate, metadata_quota_wait: waiting ?? capacity};
}
- return {...candidate, recovery_eligible: recovery.source_verified === true,
- metadata_handoffs: payload.context_handoff};
+ return {...candidate, recovery_eligible: true};
}
/** A completed model Turn can explicitly report unfinished business. The
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index d746824414..44307c4483 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -230,7 +230,12 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
response: {...proof.response, gate: {kind: "response_metadata_invalid", summary: "Unsubmitted handoffs"}}};
const owed = {...delivered, native_result_recovery: {...adopted, native_observation: frozen}};
const metadata = {...packet, request: owed, context_recovery: {...packet.context_recovery,
- queued_input_present: false, native_observation: {...frozen, raw_response: raw}}};
+ observed_at: "2026-01-01T00:00:30Z",
+ queued_input_present: false, native_observation: {...frozen, raw_response: raw},
+ metadata_capacity: {model_id: "gemini-4-argon-high", remaining_fraction: 1,
+ account_ref: "3".repeat(64),
+ reset_time: "2026-01-10T00:00:00Z", observed_at: Date.parse("2026-01-01T00:00:30Z") / 1000,
+ quota_exhausted: false, normal_completion: true}}};
const review = planBoundConversationRequest(metadata);
assert.equal(review.recovery_eligible, true);
assert.equal(review.metadata_review, true);
@@ -256,6 +261,65 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
assert.equal(planBoundConversationRequest({request: exhausted, current_session: session,
pending_observation: {turn: admitted, stop_observed: false}}).pending, true,
"A held/failed format review retains the original unresolved handoffs without retry authority");
+ const objectBrief = {Hypotheses: ["Original hypothesis"], Evidence: ["Read evidence"],
+ Constraints: ["No changes to frozen work"]};
+ const objectHandoff = {schema_version: "context_handoff_v0", goal_id: "work", agent_id: "worker", brief: objectBrief};
+ const objectRaw = "" + JSON.stringify({message: "Original prose", context_handoff: objectHandoff,
+ proposals: []}) + "";
+ const historical = {...frozen, latest_input_verified: false, end_step_index: 6,
+ answer_sha256: createHash("sha256").update(objectRaw).digest("hex")};
+ const completed = {...failed, status: "completed", error_code: null, response: historical.response};
+ const completedRow = {...packet.request};
+ const direct = {...metadata, request: completedRow, context_recovery: {...metadata.context_recovery,
+ failed_turn: completed, native_input: {thread_id: "native", start_step_index: 4, input_sha256: historical.input_sha256},
+ native_observation: {...historical, raw_response: objectRaw}}};
+ const directPlan = planBoundConversationRequest(direct);
+ assert.equal(directPlan.recovery_eligible, true);
+ assert.equal(directPlan.metadata_review, true);
+ assert.equal(directPlan.recover_native_result, undefined, "Already delivered prose cannot be adopted/sent again");
+ assert.deepEqual(directPlan.metadata_handoffs, [{goal_id: "work", agent_id: "worker", brief: JSON.stringify(objectBrief)}]);
+ for (const handoff of [{...objectHandoff, execution_binding_id: "invented"},
+ {...objectHandoff, schema_version: "unknown-version"}, {...objectHandoff, brief: {budget: 9007199254740992}},
+ {...objectHandoff, brief: {effects_settled: true}}, {...objectHandoff, brief: null}]) {
+ const invalidRaw = "" + JSON.stringify({message: "Original prose", context_handoff: handoff,
+ proposals: []}) + "";
+ assert.equal(planBoundConversationRequest({...direct, context_recovery: {...direct.context_recovery,
+ native_observation: {...historical, raw_response: invalidRaw,
+ answer_sha256: createHash("sha256").update(invalidRaw).digest("hex")}}}).recovery_eligible, false);
+ }
+ for (const change of [{source_verified: false}, {queued_input_present: true}, {stop_observed: true},
+ {steering_observed: true}, {native_input: null}, {metadata_capacity: null},
+ {native_observation: {...historical, raw_response: objectRaw, latest_input_verified: false, end_step_index: 8}},
+ {native_observation: {...historical, raw_response: objectRaw, execution_id: ""}},
+ {native_observation: {...historical, raw_response: objectRaw, input_sha256: "9".repeat(64)}},
+ {native_observation: {...historical, raw_response: objectRaw, response: {...historical.response, message: "Changed prose"}}}]) {
+ assert.equal(planBoundConversationRequest({...direct, context_recovery: {...direct.context_recovery, ...change}}).recovery_eligible, false);
+ }
+ const lowCapacity = {...metadata.context_recovery.metadata_capacity, quota_exhausted: true,
+ normal_completion: false, remaining_fraction: 0.006};
+ const low = {...direct, context_recovery: {...direct.context_recovery, metadata_capacity: lowCapacity}};
+ const held = planBoundConversationRequest(low);
+ assert.equal(held.recovery_eligible, false, "A small positive fraction cannot override the actual quota failure");
+ assert.deepEqual(held.metadata_quota_wait, lowCapacity);
+ const waitingRow = {...completedRow, metadata_quota_wait: held.metadata_quota_wait};
+ assert.equal(planBoundConversationRequest({...low, request: waitingRow}).recovery_eligible, false);
+ assert.equal(planBoundConversationRequest({...low, request: waitingRow,
+ context_recovery: {...low.context_recovery, metadata_capacity: {...lowCapacity, remaining_fraction: 1}}}).recovery_eligible, false);
+ assert.equal(planBoundConversationRequest({...low, request: waitingRow,
+ context_recovery: {...low.context_recovery, metadata_capacity: {...lowCapacity, remaining_fraction: 1,
+ reset_time: "2026-01-11T00:00:00Z"}}}).recovery_eligible, true);
+ assert.equal(planBoundConversationRequest({...low, request: waitingRow,
+ context_recovery: {...low.context_recovery, metadata_capacity: {...lowCapacity, remaining_fraction: 1,
+ account_ref: "4".repeat(64)}}}).recovery_eligible, true);
+ assert.equal(planBoundConversationRequest({...low, request: waitingRow,
+ context_recovery: {...low.context_recovery, metadata_capacity: {...lowCapacity,
+ reset_time: "2026-01-10T00:00:00+00:00"}}}).recovery_eligible, false);
+ for (const capacity of [{...lowCapacity, model_id: "another-model"}, {...lowCapacity, observed_at: 0},
+ {...lowCapacity, remaining_fraction: -1}, {...lowCapacity, remaining_fraction: 2},
+ {...lowCapacity, quota_exhausted: null}, {...lowCapacity, reset_time: "unknown"}]) {
+ assert.equal(planBoundConversationRequest({...low, request: waitingRow,
+ context_recovery: {...low.context_recovery, metadata_capacity: capacity}}).recovery_eligible, false);
+ }
const unfinished = {...delivered, native_result_recovery: {...adopted,
native_observation: {...proof, response: {...proof.response,
unfinished_work: {remaining: "Still owed", dependency: "Original worker", effects_settled: true}}}}};
diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py
index 552e458852..48adf17a02 100644
--- a/tests/test_antigravity_steward_adapter.py
+++ b/tests/test_antigravity_steward_adapter.py
@@ -125,6 +125,15 @@ def do_POST(self):
{"modelId": f"gemini-4-argon-{tier}",
"modelOrAlias": {"model": f"MODEL_PLACEHOLDER_M{number}"}}
for tier, number in (("high", 417), ("medium", 418), ("low", 419))]}
+ elif method == "GetUserStatus":
+ from datetime import timedelta
+ reset = state.setdefault("quota_reset", (datetime.now(timezone.utc) + timedelta(days=7)).isoformat())
+ response = {"userStatus": {"email": state.get("quota_account", "synthetic@invalid.example"),
+ "cascadeModelConfigData": {"clientModelConfigs": [{
+ "modelId": "gemini-4-argon-high", "quotaInfo": {
+ "remainingFraction": state.get("quota_fraction", 1), "resetTime": reset}}]}}}
+ if state.get("quota_unknown"):
+ response = {}
elif method == "StartCascade":
response = {"cascadeId": "synthetic-cascade"}
elif method == "GetCascadeTrajectory":
@@ -181,9 +190,18 @@ def do_POST(self):
"status": "CORTEX_STEP_STATUS_WAITING" if state["mode"] == "waiting" else "CORTEX_STEP_STATUS_DONE",
"plannerResponse": {"thinking": "synthetic status", "response": answer,
"stopReason": "STOP_REASON_STOP_PATTERN"}}])
+ if index in state.get("quota_error_indices", []):
+ final = response["steps"][-1]
+ final["type"] = "CORTEX_STEP_TYPE_ERROR_MESSAGE"
+ final.pop("plannerResponse")
+ final["errorMessage"] = {"error": {"userErrorMessage": "Individual quota reached. Resets in 123h."}}
+ if index in state.get("cleared_input_indices", []):
+ for step in response["steps"][-2:]:
+ step["status"] = "CORTEX_STEP_STATUS_CLEARED"
+ response["steps"][-1].get("plannerResponse", {}).pop("stopReason", None)
if index > 0:
final = response["steps"][-1]
- fault = state.get("review_native_fault")
+ fault = state.get("review_native_fault") if index >= state.get("review_fault_start_index", 1) else None
if fault == "tool-call":
final["plannerResponse"]["toolCalls"] = [{"name": "run_command"}]
elif fault == "tool-step":
diff --git a/tests/test_chat_store_input_validation.py b/tests/test_chat_store_input_validation.py
index b3f9cc421e..bdf468df5c 100644
--- a/tests/test_chat_store_input_validation.py
+++ b/tests/test_chat_store_input_validation.py
@@ -52,6 +52,44 @@ def _dedup_store(tmp_path: Path) -> tuple[ChatSessionStore, str]:
return store, session_id
+@pytest.mark.parametrize("fault", [None, "changed-turn", "failed", "wrong-invocation", "immutable"])
+def test_completed_native_retention_is_fenced_and_private(tmp_path: Path, fault: str | None) -> None:
+ store, sid = _dedup_store(tmp_path)
+ turn, _ = store.create_turn(sid, client_turn_id="original-input", message="Original visible request")
+ tid = turn["turn_id"]
+ store.update_turn(sid, tid, status="completed", upstream_turn_id="invocation-one",
+ response={"message": "Original visible answer"})
+ original = store.load_turn(sid, tid)
+ result = {"thread_id": "thread-one", "upstream_turn_id": "invocation-one",
+ "steps": [{"text": "PRIVATE_NATIVE_COMPLETION_SENTINEL"}]}
+ if fault == "changed-turn":
+ store.update_turn(sid, tid, response={"message": "A changed result"})
+ elif fault == "failed":
+ store.update_turn(sid, tid, status="failed")
+ original = store.load_turn(sid, tid)
+ elif fault == "wrong-invocation":
+ result["upstream_turn_id"] = "foreign-invocation"
+ before = store._turn_path(sid, tid).read_bytes()
+ if fault in {"changed-turn", "failed", "wrong-invocation"}:
+ with pytest.raises(ValueError):
+ store.retain_native_result(sid, tid, result, completed_turn=original)
+ assert not store.native_result(sid, tid)
+ else:
+ store.retain_native_result(sid, tid, result, completed_turn=original)
+ path = store._session_dir(sid) / "native-results" / f"{tid}.json"
+ stamp = path.stat().st_mtime_ns
+ assert path.stat().st_mode & 0o777 == 0o600
+ assert path.parent.stat().st_mode & 0o777 == 0o700
+ store.retain_native_result(sid, tid, result, completed_turn=original)
+ assert path.stat().st_mtime_ns == stamp
+ if fault == "immutable":
+ with pytest.raises(ValueError, match="immutable"):
+ store.retain_native_result(sid, tid, {**result, "steps": []}, completed_turn=original)
+ assert store.native_result(sid, tid) == result
+ assert store._turn_path(sid, tid).read_bytes() == before
+ assert "PRIVATE_NATIVE_COMPLETION_SENTINEL" not in json.dumps(store.messages(sid))
+
+
def test_history_cursor_rejects_other_session_and_rewritten_prefix(tmp_path: Path) -> None:
store, sid = _dedup_store(tmp_path)
for i in range(5):
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 5735e75d0f..0684fc02ef 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -255,7 +255,9 @@ def lost_core_terminal(adapter, message, event_sink):
assert transport.health()[row["binding_id"]]["pending_count"] == 1
assert "Argon answer 1" in provider.writes[-1][1]
assert "相关操作和委派尚未提交" in provider.writes[-1][1]
- assert not transport.core.context_recovery_candidate(ref) # No business replay.
+ assert transport.core.context_recovery_candidate(ref)
+ plan = transport.core._context_recovery_plan(transport.core.read_request(ref), discovery_only=True)
+ assert plan["metadata_review"] and not plan.get("recover_native_result") # Format only, no business replay.
assert language_server["sent_count"] == 1
return
assert failed["status"] == "failed" and failed["error_code"] == "antigravity_rpc_error"
@@ -809,6 +811,160 @@ def crash(**kwargs):
fixture.close()
+@pytest.mark.parametrize("fault", [None, "quota-restore", "quota-unchanged", "quota-unknown",
+ "stop", "new-input", "no-anchor", "changed-proof", "unknown-outer", "wrong-target", "changed-brief", "tool-step",
+ "cleared-before-proof", "retention-io"])
+def test_completed_invalid_metadata_uses_one_original_format_review(tmp_path, language_server, monkeypatch, fault):
+ """An already delivered, historical answer must not be sent/executed twice."""
+ from copy import deepcopy
+ from datetime import timezone
+ from test_native_steward_private import steward
+ from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter
+ from loopx.capabilities.manager_context import POLICY_SCHEMA, _root, _write, pending
+
+ 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, _, workspace = next(fixture)
+ try:
+ registry = json.loads(runtime.registry_path.read_text())
+ registry["goals"] = [{"id": "work", "repo": str(workspace), "coordination": {"registered_agents": ["worker"]}}]
+ runtime.registry_path.write_text(json.dumps(registry))
+ binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
+ executor_endpoint_id="antigravity", context_kind="steward")
+ original_brief = {"schema_version": "collaboration_brief_v0", "Hypotheses": ["Assess this hypothesis"],
+ "Evidence": ["Original authorized evidence"], "Constraints": ["Return here; do not change frozen work"]}
+ handoff = {"schema_version": "context_handoff_v0", "goal_id": "work", "agent_id": "worker", "brief": original_brief}
+ if fault == "unknown-outer":
+ handoff["execution_binding_id"] = "unapproved-binding"
+ language_server.update(invalid_handoff=True, original_handoffs=handoff,
+ answers={0: "Original complete visible answer."})
+ # Emulate an original completion from before durable result capture;
+ # keep its real Core input anchor and same-invocation native history.
+ start_turn = AntigravityAdapter.start_turn
+ def legacy_completion(adapter, message, event_sink, **kwargs):
+ return start_turn(adapter, message, lambda kind, payload: event_sink(kind, payload)
+ if kind != "turn.native_result" else None, **kwargs)
+ monkeypatch.setattr(AntigravityAdapter, "start_turn", legacy_completion)
+ event = provider.event("steward-app", "completed-metadata", "Ask the original worker to assess this hypothesis")
+ transport.admit("steward-app", event)
+ ref = identity_ref(binding["provider_ref"], event["message_id"])
+ original = transport.core.read_request(ref)
+ completed = runtime.wait_for_turn(session_id=original["session_id"], turn_id=original["turn_id"], timeout_sec=15)
+ assert completed["status"] == "completed"
+ assert completed["response"]["gate"]["kind"] == "response_metadata_invalid"
+ monkeypatch.setattr(AntigravityAdapter, "start_turn", start_turn)
+ assert transport.reconcile_request(transport.root / f"{ref}.json") == 1
+ original_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes()
+ original_events = store.events_after(original["session_id"], original["turn_id"], None)
+ assert not store.native_result(original["session_id"], original["turn_id"])
+ # Another native input is historical context, not authority to replay it.
+ language_server["inputs"].append({"text": "An unrelated later native input", "created_at": datetime.now(timezone.utc).isoformat()})
+ language_server["sent_count"] += 1
+ if fault in {"quota-restore", "quota-unchanged"}:
+ language_server.update(quota_error_indices=[1], quota_fraction=0.006)
+ elif fault == "quota-unknown":
+ language_server["quota_unknown"] = True
+ elif fault == "cleared-before-proof":
+ language_server["cleared_input_indices"] = [0]
+ elif fault == "retention-io":
+ def cannot_retain(*args, **kwargs):
+ raise OSError("Synthetic supplemental storage unavailable")
+ monkeypatch.setattr(store, "retain_native_result", cannot_retain)
+ context = json.dumps(original_brief, ensure_ascii=False, separators=(",", ":"))
+ brief = {"schema_version": "collaboration_brief_v0", "purpose": "Assess the original hypothesis", "context": context,
+ "constraints": ["Do not change frozen work"], "inputs": [], "acceptance": ["Return the assessment"],
+ "return_requirement": "Return to the original conversation"}
+ response = {"message": "Unverified model delivery claim", "proposals": [], "gate": None,
+ "context_handoff": [{"goal_id": "work", "agent_id": "worker", "brief": brief}]}
+ if fault == "wrong-target":
+ response["context_handoff"][0]["agent_id"] = "other-worker"
+ elif fault == "changed-brief":
+ brief["context"] = "Changed original evidence"
+ elif fault == "tool-step":
+ language_server["review_native_fault"] = "tool-step"
+ language_server["review_fault_start_index"] = 2
+ language_server["review_response"] = response
+ session = store.load_session(original["session_id"])
+ _write(_root(runtime.coordination_runtime_root) / "policy.json", {"schema_version": POLICY_SCHEMA, "sources": {
+ session["channel_id"]: {"local_delivery_scope": "selected", "sender_ids": [binding["operator_ref"]],
+ "targets": [{"goal_id": "work", "agent_id": "worker"}]}}})
+ if fault == "stop":
+ store.append_event(session["session_id"], original["turn_id"], kind="turn.interrupt_requested", payload={})
+ original_events = store.events_after(original["session_id"], original["turn_id"], None)
+ elif fault == "new-input":
+ store.create_queued_turn(session["session_id"], client_turn_id="new-correction", message="A correction", origin="lark")
+ elif fault == "no-anchor":
+ events_after = store.events_after
+ def without_anchor(*args):
+ rows = deepcopy(events_after(*args))
+ for item in rows:
+ item.get("payload", {}).pop("native_input", None)
+ return rows
+ monkeypatch.setattr(store, "events_after", without_anchor)
+ elif fault == "changed-proof":
+ observe = AntigravityAdapter.completed_result_observation
+ def changed(**kwargs):
+ result = observe(**kwargs)
+ if result:
+ result["response"]["message"] += "Changed original answer"
+ return result
+ monkeypatch.setattr(AntigravityAdapter, "completed_result_observation", staticmethod(changed))
+ path = transport.root / f"{ref}.json"
+ transport.reconcile_request(path)
+ if fault in {"quota-restore", "quota-unchanged"}:
+ held = transport.core.read_request(ref)
+ assert held["metadata_quota_wait"]["remaining_fraction"] == 0.006
+ assert not held.get("context_recovery")
+ retained = store.native_result(original["session_id"], original["turn_id"])
+ assert retained["upstream_turn_id"] == completed["upstream_turn_id"]
+ retained_bytes = (store._session_dir(original["session_id"]) / "native-results" / (original["turn_id"] + ".json")).read_bytes()
+ # The original service retained its own verified facts before quota
+ # recovery. Later native compaction cannot erase this obligation.
+ language_server["cleared_input_indices"] = [0]
+ transport.reconcile_request(path)
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+ if fault == "quota-restore":
+ language_server["quota_fraction"] = 1
+ transport.reconcile_request(path)
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+ from datetime import timedelta
+ language_server["quota_reset"] = (datetime.now(timezone.utc) + timedelta(days=14)).isoformat()
+ transport.reconcile_request(path)
+ assert (store._session_dir(original["session_id"]) / "native-results" / (original["turn_id"] + ".json")).read_bytes() == retained_bytes
+ row = transport.core.read_request(ref)
+ assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_bytes
+ if fault in {"quota-unchanged", "quota-unknown", "stop", "new-input", "no-anchor", "changed-proof", "unknown-outer",
+ "cleared-before-proof", "retention-io"}:
+ assert row["turn_id"] == original["turn_id"]
+ assert not pending(runtime.coordination_runtime_root, "work", "worker")["items"]
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+ return
+ assert row["turn_id"] != original["turn_id"], "Completed invalid metadata was not returned to its original worker"
+ reviewed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)
+ transport.reconcile_request(path)
+ entries = pending(runtime.coordination_runtime_root, "work", "worker")["items"]
+ if fault in {None, "quota-restore"}:
+ assert reviewed["status"] == "completed", reviewed
+ assert len(entries) == 1 and entries[0]["source_id"] == ref
+ assert entries[0]["brief"]["context"] == context
+ else:
+ assert reviewed["status"] == "failed" and not entries
+ assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_bytes
+ assert store.events_after(original["session_id"], original["turn_id"], None) == original_events
+ assert "native_result_recovery" not in row # Already delivered prose is not sent a second time.
+ assert sum("Original complete visible answer." in str(item.get("body", {}).get("content", item.get("content")))
+ for item in provider.messages.values()) == 1
+ assert not any("Unverified model delivery claim" in write[1] for write in provider.writes)
+ sends = [body for method, body in language_server["calls"] if method == "SendUserCascadeMessage"]
+ assert len(sends) == 2 and sends[-1]["cascadeConfig"]["plannerConfig"]["toolConfig"]["forceDisableToolCalling"] is True
+ before = (len(sends), len(provider.writes))
+ transport.reconcile_request(path)
+ assert (len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]), len(provider.writes)) == before
+ finally:
+ fixture.close()
+
+
def connect_group(fixture):
store, runtime, provider, transport = connect(fixture)
runtime.codex_home = fixture[-1].parent / "isolated-native-account"