From c656e063497ca03e306f7097b26c7e4b24fce15f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 23:39:53 +0530
Subject: [PATCH] fix(chat): retain pending partial results after quota
recovery
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
.../native_chat/external_conversations.py | 4 +-
.../collaboration/conversation_binding.ts | 47 +++++---
.../extensions/lark/private_conversations.py | 4 +-
.../conversation_binding.test.ts | 41 +++++++
tests/test_lark_private_conversations.py | 100 ++++++++++++++++++
5 files changed, 179 insertions(+), 17 deletions(-)
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index 950481f696..77f831fd2b 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -112,7 +112,7 @@ def read_controls() -> list[dict[str, Any]]:
def quota_backlog_pending(self, row: dict[str, Any], *,
source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
- """Project existing work without resolving providers or admitting recovery."""
+ """Retain quota and declared unfinished work without admitting recovery."""
from ...control_plane.effect_runtime import effect_runtime_result
store = self.controller.store
@@ -124,7 +124,7 @@ def quota_backlog_pending(self, row: dict[str, Any], *,
"current_session": session,
"pending_observation": observation,
}
- # The typed owner first excludes work that is not quota backlog even
+ # The typed owner first excludes work that has no retained obligation even
# without a stop. Stop evidence can only remove a pending request;
# never expose a positive result before reading those native facts.
if effect_runtime_result("collaboration.conversation.request", params).get("pending") is not True:
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 845cd4fd8c..5de7820e6f 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -442,10 +442,26 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
const turn = observation.turn == null ? null : requireJsonObject(observation.turn, "observed Turn");
// Pending work is not current retry eligibility: backoff, disabled bindings,
// exhausted attempts and unavailable providers do not retire the request.
- // Only explicit lifecycle evidence removes a delivered quota failure.
+ // A delivered partial answer is not completion either. Keep explicit
+ // unfinished work and a correlated quota continuation's unresolved gate
+ // visible, without changing recovery eligibility or admitting execution.
+ const response = turn?.response && typeof turn.response === "object" && !Array.isArray(turn.response)
+ ? turn.response as JsonObject : null;
+ const reservation = row.context_recovery && typeof row.context_recovery === "object"
+ && !Array.isArray(row.context_recovery) ? row.context_recovery as JsonObject : null;
+ const gate = response?.gate;
+ const quotaGate = reservation !== null && Number.isSafeInteger(reservation.quota_attempt)
+ && Number(reservation.quota_attempt) >= 1 && Number(reservation.quota_attempt) <= 3
+ && typeof reservation.failed_turn_id === "string" && reservation.failed_turn_id !== row.turn_id
+ && typeof reservation.client_turn_id === "string" && reservation.client_turn_id.length > 0
+ && turn?.client_turn_id === reservation.client_turn_id
+ && gate !== null && typeof gate === "object" && !Array.isArray(gate)
+ && typeof (gate as JsonObject).kind === "string" && typeof (gate as JsonObject).summary === "string";
+ const unfinished = turn?.status === "completed" && response !== null
+ && (unfinishedProjectWork(response) !== null || quotaGate);
return {pending: row.command === null && row.status === "accepted"
&& turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
- && turn?.status === "failed" && turn.error_code === "usage_limit_exceeded"
+ && (unfinished || turn?.status === "failed" && turn.error_code === "usage_limit_exceeded")
&& !(current?.session_id === row.session_id && current?.status === "closed")
&& observation.stop_observed !== true};
}
@@ -624,25 +640,30 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
* proposal alone grants nothing: only the original worker may reserve one
* continuation after stable latest-native/source/grant/stop checks.
*/
+function unfinishedProjectWork(result: JsonObject): JsonObject | null {
+ const raw = result.unfinished_work;
+ if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null;
+ const work = raw as JsonObject;
+ if (Object.keys(work).sort().join(",") !== "dependency,effects_settled,remaining"
+ || work.effects_settled !== true
+ || [work.remaining, work.dependency].some(value => typeof value !== "string"
+ || !value.trim() || value.length > 1000)
+ || [result.gate, result.protected_action, result.context_handoff, result.goal_draft].some(value => value != null)
+ || !Array.isArray(result.proposals) || result.proposals.length) return null;
+ return work;
+}
+
function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: JsonObject,
completed: JsonObject, previous: JsonObject | null, version: number): JsonObject {
const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
const response = completed.response;
if (!response || typeof response !== "object" || Array.isArray(response)) return no;
- const result = response as JsonObject;
- const raw = result.unfinished_work;
- if (!raw || typeof raw !== "object" || Array.isArray(raw)) return no;
- const work = raw as JsonObject;
+ const work = unfinishedProjectWork(response as JsonObject);
+ if (work === null) return no;
const matchesWork = (value: unknown) => value !== null && typeof value === "object" && !Array.isArray(value)
&& Object.keys(value).sort().join(",") === "dependency,effects_settled,remaining"
&& ["remaining", "dependency", "effects_settled"].every(key => (value as JsonObject)[key] === work[key]);
- if (Object.keys(work).sort().join(",") !== "dependency,effects_settled,remaining"
- || work.effects_settled !== true
- || [work.remaining, work.dependency].some(value => typeof value !== "string"
- || !value.trim() || value.length > 1000)
- || [result.gate, result.protected_action, result.context_handoff, result.goal_draft].some(value => value != null)
- || !Array.isArray(result.proposals) || result.proposals.length
- || completed.session_id !== current.session_id || completed.turn_id !== row.turn_id
+ if (completed.session_id !== current.session_id || completed.turn_id !== row.turn_id
|| recovery.steering_observed !== false || recovery.stop_observed !== false
|| recovery.attachments_present !== false
|| typeof completed.upstream_turn_id !== "string" || !completed.upstream_turn_id) return no;
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index e7e65b9b5f..586ca46cd6 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -230,8 +230,8 @@ def health(self) -> dict[str, dict[str, int]]:
pending = row["status"] != "delivered"
if (not pending and path.parent == self.root
and row.get("session_id") and row.get("turn_id")):
- # Delivery of a quota-failure notice does not finish the user's
- # request. Read only the existing Core correlation and Turn;
+ # Delivery of a quota failure or partial result does not finish
+ # the request. Read only the existing Core correlation and Turn;
# recovery eligibility, allowance and effects still belong to
# the original worker's fresh admission checks.
native = self.core.read_request(row["request_ref"])
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 9059b11ae3..1bec0673a4 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -730,3 +730,44 @@ test("quota backlog projection retires explicit lifecycle stops, not unavailable
assert.equal(planBoundConversationRequest({...packet, request: changed}).pending, false);
}
});
+
+test("delivered partial results remain pending without authorizing another execution", () => {
+ const original = {request_ref: "f".repeat(24), command: null, status: "accepted",
+ session_id: "original", turn_id: "partial"};
+ const work = {remaining: "Finish the retained result", dependency: "Temporary source outage", effects_settled: true};
+ const response = {message: "Partial result", proposals: [], gate: null, protected_action: null, unfinished_work: work};
+ const turn = {session_id: "original", turn_id: "partial", status: "completed", response};
+ const packet = {request: original, current_session: {session_id: "original", status: "ready"},
+ pending_observation: {turn, stop_observed: false}};
+ assert.deepEqual(planBoundConversationRequest(packet), {pending: true});
+ for (const unfinished_work of [null, {...work, effects_settled: false}, {...work, remaining: ""}, {...work, extra: true}]) {
+ assert.equal(planBoundConversationRequest({...packet, pending_observation: {turn: {...turn,
+ response: {...response, unfinished_work}}, stop_observed: false}}).pending, false);
+ }
+ for (const changed of [{...turn, response: {...response, gate: {kind: "approval"}}},
+ {...turn, response: {...response, protected_action: {kind: "publish"}}},
+ {...turn, response: {...response, proposals: [{kind: "new-work"}]}},
+ {...turn, session_id: "foreign"}, {...turn, turn_id: "foreign"}]) {
+ assert.equal(planBoundConversationRequest({...packet,
+ pending_observation: {turn: changed, stop_observed: false}}).pending, false);
+ }
+ assert.equal(planBoundConversationRequest({...packet, pending_observation: {turn, stop_observed: true}}).pending, false);
+ assert.equal(planBoundConversationRequest({...packet,
+ current_session: {...packet.current_session, status: "closed"}}).pending, false);
+
+ const recovery = {quota_attempt: 1, failed_turn_id: "quota-failure", client_turn_id: "original-quota-continuation"};
+ const gated = {...turn, client_turn_id: recovery.client_turn_id,
+ response: {...response, unfinished_work: null, gate: {kind: "authorization", summary: "Verification remains blocked"}}};
+ const waiting = {...packet, request: {...original, context_recovery: recovery},
+ pending_observation: {turn: gated, stop_observed: false}};
+ assert.deepEqual(planBoundConversationRequest(waiting), {pending: true});
+ for (const context_recovery of [null, {...recovery, quota_attempt: 0}, {...recovery, quota_attempt: 4},
+ {...recovery, failed_turn_id: "partial"}, {...recovery, client_turn_id: "unrelated"}]) {
+ assert.equal(planBoundConversationRequest({...waiting,
+ request: {...original, context_recovery}}).pending, false);
+ }
+ assert.equal(planBoundConversationRequest({...waiting, pending_observation: {turn: {...gated,
+ response: {...gated.response, gate: null}}, stop_observed: false}}).pending, false);
+ assert.equal(planBoundConversationRequest({...waiting,
+ pending_observation: {turn: gated, stop_observed: true}}).pending, false);
+});
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 350457f807..2eceddc7f9 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -275,6 +275,106 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(
runtime.close()
+@pytest.mark.parametrize("command", ["stop", "new"])
+def test_delivered_unfinished_work_remains_visible_without_admitting_recovery(ordinary, monkeypatch, command): # noqa: F811
+ store, runtime, _, _, _, fake, _ = ordinary
+ work = {"remaining": "Finish the verified remaining result", "dependency": "Temporary source outage", "effects_settled": True}
+ partial = "Partial result; authorized work remains.\n" + json.dumps({
+ "message": "", "unfinished_work": work}) + ""
+ source = fake.read_text()
+ start = source.index(" response = 'Visible runtime response.")
+ end = source.index(' print(json.dumps({"method": "item/agentMessage/delta"', start)
+ fake.write_text(source[:start] + f" response = {partial!r}\n" + source[end:])
+ store, runtime, provider, transport = connect(ordinary)
+ try:
+ event = provider.event("notes-app", "unfinished-visible", "Finish this synthetic request.")
+ transport.admit("notes-app", event)
+ row = transport.core.pending()[0]
+ turn = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)
+ assert turn["response"]["unfinished_work"] == work
+ transport.reconcile()
+ assert transport.core.read_request(row["request_ref"])["delivery_verified"]
+ restarted = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent,
+ runner=provider, cli_bin="lark-cli")
+ starts = ordinary[4].read_bytes(), list(provider.writes)
+ with monkeypatch.context() as observation_only:
+ observation_only.setattr(runtime, "enqueue_turn", lambda **_: pytest.fail("health admitted work"))
+ observation_only.setattr(transport.core, "recover_failed_request", lambda *_args, **_: pytest.fail("health replayed work"))
+ assert transport.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
+ assert restarted.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
+ assert (ordinary[4].read_bytes(), provider.writes) == starts
+ assert store.load_turn(row["session_id"], row["turn_id"]) == turn
+ transport.admit("notes-app", provider.event("notes-app", "retire-unfinished", f"/{command}"))
+ transport.reconcile()
+ assert transport.health()[row["binding_id"]]["pending_count"] == 0
+ finally:
+ runtime.close()
+
+
+@pytest.mark.parametrize("command", ["stop", "new"])
+def test_original_quota_recovery_with_gate_remains_pending_after_verified_reply(ordinary, monkeypatch, command): # noqa: F811
+ from datetime import timedelta, timezone
+ from loopx.capabilities.native_chat import external_conversations
+ from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
+
+ store, runtime, _, _, _, fake, _ = ordinary
+ source = fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', '"codexErrorInfo": "usageLimitExceeded"')
+ source = source.replace(' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '')
+ fake.write_text(source)
+ store, runtime, provider, transport = connect(ordinary)
+ try:
+ event = provider.event("notes-app", "quota-gate-visible", "typed terminal failure")
+ transport.admit("notes-app", event)
+ row = transport.core.pending()[0]
+ failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)
+ assert failed["error_code"] == "usage_limit_exceeded"
+ transport.reconcile()
+ failed_path = store._turn_path(row["session_id"], row["turn_id"])
+ failed_bytes = failed_path.read_bytes()
+ runtime.adapters.pop(row["session_id"]).close_session()
+ gate = {"kind": "authorization", "summary": "Original workspace authorization needs verification", "next_action": ""}
+ answer = "Recovery reached the original authorization boundary.\n" + json.dumps({
+ "message": "", "gate": gate}) + ""
+ source = source.replace('if "typed terminal failure" in json.dumps(request.get("params") or {}):', 'if False:')
+ start = source.index(" response = 'Visible runtime response.")
+ end = source.index(' print(json.dumps({"method": "item/agentMessage/delta"', start)
+ fake.write_text(source[:start] + f" response = {answer!r}\n" + source[end:])
+ clock = datetime.now(timezone.utc) + timedelta(seconds=61)
+ class AfterBackoff(datetime):
+ @classmethod
+ def now(cls, tz=None):
+ return clock
+ monkeypatch.setattr(external_conversations, "datetime", AfterBackoff)
+ monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **kw: {
+ "thread_id": "durable-thread", "turn_id": kw["turn_id"], "codex_error_info": "usageLimitExceeded",
+ "only_original_user_item": True, "quota_available": True,
+ "quota_observed_at": int(clock.timestamp() * 1000) / 1000}))
+ # The original scanner owns recovery; observing health never admits it.
+ transport = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent,
+ runner=provider, cli_bin="lark-cli")
+ transport.reconcile()
+ continued = transport.core.read_request(row["request_ref"])
+ assert continued["turn_id"] != row["turn_id"] and continued["context_recovery"]["quota_attempt"] == 1
+ completed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=continued["turn_id"], timeout_sec=10)
+ assert completed["status"] == "completed" and completed["response"]["gate"] == gate
+ transport.reconcile_request(transport.root / f"{row['request_ref']}.json")
+ assert transport.core.read_request(row["request_ref"])["delivery_verified"]
+ restarted = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent,
+ runner=provider, cli_bin="lark-cli")
+ before = ordinary[4].read_bytes(), list(provider.writes)
+ assert transport.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
+ assert restarted.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
+ restarted.reconcile()
+ assert (ordinary[4].read_bytes(), provider.writes) == before
+ assert failed_path.read_bytes() == failed_bytes
+ assert store.load_turn(row["session_id"], continued["turn_id"]) == completed
+ transport.admit("notes-app", provider.event("notes-app", "retire-quota-gate", f"/{command}"))
+ transport.reconcile()
+ assert transport.health()[row["binding_id"]]["pending_count"] == 0
+ finally:
+ runtime.close()
+
+
@pytest.mark.parametrize("command", ["stop", "new"])
def test_health_reads_source_controls_once_and_keeps_fresh_recovery(ordinary, monkeypatch, command): # noqa: F811
from loopx.capabilities.native_chat import external_conversations