diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index e35f070e45..950481f696 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -88,29 +88,53 @@ def _context_recovery_plan(self, row: dict[str, Any], *, }) def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, Any], - events: list[dict[str, Any]]) -> bool: + events: list[dict[str, Any]], *, + source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool: """Read the same durable source controls for recovery and display.""" return (any(event.get("kind") in {"turn.interrupt_requested", "turn.interrupted"} for event in events) or (turn.get("status") in {"failed", "completed"} and any( item.get("binding_id") == row["binding_id"] and item.get("source") == row["source"] and item.get("command") in {"stop", "new"} and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"]) - for item in self.pending()))) + for item in (source_controls or self.pending)()))) - def quota_backlog_pending(self, row: dict[str, Any]) -> bool: + def quota_backlog_observer(self) -> Callable[[dict[str, Any]], bool]: + """Share lazy journal IO within one read-only observation, never admission.""" + controls: list[dict[str, Any]] | None = None + + def read_controls() -> list[dict[str, Any]]: + nonlocal controls + if controls is None: + controls = self.pending() + return controls + + return lambda row: self.quota_backlog_pending(row, source_controls=read_controls) + + 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.""" from ...control_plane.effect_runtime import effect_runtime_result store = self.controller.store session = store.load_session(row["session_id"]) turn = store.load_turn(row["session_id"], row["turn_id"]) - events = store.events_after(row["session_id"], row["turn_id"], None) if turn else [] - return effect_runtime_result("collaboration.conversation.request", { + observation = {"turn": turn, "stop_observed": False} + params = { "request": {key: value for key, value in row.items() if key != "attachments"}, "current_session": session, - "pending_observation": {"turn": turn, - "stop_observed": self._source_stop_observed(row, turn, events) if turn else False}, - }).get("pending") is True + "pending_observation": observation, + } + # The typed owner first excludes work that is not quota backlog 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: + return False + assert turn is not None # Required by the typed pending projection. + events = store.events_after(row["session_id"], row["turn_id"], None) + if self._source_stop_observed(row, turn, events, source_controls=source_controls): + observation["stop_observed"] = True + return effect_runtime_result("collaboration.conversation.request", params).get("pending") is True + return True def context_recovery_candidate(self, request_ref: str) -> bool: """Discovery only; every authority and native fact is re-read at admission.""" diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index a67d880860..e7e65b9b5f 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -18,6 +18,7 @@ from ...capabilities.native_chat.external_conversations import ChatExternalConversations from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json from ...control_plane.collaboration import conversation_reply_context +from ...control_plane.effect_runtime import effect_runtime_request_scope from ...file_lock import exclusive_file_lock from ...presentation.markdown import markdown_scalar from ...presentation.renderers.conversation_status_markdown import render_conversation_status @@ -219,8 +220,10 @@ def profiles(self) -> dict[str, dict[str, str]]: "binding_id": row["binding_id"], "consumer_ref": hashlib.sha256(app_id.encode("utf-8")).hexdigest()[:32]} return profiles + @effect_runtime_request_scope() def health(self) -> dict[str, dict[str, int]]: health: dict[str, dict[str, int]] = {} + quota_pending = self.core.quota_backlog_observer() for path in [*self.root.glob("*.json"), *self.root.glob("web-projections/*.json")]: row = _read_json(path) entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0}) @@ -235,7 +238,7 @@ def health(self) -> dict[str, dict[str, int]]: if (native.get("status") == "accepted" and native.get("command") is None and all(native.get(key) == row.get(key) for key in ("binding_id", "source", "session_id", "turn_id"))): - pending = self.core.quota_backlog_pending(native) + pending = quota_pending(native) if pending: entry["pending_count"] += 1 entry["recovery_count"] += int(any(phase.get("started") and not phase.get("verified") diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 928448c658..350457f807 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -242,7 +242,7 @@ def connect_group(fixture): @pytest.mark.parametrize("info, pending", [("usageLimitExceeded", 1), ("cyberPolicy", 0), (None, 0)]) -def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(ordinary, info, pending): # noqa: F811 +def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(ordinary, monkeypatch, info, pending): # noqa: F811 store, runtime, provider, transport = connect(ordinary) fake = ordinary[-2] fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', @@ -259,6 +259,11 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying( transport.reconcile() saved = json.loads((transport.root / f"{row['request_ref']}.json").read_text()) assert saved["status"] == "delivered" and saved["deliveries"]["terminal"]["verified"] + if not pending: + # The typed owner can exclude completed/non-quota work without + # scanning unrelated events or source-control journals. + monkeypatch.setattr(store, "events_after", lambda *_: pytest.fail("not quota backlog")) + monkeypatch.setattr(transport.core, "pending", lambda: pytest.fail("not quota backlog")) assert transport.health()[row["binding_id"]]["pending_count"] == pending before = ordinary[4].read_bytes(), list(provider.writes) restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root, @@ -270,6 +275,87 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying( 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 + from loopx.control_plane import effect_runtime + + store, runtime, provider, transport = connect(ordinary) + fake = ordinary[-2] + fake.write_text(fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', + '"codexErrorInfo": "usageLimitExceeded"').replace( + ' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '')) + try: + binding = transport._binding("notes-app") + rows = [] + for index in range(8): + event = provider.event("notes-app", f"health-quota-{index}", "typed terminal failure") + transport.admit("notes-app", event) + row = transport.core.read_request(identity_ref(binding["provider_ref"], event["message_id"])) + turn = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10) + assert turn["status"] == "failed" and turn["error_code"] == "usage_limit_exceeded" + transport.reconcile() + rows.append(row) + + def durable_state(): + paths = {path for root in (transport.root, transport.core.root) for path in root.glob("*.json")} + paths.update(path for path in store.sessions_root.rglob("*") if path.suffix in {".json", ".jsonl"}) + return {path: path.read_bytes() for path in paths} + + before = durable_state() + effects = ordinary[4].read_bytes(), list(provider.writes) + reads = [] + revisions = [] + read_json = external_conversations._read_json + fingerprint = effect_runtime._runtime_fingerprint + + def counted_read(path): + if path.parent == transport.core.root and re.fullmatch(r"[a-f0-9]{24}", path.stem): + reads.append(path) + return read_json(path) + + monkeypatch.setattr(external_conversations, "_read_json", counted_read) + def counted_revision(): + revisions.append(True) + return fingerprint() + + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", counted_revision) + expected = {"pending_count": len(rows), "recovery_count": 0} + assert transport.health()[binding["binding_id"]] == expected + # Correlation reads plus one complete control observation, rather than + # a complete journal scan for each historical terminal request. + assert len(reads) <= 3 * len(rows) + assert len(revisions) <= 1 # One managed runtime revision for the health request. + assert durable_state() == before + assert (ordinary[4].read_bytes(), provider.writes) == effects + + # Durable control observations must keep all three existing fences. + # These synthetic journal records are inputs, not new admissions. + for index, change in enumerate([ + {"binding_id": "f" * 24}, + {"source": dict(rows[0]["source"], source_ref="e" * 24)}, + {"created_at": "2000-01-01T00:00:00+00:00"}, + ]): + control_row = dict(rows[0], command=command, **change) + (transport.core.root / f"{index + 1:024x}.json").write_text(json.dumps(control_row)) + # The journal reader already tolerates damaged JSON and ignores + # non-request metadata; observation reuse must retain that behavior. + (transport.core.root / ("d" * 24 + ".json")).write_text("{") + (transport.core.root / "metadata.json").write_text(json.dumps(dict(rows[0], command=command))) + assert transport.health()[binding["binding_id"]] == expected + assert all(transport.core.quota_backlog_pending(row) for row in rows) + + # An actual later control must reach both a new health observation and + # the default recovery check, independently of the previous snapshot. + control = provider.event("notes-app", "health-control", f"/{command}") + transport.admit("notes-app", control) + assert transport.health()[binding["binding_id"]]["pending_count"] == 1 # undelivered control + assert all(not transport.core.quota_backlog_pending(row) for row in rows) + assert all(not transport.core.context_recovery_candidate(row["request_ref"]) for row in rows) + finally: + runtime.close() + + @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-input", "private-path", "still-incomplete"]) def test_original_worker_continues_explicit_unfinished_project_work_once_after_restart(ordinary, monkeypatch, fault): # noqa: F811