fix(chat): retain pending partial results after quota recovery

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:

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;

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"])

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);
+});

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<loopx-review-json>" + json.dumps({
+        "message": "", "unfinished_work": work}) + "</loopx-review-json>"
+    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<loopx-review-json>" + json.dumps({
+            "message": "", "gate": gate}) + "</loopx-review-json>"
+        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