diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 75251ded78..7b789aa550 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1921,3 +1921,26 @@ Reservation/enqueue crashes adopt the same queued Turn. A failed, partial or unknown review stays pending and cannot become another review or business retry. Other malformed actions and completed Turns without a recovered native result remain explicit holds; this path is not a generic action or execution replayer. + +### File-backed idle recovery observations + +Repeated recovery checks must not reparse every historical Turn and source body +for each old request. The File Chat provider retains only validated terminal-Turn +identity/status and the source-control fields used by the existing stop/new and +later-input checks. Reuse requires an unchanged device, inode, nanosecond mtime, +ctime and size, checked before and after the original read. Changed or unreadable +files use the original reader; nonterminal Turns are always read in full. Deleted +entries are pruned, concurrent callers receive independent metadata, and a read +racing a replacement cannot seed a reusable observation. +Windows keeps the original fresh-read path because its Python stat ctime can +represent creation time rather than a usable change stamp; see the +[Python stat time fields](https://docs.python.org/3.12/library/os.html#os.stat_result.st_ctime_ns). +The private-reply journal observer shares the same change-token primitive while +retaining its own legacy-repair projection and original effect checks. + +These observations are neither recovery decisions nor grants. Exact requests, +selected Turns, current bindings, native effects and typed admission remain fresh. +Source qualification compares the same File history and complete recovery plans, +including newly queued input, strict read failures, stop/new changes and replaced +files with preserved mtime. Installed responsiveness, provider/model availability +and actual original-audience replies remain separate acceptance obligations. diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 49ac3a5edf..7d80154f67 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -1260,3 +1260,20 @@ worker 使用独立展示阶段返回这些回执，保留原失败 Turn、失 预约或入队后的崩溃只采用同一个队列 Turn；失败、部分提交或结果未知的复核继续 记为欠办，不能变成又一次复核或业务重试。其他错误动作、没有已恢复原生结果的 completed Turn 继续明确保留，本路径不提供通用动作或执行重放。 + +### File 后端的空闲恢复观测 + +逐条检查旧请求时，不应反复解析全部历史 Turn 和来源正文。File Chat provider +只保留经过验证的终态 Turn 身份/状态，以及既有 stop/new 和后续输入检查所需的 +来源控制字段。复用要求 device、inode、纳秒 mtime、ctime 和文件大小均未改变， +并在原读取前后核对。变化或不可读的文件仍走原 reader；非终态 Turn 每次读取 +完整正文。删除项清理、并发调用返回独立元数据，读取期间发生替换时不保存可复用观测。 +Windows 的 Python stat ctime 可能代表创建时间，不能作为可靠的变化标识，因此 +继续走原来的新鲜读取路径，见 +[Python stat 时间字段说明](https://docs.python.org/3.12/library/os.html#os.stat_result.st_ctime_ns)。 +私聊回复 journal 的观测复用同一个变化标识实现，仍保留自身的旧修复投影和原效果核验。 + +这些观测不是恢复决定或授权。精确请求、选中的 Turn、当前 binding、原生效果及 +typed 准入仍保持新鲜读取。源码资格以同一份 File 历史和完整恢复计划做对照， +覆盖新入队输入、严格读取失败、stop/new 变化及保留 mtime 的文件替换。实际安装后的 +响应速度、provider/模型可用性和原受众真正收到答复，仍须分别验收。 diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 3177da139c..47d5eeaa68 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -24,6 +24,7 @@ ) from ...file_lock import exclusive_file_lock, try_exclusive_file_lock from .codex_context import checked_session_home +from .file_discovery import FileChatDiscovery # Request journals are named by their 24-hex request_ref; lock metadata and # other siblings in this directory are not requests. @@ -82,7 +83,7 @@ def user_messages(self, session_id: str) -> list[dict[str, Any]]: def source_controls(self) -> list[dict[str, Any]]: if self.controls is None: - self.controls = self.core.pending() + self.controls = self.core.source_controls() return self.controls @@ -92,6 +93,7 @@ def __init__(self, controller: Any) -> None: self.bindings = controller.project_contexts.conversation_bindings self.bindings.controller = controller self.root = controller.store.root / "external-requests" + self._source_control_discovery = FileChatDiscovery() self.actions: Any | None = None def _context_recovery_plan(self, row: dict[str, Any], *, @@ -182,7 +184,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *, and item.get("source") == row["source"] and item.get("command") is None and datetime.fromisoformat(item["created_at"]) > datetime.fromisoformat(row["created_at"]) for item in (discovery_reads.source_controls() if discovery_reads is not None - else self.pending())) if turn.get("error_code") == "antigravity_step_error" else None, + else self.source_controls())) if turn.get("error_code") == "antigravity_step_error" else None, "native_input": anchors[0] if len(anchors) == 1 else None, "response_literal_open": bool(open_literal), "metadata_capacity": metadata_capacity, @@ -212,7 +214,7 @@ def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, 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 (source_controls or self.pending)()))) + for item in (source_controls or self.source_controls)()))) def quota_backlog_observer(self) -> Callable[[Sequence[dict[str, Any]]], list[bool]]: """Batch the existing typed projection; share controls only within this read.""" @@ -221,7 +223,7 @@ def quota_backlog_observer(self) -> Callable[[Sequence[dict[str, Any]]], list[bo def read_controls() -> list[dict[str, Any]]: nonlocal controls if controls is None: - controls = self.pending() + controls = self.source_controls() return controls def observe(rows: Sequence[dict[str, Any]]) -> list[bool]: @@ -1248,6 +1250,25 @@ def pending(self) -> list[dict[str, Any]]: return [_read_json(path) for path in sorted(self.root.glob("*.json")) if _REQUEST_JOURNAL_STEM.fullmatch(path.stem)] + def source_controls(self) -> list[dict[str, Any]]: + """Observe exact source-control fields without reparsing unchanged bodies. + + Recovery/stop rules remain with their original typed owner and callers. + A new, replaced or modified journal is read again; pending() and exact + request reads still expose the complete current canonical records. + """ + paths = (path for path in sorted(self.root.glob("*.json")) + if _REQUEST_JOURNAL_STEM.fullmatch(path.stem)) + + def project(_path, row): + fields = ("binding_id", "source", "command", "created_at") + # The tolerant reader can return {} after transient IO failure. + # An incomplete read must not seed an unchanged-file observation. + return {key: row[key] for key in fields} if all(key in row for key in fields) else None + + return [row for _path, row in self._source_control_discovery.rows(self.root, paths, + read=_read_json, project=project)] + def read_request(self, request_ref: str) -> dict[str, Any]: return _read_json(self.root / f"{request_ref}.json") diff --git a/loopx/capabilities/native_chat/file_discovery.py b/loopx/capabilities/native_chat/file_discovery.py new file mode 100644 index 0000000000..625ee8ad6b --- /dev/null +++ b/loopx/capabilities/native_chat/file_discovery.py @@ -0,0 +1,69 @@ +"""Small, invalidatable File Chat journal observations, never execution state.""" +from __future__ import annotations + +from collections.abc import Callable, Iterable, Iterator +from copy import deepcopy +from pathlib import Path +import sys +from threading import Lock +from typing import Any + + +def file_change_fingerprint(path: Path) -> tuple[int, ...] | None: + """A local mutation token, or no reuse when the platform lacks one.""" + # Windows stat ctime can be creation time, not a mutation stamp. Keep + # the original fresh reader there rather than trusting an mtime match. + if sys.platform == "win32": + return None + stat = path.stat() + return stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size + + +class FileChatDiscovery: + """Keep only caller-selected metadata while the complete file is unchanged. + + Native Chat uses this for terminal-Turn discovery and source-control fields. + Selected Turns/requests and every admission still use their original fresh + readers. No body, grant, queue decision or provider effect is retained here. + """ + + def __init__(self) -> None: + self._lock = Lock() + self._directories: dict[Path, dict[Path, tuple[tuple[int, ...], dict[str, Any]]]] = {} + + def rows(self, directory: Path, paths: Iterable[Path], *, + read: Callable[[Path], dict[str, Any]], + project: Callable[[Path, dict[str, Any]], dict[str, Any] | None], + ) -> Iterator[tuple[Path, dict[str, Any]]]: + seen = set() + with self._lock: + entries = self._directories.setdefault(directory, {}) + for path in paths: + seen.add(path) + try: + before = file_change_fingerprint(path) + except OSError: + before = None + with self._lock: + saved = entries.get(path) + if before is None or saved is None or saved[0] != before: + entries.pop(path, None) + saved = None + if saved is not None: + yield path, deepcopy(saved[1]) + continue + # Preserve the owning reader's missing/unreadable/strict behavior. + value = read(path) + view = project(path, value) + if before is not None and view is not None: + try: + unchanged = file_change_fingerprint(path) == before + except OSError: + unchanged = False + if unchanged: + with self._lock: + entries[path] = before, deepcopy(view) + yield path, value if view is None else view + with self._lock: + for path in entries.keys() - seen: + del entries[path] diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 571576b60c..21858e7ed4 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -2,6 +2,7 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone +from functools import cached_property import os import json from pathlib import Path @@ -9,6 +10,7 @@ from typing import Any from .chat import require_matching_message_replay, require_matching_replay from .file_lock import exclusive_file_lock +from .capabilities.native_chat.file_discovery import FileChatDiscovery SESSION_QUEUE_MAX_PENDING = 20 @@ -69,6 +71,12 @@ def chat_runtime_activation_state(runtime_root: Path) -> dict[str, int]: class ChatIngressStore: + @cached_property + def _terminal_turn_discovery(self) -> FileChatDiscovery: + # File queue discovery owns this per-store hint. Concurrent first use + # may build two empty observers; neither holds execution authority. + return FileChatDiscovery() + def runtime_activation_lock(self): return chat_runtime_activation_lock(self.root.parent) @@ -81,9 +89,8 @@ def queued_turns(self, session_id: str, *, strict: bool = False, include_reserve active_turn_id = str(session.get("active_turn_id") or "") turns_dir = self._session_dir(session_id) / "turns" rows: list[dict[str, Any]] = [] - for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []: - if path.name.endswith(".events.json"): - continue + + def read(path): if strict: try: payload = json.loads(path.read_text(encoding="utf-8")) @@ -100,6 +107,18 @@ def queued_turns(self, session_id: str, *, strict: bool = False, include_reserve raise OSError("Queued Turn state is invalid") else: payload = _read_json(path) + return payload + + def terminal(path, payload): + if (payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION + and payload.get("session_id") == session_id and payload.get("turn_id") == path.stem + and isinstance(payload.get("status"), str) + and payload.get("status") in TERMINAL_TURN_STATES): + return {key: payload[key] for key in ("schema_version", "session_id", "turn_id", "status")} + return None + + paths = (path for path in turns_dir.glob("*.json") if not path.name.endswith(".events.json")) + for path, payload in self._terminal_turn_discovery.rows(turns_dir, paths, read=read, project=terminal): if ( payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION and payload.get("status") == "queued" diff --git a/loopx/extensions/lark/private_delivery_discovery.py b/loopx/extensions/lark/private_delivery_discovery.py index 1c61a895c9..b58baf6766 100644 --- a/loopx/extensions/lark/private_delivery_discovery.py +++ b/loopx/extensions/lark/private_delivery_discovery.py @@ -6,6 +6,7 @@ from typing import Any from ...chat_store import _read_json +from ...capabilities.native_chat.file_discovery import file_change_fingerprint from .private_progress import legacy_recovery_repairs @@ -22,15 +23,11 @@ def rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]: """ seen = set() - def fingerprint(path: Path) -> tuple[int, ...]: - stat = path.stat() - return (stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size) - for path in sorted(self.root.glob("*.json")): seen.add(path) - before = fingerprint(path) + before = file_change_fingerprint(path) saved = self._completed.get(path) - if saved is not None and saved[0] == before: + if before is not None and saved is not None and saved[0] == before: yield path, saved[1], saved[2] continue self._completed.pop(path, None) @@ -42,7 +39,7 @@ def fingerprint(path: Path) -> tuple[int, ...]: ("status", "request_ref", "binding_id", "session_id", "turn_id")} view.update(agent_target=bool(row.get("agent_target")), event={"chat_type": row.get("event", {}).get("chat_type")}) - if fingerprint(path) == before: + if before is not None and file_change_fingerprint(path) == before: self._completed[path] = (before, view, repairs) row = view yield path, row, repairs diff --git a/tests/test_chat_file_discovery.py b/tests/test_chat_file_discovery.py new file mode 100644 index 0000000000..baca091762 --- /dev/null +++ b/tests/test_chat_file_discovery.py @@ -0,0 +1,246 @@ +"""File-backed parity at the real queue/control readers, including invalidation.""" +import json +import os +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from loopx.chat_store import ChatSessionStore, _atomic_write_json +from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations +from loopx.capabilities.native_chat.file_discovery import FileChatDiscovery + + +@pytest.fixture +def original_store(tmp_path): + store = ChatSessionStore(tmp_path, startup_maintenance=False) + session = store.create_session(goal_id="goal-one", agent_id="codex", + executor_endpoint_id="codex", adapter_kind="codex_app_server", + upstream_thread_id="thread-one", upstream_mode="chat") + return store, session["session_id"] + + +def _turn(store, sid, client, status): + turn, _ = store.create_turn(sid, client_turn_id=client, message="Original input") + store.update_turn(sid, turn["turn_id"], status=status) + store.update_session(sid, active_turn_id=None) + return turn["turn_id"] + + +def test_terminal_history_stays_out_of_warm_queue_body_reads(original_store, monkeypatch): + store, sid = original_store + terminal = [_turn(store, sid, f"history-{i}", "completed") for i in range(12)] + pending = _turn(store, sid, "pending", "queued") + assert [t["turn_id"] for t in store.queued_turns(sid)] == [pending] + reads = [] + original = Path.read_text + + def read(path, *args, **kwargs): + if path.parent.name == "turns": + reads.append(path.stem) + return original(path, *args, **kwargs) + + monkeypatch.setattr(Path, "read_text", read) + assert [t["turn_id"] for t in store.queued_turns(sid, strict=True)] == [pending] + assert not set(terminal) & set(reads) + assert pending in reads + # Exact reads retain complete fresh records, independent of the hint. + assert store.load_turn(sid, terminal[0])["message"] == "Original input" + + +@pytest.mark.parametrize("replace", [False, True]) +def test_rewritten_terminal_turn_is_visible_even_with_preserved_mtime(original_store, replace): + store, sid = original_store + tid = _turn(store, sid, "original", "failed") + assert store.queued_turns(sid, strict=True) == [] + path = store._turn_path(sid, tid) + stamp = path.stat() + value = store.load_turn(sid, tid) + value["status"] = "queued" + if replace: + _atomic_write_json(path, value) + else: + path.write_text(json.dumps(value)) + os.utime(path, ns=(stamp.st_atime_ns, stamp.st_mtime_ns)) + assert [t["turn_id"] for t in store.queued_turns(sid, strict=True)] == [tid] + + +def test_new_queued_input_and_active_exclusion_stay_fresh(original_store): + store, sid = original_store + _turn(store, sid, "old", "completed") + assert store.queued_turns(sid) == [] + tid = _turn(store, sid, "new", "queued") + assert [t["turn_id"] for t in store.queued_turns(sid)] == [tid] + store.update_session(sid, active_turn_id=tid) + assert store.queued_turns(sid) == [] + + +def test_concurrent_first_queue_observations_do_not_change_selection(original_store): + store, sid = original_store + _turn(store, sid, "history", "completed") + tid = _turn(store, sid, "pending", "queued") + with ThreadPoolExecutor(max_workers=2) as pool: + rows = list(pool.map(lambda _: store.queued_turns(sid, strict=True), range(2))) + assert [[turn["turn_id"] for turn in result] for result in rows] == [[tid], [tid]] + + +@pytest.mark.parametrize("fault", ["malformed", "foreign-session", "invalid-status"]) +def test_warmed_terminal_does_not_hide_strict_read_failures(original_store, fault): + store, sid = original_store + tid = _turn(store, sid, "old", "completed") + assert store.queued_turns(sid, strict=True) == [] + path = store._turn_path(sid, tid) + value = store.load_turn(sid, tid) + if fault == "malformed": + path.write_text("{") + else: + value["session_id" if fault == "foreign-session" else "status"] = "foreign" + _atomic_write_json(path, value) + with pytest.raises(OSError, match="Queued Turn state"): + store.queued_turns(sid, strict=True) + + +def _core(tmp_path): + controller = SimpleNamespace(store=SimpleNamespace(root=tmp_path), + project_contexts=SimpleNamespace(conversation_bindings=SimpleNamespace())) + core = ChatExternalConversations(controller) + core.root.mkdir() + return core + + +def test_source_controls_reuse_only_fields_and_keep_stop_and_new_changes(tmp_path, monkeypatch): + import loopx.capabilities.native_chat.external_conversations as module + + core = _core(tmp_path) + source = {"source_ref": "original", "sender_ref": "owner"} + first = {"binding_id": "binding", "source": source, "command": None, + "created_at": "2026-01-01T00:00:00+00:00", "message": "BODY" * 1000} + path = core.root / ("a" * 24 + ".json") + _atomic_write_json(path, first) + expected = [{key: value for key, value in first.items() if key != "message"}] + assert core.source_controls() == expected + original = module._read_json + reads = [] + + def read(path): + reads.append(path) + return original(path) + + monkeypatch.setattr(module, "_read_json", read) + view = core.source_controls() + assert view == expected and reads == [] + view[0]["source"]["source_ref"] = "mutated" + assert core.source_controls() == expected + row = {"binding_id": "binding", "source": source, "created_at": first["created_at"]} + assert not core._source_stop_observed(row, {"status": "failed"}, []) + _atomic_write_json(path, {**first, "command": "stop"}) + assert core._source_stop_observed(row, {"status": "failed"}, []) + assert reads == [path] + # A new independent source is still observed; it does not stop this input. + other = core.root / ("b" * 24 + ".json") + _atomic_write_json(other, {**first, "source": {"source_ref": "other"}, "command": "new"}) + path.unlink() + assert not core._source_stop_observed(row, {"status": "failed"}, []) + assert core.source_controls() == [{key: value for key, value in original(other).items() if key != "message"}] + assert core.pending()[0]["message"] == first["message"] + + +def test_changed_during_read_is_not_saved_as_an_unchanged_hint(tmp_path): + path = tmp_path / "journal.json" + path.write_text('{"value":1}') + discovery = FileChatDiscovery() + reads = [] + + def read(path): + row = json.loads(path.read_text()) + reads.append(row) + if len(reads) == 1: + _atomic_write_json(path, {"value": 2}) + return row + + def observe(): + return list(discovery.rows(tmp_path, [path], read=read, project=lambda _p, row: row)) + + assert observe()[0][1] == {"value": 1} + assert observe()[0][1] == {"value": 2} + assert observe()[0][1] == {"value": 2} + assert len(reads) == 2 + + +def test_transient_empty_read_cannot_hide_an_unchanged_stop(tmp_path, monkeypatch): + import loopx.capabilities.native_chat.external_conversations as module + + core = _core(tmp_path) + source = {"source_ref": "original", "sender_ref": "owner"} + control = {"binding_id": "binding", "source": source, "command": "stop", + "created_at": "2026-01-01T00:00:00+00:00"} + _atomic_write_json(core.root / ("a" * 24 + ".json"), control) + original, reads = module._read_json, [] + + def read(path): + reads.append(path) + return {} if len(reads) == 1 else original(path) + + monkeypatch.setattr(module, "_read_json", read) + assert core.source_controls() == [{}] + # No file write or timestamp change between the failed and healthy reads. + assert core.source_controls() == [control] + assert len(reads) == 2 + + +def test_unavailable_stat_uses_original_reader_instead_of_saved_view(tmp_path, monkeypatch): + path = tmp_path / "journal.json" + path.write_text('{"value":1}') + discovery = FileChatDiscovery() + + def observe(): + return list(discovery.rows(tmp_path, [path], read=lambda p: json.loads(p.read_text()), + project=lambda _p, row: row)) + + assert observe()[0][1] == {"value": 1} + path.unlink() + with pytest.raises(FileNotFoundError): + observe() + + +def test_windows_creation_clock_does_not_hide_same_size_preserved_stamp_stop(tmp_path, monkeypatch): + import loopx.capabilities.native_chat.file_discovery as discovery_module + + core = _core(tmp_path) + source = {"source_ref": "original", "sender_ref": "owner"} + first = {"binding_id": "binding", "source": source, "command": None, + "created_at": "2026-01-01T00:00:00+00:00", "padding": "aa"} + path = core.root / ("a" * 24 + ".json") + before = json.dumps(first) + path.write_text(before) + stamp, original_stat = path.stat(), Path.stat + # Simulate Windows' creation clock and an in-place, same-size rewrite that + # preserves mtime. Other filesystem operations retain their actual stat. + monkeypatch.setattr(discovery_module, "sys", SimpleNamespace(platform="win32")) + monkeypatch.setattr(Path, "stat", lambda p, *a, **k: stamp if p == path else original_stat(p, *a, **k)) + assert core.source_controls()[0]["command"] is None + after = json.dumps({**first, "command": "stop", "padding": ""}) + assert len(before.encode()) == len(after.encode()) + path.write_text(after) + os.utime(path, ns=(stamp.st_atime_ns, stamp.st_mtime_ns)) + row = {"binding_id": "binding", "source": source, "created_at": first["created_at"]} + assert core._source_stop_observed(row, {"status": "failed"}, []) + + +def test_private_reply_discovery_uses_the_same_windows_fresh_read_fallback(tmp_path, monkeypatch): + import loopx.capabilities.native_chat.file_discovery as discovery_module + from loopx.extensions.lark.private_delivery_discovery import PrivateDeliveryDiscovery + + path = tmp_path / ("a" * 24 + ".json") + row = {"status": "delivered", "request_ref": "a" * 24, "binding_id": "binding", + "session_id": "session", "turn_id": "turn", "event": {"chat_type": "p2p"}} + path.write_text(json.dumps(row)) + stamp, original_stat = path.stat(), Path.stat + monkeypatch.setattr(discovery_module, "sys", SimpleNamespace(platform="win32")) + monkeypatch.setattr(Path, "stat", lambda p, *a, **k: stamp if p == path else original_stat(p, *a, **k)) + discovery = PrivateDeliveryDiscovery(tmp_path) + assert list(discovery.rows())[0][1]["binding_id"] == "binding" + path.write_text(json.dumps({**row, "binding_id": "changed"})) + os.utime(path, ns=(stamp.st_atime_ns, stamp.st_mtime_ns)) + assert list(discovery.rows())[0][1]["binding_id"] == "changed" diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 322d4d3cb2..3a75a32ef8 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -182,15 +182,15 @@ def checked_update(session_id, **changes): return update_session(session_id, **changes) monkeypatch.setattr(store, "update_session", checked_update) if fault in {None, "later-source-after-discovery"}: - pending = transport.core.pending + source_controls = transport.core.source_controls scans = [] - def counted_pending(): + def counted_controls(): scans.append(True) - return pending() - monkeypatch.setattr(transport.core, "pending", counted_pending) + return source_controls() + monkeypatch.setattr(transport.core, "source_controls", counted_controls) discovery = transport.core.context_recovery_observer() assert discovery(ref) and discovery(ref) - assert len(scans) == 1 # One full source inventory per discovery pass. + assert len(scans) == 1 # One complete source-control view per discovery pass. if fault in {"stop-after-discovery", "source-changed-after-discovery"}: assert path in transport.pending_delivery_paths() if fault in {"stop", "stop-after-discovery"}: @@ -1313,7 +1313,7 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying( # 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")) + monkeypatch.setattr(transport.core, "source_controls", 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,