From cee7414f08330788d2bbc62846311051dd521aa3 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 20:00:55 +0800 Subject: [PATCH 1/2] fix(chat): reuse history reads during recovery discovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/external_conversations.py | 56 +++++- .../extensions/lark/private_conversations.py | 4 +- tests/test_context_recovery_discovery.py | 164 ++++++++++++++++++ 3 files changed, 217 insertions(+), 7 deletions(-) create mode 100644 tests/test_context_recovery_discovery.py diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index b4078e6fdc..d0b01e2413 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -38,6 +38,37 @@ def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]: return identity +class _RecoveryDiscoveryReads: + """Lazy facts for one queue-discovery pass, never recovery admission. + + Historical requests share Session history. Retain queue/control/history + hints, never provider identity or a recovery decision. A new + pass builds new observations; normal admission never receives this object. + """ + + def __init__(self, core: ChatExternalConversations) -> None: + self.core = core + self.queues: dict[str, bool] = {} + self.histories: dict[str, list[dict[str, Any]]] = {} + self.controls: list[dict[str, Any]] | None = None + + def queued(self, session_id: str) -> bool: + if session_id not in self.queues: + self.queues[session_id] = bool(self.core.controller.store.queued_turns(session_id)) + return self.queues[session_id] + + def user_messages(self, session_id: str) -> list[dict[str, Any]]: + if session_id not in self.histories: + self.histories[session_id] = [message for message in self.core.controller.store.messages(session_id) + if message.get("role") == "user"] + return self.histories[session_id] + + def source_controls(self) -> list[dict[str, Any]]: + if self.controls is None: + self.controls = self.core.pending() + return self.controls + + class ChatExternalConversations: def __init__(self, controller: Any) -> None: self.controller = controller @@ -52,10 +83,13 @@ def _context_recovery_plan(self, row: dict[str, Any], *, selected_binding: dict[str, Any] | None = None, discovery_only: bool = False, preparation_scope_available: bool = False, - metadata_capacity: dict[str, Any] | None = None) -> dict[str, Any]: + metadata_capacity: dict[str, Any] | None = None, + discovery_reads: _RecoveryDiscoveryReads | None = None) -> dict[str, Any]: from ...chat_manager import MANAGER_CONTEXT_VERSION from ...control_plane.effect_runtime import effect_runtime_result + if discovery_reads is not None and not discovery_only: + raise ValueError("discovery observations cannot authorize recovery") store = self.controller.store session = store.load_session(str(row.get("session_id") or "")) turn = store.load_turn(str(row.get("session_id") or ""), str(row.get("turn_id") or "")) @@ -99,7 +133,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *, later_equivalent = any(item.get("role") == "user" and item.get("turn_id") != turn["turn_id"] and item.get("text") == row.get("message") and item.get("origin") in {None, "lark", "web"} and datetime.fromisoformat(item["created_at"]) > created - for item in store.messages(session["session_id"])) + for item in (discovery_reads.user_messages(session["session_id"]) + if discovery_reads is not None else store.messages(session["session_id"]))) from ...chat_manager import steward_machine_defaults return effect_runtime_result("collaboration.conversation.request", { "request": {key: value for key, value in row.items() if key != "attachments"}, @@ -107,10 +142,12 @@ def _context_recovery_plan(self, row: dict[str, Any], *, "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn, "observed_at": datetime.now(timezone.utc).isoformat(), "admitted_turn": admitted, - "queued_input_present": bool(store.queued_turns(session["session_id"])), + "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")), "steering_observed": any(event.get("kind") == "turn.steered" for event in events), - "stop_observed": self._source_stop_observed(row, turn, events), + "stop_observed": self._source_stop_observed(row, turn, events, + source_controls=discovery_reads.source_controls if discovery_reads is not None else None), "preparation_observation": preparation, "later_equivalent_input_present": later_equivalent, "preparation_scope_available": preparation_scope_available, @@ -187,7 +224,13 @@ def quota_backlog_pending(self, row: dict[str, Any], *, return effect_runtime_result("collaboration.conversation.request", params).get("pending") is True return True - def context_recovery_candidate(self, request_ref: str) -> bool: + def context_recovery_observer(self) -> Callable[[str], bool]: + """Reuse scan-local hints; reread each request, Turn and binding.""" + reads = _RecoveryDiscoveryReads(self) + return lambda ref: self.context_recovery_candidate(ref, discovery_reads=reads) + + def context_recovery_candidate(self, request_ref: str, *, + discovery_reads: _RecoveryDiscoveryReads | None = None) -> bool: """Discovery only; every authority and native fact is re-read at admission.""" try: row = self.read_request(request_ref) @@ -209,7 +252,8 @@ def context_recovery_candidate(self, request_ref: str) -> bool: 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 + return self._context_recovery_plan(row, discovery_only=True, + discovery_reads=discovery_reads).get("recovery_candidate") is True except (KeyError, ValueError, RuntimeError, OSError): return False diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index ff16ac1633..989359bd48 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -1290,9 +1290,11 @@ def inbox() -> Path: except (KeyError, ValueError, OSError): return # Unknown authority or provider effects stay pending. + @effect_runtime_request_scope() def pending_delivery_paths(self) -> list[Path]: """The durable transport store is the queue; no in-memory admission.""" paths = [] + recovery_candidate = self.core.context_recovery_observer() for path in sorted(self.root.glob("*.json")): row = _read_json(path) if row["status"] != "delivered": @@ -1303,7 +1305,7 @@ def pending_delivery_paths(self) -> list[Path]: # The canonical request, provider effect inventory and source # are independently re-read by recover_failed_request. self._context_recovery_probe_at[path] = time.monotonic() + 60 - if legacy_recovery_repairs(row) or self.core.context_recovery_candidate(row["request_ref"]): + if legacy_recovery_repairs(row) or recovery_candidate(row["request_ref"]): paths.append(path) return [*paths, *self._web_projection_paths()] diff --git a/tests/test_context_recovery_discovery.py b/tests/test_context_recovery_discovery.py new file mode 100644 index 0000000000..94db5ec1de --- /dev/null +++ b/tests/test_context_recovery_discovery.py @@ -0,0 +1,164 @@ +"""Discovery amortizes history IO; execution still needs current authority.""" +from collections import Counter +from datetime import datetime, timedelta +from pathlib import Path + +import pytest + +from test_native_steward_private import steward, finish # noqa: F401 +from loopx.capabilities.native_chat import external_conversations +from loopx.extensions.lark.conversation_identity import identity_ref + + +@pytest.fixture +def failed_source(steward): # noqa: F811 + store, runtime, provider, transport, binding, capture, _ = steward + fake = Path(runtime.codex_bin) + fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', + '"codexErrorInfo": "contextWindowExceeded"').replace( + ' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '')) + event = provider.event("steward-app", "discovery-history", "typed terminal failure") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + assert finish(runtime, row)["status"] == "failed" + store.update_session(row["session_id"], manager_context_version=20) + assert transport.reconcile() == 1 + return store, runtime, provider, transport, binding, capture, ref, row + + +def test_one_discovery_pass_reuses_queue_history_and_controls(failed_source, monkeypatch): + store, _, provider, transport, _, capture, ref, row = failed_source + # Model a retained pre-dispatch failure after its existing backoff. Discovery + # is only a hint: the original native event sequence still owns admission. + store.update_turn(row["session_id"], row["turn_id"], upstream_turn_id=None, + error_code="manager_authorization_unavailable", delta_count=0, response=None) + real_datetime = datetime + class AfterBackoff(datetime): + @classmethod + def now(cls, tz=None): + return real_datetime.now(tz) + timedelta(minutes=2) + monkeypatch.setattr(external_conversations, "datetime", AfterBackoff) + counts = Counter() + for owner, name in [(store, "queued_turns"), (store, "messages"), (transport.core, "pending")]: + original = getattr(owner, name) + def counted(*args, _name=name, _read=original, **kwargs): + counts[_name] += 1 + return _read(*args, **kwargs) + monkeypatch.setattr(owner, name, counted) + before_calls, before_requests = len(provider.calls), capture.read_bytes() + observe = transport.core.context_recovery_observer() + assert [observe(ref) for _ in range(4)] == [True] * 4 + assert counts == {"queued_turns": 1, "messages": 1, "pending": 1} + # The next pass rereads rather than retaining a negative/positive cache. + assert transport.core.context_recovery_observer()(ref) + assert counts == {"queued_turns": 2, "messages": 2, "pending": 2} + assert len(provider.calls) == before_calls and capture.read_bytes() == before_requests + + +@pytest.mark.parametrize("change", ["queued-input", "revoked", "owner-unavailable"]) +def test_discovery_hint_never_authorizes_recovery(failed_source, monkeypatch, change): + store, _, provider, transport, binding, capture, ref, row = failed_source + observe = transport.core.context_recovery_observer() + assert observe(ref) + original = store._turn_path(row["session_id"], row["turn_id"]).read_bytes() + before = capture.read_bytes() + if change == "queued-input": + store.create_queued_turn(row["session_id"], client_turn_id="new-original-input", + message="Resolve the newer input first", origin="web") + queued = store.queued_turns + observed = [] + deny_after_read = False + def current_queue(session_id): + result = queued(session_id) + observed.append(bool(result)) + if deny_after_read: + provider.owner_ready = False + return result + monkeypatch.setattr(store, "queued_turns", current_queue) + transport.core.context_recovery_observer()(ref) + assert observed == [True], "The next observation reads the new queue" + # A candidate is not an eligibility verdict. Normal admission must read + # the new queue itself, then reject authority lost during that read. + deny_after_read = True + with pytest.raises(ValueError): + transport.core.recover_failed_request(ref, + source_observation=lambda _selected: pytest.fail("source IO after denied authority")) + assert observed == [True, True] + else: + if change == "revoked": + transport.bindings.disconnect(binding["binding_id"], + expected_revision=transport.bindings.read()["revision"]) + assert not observe(ref), "Even the same discovery pass rereads the binding" + else: + provider.owner_ready = False + with pytest.raises(ValueError): + transport.core.recover_failed_request(ref, + source_observation=lambda _selected: pytest.fail("source IO after denied authority")) + assert capture.read_bytes() == before + assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == original + + +def test_unavailable_queue_is_not_cached_as_empty(failed_source, monkeypatch): + store, _, _, transport, _, _, ref, _ = failed_source + original = store.queued_turns + calls = 0 + def flaky(session_id): + nonlocal calls + calls += 1 + if calls == 1: + raise OSError("synthetic unavailable queue") + return original(session_id) + monkeypatch.setattr(store, "queued_turns", flaky) + observe = transport.core.context_recovery_observer() + assert not observe(ref) + assert observe(ref) + assert calls == 2 + + +def test_admission_rejects_discovery_observations(failed_source): + _, _, _, transport, _, _, _, row = failed_source + reads = external_conversations._RecoveryDiscoveryReads(transport.core) + with pytest.raises(ValueError, match="cannot authorize recovery"): + transport.core._context_recovery_plan(row, discovery_reads=reads) + + +def test_discovery_facts_do_not_cross_sessions(failed_source): + store, _, _, transport, _, _, _, row = failed_source + second = store.create_session(goal_id="separate-synthetic-goal", agent_id="codex", channel_id="separate-source", + adapter_kind="codex_app_server", upstream_thread_id="separate-synthetic-thread") + queued, created = store.create_queued_turn(second["session_id"], client_turn_id="other-source-input", + message="Only the second source has queued input", origin="web") + assert created + reads = external_conversations._RecoveryDiscoveryReads(transport.core) + assert not reads.queued(row["session_id"]) + assert reads.queued(second["session_id"]) + assert {message["turn_id"] for message in reads.user_messages(second["session_id"])} == {queued["turn_id"]} + assert all(message["turn_id"] != queued["turn_id"] for message in reads.user_messages(row["session_id"])) + + +def test_lark_scanner_shares_only_one_pass_and_pins_one_code_revision(failed_source, monkeypatch): + from loopx.control_plane import effect_runtime + store, runtime, provider, transport, binding, _, ref, row = failed_source + event = provider.event("steward-app", "second-discovery-history", "typed terminal failure") + transport.admit("steward-app", event) + second = identity_ref(binding["provider_ref"], event["message_id"]) + next_row = transport.core.read_request(second) + assert finish(runtime, next_row)["status"] == "failed" + assert transport.reconcile_request(transport.root / f"{second}.json") == 1 + store.update_session(row["session_id"], manager_context_version=20) + calls = Counter() + for owner, name in [(store, "queued_turns"), (effect_runtime, "_runtime_fingerprint")]: + original = getattr(owner, name) + def counted(*args, _name=name, _read=original, **kwargs): + calls[_name] += 1 + return _read(*args, **kwargs) + monkeypatch.setattr(owner, name, counted) + provider_calls = len(provider.calls) + expected = {transport.root / f"{request}.json" for request in (ref, second)} + for number in (1, 2): + # Advance the existing discovery clock, not a new retry mechanism. + transport._context_recovery_probe_at.clear() + assert set(transport.pending_delivery_paths()) == expected + assert calls == {"queued_turns": number, "_runtime_fingerprint": number} + assert len(provider.calls) == provider_calls From 8ed84e740bed68d9148435125c054bb9ddf6c28a Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 3 Nov 2026 23:12:38 +0800 Subject: [PATCH 2/2] test(capabilities): include anti-entropy in builtin inventory Signed-off-by: huangruiteng --- examples/capability-extension-registry-smoke.py | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/capability-extension-registry-smoke.py b/examples/capability-extension-registry-smoke.py index e176471f0e..1b07f496ad 100644 --- a/examples/capability-extension-registry-smoke.py +++ b/examples/capability-extension-registry-smoke.py @@ -84,6 +84,7 @@ def run_cli(runtime_root: Path, *args: str) -> dict[str, object]: "reliability-diagnostics", "progress-review-sentinel", "goal-capability-organization", + "anti-entropy-governance", ] assert all(item["provider_id"] == "loopx-core" for item in builtin_capabilities) value_summary = next(