From bea837f53252d0f6b284af663a8ae8db0fac572a Mon Sep 17 00:00:00 2001 From: Lan_zhijiang Date: Sat, 31 Oct 2026 14:47:26 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(extensions):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E5=88=86=E5=8F=91=E6=94=B6=E5=B0=BE=E4=B8=8E=E9=87=87=E9=9B=86?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 保留取消期间的安装资源并记录失败加载版本 - 保护 Mail 并发进度并完成 Twitter 全量分页 - 集中 GitHub 游标规则，补充行为回归与发行说明 --- .changes/maintainability.fixed.md | 1 + app/business/extension/distribution.py | 25 +- app/business/extension/main.py | 160 ++++++--- app/business/extension/state.py | 12 +- docs/30-unit-tdd/mail-extension.md | 4 + docs/30-unit-tdd/twitter-extension.md | 2 + .../first-party-extension-distribution.md | 13 + .../github/.changes/maintainability.fixed.md | 1 + extensions/github/README.md | 4 + extensions/github/adapter.py | 71 ++-- .../mail/.changes/maintainability.fixed.md | 1 + extensions/mail/pyproject.toml | 1 + extensions/mail/source.py | 37 +- .../twitter/.changes/maintainability.fixed.md | 1 + extensions/twitter/bookmark.py | 80 +++-- .../twitter/integration/test_archive_graph.py | 89 +++++ tests/extensions/github/test_snapshot.py | 127 +++++++ tests/extensions/mail/test_source_state.py | 145 ++++++++ tests/test_extension_startup_provision.py | 323 +++++++++++++++++- 19 files changed, 926 insertions(+), 171 deletions(-) create mode 100644 .changes/maintainability.fixed.md create mode 100644 extensions/github/.changes/maintainability.fixed.md create mode 100644 extensions/mail/.changes/maintainability.fixed.md create mode 100644 extensions/twitter/.changes/maintainability.fixed.md create mode 100644 tests/extensions/github/test_snapshot.py create mode 100644 tests/extensions/mail/test_source_state.py diff --git a/.changes/maintainability.fixed.md b/.changes/maintainability.fixed.md new file mode 100644 index 00000000..1d1d161b --- /dev/null +++ b/.changes/maintainability.fixed.md @@ -0,0 +1 @@ +在修改 Python 环境前拒绝替换仍启用的扩展，并在取消操作时保留安装所需资源直到收尾完成。模块加载或启动失败后仍保留已加载版本，避免在同一进程中误用另一版本。 diff --git a/app/business/extension/distribution.py b/app/business/extension/distribution.py index 99ac8e5a..42d07f84 100644 --- a/app/business/extension/distribution.py +++ b/app/business/extension/distribution.py @@ -5,7 +5,7 @@ from inkcre_extension_runtime_core_py.distribution import ( AcquiredDistribution, - PipDistributionConsumer as RuntimePipDistributionConsumer, + PipDistributionConsumer, ) from inkcre_extension_runtime_core_py.modules import DistributionModules @@ -30,29 +30,6 @@ def apply( ) -> AcquiredDistribution: ... -class PipDistributionConsumer: - def __init__(self, origin: str) -> None: - self._runtime = RuntimePipDistributionConsumer(origin) - - def download( - self, - release: ExtensionReleaseDescriptor, - association: PythonReleaseDescriptor, - host_version: str, - wheel_dir: Path, - ) -> Path: - return self._runtime.download(release, association, host_version, wheel_dir) - - def apply( - self, - extension_wheel: Path, - release: ExtensionReleaseDescriptor, - association: PythonReleaseDescriptor, - host_version: str, - ) -> AcquiredDistribution: - return self._runtime.apply(extension_wheel, release, association, host_version) - - __all__ = [ "AcquiredDistribution", "DistributionConsumer", diff --git a/app/business/extension/main.py b/app/business/extension/main.py index b23bab72..3ac3fd46 100644 --- a/app/business/extension/main.py +++ b/app/business/extension/main.py @@ -78,6 +78,33 @@ EXTENSION_MANAGEMENT_CAPABILITY = "core.extension.management.v2" +async def _finish_before_cancelling[T]( + work: typing.Coroutine[typing.Any, typing.Any, T], +) -> T: + """Keep resources alive until work that cannot be abandoned has finished.""" + task = asyncio.create_task(work) + cancellation: asyncio.CancelledError | None = None + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError as error: + # Another cancellation must not release the caller's lock or temporary + # directory while its worker still owns them. + cancellation = cancellation or error + except Exception: + break + try: + result = task.result() + except Exception as error: + if cancellation is None: + raise + LOGGER.exception("Extension operation failed while its caller was cancelling") + raise cancellation from error + if cancellation is not None: + raise cancellation + return result + + class ExtensionDelegationError(RuntimeError): """The selected Peer did not return the Extension management contract.""" @@ -364,6 +391,7 @@ async def _select_locked( raise ExtensionCompatibilityError( "This Core Peer supports only Python Extension distributions" ) + await self._require_selectable(name, version, peer_id=peer_id) release_client, consumer = await self._operation_consumers() release = await asyncio.to_thread( self._resolve, @@ -387,48 +415,28 @@ async def _select_locked( f"{name}@{version} must be applied while the Peer is paused; " "the process restart lifecycle is unavailable" ) - try: - self._operations[name] = ("downloading", None) - with tempfile.TemporaryDirectory(prefix="inkcre-extension-wheel-") as directory: - extension_wheel = await asyncio.to_thread( - consumer.download, - release, - association, - CORE_VERSION, - Path(directory), - ) - if self._environment_application_pending: - raise ExtensionStateConflictError( - "Another Extension operation is already applying a Python environment" + self._operations[name] = ("downloading", None) + with tempfile.TemporaryDirectory(prefix="inkcre-extension-wheel-") as directory: + try: + extension_wheel = await _finish_before_cancelling( + asyncio.to_thread( + consumer.download, release, association, CORE_VERSION, Path(directory) ) - self._environment_application_pending = True - self._operations[name] = ("applying", None) - await self.pause_peer_execution() - acquired = await asyncio.to_thread( - consumer.apply, extension_wheel, release, association, CORE_VERSION ) - self._validate_installed_association(acquired, association) - except Exception as error: - # Once application starts, package managers may partially change shared - # dependencies. Keep the Peer paused; never claim environment rollback. - self._operations[name] = ("failed", str(error)) - raise ExtensionEnvironmentApplicationRequiredError( - f"Preparing or applying {name}@{version} failed; " - f"the Peer remains paused if application had started: {error}" - ) from error - try: - await self.store.ensure(name, release.nickname) - selected = await self.store.select_participation(peer_id, name, "python", version) - except Exception as error: - if install_required: - self._operations[name] = ("failed", str(error)) - request_process_restart() - raise ExtensionEnvironmentApplicationRequiredError( - f"Applied {name}@{version}, but could not save the Peer selection; " - f"the process is restarting to reload the changed environment: {error}" - ) from error - raise - if install_required or restart_required: + except asyncio.CancelledError: + self._operations[name] = (None, "Extension download was cancelled") + raise + except Exception as error: + self._operations[name] = ("failed", str(error)) + raise ExtensionEnvironmentApplicationRequiredError( + f"Preparing {name}@{version} failed: {error}" + ) from error + return await _finish_before_cancelling( + self._apply_selection(release, association, consumer, extension_wheel, peer_id) + ) + await self.store.ensure(name, release.nickname) + selected = await self.store.select_participation(peer_id, name, "python", version) + if restart_required: self._environment_application_pending = True self._operations[name] = ("restarting", None) request_process_restart() @@ -436,6 +444,64 @@ async def _select_locked( self._operations[name] = (None, None) return selected + async def _require_selectable( + self, name: str, version: str, *, peer_id: typing.Any + ) -> None: + selection = await self.store.get_participation(peer_id, name) + if ( + selection is not None + and selection.enabled + and ( + selection.distribution, + selection.version, + ) + != ("python", version) + ): + raise ExtensionStateConflictError( + f"Cannot change {name} selection while it is enabled on this Peer" + ) + + async def _apply_selection( + self, + release: ExtensionReleaseDescriptor, + association: PythonReleaseDescriptor, + consumer: DistributionConsumer, + wheel: Path, + peer_id: typing.Any, + ) -> ExtensionParticipation: + """Finish environment mutation and its durable selection before releasing the lock.""" + name, version = release.name, release.version + assert self.pause_peer_execution is not None + self._environment_application_pending = True + self._operations[name] = ("applying", None) + try: + await self.pause_peer_execution() + # Other admitted Peers can change intent during the drain. This check + # avoids known conflicts; the store still arbitrates the final write. + await self._require_selectable(name, version, peer_id=peer_id) + acquired = await asyncio.to_thread( + consumer.apply, wheel, release, association, CORE_VERSION + ) + self._validate_installed_association(acquired, association) + except Exception as error: + self._operations[name] = ("failed", str(error)) + raise ExtensionEnvironmentApplicationRequiredError( + f"Applying {name}@{version} failed; the Peer remains paused: {error}" + ) from error + try: + await self.store.ensure(name, release.nickname) + selected = await self.store.select_participation(peer_id, name, "python", version) + except Exception as error: + self._operations[name] = ("failed", str(error)) + request_process_restart() + raise ExtensionEnvironmentApplicationRequiredError( + f"Applied {name}@{version}, but could not save the Peer selection; " + f"the process is restarting to reload the changed environment: {error}" + ) from error + self._operations[name] = ("restarting", None) + request_process_restart() + return selected + @staticmethod def _environment_package_versions() -> dict[str, str]: return { @@ -653,16 +719,18 @@ async def provision_enabled_distributions(self) -> None: await self._require_current_enabled_selection(peer_id, selection) selection_wheel_dir = wheel_dir / str(index) selection_wheel_dir.mkdir() - wheel = await asyncio.to_thread( - consumer.download, release, association, CORE_VERSION, selection_wheel_dir + wheel = await _finish_before_cancelling( + asyncio.to_thread( + consumer.download, release, association, CORE_VERSION, selection_wheel_dir + ) ) wheels.append((selection, release, association, wheel)) for selection, release, association, wheel in wheels: await self._require_current_enabled_selection(peer_id, selection) environment_before = self._environment_package_versions() - acquired = await asyncio.to_thread( - consumer.apply, wheel, release, association, CORE_VERSION + acquired = await _finish_before_cancelling( + asyncio.to_thread(consumer.apply, wheel, release, association, CORE_VERSION) ) self._assert_environment_packages_preserved(environment_before, association.project) self._validate_installed_association(acquired, association) @@ -790,6 +858,9 @@ async def _start_acquired( raise extension_class: type[ExtensionBase] | None = None try: + # Import is irreversible even when the module or its startup later fails. + # A different selected version therefore still needs a fresh process. + self._loaded_versions[state.name] = selection.version extension_class = typing.cast(type[ExtensionBase], modules.load(ExtensionBase)) if extension_class.__extid__ != association.entry_point.name: raise ExtensionCompatibilityError( @@ -828,7 +899,6 @@ async def _start_acquired( claim=claim, ) self.running[state.name] = running - self._loaded_versions[state.name] = selection.version return running async def _get_current_enabled_selection( diff --git a/app/business/extension/state.py b/app/business/extension/state.py index 8f37aa72..b99b8da3 100644 --- a/app/business/extension/state.py +++ b/app/business/extension/state.py @@ -176,25 +176,19 @@ async def uninstall(self, name: str) -> None: ) await repository.delete(row) - async def _update_json( + async def update_config( self, name: str, - field: typing.Literal["config"], - value: dict[str, typing.Any], + config: dict[str, typing.Any], ) -> InstalledExtension: async with extension_uow() as repository: row = await repository.get(name, lock=True) if row is None: raise ExtensionNotInstalledError(f"{name} is not installed") - setattr(row, field, value) + row.config = config await repository.save(row) return self._extension(row) - async def update_config( - self, name: str, config: dict[str, typing.Any] - ) -> InstalledExtension: - return await self._update_json(name, "config", config) - async def read_config(self, name: str) -> dict[str, typing.Any]: state = await self.get(name) if state is None: diff --git a/docs/30-unit-tdd/mail-extension.md b/docs/30-unit-tdd/mail-extension.md index 3a683a01..eaaca8d1 100644 --- a/docs/30-unit-tdd/mail-extension.md +++ b/docs/30-unit-tdd/mail-extension.md @@ -76,6 +76,10 @@ synchronize_deletions: boolean Source state 只保存 non-secret access binding 与 per-Mailbox ordinary checkpoints。Binding 改变不能静默接续旧 state/graph effects。Checkpoint 由 adapter 解释，Source 决定 accepted-effect boundary 与持久推进时机。 +Adapter 连接完成后，Source 在行锁内重新读取 state 并接受 binding，仅在首次绑定时写入 binding。 +连接可能耗时较长，因此不能把连接前的完整 state 写回，否则会覆盖并发采集已经提交的 checkpoint。 +各 Mailbox 仍通过 compare-and-swap 推进 checkpoint；Backfill 的连接与绑定不会改写 ordinary 进度。 + Ordinary collection 默认只接纳 Source creation time 之后的信息；第一次建立 checkpoint 后使用： 1. QRESYNC（可靠 VANISHED + changed flags）； diff --git a/docs/30-unit-tdd/twitter-extension.md b/docs/30-unit-tdd/twitter-extension.md index 358ec2fe..a4dcdcbe 100644 --- a/docs/30-unit-tdd/twitter-extension.md +++ b/docs/30-unit-tdd/twitter-extension.md @@ -8,6 +8,8 @@ Python Twitter distribution 把 x-monkey 提交和 Bookmark 采集写入同一 一次 submit 使用一个 `graph_uow()`。Bookmark 先完成 provider 请求，再进入 `source_uow()`，锁定 Source，使用其 `graph` 合并新主帖及实际捕获的上下文，并在同一事务内推进 cursor。主帖顺序与展开上下文分开，因此引用帖不会成为下一次书签采集的 watermark；旧的数值 cursor 在比较边界转换为十进制字符串，不重置历史状态。 +普通 Bookmark 采集读取一个近期页面，并在遇到已保存的主帖 ID 时截断。显式 `full=true` 则在同一个 Job 内遍历全部页面；每页先独立提交 graph，再更新 Job 的 `page` 诊断字段。Job 只执行一次，保存 continuation 本身不会调度后续页面。全量采集不改变普通采集的 watermark；后页请求失败或游标成环时，任务明确失败，前面已提交的页面保留，后续显式全量采集可以幂等重扫。 + 两条写入路径都先取得 `extensions.twitter.reconcile` 全局事务 advisory 锁，再批量重读原生身份。根内容和受管关系一起提交；异常让整批回滚，调用方可安全重放。该锁仅约束正常 Twitter 写入入口，不能替代所有 info-base 原始写入的约束。provider 和媒体网络访问都不能放进此事务。 原生身份、Blocks 和引用边使用批量查询与创建。不要在每个媒体项内单独查询、创建和 flush：Core 与数据库跨网络部署时，这会把一次正常批次放大为数百次串行往返，并在持锁期间耗尽客户端的同步请求期限。 diff --git a/docs/40-deployment/first-party-extension-distribution.md b/docs/40-deployment/first-party-extension-distribution.md index 6e062243..316d20aa 100644 --- a/docs/40-deployment/first-party-extension-distribution.md +++ b/docs/40-deployment/first-party-extension-distribution.md @@ -26,6 +26,16 @@ used by other Extensions. After apply, Core replaces the process with a fresh on cannot continue using modules loaded before the environment change. An apply failure is reported and leaves the Peer paused; Core does not claim package rollback. Exact installed-record reuse avoids this lifecycle only when the installed project, entry point, and Host SDK match the Registry record. + +Host 在查询 Registry 前拒绝替换当前 Peer 仍启用的不同版本，并在排空 Peer 后、调用 pip 前再次核对。 +其他已准入 Peer 仍可直接修改参与记录，因此这两次检查不代替保存 selection 时的事务校验，也不跨网络或 +pip 持有数据库事务。安装后保存失败仍会报告错误并请求重启，以重新核对实际环境和持久意图。 + +取消下载会等待已经启动的下载线程结束，再清理临时目录，不会继续安装。开始暂停 Peer 后，Host 则完成 +暂停、安装、selection 保存和重启请求，或者进入明确失败状态，随后才向调用者传播取消。整个过程保留 +运行锁和临时目录，避免后台 pip 在资源被释放后继续修改解释器。冷启动恢复被取消时只等待当前下载或安装 +结束，不继续安装后续 wheel。Runtime SDK 的 pip 调用目前没有总时限，因此此规则保证收尾顺序，不承诺快速取消。 + The standard `inkcre.core.extensions` entry point is discovered globally and every loaded Extension module is verified against the installed wheel file record. There is no per-Extension target directory, custom ZIP loader or module search-path rewrite. Registry failure or a blocked Release @@ -39,6 +49,9 @@ new schema. A process that already imported another version must restart; Core d new class. `running` and `loaded_version` report observed runtime facts and are never inferred from the durable enabled intent. +模块加载或 Extension startup 失败也不能撤销已经执行的 Python import。Host 会保留该版本的加载记录， +后续选择另一版本仍需重启；失败的 runtime 不会因此被报告为 running。 + V2 management commands select or remove a Peer’s exact pin, enable or disable that Peer’s intent, and report actual runtime status. A disabled pin may be removed; an enabled participation must first be disabled. A legacy enabled member migrates to its exact prior version with an unresolved diff --git a/extensions/github/.changes/maintainability.fixed.md b/extensions/github/.changes/maintainability.fixed.md new file mode 100644 index 00000000..8346dce6 --- /dev/null +++ b/extensions/github/.changes/maintainability.fixed.md @@ -0,0 +1 @@ +统一遍历 Stars、Lists 和 List items 的 GraphQL 分页；分页缺少 continuation 或游标成环时明确失败，避免采集任务持续请求且无法完成。 diff --git a/extensions/github/README.md b/extensions/github/README.md index d903cbb5..f91bd8e6 100644 --- a/extensions/github/README.md +++ b/extensions/github/README.md @@ -34,3 +34,7 @@ a new Source。 Dispatch the ordinary `core.source.collect.v1` Job through the generic Source/Job surface。Each run fetches a complete current Stars and Lists snapshot before reconciling graph facts。There is no extension-specific collection endpoint、incremental cursor、`full` option or historical backfill mode。 + +Stars、Lists 和各 List 的 items 都必须完成分页后才进入 graph reconciliation。 +如果 GitHub 声明还有下一页却没有提供有效游标，或者游标形成循环，当前任务会失败； +系统不会把这次不完整的观察当成完整快照，也不会据此移除已有关系。 diff --git a/extensions/github/adapter.py b/extensions/github/adapter.py index 8df490d0..3fe80cec 100644 --- a/extensions/github/adapter.py +++ b/extensions/github/adapter.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import AsyncIterator import typing import httpx @@ -114,7 +115,7 @@ def _page_info(connection: dict[str, typing.Any]) -> tuple[bool, str | None]: cursor = page.get("endCursor") if not isinstance(has_next, bool): raise GitHubGraphQLError("GitHub GraphQL returned invalid pageInfo") - if has_next and not isinstance(cursor, str): + if has_next and (not isinstance(cursor, str) or not cursor): raise GitHubGraphQLError("GitHub GraphQL pagination omitted endCursor") return has_next, cursor if isinstance(cursor, str) else None @@ -217,16 +218,14 @@ async def fetch_snapshot(self) -> GitHubSnapshot: repositories: dict[str, GitHubRepositoryFact] = {} starred_ids: list[str] = [] account: GitHubAccount | None = None - cursor: str | None = None - - while True: - data = await self._execute(_ACCOUNT_AND_STARS_QUERY, {"cursor": cursor}) + async for data, connection in self._connection_pages( + _ACCOUNT_AND_STARS_QUERY, ("viewer", "starredRepositories") + ): viewer = _object(data.get("viewer"), "viewer") observed_account = _account(viewer, viewer=True) if account is not None and observed_account.node_id != account.node_id: raise GitHubGraphQLError("GitHub viewer changed during pagination") account = observed_account - connection = _object(viewer.get("starredRepositories"), "starredRepositories") nodes = connection.get("nodes") if not isinstance(nodes, list): raise GitHubGraphQLError("GitHub GraphQL returned invalid starredRepositories") @@ -234,33 +233,22 @@ async def fetch_snapshot(self) -> GitHubSnapshot: fact = _repository(_object(node, "starred Repository")) repositories[fact.repository.node_id] = fact starred_ids.append(fact.repository.node_id) - has_next, cursor = _page_info(connection) - if not has_next: - break lists: list[GitHubListFact] = [] - cursor = None - while True: - data = await self._execute(_LISTS_QUERY, {"cursor": cursor}) - viewer = _object(data.get("viewer"), "viewer") - connection = _object(viewer.get("lists"), "viewer.lists") + async for _, connection in self._connection_pages(_LISTS_QUERY, ("viewer", "lists")): nodes = connection.get("nodes") if not isinstance(nodes, list): raise GitHubGraphQLError("GitHub GraphQL returned invalid Lists") for node_value in nodes: node = _object(node_value, "List") - item_connection = _object(node.get("items"), "List.items") - item_ids = self._collect_repository_ids(item_connection) - has_more_items, item_cursor = _page_info(item_connection) - while has_more_items: - item_data = await self._execute( - _LIST_ITEMS_QUERY, - {"id": node.get("id"), "cursor": item_cursor}, - ) - list_node = _object(item_data.get("node"), "UserList node") - item_connection = _object(list_node.get("items"), "List.items") + item_ids: list[str] = [] + async for _, item_connection in self._connection_pages( + _LIST_ITEMS_QUERY, + ("node", "items"), + variables={"id": node.get("id")}, + initial_response={"node": node}, + ): item_ids.extend(self._collect_repository_ids(item_connection)) - has_more_items, item_cursor = _page_info(item_connection) lists.append( GitHubListFact( list=GitHubList.model_validate( @@ -275,9 +263,6 @@ async def fetch_snapshot(self) -> GitHubSnapshot: repository_node_ids=tuple(dict.fromkeys(item_ids)), ) ) - has_next, cursor = _page_info(connection) - if not has_next: - break if account is None: # pragma: no cover - at least one Stars page is required raise GitHubGraphQLError("GitHub GraphQL omitted the authenticated account") @@ -307,6 +292,36 @@ async def fetch_snapshot(self) -> GitHubSnapshot: lists=tuple(lists), ) + async def _connection_pages( + self, + query: str, + path: tuple[str, ...], + *, + variables: dict[str, typing.Any] | None = None, + initial_response: dict[str, typing.Any] | None = None, + ) -> AsyncIterator[tuple[dict[str, typing.Any], dict[str, typing.Any]]]: + """Traverse one connection, including a page embedded in an outer query.""" + cursor: str | None = None + seen_cursors: set[str] = set() + data = initial_response + while True: + if data is None: + data = await self._execute(query, {**(variables or {}), "cursor": cursor}) + connection = data + for key in path: + connection = _object(connection.get(key), ".".join(path)) + has_next, next_cursor = _page_info(connection) + if has_next: + # A repeated cursor cannot produce a complete observation. Abort before + # reconciliation rather than looping until a rate limit or Job timeout. + if next_cursor in seen_cursors: + raise GitHubGraphQLError("GitHub GraphQL pagination repeated endCursor") + seen_cursors.add(typing.cast(str, next_cursor)) + yield data, connection + if not has_next: + return + cursor, data = next_cursor, None + @staticmethod def _collect_repository_ids(connection: dict[str, typing.Any]) -> list[str]: nodes = connection.get("nodes") diff --git a/extensions/mail/.changes/maintainability.fixed.md b/extensions/mail/.changes/maintainability.fixed.md new file mode 100644 index 00000000..77e076f5 --- /dev/null +++ b/extensions/mail/.changes/maintainability.fixed.md @@ -0,0 +1 @@ +连接邮箱后接受访问绑定时保留并发采集提交的 checkpoint，避免普通采集或历史回填覆盖最新进度；独立安装 Mail 时显式安装 IMAPClient。 diff --git a/extensions/mail/pyproject.toml b/extensions/mail/pyproject.toml index e247f737..472a5eb4 100644 --- a/extensions/mail/pyproject.toml +++ b/extensions/mail/pyproject.toml @@ -10,6 +10,7 @@ authors = [{ name = "Lan_zhijiang", email = "lanzhijiang@foxmail.com" }] dependencies = [ "beautifulsoup4>=4.12.0,<5.0.0", "fastapi>=0.139.2,<0.140.0", + "imapclient>=3.1,<4", "pydantic>=2.11.7,<3.0.0", "sqlmodel>=0.0.24,<0.0.25", ] diff --git a/extensions/mail/source.py b/extensions/mail/source.py index 9aa90cc2..52a1343f 100644 --- a/extensions/mail/source.py +++ b/extensions/mail/source.py @@ -16,6 +16,7 @@ from .reconcile import MailGraphReconciler from .schema import ( IMAPCheckpoint, + MailAccessBinding, MailBackfillConfig, MailCollectConfig, MailboxExclusionPolicy, @@ -55,15 +56,13 @@ class Source( async def collect(self, job: JobModel, config: pydantic.BaseModel) -> None: source, setup = await self._load_effective_source() - state = MailSourceState.model_validate(source.state or {}) diagnostics: list[dict[str, typing.Any]] = [] counts = {"messages": 0, "flag_changes": 0, "removals": 0, "mailboxes": 0} job.state = {"diagnostics": diagnostics, "counts": counts} adapter = create_mail_adapter(setup.protocol, setup.parameters) async with adapter: - state = self._accept_binding(state, adapter.binding) - await self._persist_state(state) + state = await self._accept_binding(adapter.binding) exclusions = setup.excluded_mailboxes if exclusions is None: # pragma: no cover - materialization postcondition raise RuntimeError("Mail Source exclusions were not materialized") @@ -126,15 +125,13 @@ async def collect(self, job: JobModel, config: pydantic.BaseModel) -> None: async def backfill(self, job: JobModel, config: pydantic.BaseModel) -> None: interval = typing.cast(MailBackfillConfig, config) _source, setup = await self._load_effective_source() - state = MailSourceState.model_validate(await self.get_state()) diagnostics: list[dict[str, typing.Any]] = [] count = 0 job.state = {"diagnostics": diagnostics, "messages": count} adapter = create_mail_adapter(setup.protocol, setup.parameters) async with adapter: - state = self._accept_binding(state, adapter.binding) - await self._persist_state(state) + state = await self._accept_binding(adapter.binding) exclusions = setup.excluded_mailboxes if exclusions is None: # pragma: no cover - materialization postcondition raise RuntimeError("Mail Source exclusions were not materialized") @@ -196,16 +193,24 @@ async def _load_effective_source(self) -> tuple[SourceModel, MailSourceConfig]: await uow.sources.save(source) return source, config - @staticmethod - def _accept_binding(state: MailSourceState, binding) -> MailSourceState: - if state.binding is not None and state.binding != binding: - raise MailSourceBindingError( - "Mail Source access binding changed; create a new Source for a different context" - ) - return state.model_copy(update={"binding": binding}) - - async def _persist_state(self, state: MailSourceState) -> None: - await self.set_state(state.model_dump(mode="json")) + async def _accept_binding(self, binding: MailAccessBinding) -> MailSourceState: + """Accept the remote context without replacing another Job's progress.""" + async with source_uow() as uow: + source = await uow.sources.get(self._id, lock=True) + if source is None: + raise MailSourceBindingError("Mail Source no longer exists") + # Connecting can outlive another collect/backfill Job. Read checkpoints + # under the same row lock that protects their compare-and-swap updates. + state = MailSourceState.model_validate(source.state or {}) + if state.binding is not None and state.binding != binding: + raise MailSourceBindingError( + "Mail Source access binding changed; create a new Source for a different context" + ) + if state.binding is None: + state.binding = binding + source.state = state.model_dump(mode="json") + await uow.sources.save(source) + return state async def _ensure_mailbox(self, fact, state: MailSourceState) -> int: async with source_uow() as uow: diff --git a/extensions/twitter/.changes/maintainability.fixed.md b/extensions/twitter/.changes/maintainability.fixed.md new file mode 100644 index 00000000..d7c3818e --- /dev/null +++ b/extensions/twitter/.changes/maintainability.fixed.md @@ -0,0 +1 @@ +全量书签采集在同一个任务内遍历所有页面，并逐页提交已采集的信息；后续页面失败或游标成环时明确失败，避免第一页后直接结束并遗漏历史书签。 diff --git a/extensions/twitter/bookmark.py b/extensions/twitter/bookmark.py index bf6aa057..68be43f1 100644 --- a/extensions/twitter/bookmark.py +++ b/extensions/twitter/bookmark.py @@ -64,51 +64,49 @@ class Source( API_BASE_URL = "https://api.x.com/2" async def collect(self, job: JobModel, config: pydantic.BaseModel) -> None: - """Collect all new bookmarks and its notes. - - What is new bookmarks? - The tweets before the last collected tweet. The last collected tweet - is the latest created_at tweet block. - (Potential issue if bookmarks order changes) - - What is bookmark note? - User can add a note to a bookmark by replying the bookmark tweet. - - Docs https://docs.x.com/x-api/bookmarks/get-bookmarks - """ + """Collect one recent page or traverse all pages for an explicit full Job.""" collect_config = typing.cast(CollectConfig, config) full = collect_config.full result_limit = collect_config.result_limit page = job.state.get("page") if job.state else None - + seen_pages = {page} if page else set() api_client = await TwitterAPI.new() - bookmarks_res = await api_client.get_bookmarks(page=page, max_results=result_limit) - - async with source_uow() as uow: - source = await uow.sources.get(self._id, lock=True) - if source is None: - raise ValueError("Twitter Source no longer exists") - primary_ids = bookmarks_res.primary_post_ids - selected = primary_ids + while True: + bookmarks_res = await api_client.get_bookmarks(page=page, max_results=result_limit) + next_page = bookmarks_res.next_page + if full and next_page and next_page in seen_pages: + raise ValueError("Twitter bookmark pagination repeated a cursor") + + async with source_uow() as uow: + source = await uow.sources.get(self._id, lock=True) + if source is None: + raise ValueError("Twitter Source no longer exists") + primary_ids = bookmarks_res.primary_post_ids + selected = primary_ids + if not full: + latest_id = str((source.state or {}).get("latest_tweet_id", "")) + if latest_id in primary_ids: + selected = primary_ids[: primary_ids.index(latest_id)] + if selected: + await persist_twitter( + _selected_batch(bookmarks_res.batch, selected), + uow.graph, + text_policy="supplement", + ) + if not full and primary_ids: + source.state = { + **(source.state or {}), + "latest_tweet_id": primary_ids[0], + } + await uow.sources.save(source) + if not full: - latest_id = str((source.state or {}).get("latest_tweet_id", "")) - if latest_id in primary_ids: - selected = primary_ids[: primary_ids.index(latest_id)] - if selected: - await persist_twitter( - _selected_batch(bookmarks_res.batch, selected), - uow.graph, - text_policy="supplement", - ) - if not full and primary_ids: - source.state = { - **(source.state or {}), - "latest_tweet_id": primary_ids[0], - } - await uow.sources.save(source) - - # Update job state for next page if full and has next_page - if full and bookmarks_res.next_page and bookmarks_res.next_page != page: - job.state = job.state or {} - job.state["page"] = bookmarks_res.next_page + return + # Jobs execute once. Persist each page before advancing diagnostics, and + # finish traversal here rather than leaving continuation to a future Job. + job.state = {**(job.state or {}), "page": next_page} + if not next_page: + return + seen_pages.add(next_page) + page = next_page diff --git a/scripts/tests/twitter/integration/test_archive_graph.py b/scripts/tests/twitter/integration/test_archive_graph.py index 739ee49c..ad3227f4 100644 --- a/scripts/tests/twitter/integration/test_archive_graph.py +++ b/scripts/tests/twitter/integration/test_archive_graph.py @@ -683,3 +683,92 @@ async def remember(): enriched = async_runner.run(_write(ArchiveBatch.model_validate(native), tracked)) assert enriched.block_ids["posts"][post["id"]] == root.id assert json.loads(_record(post["id"]).content)["text"] == post["text"] + + +@pytest.mark.parametrize("ending", ["complete", "provider_failure", "cursor_cycle"]) +def test_full_bookmarks_commit_each_page_and_finish_or_fail_explicitly( + async_runner, native, tracked, ending +): + source_type = f"{Source.__module__}.{Source.__qualname__}" + async_runner.run(SourceManager.sync_source_types_async({source_type: Source})) + source = async_runner.run(SourceManager.create(source_type, nickname="paged bookmarks")) + posts = [ + {**_minimal(native["posts"][0]), "id": str(int(native["posts"][0]["id"]) + index)} + for index in range(3) + ] + calls = [] + job = JobModel(type="core.source.collect.v1", parameters={}) + + async def get_bookmarks(*, page, max_results): + assert max_results == 5 + calls.append(page) + index = len(calls) - 1 + assert index < 3, "pagination must terminate" + if index: + # Read from a different connection while the next network call is active. + # This proves the previous page committed without holding a transaction. + assert _record(posts[index - 1]["id"]).id is not None + assert job.state["page"] == page + if index == 1 and ending == "provider_failure": + raise RuntimeError("provider temporarily unavailable") + return TwitterAPIResult( + batch=TwitterBatch.model_validate({"posts": [posts[index]]}), + primary_post_ids=(posts[index]["id"],), + next_page=( + "a" + if index == 0 or (index == 2 and ending == "cursor_cycle") + else "b" + if index == 1 + else None + ), + ) + + provider = SimpleNamespace(get_bookmarks=get_bookmarks) + try: + with mock.patch.object( + TwitterAPI, "new", new_callable=mock.AsyncMock, return_value=provider + ): + command = Source(source.id).collect(job, CollectConfig(full=True, result_limit=5)) + if ending == "complete": + async_runner.run(command) + else: + error = RuntimeError if ending == "provider_failure" else ValueError + message = ( + "provider temporarily unavailable" + if ending == "provider_failure" + else "repeated a cursor" + ) + with pytest.raises(error, match=message): + async_runner.run(command) + accepted = {"complete": 3, "provider_failure": 1, "cursor_cycle": 2}[ending] + assert calls == [None, "a", "b"][: 2 if ending == "provider_failure" else 3] + assert ( + job.state["page"] + == {"complete": None, "provider_failure": "a", "cursor_cycle": "b"}[ending] + ) + + async def observed(): + async with graph_uow() as uow: + return await uow.blocks.find_json_field_many( + (POST_RESOLVER,), "id", tuple(post["id"] for post in posts) + ) + + roots = async_runner.run(observed()) + assert {json.loads(root.content)["id"] for root in roots} == { + post["id"] for post in posts[:accepted] + } + finally: + + async def remember(): + async with graph_uow() as uow: + rows = await uow.blocks.find_json_field_many( + (POST_RESOLVER, USER_RESOLVER), + "id", + (*[post["id"] for post in posts], posts[0]["authorId"]), + ) + tracked.update(row.id for row in rows) + + async_runner.run(remember()) + with TestSession() as db: + db.delete(db.get(SourceModel, source.id)) + db.commit() diff --git a/tests/extensions/github/test_snapshot.py b/tests/extensions/github/test_snapshot.py new file mode 100644 index 00000000..ea57500a --- /dev/null +++ b/tests/extensions/github/test_snapshot.py @@ -0,0 +1,127 @@ +"""GraphQL pagination must produce one complete snapshot or an explicit failure.""" + +import asyncio +import json +import typing + +import httpx +import pytest + +from extensions.github.adapter import GitHubGraphQLAdapter, GitHubGraphQLError + + +ACCOUNT: dict[str, typing.Any] = { + "id": "U1", + "login": "reader", + "url": "https://github.com/reader", +} + + +def _connection(nodes, cursor=None): + return { + "nodes": nodes, + "pageInfo": {"hasNextPage": cursor is not None, "endCursor": cursor}, + } + + +def _repository(node_id): + return { + "id": node_id, + "nameWithOwner": f"reader/{node_id}", + "url": f"https://github.com/reader/{node_id}", + "isPrivate": False, + "isArchived": False, + "repositoryTopics": {"nodes": []}, + "owner": {**ACCOUNT, "__typename": "User"}, + } + + +def _list(node_id, items): + return { + "id": node_id, + "name": node_id, + "slug": node_id, + "isPrivate": False, + "items": items, + } + + +def _fetch(responses): + requests = [] + + def respond(request): + body = json.loads(request.content) + requests.append(body) + assert len(requests) <= len(responses), "pagination did not terminate" + return httpx.Response(200, json={"data": responses[len(requests) - 1]}) + + async def run(): + async with httpx.AsyncClient( + base_url="https://api.github.com", transport=httpx.MockTransport(respond) + ) as client: + async with GitHubGraphQLAdapter("test-token", client=client) as adapter: + return await adapter.fetch_snapshot() + + return asyncio.run(run()), requests + + +def test_snapshot_traverses_stars_lists_and_embedded_items(): + snapshot, requests = _fetch( + [ + { + "viewer": {**ACCOUNT, "starredRepositories": _connection([_repository("R1")], "s1")} + }, + {"viewer": {**ACCOUNT, "starredRepositories": _connection([_repository("R2")])}}, + { + "viewer": { + "lists": _connection([_list("L1", _connection([{"id": "R1"}], "i1"))], "l1") + } + }, + {"node": {"items": _connection([{"id": "R1"}, {"id": "R3"}])}}, + {"viewer": {"lists": _connection([_list("L2", _connection([]))])}}, + {"nodes": [_repository("R3")]}, + ] + ) + assert snapshot.account.node_id == "U1" + assert snapshot.starred_repository_node_ids == ("R1", "R2") + assert {fact.repository.node_id for fact in snapshot.repositories} == {"R1", "R2", "R3"} + assert [(fact.list.node_id, fact.repository_node_ids) for fact in snapshot.lists] == [ + ("L1", ("R1", "R3")), + ("L2", ()), + ] + assert [request["variables"] for request in requests] == [ + {"cursor": None}, + {"cursor": "s1"}, + {"cursor": None}, + {"id": "L1", "cursor": "i1"}, + {"cursor": "l1"}, + {"ids": ["R3"]}, + ] + + +@pytest.mark.parametrize("level", ["stars", "lists", "items"]) +@pytest.mark.parametrize("cursors", [("a", "a"), ("a", "b", "a")]) +def test_repeated_cursor_rejects_incomplete_snapshot(level, cursors): + responses = [] + if level != "stars": + responses.append({"viewer": {**ACCOUNT, "starredRepositories": _connection([])}}) + for index, cursor in enumerate(cursors): + connection = _connection([], cursor) + if level == "stars": + responses.append({"viewer": {**ACCOUNT, "starredRepositories": connection}}) + elif level == "lists": + responses.append({"viewer": {"lists": connection}}) + elif index == 0: + responses.append({"viewer": {"lists": _connection([_list("L1", connection)])}}) + else: + responses.append({"node": {"items": connection}}) + with pytest.raises(GitHubGraphQLError, match="repeated endCursor"): + _fetch(responses) + + +@pytest.mark.parametrize("cursor", [None, "", 42]) +def test_missing_continuation_is_not_a_complete_snapshot(cursor): + connection = _connection([]) + connection["pageInfo"] = {"hasNextPage": True, "endCursor": cursor} + with pytest.raises(GitHubGraphQLError, match="omitted endCursor"): + _fetch([{"viewer": {**ACCOUNT, "starredRepositories": connection}}]) diff --git a/tests/extensions/mail/test_source_state.py b/tests/extensions/mail/test_source_state.py new file mode 100644 index 00000000..6e1a4e51 --- /dev/null +++ b/tests/extensions/mail/test_source_state.py @@ -0,0 +1,145 @@ +"""Mail connection setup preserves progress committed by concurrent Jobs.""" + +import asyncio +import datetime +import os +from unittest import mock + +import pytest + +from app.business.source import SourceManager +from app.persistence.source.uow import source_uow +from app.schemas.job import JobModel +from app.schemas.source import SourceModel +from extensions.mail.schema import ( + IMAPCheckpoint, + MailAccessBinding, + MailBackfillConfig, + MailCollectConfig, + MailSourceState, +) +from extensions.mail.source import MailSourceBindingError, Source +from tests.database import TestSession + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not os.getenv("INKCRE_TEST_DATABASE_URL"), + reason="requires an explicitly selected migrated disposable PostgreSQL runtime", + ), +] + +BINDING = MailAccessBinding( + protocol="imap", host="mail.invalid", port=993, security="tls", username="reader" +) + + +@pytest.fixture +def source_id(async_runner): + source_type = f"{Source.__module__}.{Source.__qualname__}" + async_runner.run(SourceManager.sync_source_types_async({source_type: Source})) + source = async_runner.run( + SourceManager.create( + source_type, + config={ + "parameters": {"host": "mail.invalid", "username": "reader", "password": "test"}, + "excluded_mailboxes": {"names": [], "special_uses": []}, + }, + ) + ) + yield source.id + with TestSession() as session: + session.delete(session.get(SourceModel, source.id)) + session.commit() + + +@pytest.mark.parametrize("command", ["collect", "backfill"]) +@pytest.mark.parametrize("already_bound", [False, True]) +def test_connecting_job_preserves_concurrent_checkpoint( + async_runner, source_id, command, already_bound +): + async def exercise(): + connecting, release_connection = asyncio.Event(), asyncio.Event() + before = MailSourceState(binding=BINDING if already_bound else None) + after = before.model_copy( + update={"checkpoints": {"123": IMAPCheckpoint(uid_validity=7, last_uid=900)}} + ) + + class Adapter: + binding = BINDING + + async def __aenter__(self): + connecting.set() + await release_connection.wait() + return self + + async def __aexit__(self, *_exc): + pass + + async def discover_mailboxes(self, _exclusions): + return [] + + async with source_uow() as uow: + row = await uow.sources.get(source_id, lock=True) + assert row is not None + row.state = before.model_dump(mode="json") + await uow.sources.save(row) + config = ( + MailCollectConfig() + if command == "collect" + else MailBackfillConfig(since=datetime.date(2020, 1, 1)) + ) + with mock.patch("extensions.mail.source.create_mail_adapter", return_value=Adapter()): + pending = asyncio.create_task( + getattr(Source(source_id), command)( + JobModel(type=f"core.source.{command}.v1", parameters={}), config + ) + ) + try: + await asyncio.wait_for(connecting.wait(), timeout=5) + async with source_uow() as uow: + row = await uow.sources.get(source_id, lock=True) + assert row is not None + row.state = after.model_dump(mode="json") + await uow.sources.save(row) + finally: + release_connection.set() + await pending + async with source_uow() as uow: + row = await uow.sources.get(source_id) + assert row is not None + state = MailSourceState.model_validate(row.state) + assert state.binding == BINDING + assert state.checkpoints == after.checkpoints + + async_runner.run(exercise()) + + +def test_changed_remote_binding_fails_without_overwriting_state(async_runner, source_id): + state = MailSourceState( + binding=BINDING, + checkpoints={"123": IMAPCheckpoint(uid_validity=7, last_uid=900)}, + ) + + async def exercise(): + async with source_uow() as uow: + row = await uow.sources.get(source_id, lock=True) + assert row is not None + row.state = state.model_dump(mode="json") + await uow.sources.save(row) + adapter = mock.AsyncMock() + adapter.__aenter__.return_value = adapter + adapter.binding = BINDING.model_copy(update={"username": "different-account"}) + with mock.patch("extensions.mail.source.create_mail_adapter", return_value=adapter): + with pytest.raises(MailSourceBindingError, match="access binding changed"): + await Source(source_id).collect( + JobModel(type="core.source.collect.v1", parameters={}), MailCollectConfig() + ) + async with source_uow() as uow: + row = await uow.sources.get(source_id) + assert row is not None + assert MailSourceState.model_validate(row.state) == state + adapter.discover_mailboxes.assert_not_called() + + async_runner.run(exercise()) diff --git a/tests/test_extension_startup_provision.py b/tests/test_extension_startup_provision.py index 858aa110..dc016de9 100644 --- a/tests/test_extension_startup_provision.py +++ b/tests/test_extension_startup_provision.py @@ -1,4 +1,6 @@ +import asyncio from pathlib import Path +import threading import typing import uuid @@ -61,6 +63,34 @@ async def get_participation(self, peer_id: uuid.UUID, name: str): async def get(self, name: str): return InstalledExtension(name=name) + async def list(self): + return tuple(InstalledExtension(name=row.name) for row in self.selections) + + async def set_participation_enabled(self, peer_id, name, enabled): + current = await self.get_participation(peer_id, name) + assert current is not None + selected = current.model_copy(update={"enabled": enabled}) + self.selections = [selected if row.name == name else row for row in self.selections] + return selected + + async def ensure(self, name: str, nickname: str): + return InstalledExtension(name=name, nickname=nickname) + + async def select_participation( + self, peer_id: uuid.UUID, name: str, distribution: str, version: str + ): + current = await self.get_participation(peer_id, name) + if current is not None and current.enabled: + raise extension_main.ExtensionStateConflictError("selection is enabled") + selected = participation(name, version, enabled=False) + self.selections = [row for row in self.selections if row.name != name] + self.selections.append(selected) + return selected + + async def remove_participation(self, peer_id: uuid.UUID, name: str): + assert peer_id == PEER_ID + self.selections = [row for row in self.selections if row.name != name] + async def resolve_legacy_python_participation( self, peer_id: uuid.UUID, name: str, version: str ): @@ -97,20 +127,20 @@ def __init__(self) -> None: self.on_download = None self.on_apply = None - def download(self, descriptor, _association, _host_version, wheel_dir: Path) -> Path: - self.downloads.append((descriptor.name, descriptor.version)) + def download(self, release, association, host_version, wheel_dir: Path) -> Path: + self.downloads.append((release.name, release.version)) self.download_dirs.append(wheel_dir) if self.on_download is not None: - self.on_download(descriptor) - return wheel_dir / f"{descriptor.name.rsplit('/', 1)[-1]}-{descriptor.version}.whl" + self.on_download(release) + return wheel_dir / f"{release.name.rsplit('/', 1)[-1]}-{release.version}.whl" - def apply(self, _wheel, descriptor, _association, _host_version): - key = (descriptor.name, descriptor.version) + def apply(self, extension_wheel, release, association, host_version): + key = (release.name, release.version) self.applications.append(key) - acquired = acquired_distribution(descriptor) + acquired = acquired_distribution(release) self.installed[key] = acquired if self.on_apply is not None: - self.on_apply(descriptor) + self.on_apply(release) return acquired @@ -386,3 +416,280 @@ def test_environment_application_rejects_core_runtime_or_base_downgrade( ExtensionHost._assert_environment_packages_preserved( ExtensionHost._environment_package_versions(), "inkcre-ext-mail" ) + + +def test_select_rejects_enabled_replacement_before_registry_or_environment_work( + async_runner, monkeypatch +): + store = SelectionStore(participation("inkcre/mail", "0.3.2", enabled=True)) + reader = ReleaseReader({("inkcre/mail", "0.3.3"): release("inkcre/mail", "0.3.3")}) + consumer = DistributionConsumer() + host = ExtensionHost( + store=typing.cast(ExtensionStore, store), + release_client=reader, + distribution_consumer=consumer, + ) + monkeypatch.setattr(extension_main.AcquiredDistribution, "discover", lambda *_: None) + + with pytest.raises(extension_main.ExtensionStateConflictError, match="enabled"): + async_runner.run(host.select("inkcre/mail", "python", "0.3.3", peer_id=PEER_ID)) + + assert reader.lookups == [] + assert consumer.applications == [] + assert store.selections[0].version == "0.3.2" + + +def test_select_rechecks_enabled_intent_after_peer_drain(async_runner, monkeypatch): + store = SelectionStore(participation("inkcre/mail", "0.3.2", enabled=False)) + reader = ReleaseReader({("inkcre/mail", "0.3.3"): release("inkcre/mail", "0.3.3")}) + consumer = DistributionConsumer() + host = ExtensionHost( + store=typing.cast(ExtensionStore, store), + release_client=reader, + distribution_consumer=consumer, + ) + monkeypatch.setattr(extension_main.AcquiredDistribution, "discover", lambda *_: None) + + async def pause(): + store.selections[0] = store.selections[0].model_copy(update={"enabled": True}) + + host.pause_peer_execution = pause + with pytest.raises(extension_main.ExtensionEnvironmentApplicationRequiredError): + async_runner.run(host.select("inkcre/mail", "python", "0.3.3", peer_id=PEER_ID)) + + assert consumer.applications == [] + status = async_runner.run(host.status(PEER_ID)) + assert status.extensions[0].operation == "failed" + assert "enabled" in (status.extensions[0].error or "") + + +def test_cancelled_download_keeps_directory_and_serializes_until_worker_finishes( + async_runner, monkeypatch +): + async def scenario(): + entered = asyncio.Event() + release_worker = threading.Event() + loop = asyncio.get_running_loop() + store = SelectionStore(participation("inkcre/mail", "0.3.2", enabled=False)) + reader = ReleaseReader({("inkcre/mail", "0.3.3"): release("inkcre/mail", "0.3.3")}) + + class Consumer(DistributionConsumer): + def download(self, release, association, host_version, wheel_dir): + wheel = super().download(release, association, host_version, wheel_dir) + loop.call_soon_threadsafe(entered.set) + if not release_worker.wait(10): + raise TimeoutError("test did not release download") + # Writing after cancellation detects premature TemporaryDirectory cleanup. + wheel.write_bytes(b"download completed") + return wheel + + consumer = Consumer() + host = ExtensionHost( + store=typing.cast(ExtensionStore, store), + release_client=reader, + distribution_consumer=consumer, + ) + + async def pause(): + raise AssertionError("a cancelled download must not pause the Peer") + + host.pause_peer_execution = pause + monkeypatch.setattr(extension_main.AcquiredDistribution, "discover", lambda *_: None) + task = asyncio.create_task( + host.select("inkcre/mail", "python", "0.3.3", peer_id=PEER_ID) + ) + await entered.wait() + removal = asyncio.create_task(host.unselect("inkcre/mail", peer_id=PEER_ID)) + try: + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + assert not removal.done() + finally: + release_worker.set() + with pytest.raises(asyncio.CancelledError): + await task + await removal + assert not consumer.download_dirs[0].exists() + assert consumer.applications == [] + + async_runner.run(scenario()) + + +@pytest.mark.parametrize( + ("phase", "fails"), + [("pause", False), ("apply", False), ("save", False), ("apply", True), ("save", True)], +) +def test_cancelled_application_reaches_an_observable_terminal_state( + async_runner, monkeypatch, phase, fails +): + async def scenario(): + entered, resume = asyncio.Event(), asyncio.Event() + release_worker = threading.Event() + loop = asyncio.get_running_loop() + restarted: list[bool] = [] + + class Store(SelectionStore): + async def select_participation(self, peer_id, name, distribution, version): + if phase == "save": + entered.set() + await resume.wait() + if fails: + raise RuntimeError("selection write unavailable") + return await super().select_participation(peer_id, name, distribution, version) + + class Consumer(DistributionConsumer): + def download(self, release, association, host_version, wheel_dir): + wheel = super().download(release, association, host_version, wheel_dir) + wheel.write_bytes(b"wheel") + return wheel + + def apply(self, extension_wheel, release, association, host_version): + if phase == "apply": + loop.call_soon_threadsafe(entered.set) + if not release_worker.wait(10): + raise TimeoutError("test did not release apply") + assert extension_wheel.read_bytes() == b"wheel" + if fails and phase == "apply": + raise RuntimeError("package application failed") + return super().apply(extension_wheel, release, association, host_version) + + store = Store(participation("inkcre/mail", "0.3.2", enabled=False)) + reader = ReleaseReader({("inkcre/mail", "0.3.3"): release("inkcre/mail", "0.3.3")}) + consumer = Consumer() + host = ExtensionHost( + store=typing.cast(ExtensionStore, store), + release_client=reader, + distribution_consumer=consumer, + ) + + async def pause(): + if phase == "pause": + entered.set() + await resume.wait() + + host.pause_peer_execution = pause + monkeypatch.setattr( + extension_main, "request_process_restart", lambda: restarted.append(True) + ) + monkeypatch.setattr(extension_main.AcquiredDistribution, "discover", lambda *_: None) + task = asyncio.create_task( + host.select("inkcre/mail", "python", "0.3.3", peer_id=PEER_ID) + ) + await entered.wait() + try: + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + finally: + resume.set() + release_worker.set() + with pytest.raises(asyncio.CancelledError): + await task + status = await host.status(PEER_ID) + if fails: + assert store.selections[0].version == "0.3.2" + assert status.extensions[0].operation == "failed" + assert status.extensions[0].error + assert restarted == ([True] if phase == "save" else []) + else: + assert store.selections[0].version == "0.3.3" + assert status.extensions[0].operation == "restarting" + assert restarted == [True] + assert not consumer.download_dirs[0].exists() + + async_runner.run(scenario()) + + +def test_cancelled_cold_restore_finishes_current_apply_without_starting_next( + async_runner, monkeypatch +): + async def scenario(): + entered = asyncio.Event() + resume = threading.Event() + loop = asyncio.get_running_loop() + selected = ( + participation("inkcre/mail", "0.3.3", enabled=True), + participation("inkcre/rss", "0.2.2", enabled=True), + ) + reader = ReleaseReader( + {(row.name, row.version): release(row.name, row.version) for row in selected} + ) + consumer = DistributionConsumer() + + def on_apply(_descriptor): + loop.call_soon_threadsafe(entered.set) + if not resume.wait(10): + raise TimeoutError("test did not release apply") + assert all(directory.exists() for directory in consumer.download_dirs) + + consumer.on_apply = on_apply + host = ExtensionHost( + store=typing.cast(ExtensionStore, SelectionStore(*selected)), + release_client=reader, + distribution_consumer=consumer, + ) + monkeypatch.setattr(extension_main.AcquiredDistribution, "discover", lambda *_: None) + task = asyncio.create_task(host.provision_enabled_distributions()) + await entered.wait() + try: + task.cancel() + await asyncio.sleep(0) + assert not task.done() + finally: + resume.set() + with pytest.raises(asyncio.CancelledError): + await task + assert consumer.applications == [("inkcre/mail", "0.3.3")] + assert all(not directory.exists() for directory in consumer.download_dirs) + + async_runner.run(scenario()) + + +def test_failed_import_still_requires_restart_to_select_a_different_version( + async_runner, monkeypatch +): + name = "inkcre/failed-import" + store = SelectionStore(participation(name, "1.0.0", enabled=False)) + releases = {(name, version): release(name, version) for version in ("1.0.0", "1.1.0")} + reader = ReleaseReader(releases) + restarted: list[bool] = [] + + class BrokenModule: + def __init__(self, _acquired): + pass + + def load(self, _base): + raise RuntimeError("partial import failed") + + host = ExtensionHost( + store=typing.cast(ExtensionStore, store), + release_client=reader, + distribution_consumer=DistributionConsumer(), + ) + monkeypatch.setattr(extension_main, "DistributionModules", BrokenModule) + monkeypatch.setattr( + extension_main.AcquiredDistribution, + "discover", + lambda name, version, _host: acquired_distribution(releases[(name, version)]), + ) + monkeypatch.setattr( + extension_main, "request_process_restart", lambda: restarted.append(True) + ) + + async def scenario(): + with pytest.raises(RuntimeError, match="partial import"): + await host.enable(name, peer_id=PEER_ID, app=extension_main.fastapi.FastAPI()) + await host.disable(name, peer_id=PEER_ID) + await host.select(name, "python", "1.1.0", peer_id=PEER_ID) + status = await host.status(PEER_ID) + assert status.extensions[0].loaded_version == "1.0.0" + assert status.extensions[0].restart_required + assert not status.extensions[0].running + + async_runner.run(scenario()) + assert restarted == [True] From ba69717fba21e0a9597bf3d2f05567e6cf50bdc3 Mon Sep 17 00:00:00 2001 From: Lan_zhijiang Date: Sat, 31 Oct 2026 19:17:04 +0800 Subject: [PATCH 2/2] =?UTF-8?q?ref(core):=20=E9=9B=86=E4=B8=AD=E5=90=88?= =?UTF-8?q?=E5=90=8C=E5=AE=9A=E4=B9=89=E4=B8=8E=E8=BF=90=E8=A1=8C=E8=B5=84?= =?UTF-8?q?=E6=BA=90=E6=89=80=E6=9C=89=E6=9D=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .changes/maintainability-contracts.changed.md | 1 + .pre-commit-config.yaml | 8 +- app/business/extension/AGENTS.md | 3 + app/business/extension/main.py | 100 ++++--- app/business/job.py | 39 +-- app/business/lexical_retrieval/job.py | 15 +- app/business/organization/jobs.py | 45 +-- app/business/organization_job.py | 3 - app/business/organization_media.py | 5 +- app/business/semantic_retrieval/job.py | 15 +- app/business/sink/agent_query.py | 15 +- app/business/source/AGENTS.md | 3 + app/business/source/job.py | 15 +- app/business/source/main.py | 13 +- app/contracts/__init__.py | 1 + app/contracts/jobs.py | 215 +++++++++++++++ app/database_contract/profile.py | 258 +----------------- app/schemas/lexical_retrieval.py | 21 +- app/schemas/organization.py | 18 +- app/schemas/organization_behavior.py | 10 +- app/schemas/semantic_retrieval.py | 25 +- app/schemas/source/main.py | 22 +- .../business-pipeline-and-authority.md | 8 + docs/40-deployment/development-environment.md | 6 +- docs/40-deployment/runtime-orchestration.md | 7 + .../.changes/source-identity.changed.md | 1 + extensions/github/stars.py | 7 +- .../mail/.changes/source-identity.changed.md | 1 + extensions/mail/source.py | 2 + .../rss/.changes/source-identity.changed.md | 1 + extensions/rss/atom.py | 2 + extensions/rss/rss.py | 2 + .../.changes/source-identity.changed.md | 1 + extensions/telegram/source.py | 7 +- .../.changes/source-identity.changed.md | 1 + extensions/twitter/bookmark.py | 2 + pyproject.toml | 1 + run.py | 14 +- scripts/automation/repository_check.sh | 19 +- scripts/tests/caldav/test_dav_graph.py | 55 +++- tests/test_builtin_job_contracts.py | 24 ++ ...extension_participation_lifecycle_races.py | 5 +- tests/test_extension_startup_provision.py | 100 +++++++ tests/test_source_identity.py | 78 ++++++ 44 files changed, 703 insertions(+), 491 deletions(-) create mode 100644 .changes/maintainability-contracts.changed.md create mode 100644 app/contracts/__init__.py create mode 100644 app/contracts/jobs.py create mode 100644 extensions/github/.changes/source-identity.changed.md create mode 100644 extensions/mail/.changes/source-identity.changed.md create mode 100644 extensions/rss/.changes/source-identity.changed.md create mode 100644 extensions/telegram/.changes/source-identity.changed.md create mode 100644 extensions/twitter/.changes/source-identity.changed.md create mode 100644 tests/test_builtin_job_contracts.py create mode 100644 tests/test_source_identity.py diff --git a/.changes/maintainability-contracts.changed.md b/.changes/maintainability-contracts.changed.md new file mode 100644 index 00000000..6ee86527 --- /dev/null +++ b/.changes/maintainability-contracts.changed.md @@ -0,0 +1 @@ +内置 Job 的输入与发布目录共用同一合同；Source 可以显式保留稳定类型身份，扩展资源收尾与进程排空顺序集中维护。 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ff5ed50e..31ba375b 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -25,25 +25,25 @@ repos: pass_filenames: false - id: typecheck name: Check Pyrefly types - entry: pyrefly check --min-severity=warn --output-format=min-text --progress-bar=no + entry: pdm run typecheck language: system files: ^(app/|extensions/|libs/|migrations/|scripts/|tests/|utils/|run\.py$|pyproject\.toml$|pdm\.lock$) pass_filenames: false - id: check-lock name: Check PDM lock - entry: python scripts/check_lock.py + entry: pdm run check:lock language: system files: ^(pyproject\.toml|pdm\.lock)$ pass_filenames: false - id: lint-migrations name: Lint migration contract - entry: ruff check --no-cache migrations/env.py migrations/metadata.py migrations/settings.py scripts/_tooling.py scripts/doctor.py scripts/check_lock.py scripts/check_migration_history.py scripts/database_manifest.py scripts/sanitize_preview_base.py tests/migrations + entry: pdm run lint:migrations language: system files: ^(alembic\.ini|migrations/|scripts/|tests/migrations/) pass_filenames: false - id: test-migrations name: Test migration contract - entry: python -m pytest -q tests/migrations + entry: pdm run test:migrations language: system files: ^(alembic\.ini|migrations/|tests/migrations/) pass_filenames: false diff --git a/app/business/extension/AGENTS.md b/app/business/extension/AGENTS.md index 69fd7c97..155ebdcd 100644 --- a/app/business/extension/AGENTS.md +++ b/app/business/extension/AGENTS.md @@ -11,6 +11,9 @@ - `ExtensionBase` 保持配置校验、`on_start_async`、`on_close` 与 active-effect publication。Route、Peer inbound 和 public claim 可逆；已 import 的 Source/Resolver/Sink class registration 在当前进程内单调保留。 - Extension wheel 直接 import Core 模块；Host 只接受标准 `inkcre.core.extensions` entry point。 +- `ActiveExtensionEffects` 拥有已取得的 claim/class 效果及释放顺序；新增 active effect 只修改该所有者。 + 普通停止遇错立即退出；intent 冲突清理聚合 close/unpublish/unbind 错误；启动失败保留原错误并尽力 + 收尾。claim 释放错误不被吞掉。环境应用由 Host 在已持锁处关闭 Extension，bootstrap callback 只排空生产者。 - Registry Simple URL 必须与配置的 Registry 同源且路径精确匹配 Project。 - runtime 只能把普通 wheel 安装到当前 Core interpreter/site-packages；禁止 `pip --target`、 per-Extension overlay、`sys.path` 或 `extensions.__path__` 改写。 diff --git a/app/business/extension/main.py b/app/business/extension/main.py index 3ac3fd46..0fb1048e 100644 --- a/app/business/extension/main.py +++ b/app/business/extension/main.py @@ -3,9 +3,9 @@ from __future__ import annotations import asyncio -import contextlib from dataclasses import dataclass import importlib.metadata +import inspect from pathlib import Path import tempfile import typing @@ -193,14 +193,58 @@ async def mutate_config_and_state_async( return result +@dataclass +class ActiveExtensionEffects: + """Own the acquired runtime effects and their one release order. + + Startup preserves its original error, conflict cleanup collects recoverable + hook failures, and ordinary stop fails immediately. Releasing the Host claim + is always required to succeed; it must never be silently discarded. + """ + + claim: ExtensionRuntimeClaim + extension_class: type[ExtensionBase] | None = None + + async def close( + self, + *, + policy: typing.Literal["stop", "conflict", "startup_failure"] = "stop", + ) -> list[Exception]: + failures: list[Exception] = [] + if self.extension_class is not None: + extension = self.extension_class + recover_hook_errors = policy != "stop" + for release, recover_error in ( + (extension.on_close, recover_hook_errors), + (extension.unpublish, recover_hook_errors), + (extension.unbind, recover_hook_errors), + (extension.release_runtime, policy == "startup_failure"), + ): + try: + result = release() + if inspect.isawaitable(result): + await result + except Exception as error: + if not recover_error: + raise + failures.append(error) + self.claim.release() + return failures + + @dataclass class RunningExtension: name: str version: str association: PythonReleaseDescriptor acquired: AcquiredDistribution - extension_class: type[ExtensionBase] - claim: ExtensionRuntimeClaim + effects: ActiveExtensionEffects + + @property + def extension_class(self) -> type[ExtensionBase]: + extension = self.effects.extension_class + assert extension is not None + return extension class ExtensionHost: @@ -476,6 +520,9 @@ async def _apply_selection( self._operations[name] = ("applying", None) try: await self.pause_peer_execution() + # The callback drains Core producers; Host lock ownership never escapes + # into bootstrap code. This operation already owns the runtime lock. + await self._close_running_locked() # Other admitted Peers can change intent during the drain. This check # avoids known conflicts; the store still arbitrates the final write. await self._require_selectable(name, version, peer_id=peer_id) @@ -851,17 +898,14 @@ async def _start_acquired( claim = ExtensionRuntimeClaim.acquire(association.entry_point.name) except ExtensionRuntimeClaimConflictError as error: raise ExtensionStateConflictError(str(error)) from error + effects = ActiveExtensionEffects(claim) try: modules = DistributionModules(acquired) - except Exception: - claim.release() - raise - extension_class: type[ExtensionBase] | None = None - try: # Import is irreversible even when the module or its startup later fails. # A different selected version therefore still needs a fresh process. self._loaded_versions[state.name] = selection.version extension_class = typing.cast(type[ExtensionBase], modules.load(ExtensionBase)) + effects.extension_class = extension_class if extension_class.__extid__ != association.entry_point.name: raise ExtensionCompatibilityError( "ExtensionBase identity differs from the declared entry point" @@ -879,24 +923,14 @@ async def _start_acquired( await extension_class.on_start_async(app) modules.assert_origins() except BaseException: - if extension_class is not None: - with contextlib.suppress(Exception): - await extension_class.on_close() - with contextlib.suppress(Exception): - extension_class.unpublish() - with contextlib.suppress(Exception): - extension_class.unbind() - with contextlib.suppress(Exception): - extension_class.release_runtime() - claim.release() + await effects.close(policy="startup_failure") raise running = RunningExtension( name=state.name, version=selection.version, association=association, acquired=acquired, - extension_class=extension_class, - claim=claim, + effects=effects, ) self.running[state.name] = running return running @@ -916,29 +950,11 @@ async def _get_current_enabled_selection( return current async def _stop(self, running: RunningExtension) -> None: - await running.extension_class.on_close() - running.extension_class.unpublish() - running.extension_class.unbind() - running.extension_class.release_runtime() - running.claim.release() + await running.effects.close() self.running.pop(running.name, None) async def _force_stop(self, running: RunningExtension) -> typing.List[Exception]: - failures: list[Exception] = [] - try: - await running.extension_class.on_close() - except Exception as error: - failures.append(error) - try: - running.extension_class.unpublish() - except Exception as error: - failures.append(error) - try: - running.extension_class.unbind() - except Exception as error: - failures.append(error) - running.extension_class.release_runtime() - running.claim.release() + failures = await running.effects.close(policy="conflict") self.running.pop(running.name, None) return failures @@ -1080,10 +1096,6 @@ async def close_running(self) -> None: async with self._runtime_lock: await self._close_running_locked() - async def close_running_for_environment_application(self) -> None: - """Close extensions while select owns the lock and fences other mutations.""" - await self._close_running_locked() - async def _close_running_locked(self) -> None: failures: list[Exception] = [] for running in tuple(self.running.values()): diff --git a/app/business/job.py b/app/business/job.py index a85d1bbc..596326b9 100644 --- a/app/business/job.py +++ b/app/business/job.py @@ -8,7 +8,7 @@ import jsonschema # pyrefly: ignore[untyped-import] import pydantic -from app.database_contract.profile import BUILTIN_JOB_TYPES_BY_ID +from app.contracts.jobs import BUILTIN_JOB_TYPES_BY_ID from app.persistence.job.uow import JobUnitOfWork, job_uow from app.scheduler import scheduler, with_trace_id from app.schemas.job import JobID, JobModel, JobStatus, JobTypeID, JobTypeModel @@ -52,11 +52,27 @@ def __init_subclass__( cls, *, job_type: JobTypeID, - description: str, - parameters_model: type[ParametersTV], - default_timeout_seconds: int, + description: str | None = None, + parameters_model: type[ParametersTV] | None = None, + default_timeout_seconds: int | None = None, **kwargs, ) -> None: + builtin = BUILTIN_JOB_TYPES_BY_ID.get(job_type) + if builtin is not None: + if any( + supplied is not None and supplied != expected + for supplied, expected in ( + (description, builtin.description), + (parameters_model, builtin.parameters_model), + (default_timeout_seconds, builtin.default_timeout_seconds), + ) + ): + raise ValueError(f"Built-in Job {job_type!r} must use its shared contract") + description = builtin.description + parameters_model = typing.cast(type[ParametersTV], builtin.parameters_model) + default_timeout_seconds = builtin.default_timeout_seconds + if description is None or parameters_model is None or default_timeout_seconds is None: + raise TypeError("Custom Jobs require description, parameters_model and timeout") if default_timeout_seconds <= 0: raise ValueError("default_timeout_seconds must be positive") cls.type = job_type @@ -118,21 +134,12 @@ def register_handler(cls, handler: type[JobHandler]) -> None: async def sync_job_types(cls) -> None: rows = [] for handler in cls._handlers.values(): - builtin = BUILTIN_JOB_TYPES_BY_ID.get(handler.type) rows.append( dict( id=handler.type, - description=builtin.description if builtin is not None else handler.description, - parameters_schema=( - builtin.parameters_schema - if builtin is not None - else handler.parameters_model.model_json_schema() - ), - default_timeout_seconds=( - builtin.default_timeout_seconds - if builtin is not None - else handler.default_timeout_seconds - ), + description=handler.description, + parameters_schema=handler.parameters_model.model_json_schema(), + default_timeout_seconds=handler.default_timeout_seconds, ) ) async with job_uow() as uow: diff --git a/app/business/lexical_retrieval/job.py b/app/business/lexical_retrieval/job.py index a724f4db..d38c74a2 100644 --- a/app/business/lexical_retrieval/job.py +++ b/app/business/lexical_retrieval/job.py @@ -1,5 +1,10 @@ """Exact lexical projection maintenance commands hosted by Jobs.""" +from app.contracts.jobs import ( + LEXICAL_MAINTAIN_JOB_TYPE as LEXICAL_MAINTAIN_JOB_TYPE, + LEXICAL_REBUILD_JOB_TYPE as LEXICAL_REBUILD_JOB_TYPE, +) + from app.business.info_base.resolver import ResolverManager from app.business.job import JobHandler from app.schemas.job import JobModel @@ -8,16 +13,9 @@ from .main import LexicalRetrievalManager -LEXICAL_MAINTAIN_JOB_TYPE = "core.feature_retrieval.lexical.maintain.v1" -LEXICAL_REBUILD_JOB_TYPE = "core.feature_retrieval.lexical.rebuild.v1" - - class LexicalMaintainJobHandler( JobHandler[LexicalMaintenanceJobParameters], job_type=LEXICAL_MAINTAIN_JOB_TYPE, - description="Maintain missing or stale Block lexical projection records.", - parameters_model=LexicalMaintenanceJobParameters, - default_timeout_seconds=900, ): @classmethod async def can_handle(cls, parameters: LexicalMaintenanceJobParameters) -> bool: @@ -37,9 +35,6 @@ async def handle( class LexicalRebuildJobHandler( JobHandler[LexicalMaintenanceJobParameters], job_type=LEXICAL_REBUILD_JOB_TYPE, - description="Rebuild Block lexical records present before invocation.", - parameters_model=LexicalMaintenanceJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: LexicalMaintenanceJobParameters) -> bool: diff --git a/app/business/organization/jobs.py b/app/business/organization/jobs.py index b7e6b555..4d1e677d 100644 --- a/app/business/organization/jobs.py +++ b/app/business/organization/jobs.py @@ -1,5 +1,16 @@ """Independent automatic Organization routes and explicit rumination Jobs.""" +from app.contracts.jobs import ( + RUMINATION_JOB_TYPE as RUMINATION_JOB_TYPE, + EXPLICIT_RUMINATION_JOB_TYPE as EXPLICIT_RUMINATION_JOB_TYPE, + SUPERSESSION_JOB_TYPE as SUPERSESSION_JOB_TYPE, + REFINEMENT_JOB_TYPE as REFINEMENT_JOB_TYPE, + EVIDENCE_STANCE_JOB_TYPE as EVIDENCE_STANCE_JOB_TYPE, + SYNTHESIS_JOB_TYPE as SYNTHESIS_JOB_TYPE, + REFERENT_ANCHORING_JOB_TYPE as REFERENT_ANCHORING_JOB_TYPE, + DUPLICATE_ASSERTION_JOB_TYPE as DUPLICATE_ASSERTION_JOB_TYPE, +) + from app.business.job import JobHandler from app.schemas.job import JobModel from app.schemas.organization import RuminationRequest @@ -14,22 +25,9 @@ from .synthesis import SynthesisBehaviorResolver -RUMINATION_JOB_TYPE = "core.organization.rumination.automatic.v1" -EXPLICIT_RUMINATION_JOB_TYPE = "core.organization.rumination.explicit.v1" -SUPERSESSION_JOB_TYPE = "core.organization.supersession.automatic.v1" -REFINEMENT_JOB_TYPE = "core.organization.refinement.automatic.v1" -EVIDENCE_STANCE_JOB_TYPE = "core.organization.evidence-stance.automatic.v1" -SYNTHESIS_JOB_TYPE = "core.organization.synthesis.automatic.v1" -REFERENT_ANCHORING_JOB_TYPE = "core.organization.existing-referent-anchoring.automatic.v1" -DUPLICATE_ASSERTION_JOB_TYPE = "core.organization.duplicate-assertion.automatic.v1" - - class ExplicitRuminationJobHandler( JobHandler[RuminationRequest], job_type=EXPLICIT_RUMINATION_JOB_TYPE, - description="Reconsider one explicitly selected Block through rumination.", - parameters_model=RuminationRequest, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: RuminationRequest) -> bool: @@ -46,9 +44,6 @@ async def handle(cls, job: JobModel, parameters: RuminationRequest) -> None: class RuminationJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=RUMINATION_JOB_TYPE, - description="Automatically reconsider bounded information seeds through rumination.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: @@ -68,9 +63,6 @@ async def handle( class SupersessionJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=SUPERSESSION_JOB_TYPE, - description="Automatically judge bounded scoped-supersession candidates.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: @@ -90,9 +82,6 @@ async def handle( class RefinementJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=REFINEMENT_JOB_TYPE, - description="Automatically judge bounded non-dominating refinement candidates.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: @@ -112,9 +101,6 @@ async def handle( class EvidenceStanceJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=EVIDENCE_STANCE_JOB_TYPE, - description="Automatically judge bounded attributable evidence stances.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: @@ -134,9 +120,6 @@ async def handle( class SynthesisJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=SYNTHESIS_JOB_TYPE, - description="Automatically create bounded provenance-preserving syntheses.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: @@ -156,9 +139,6 @@ async def handle( class ExistingReferentAnchoringJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=REFERENT_ANCHORING_JOB_TYPE, - description="Automatically anchor bounded mentions to existing referents.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: @@ -178,9 +158,6 @@ async def handle( class DuplicateAssertionJobHandler( JobHandler[AutomaticOrganizationJobParameters], job_type=DUPLICATE_ASSERTION_JOB_TYPE, - description="Automatically judge bounded provenance-aware duplicate assertions.", - parameters_model=AutomaticOrganizationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: AutomaticOrganizationJobParameters) -> bool: diff --git a/app/business/organization_job.py b/app/business/organization_job.py index 270eea94..d3d47b71 100644 --- a/app/business/organization_job.py +++ b/app/business/organization_job.py @@ -13,9 +13,6 @@ class MediaInterpretationJobHandler( JobHandler[MediaInterpretationJobParameters], job_type=MEDIA_INTERPRETATION_JOB_TYPE, - description="Interpret missing image, audio, and video Blocks through configured Agents.", - parameters_model=MediaInterpretationJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: MediaInterpretationJobParameters) -> bool: diff --git a/app/business/organization_media.py b/app/business/organization_media.py index a32fabc3..eafe1299 100644 --- a/app/business/organization_media.py +++ b/app/business/organization_media.py @@ -1,5 +1,9 @@ """System-driven media interpretation approach owned by Organization.""" +from app.contracts.jobs import ( + MEDIA_INTERPRETATION_JOB_TYPE as MEDIA_INTERPRETATION_JOB_TYPE, +) + import json import logging import typing @@ -33,7 +37,6 @@ MEDIA_INTERPRETATION_CONFIG_KEY = "core.organization.media_interpretation" MEDIA_INTERPRETATION_CONFIG_SCHEMA = "core.organization.media_interpretation.config.v1" -MEDIA_INTERPRETATION_JOB_TYPE = "core.organization.media_interpretation.v1" _CANDIDATE_LIMIT = 20 _DIAGNOSTIC_LIMIT = 20 _RESOLVER_MODALITIES: dict[str, typing.Literal["image", "audio", "video"]] = { diff --git a/app/business/semantic_retrieval/job.py b/app/business/semantic_retrieval/job.py index fc0b5a0f..ac4a4c6c 100644 --- a/app/business/semantic_retrieval/job.py +++ b/app/business/semantic_retrieval/job.py @@ -1,5 +1,10 @@ """Exact semantic embedding maintenance commands hosted by Jobs.""" +from app.contracts.jobs import ( + SEMANTIC_MAINTAIN_JOB_TYPE as SEMANTIC_MAINTAIN_JOB_TYPE, + SEMANTIC_REBUILD_JOB_TYPE as SEMANTIC_REBUILD_JOB_TYPE, +) + from app.business.job import JobHandler from app.schemas.job import JobModel from app.schemas.semantic_retrieval import EmbeddingMaintenanceJobParameters @@ -7,16 +12,9 @@ from .main import SemanticRetrievalManager -SEMANTIC_MAINTAIN_JOB_TYPE = "core.semantic_retrieval.maintain.v1" -SEMANTIC_REBUILD_JOB_TYPE = "core.semantic_retrieval.rebuild.v1" - - class SemanticMaintainJobHandler( JobHandler[EmbeddingMaintenanceJobParameters], job_type=SEMANTIC_MAINTAIN_JOB_TYPE, - description="Maintain missing or stale semantic embedding records.", - parameters_model=EmbeddingMaintenanceJobParameters, - default_timeout_seconds=900, ): @classmethod async def can_handle(cls, parameters: EmbeddingMaintenanceJobParameters) -> bool: @@ -38,9 +36,6 @@ async def handle( class SemanticRebuildJobHandler( JobHandler[EmbeddingMaintenanceJobParameters], job_type=SEMANTIC_REBUILD_JOB_TYPE, - description="Rebuild semantic embedding records present before invocation.", - parameters_model=EmbeddingMaintenanceJobParameters, - default_timeout_seconds=1800, ): @classmethod async def can_handle(cls, parameters: EmbeddingMaintenanceJobParameters) -> bool: diff --git a/app/business/sink/agent_query.py b/app/business/sink/agent_query.py index 82e2761c..69ea07e3 100644 --- a/app/business/sink/agent_query.py +++ b/app/business/sink/agent_query.py @@ -2,6 +2,10 @@ from __future__ import annotations +from app.contracts.jobs import AGENT_QUERY_JOB_TYPE as AGENT_QUERY_JOB_TYPE + +from app.contracts.jobs import AgentQueryJobParameters as AgentQueryJobParameters + import typing import fastapi @@ -29,7 +33,6 @@ AGENT_QUERY_SINK_TYPE = "core.agent-query.v1" -AGENT_QUERY_JOB_TYPE = "core.sink.agent-query.v1" SUBMIT_QUERY_RESULT_TOOL = "submit_query_result" @@ -60,13 +63,6 @@ class AgentQueryRequest(pydantic.BaseModel): timeout_seconds: int | None = pydantic.Field(default=None, gt=0) -class AgentQueryJobParameters(pydantic.BaseModel): - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - sink: int - query: str = pydantic.Field(min_length=1) - - class AgentQueryResultMissingError(RuntimeError): """A normal Agent Turn ended without delivering its required result.""" @@ -186,9 +182,6 @@ async def execute(self, job: JobModel, query: str) -> None: class AgentQueryJobHandler( JobHandler[AgentQueryJobParameters], job_type=AGENT_QUERY_JOB_TYPE, - description="Answer one information need through an enabled Agent Query Sink.", - parameters_model=AgentQueryJobParameters, - default_timeout_seconds=300, ): @classmethod async def normalize_parameters( diff --git a/app/business/source/AGENTS.md b/app/business/source/AGENTS.md index 40e6b608..a8f44401 100644 --- a/app/business/source/AGENTS.md +++ b/app/business/source/AGENTS.md @@ -21,6 +21,9 @@ ## Registration - `SourceBase.__init_subclass__()` 只登记内存 class；import 不连接数据库。 +- Source class 的 `source_type` 是稳定持久身份，移动/重命名实现时保持其值。第一方类型显式声明既有 + path 字符串；旧第三方 wheel 未声明时仍以该 class 自己的 module/qualname 注册，不继承父类身份。 +- 同一 class 重复注册幂等，不同 class 不得抢占同一 type ID；身份变更是数据合同变化，不是普通重命名。 - `await SourceManager.sync_source_types_async()` 在显式 bootstrap reconcile catalog。 - Extension source 必须在 extension startup import；否则 source type 不存在于当前 runtime registry。 - Extension runtime disable 不撤销已 import 的 Source class registration。Registration 在进程内单调保留； diff --git a/app/business/source/job.py b/app/business/source/job.py index 3e80fc84..699951d1 100644 --- a/app/business/source/job.py +++ b/app/business/source/job.py @@ -1,5 +1,10 @@ """Exact Source command handlers hosted by the global Job runtime.""" +from app.contracts.jobs import ( + SOURCE_COLLECT_JOB_TYPE as SOURCE_COLLECT_JOB_TYPE, + SOURCE_BACKFILL_JOB_TYPE as SOURCE_BACKFILL_JOB_TYPE, +) + from app.business.job import JobHandler from app.persistence.job.uow import JobUnitOfWork from app.schemas.job import JobModel @@ -13,10 +18,6 @@ from .main import SourceManager, SourceNotFoundError -SOURCE_COLLECT_JOB_TYPE = "core.source.collect.v1" -SOURCE_BACKFILL_JOB_TYPE = "core.source.backfill.v1" - - async def _source_type(parameters: SourceCollectParameters) -> str | None: source = await SourceManager.get(parameters.source) return None if source is None else source.type @@ -25,9 +26,6 @@ async def _source_type(parameters: SourceCollectParameters) -> str | None: class SourceCollectJobHandler( JobHandler[SourceCollectParameters], job_type=SOURCE_COLLECT_JOB_TYPE, - description="Run one ordinary collection command for a configured Source.", - parameters_model=SourceCollectParameters, - default_timeout_seconds=300, ): @classmethod async def normalize_parameters(cls, parameters: dict, uow: JobUnitOfWork) -> dict: @@ -59,9 +57,6 @@ async def handle(cls, job: JobModel, parameters: SourceCollectParameters) -> Non class SourceBackfillJobHandler( JobHandler[SourceBackfillParameters], job_type=SOURCE_BACKFILL_JOB_TYPE, - description="Run one exact historical collection command for a configured Source.", - parameters_model=SourceBackfillParameters, - default_timeout_seconds=1800, ): @classmethod async def normalize_parameters(cls, parameters: dict, uow: JobUnitOfWork) -> dict: diff --git a/app/business/source/main.py b/app/business/source/main.py index 2ceb8fb2..4c11d851 100644 --- a/app/business/source/main.py +++ b/app/business/source/main.py @@ -40,6 +40,7 @@ class SourceBase(abc.ABC, typing.Generic[ConfigTV]): __configcls__: type[ConfigTV] __collectconfigcls__: type[pydantic.BaseModel] __backfillconfigcls__: type[pydantic.BaseModel] | None + source_type: typing.ClassVar[str] def __init_subclass__( cls, @@ -53,6 +54,13 @@ def __init_subclass__( cls.__configschema__ = config_cls.model_json_schema() cls.__collectconfigcls__ = collect_config_cls # pyrefly: ignore[no-access] cls.__backfillconfigcls__ = backfill_config_cls # pyrefly: ignore[no-access] + # Older third-party wheels keep their historical path-based identity. New + # declarations can retain that exact value independently of code location. + cls.source_type = cls.__dict__.get( + "source_type", cls.__module__ + "." + cls.__qualname__ + ) + if not isinstance(cls.source_type, str) or not cls.source_type: + raise ValueError("Source type must be a non-empty string") SourceManager.add_source_type(cls) return super().__init_subclass__(**kwargs) @@ -130,7 +138,10 @@ class SourceManager: @classmethod def add_source_type(cls, source_cls: type[SourceBase]) -> None: """Register a source type in memory without external side effects.""" - source_type = source_cls.__module__ + "." + source_cls.__qualname__ + source_type = source_cls.source_type + existing = cls._SOURCE_CLASSES.get(source_type) + if existing is not None and existing is not source_cls: + raise ValueError(f"Source type {source_type!r} is already registered") cls._SOURCE_CLASSES[source_type] = source_cls @classmethod diff --git a/app/contracts/__init__.py b/app/contracts/__init__.py new file mode 100644 index 00000000..bf523fb6 --- /dev/null +++ b/app/contracts/__init__.py @@ -0,0 +1 @@ +"""Pure contracts shared by artifact tooling and runtime implementations.""" diff --git a/app/contracts/jobs.py b/app/contracts/jobs.py new file mode 100644 index 00000000..b8e97d02 --- /dev/null +++ b/app/contracts/jobs.py @@ -0,0 +1,215 @@ +"""Built-in Job input and catalog facts, without runtime or ORM imports. + +Domain handlers retain admission lookups, execution eligibility and effects. This +module owns only their shared input shape and published catalog metadata. +""" + +from dataclasses import dataclass +import typing + +import pydantic + + +class SourceCollectParameters(pydantic.BaseModel): + """Parameters of the exact ordinary Source Job type.""" + + model_config = pydantic.ConfigDict(extra="forbid") + + source: int + config: dict[str, typing.Any] = pydantic.Field(default_factory=dict) + + +class SourceBackfillParameters(SourceCollectParameters): + """Parameters of the exact historical Source Job type.""" + + +class LexicalMaintenanceOptions(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + max_records: int = pydantic.Field(default=100, ge=1) + scan_page_size: int = pydantic.Field(default=100, ge=1) + diagnostic_limit: int = pydantic.Field(default=20, ge=0) + + +class LexicalMaintenanceJobParameters(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + options: LexicalMaintenanceOptions = pydantic.Field( + default_factory=LexicalMaintenanceOptions + ) + + +class EmbeddingMaintenanceOptions(pydantic.BaseModel): + """Peer-local bounds for one resumable maintenance invocation.""" + + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + max_embeddings: int = pydantic.Field(default=100, ge=1) + batch_size: int = pydantic.Field(default=20, ge=1) + scan_page_size: int = pydantic.Field(default=100, ge=1) + diagnostic_limit: int = pydantic.Field(default=20, ge=0) + + +class EmbeddingMaintenanceJobParameters(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + profile: int | None = None + options: EmbeddingMaintenanceOptions = pydantic.Field( + default_factory=EmbeddingMaintenanceOptions + ) + + +class MediaInterpretationJobParameters(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + +class RuminationRequest(pydantic.BaseModel): + """Explicit request to reconsider one focal Block.""" + + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + block: int + + +class AutomaticOrganizationJobParameters(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + max_seeds: int = pydantic.Field(default=10, ge=3, le=100) + + +class AgentQueryJobParameters(pydantic.BaseModel): + model_config = pydantic.ConfigDict(extra="forbid", frozen=True) + + sink: int + query: str = pydantic.Field(min_length=1) + + +@dataclass(frozen=True) +class JobContract: + id: str + description: str + parameters_model: type[pydantic.BaseModel] + default_timeout_seconds: int + + @property + def parameters_schema(self) -> dict[str, typing.Any]: + return self.parameters_model.model_json_schema() + + +SOURCE_BACKFILL_JOB_TYPE = "core.source.backfill.v1" +SOURCE_COLLECT_JOB_TYPE = "core.source.collect.v1" +LEXICAL_REBUILD_JOB_TYPE = "core.feature_retrieval.lexical.rebuild.v1" +LEXICAL_MAINTAIN_JOB_TYPE = "core.feature_retrieval.lexical.maintain.v1" +SEMANTIC_REBUILD_JOB_TYPE = "core.semantic_retrieval.rebuild.v1" +SEMANTIC_MAINTAIN_JOB_TYPE = "core.semantic_retrieval.maintain.v1" +DUPLICATE_ASSERTION_JOB_TYPE = "core.organization.duplicate-assertion.automatic.v1" +REFERENT_ANCHORING_JOB_TYPE = "core.organization.existing-referent-anchoring.automatic.v1" +SYNTHESIS_JOB_TYPE = "core.organization.synthesis.automatic.v1" +EVIDENCE_STANCE_JOB_TYPE = "core.organization.evidence-stance.automatic.v1" +REFINEMENT_JOB_TYPE = "core.organization.refinement.automatic.v1" +SUPERSESSION_JOB_TYPE = "core.organization.supersession.automatic.v1" +EXPLICIT_RUMINATION_JOB_TYPE = "core.organization.rumination.explicit.v1" +RUMINATION_JOB_TYPE = "core.organization.rumination.automatic.v1" +AGENT_QUERY_JOB_TYPE = "core.sink.agent-query.v1" +MEDIA_INTERPRETATION_JOB_TYPE = "core.organization.media_interpretation.v1" + +BUILTIN_JOB_TYPES = ( + JobContract( + SOURCE_COLLECT_JOB_TYPE, + "Run one ordinary collection command for a configured Source.", + SourceCollectParameters, + 300, + ), + JobContract( + SOURCE_BACKFILL_JOB_TYPE, + "Run one exact historical collection command for a configured Source.", + SourceBackfillParameters, + 1800, + ), + JobContract( + LEXICAL_MAINTAIN_JOB_TYPE, + "Maintain missing or stale Block lexical projection records.", + LexicalMaintenanceJobParameters, + 900, + ), + JobContract( + LEXICAL_REBUILD_JOB_TYPE, + "Rebuild Block lexical records present before invocation.", + LexicalMaintenanceJobParameters, + 1800, + ), + JobContract( + SEMANTIC_MAINTAIN_JOB_TYPE, + "Maintain missing or stale semantic embedding records.", + EmbeddingMaintenanceJobParameters, + 900, + ), + JobContract( + SEMANTIC_REBUILD_JOB_TYPE, + "Rebuild semantic embedding records present before invocation.", + EmbeddingMaintenanceJobParameters, + 1800, + ), + JobContract( + MEDIA_INTERPRETATION_JOB_TYPE, + "Interpret missing image, audio, and video Blocks through configured Agents.", + MediaInterpretationJobParameters, + 1800, + ), + JobContract( + EXPLICIT_RUMINATION_JOB_TYPE, + "Reconsider one explicitly selected Block through rumination.", + RuminationRequest, + 1800, + ), + JobContract( + RUMINATION_JOB_TYPE, + "Automatically reconsider bounded information seeds through rumination.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + SUPERSESSION_JOB_TYPE, + "Automatically judge bounded scoped-supersession candidates.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + REFINEMENT_JOB_TYPE, + "Automatically judge bounded non-dominating refinement candidates.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + EVIDENCE_STANCE_JOB_TYPE, + "Automatically judge bounded attributable evidence stances.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + SYNTHESIS_JOB_TYPE, + "Automatically create bounded provenance-preserving syntheses.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + REFERENT_ANCHORING_JOB_TYPE, + "Automatically anchor bounded mentions to existing referents.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + DUPLICATE_ASSERTION_JOB_TYPE, + "Automatically judge bounded provenance-aware duplicate assertions.", + AutomaticOrganizationJobParameters, + 1800, + ), + JobContract( + AGENT_QUERY_JOB_TYPE, + "Answer one information need through an enabled Agent Query Sink.", + AgentQueryJobParameters, + 300, + ), +) + +BUILTIN_JOB_TYPES_BY_ID = {item.id: item for item in BUILTIN_JOB_TYPES} diff --git a/app/database_contract/profile.py b/app/database_contract/profile.py index bfb56505..4ea1e1f1 100644 --- a/app/database_contract/profile.py +++ b/app/database_contract/profile.py @@ -7,6 +7,11 @@ from dataclasses import dataclass from typing import Any +from app.contracts.jobs import ( + BUILTIN_JOB_TYPES as BUILTIN_JOB_TYPES, + BUILTIN_JOB_TYPES_BY_ID as BUILTIN_JOB_TYPES_BY_ID, +) + JsonObject = dict[str, Any] @@ -26,14 +31,6 @@ class StorageTypeProfile: writable: bool -@dataclass(frozen=True) -class JobTypeProfile: - id: str - description: str - parameters_schema: JsonObject - default_timeout_seconds: int - - @dataclass(frozen=True) class StorageProfile: id: int @@ -132,250 +129,6 @@ def _boolean(default: bool) -> JsonObject: ) -SOURCE_COLLECT_PARAMETERS_SCHEMA = { - "additionalProperties": False, - "description": "Parameters of the exact ordinary Source Job type.", - "properties": { - "config": { - "additionalProperties": True, - "title": "Config", - "type": "object", - }, - "source": {"title": "Source", "type": "integer"}, - }, - "required": ["source"], - "title": "SourceCollectParameters", - "type": "object", -} - -SOURCE_BACKFILL_PARAMETERS_SCHEMA = { - **SOURCE_COLLECT_PARAMETERS_SCHEMA, - "description": "Parameters of the exact historical Source Job type.", - "title": "SourceBackfillParameters", -} - -LEXICAL_MAINTENANCE_PARAMETERS_SCHEMA = { - "$defs": { - "LexicalMaintenanceOptions": { - "additionalProperties": False, - "properties": { - "diagnostic_limit": { - "default": 20, - "minimum": 0, - "title": "Diagnostic Limit", - "type": "integer", - }, - "max_records": { - "default": 100, - "minimum": 1, - "title": "Max Records", - "type": "integer", - }, - "scan_page_size": { - "default": 100, - "minimum": 1, - "title": "Scan Page Size", - "type": "integer", - }, - }, - "title": "LexicalMaintenanceOptions", - "type": "object", - } - }, - "additionalProperties": False, - "properties": { - "options": {"$ref": "#/$defs/LexicalMaintenanceOptions"}, - }, - "title": "LexicalMaintenanceJobParameters", - "type": "object", -} - -SEMANTIC_MAINTENANCE_PARAMETERS_SCHEMA = { - "$defs": { - "EmbeddingMaintenanceOptions": { - "additionalProperties": False, - "description": "Peer-local bounds for one resumable maintenance invocation.", - "properties": { - "batch_size": { - "default": 20, - "minimum": 1, - "title": "Batch Size", - "type": "integer", - }, - "diagnostic_limit": { - "default": 20, - "minimum": 0, - "title": "Diagnostic Limit", - "type": "integer", - }, - "max_embeddings": { - "default": 100, - "minimum": 1, - "title": "Max Embeddings", - "type": "integer", - }, - "scan_page_size": { - "default": 100, - "minimum": 1, - "title": "Scan Page Size", - "type": "integer", - }, - }, - "title": "EmbeddingMaintenanceOptions", - "type": "object", - } - }, - "additionalProperties": False, - "properties": { - "options": {"$ref": "#/$defs/EmbeddingMaintenanceOptions"}, - "profile": { - "anyOf": [{"type": "integer"}, {"type": "null"}], - "default": None, - "title": "Profile", - }, - }, - "title": "EmbeddingMaintenanceJobParameters", - "type": "object", -} - -MEDIA_INTERPRETATION_PARAMETERS_SCHEMA = { - "additionalProperties": False, - "properties": {}, - "title": "MediaInterpretationJobParameters", - "type": "object", -} - -AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA = { - "additionalProperties": False, - "properties": { - "max_seeds": { - "default": 10, - "maximum": 100, - "minimum": 3, - "title": "Max Seeds", - "type": "integer", - } - }, - "title": "AutomaticOrganizationJobParameters", - "type": "object", -} - -BUILTIN_JOB_TYPES = ( - JobTypeProfile( - "core.source.collect.v1", - "Run one ordinary collection command for a configured Source.", - SOURCE_COLLECT_PARAMETERS_SCHEMA, - 300, - ), - JobTypeProfile( - "core.source.backfill.v1", - "Run one exact historical collection command for a configured Source.", - SOURCE_BACKFILL_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.feature_retrieval.lexical.maintain.v1", - "Maintain missing or stale Block lexical projection records.", - LEXICAL_MAINTENANCE_PARAMETERS_SCHEMA, - 900, - ), - JobTypeProfile( - "core.feature_retrieval.lexical.rebuild.v1", - "Rebuild Block lexical records present before invocation.", - LEXICAL_MAINTENANCE_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.semantic_retrieval.maintain.v1", - "Maintain missing or stale semantic embedding records.", - SEMANTIC_MAINTENANCE_PARAMETERS_SCHEMA, - 900, - ), - JobTypeProfile( - "core.semantic_retrieval.rebuild.v1", - "Rebuild semantic embedding records present before invocation.", - SEMANTIC_MAINTENANCE_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.media_interpretation.v1", - "Interpret missing image, audio, and video Blocks through configured Agents.", - MEDIA_INTERPRETATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.rumination.explicit.v1", - "Reconsider one explicitly selected Block through rumination.", - { - "additionalProperties": False, - "description": "Explicit request to reconsider one focal Block.", - "properties": {"block": {"title": "Block", "type": "integer"}}, - "required": ["block"], - "title": "RuminationRequest", - "type": "object", - }, - 1800, - ), - JobTypeProfile( - "core.organization.rumination.automatic.v1", - "Automatically reconsider bounded information seeds through rumination.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.supersession.automatic.v1", - "Automatically judge bounded scoped-supersession candidates.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.refinement.automatic.v1", - "Automatically judge bounded non-dominating refinement candidates.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.evidence-stance.automatic.v1", - "Automatically judge bounded attributable evidence stances.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.synthesis.automatic.v1", - "Automatically create bounded provenance-preserving syntheses.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.existing-referent-anchoring.automatic.v1", - "Automatically anchor bounded mentions to existing referents.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.organization.duplicate-assertion.automatic.v1", - "Automatically judge bounded provenance-aware duplicate assertions.", - AUTOMATIC_ORGANIZATION_PARAMETERS_SCHEMA, - 1800, - ), - JobTypeProfile( - "core.sink.agent-query.v1", - "Answer one information need through an enabled Agent Query Sink.", - { - "additionalProperties": False, - "properties": { - "query": {"minLength": 1, "title": "Query", "type": "string"}, - "sink": {"title": "Sink", "type": "integer"}, - }, - "required": ["sink", "query"], - "title": "AgentQueryJobParameters", - "type": "object", - }, - 300, - ), -) - - BUILTIN_STORAGES = ( StorageProfile(-1, "http", "HTTP", {}), StorageProfile(-4, "postgresql_binary", "PostgreSQL Binary", {}), @@ -383,4 +136,3 @@ def _boolean(default: bool) -> JsonObject: BUILTIN_AI_DIALECTS_BY_ID = {item.id: item for item in BUILTIN_AI_DIALECTS} BUILTIN_STORAGE_TYPES_BY_ID = {item.id: item for item in BUILTIN_STORAGE_TYPES} -BUILTIN_JOB_TYPES_BY_ID = {item.id: item for item in BUILTIN_JOB_TYPES} diff --git a/app/schemas/lexical_retrieval.py b/app/schemas/lexical_retrieval.py index 60b6576b..f10f12ee 100644 --- a/app/schemas/lexical_retrieval.py +++ b/app/schemas/lexical_retrieval.py @@ -1,5 +1,10 @@ """Block-local lexical projection, maintenance, and ranked result contracts.""" +from app.contracts.jobs import ( + LexicalMaintenanceOptions as LexicalMaintenanceOptions, + LexicalMaintenanceJobParameters as LexicalMaintenanceJobParameters, +) + import datetime import math import typing @@ -117,14 +122,6 @@ class LexicalRetrievalResult(pydantic.BaseModel): matches: tuple[LexicalRetrievalMatch, ...] -class LexicalMaintenanceOptions(pydantic.BaseModel): - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - max_records: int = pydantic.Field(default=100, ge=1) - scan_page_size: int = pydantic.Field(default=100, ge=1) - diagnostic_limit: int = pydantic.Field(default=20, ge=0) - - class LexicalMaintenanceDiagnostic(pydantic.BaseModel): model_config = pydantic.ConfigDict(extra="forbid", frozen=True) @@ -140,11 +137,3 @@ class LexicalMaintenanceReport(pydantic.BaseModel): unavailable: int = 0 failed: int = 0 diagnostics: tuple[LexicalMaintenanceDiagnostic, ...] = () - - -class LexicalMaintenanceJobParameters(pydantic.BaseModel): - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - options: LexicalMaintenanceOptions = pydantic.Field( - default_factory=LexicalMaintenanceOptions - ) diff --git a/app/schemas/organization.py b/app/schemas/organization.py index e1183482..c59e5576 100644 --- a/app/schemas/organization.py +++ b/app/schemas/organization.py @@ -1,11 +1,15 @@ """Organization rumination configuration, inbound, and Agent Tool forms.""" +from app.contracts.jobs import ( + MediaInterpretationJobParameters as MediaInterpretationJobParameters, + RuminationRequest as RuminationRequest, +) + import typing import pydantic from app.schemas.ai import JSONValue -from app.schemas.info_base.block import BlockID from app.schemas.info_base.main import GraphForm, NegativeBlockID @@ -27,10 +31,6 @@ class MediaInterpretationConfig(pydantic.BaseModel): video_agent: int -class MediaInterpretationJobParameters(pydantic.BaseModel): - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - class MediaInterpretationDiagnostic(pydantic.BaseModel): model_config = pydantic.ConfigDict(extra="forbid", frozen=True) @@ -51,14 +51,6 @@ class MediaInterpretationReport(pydantic.BaseModel): diagnostics: tuple[MediaInterpretationDiagnostic, ...] = () -class RuminationRequest(pydantic.BaseModel): - """Explicit request to reconsider one focal Block.""" - - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - block: BlockID - - class GetDraftGraphSchemaInput(pydantic.BaseModel): """Generic schema-discovery input narrowed when an Agent run binds Tools.""" diff --git a/app/schemas/organization_behavior.py b/app/schemas/organization_behavior.py index 43915bf9..b58f6d8e 100644 --- a/app/schemas/organization_behavior.py +++ b/app/schemas/organization_behavior.py @@ -1,5 +1,9 @@ """Contracts for exact Organization behaviors and their Agent adapters.""" +from app.contracts.jobs import ( + AutomaticOrganizationJobParameters as AutomaticOrganizationJobParameters, +) + import typing import pydantic @@ -19,12 +23,6 @@ class BehaviorAgentConfig(pydantic.BaseModel): agent: int -class AutomaticOrganizationJobParameters(pydantic.BaseModel): - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - max_seeds: int = pydantic.Field(default=10, ge=3, le=100) - - class RelationWriteResult(pydantic.BaseModel): model_config = pydantic.ConfigDict(extra="forbid", frozen=True) diff --git a/app/schemas/semantic_retrieval.py b/app/schemas/semantic_retrieval.py index 5b59e868..e84a52c9 100644 --- a/app/schemas/semantic_retrieval.py +++ b/app/schemas/semantic_retrieval.py @@ -1,5 +1,10 @@ """Semantic retrieval configuration, maintenance and ranked result contracts.""" +from app.contracts.jobs import ( + EmbeddingMaintenanceOptions as EmbeddingMaintenanceOptions, + EmbeddingMaintenanceJobParameters as EmbeddingMaintenanceJobParameters, +) + import math import typing @@ -90,17 +95,6 @@ class SemanticRetrievalResult(pydantic.BaseModel): matches: tuple[SemanticRetrievalMatch, ...] -class EmbeddingMaintenanceOptions(pydantic.BaseModel): - """Peer-local bounds for one resumable maintenance invocation.""" - - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - max_embeddings: int = pydantic.Field(default=100, ge=1) - batch_size: int = pydantic.Field(default=20, ge=1) - scan_page_size: int = pydantic.Field(default=100, ge=1) - diagnostic_limit: int = pydantic.Field(default=20, ge=0) - - class EmbeddingMaintenanceDiagnostic(pydantic.BaseModel): model_config = pydantic.ConfigDict(extra="forbid", frozen=True) @@ -118,12 +112,3 @@ class EmbeddingMaintenanceReport(pydantic.BaseModel): unavailable: int = 0 failed: int = 0 diagnostics: tuple[EmbeddingMaintenanceDiagnostic, ...] = () - - -class EmbeddingMaintenanceJobParameters(pydantic.BaseModel): - model_config = pydantic.ConfigDict(extra="forbid", frozen=True) - - profile: EmbeddingProfileID | None = None - options: EmbeddingMaintenanceOptions = pydantic.Field( - default_factory=EmbeddingMaintenanceOptions - ) diff --git a/app/schemas/source/main.py b/app/schemas/source/main.py index 04792aaf..17114f51 100644 --- a/app/schemas/source/main.py +++ b/app/schemas/source/main.py @@ -1,3 +1,8 @@ +from app.contracts.jobs import ( + SourceCollectParameters as SourceCollectParameters, + SourceBackfillParameters as SourceBackfillParameters, +) + __all__ = [ "SourceID", "SourceModel", @@ -63,26 +68,13 @@ def process_result_value(self, value, dialect): return CollectAt.model_validate(value) if value else None -class SourceCollectParameters(pydantic.BaseModel): - """Parameters of the exact ordinary Source Job type.""" - - model_config = pydantic.ConfigDict(extra="forbid") - - source: SourceID - config: dict[str, typing.Any] = pydantic.Field(default_factory=dict) - - -class SourceBackfillParameters(SourceCollectParameters): - """Parameters of the exact historical Source Job type.""" - - class SourceTypesModel(sqlmodel.SQLModel, table=True): __tablename__: str = "sources_types" # type: ignore id: str = sqlmodel.Field(sa_column=sqlalchemy.Column(sqlalchemy.Text, primary_key=True)) """Type of source. - An absolute import path to the module where souce class at. + Stable execution identity. Historical values retain their original Python path. """ description: str = sqlmodel.Field(sa_column=sqlalchemy.Column(sqlalchemy.Text)) """Description of this source type.""" @@ -126,7 +118,7 @@ class SourceModel(sqlmodel.SQLModel, table=True): ) """Type of source. - An absolute import path to the module where souce class at. + Stable execution identity, independent of the implementing class location. When delete a source type, all sources of this type will be deleted too. """ nickname: Opt[str] = sqlmodel.Field( diff --git a/docs/30-unit-tdd/business-pipeline-and-authority.md b/docs/30-unit-tdd/business-pipeline-and-authority.md index caab6cf9..11c3dd25 100644 --- a/docs/30-unit-tdd/business-pipeline-and-authority.md +++ b/docs/30-unit-tdd/business-pipeline-and-authority.md @@ -46,6 +46,8 @@ implementation direction; it must not redefine Peer wire behavior or shared capa 例子，它不是 `SourceBase` instance，也不因此获得单独的 memo object store。 - source 或 protocol adapter 负责 native shape 与 extension-owned canonical command 之间的映射， 但不是 persistence owner。 +- Source class 显式 `source_type` 声明持久身份，第一方继续使用已发布的 path 字符串；模块位置和类名 + 变化不再改变身份。未声明的旧 wheel 保持原 path 注册。相同 class 重注册幂等，冲突 class 明确失败。 ### 3. Info-Base Owns Graph Persistence And Cross-Retrieval Reads @@ -97,6 +99,12 @@ persistence。Job handler 和外部 Sink lifecycle 在提交后的作用域外 共享 CronUnitOfWork，任一步失败回滚完整用例；不同 Cron 独立处理。Job 的取消／超时收尾另开短事务， 只更新仍为 RUNNING 的记录，并限制清理等待时间。执行资格检查的 Source、Agent、AI 与配置读取也可 await。 +内置 Job 的输入模型、exact ID、描述和默认 timeout 统一由 `app/contracts/jobs.py` 声明；该模块只依赖 +Pydantic，不加载 ORM、settings、FastAPI 或 scheduler。数据库 catalog profile 和已注册 builtin handler +消费同一合同，旧 schema import 入口保留 re-export。Handler 不得覆盖 builtin 元数据；Source/Sink 等 +需要读取领域记录的 admission normalization、`can_handle` 与执行效果仍属于原 owning unit。自定义 +Job 继续在 handler 声明自己的输入模型和元数据，不要求加入 builtin catalog。 + GitHub、RSS、Mail、Telegram、Twitter 和 Memos 的采集与物化已异步化。扩展的协调代码通过必需的 Graph/Source UoW 组合 Core persistence，不创建 session。Twitter page graph 与 Source cursor 同事务； RSS/Mail 保持逐 item/occurrence 的既有 partial effects，Memos 主删除后的 best-effort cleanup 仍独立提交。 diff --git a/docs/40-deployment/development-environment.md b/docs/40-deployment/development-environment.md index 42dcbd34..bfd2588b 100644 --- a/docs/40-deployment/development-environment.md +++ b/docs/40-deployment/development-environment.md @@ -103,7 +103,7 @@ pdm run dev repository contract checks tool versions, lock/export consistency, migration integrity, formatting, lint, settings isolation, and the complete unit-test suite. -格式与 lint 的范围由 `ruff.toml` 拥有，覆盖正式源码、测试与脚本。`tasks/` 是临时实验材料，`.agents/` 是独立维护的技能；二者不参与产品门禁。成为正式验证步骤的实验应迁入 `scripts/` 或 `tests/`。已发布迁移保持不可变；独立 CLI 使用 `pdm run --project cli check` 及自己的 CI。pre-commit 的格式与 lint 钩子复用 `pdm run format:check` 和 `pdm run lint`，CI 的完整入口仍为 `pdm run check`。 +格式与 lint 的范围由 `ruff.toml` 拥有，覆盖正式源码、测试与脚本。`tasks/` 是临时实验材料，`.agents/` 是独立维护的技能；二者不参与产品门禁。成为正式验证步骤的实验应迁入 `scripts/` 或 `tests/`。已发布迁移保持不可变；独立 CLI 使用 `pdm run --project cli check` 及自己的 CI。pre-commit 的格式、lint、类型、锁文件和迁移钩子均复用 `pdm run` 中的对应检查，锁定依赖已安装后，普通 `git commit` 不要求先激活虚拟环境。检查参数由 `pyproject.toml` 中的命令拥有；CI 的完整入口仍为 `pdm run check`。 Use narrower read-only checks while iterating: @@ -163,6 +163,10 @@ authority. 这保留实际 migration 编排和应用下载读回，不会回退当前业务测试库或声称验证了整条历史 downgrade 链。 CI 在自身临时 PostgreSQL service 上先初始化再执行这一组；外部服务和真实凭据验收继续由各自专项入口负责。 +扩展持久化的定向入口是 `pdm run test:extensions:database`，由上述完整数据库入口调用同一份场景清单：Memos 附件图、RSS 条件采集、Mail 并发 checkpoint、Twitter archive HTTP/图与 full traversal，以及 CalDAV 原生图/事务/同步恢复。它们使用真实 PostgreSQL 和受控提供方输入，不需要 IMAP、Twitter 或 CalDAV 外部账号。可附加 pytest 参数，例如 `pdm run test:extensions:database -k full_bookmarks`。改变这些合同后应使用这个入口；只运行默认 `pdm run check`，其数据库 skip 不代表这些行为已验证。 + +该清单不自动接纳所有历史测试。需要真实账号、指定语料库或独立服务的 Mail vertical/newsletter、live RSS/Atom 和语义质量验收仍使用所属领域的专项入口。增加或移除持续回归时修改 `scripts/automation/repository_check.sh` 的唯一选择入口，说明其公开行为与所需资源；本地和 CI 不再各维护一份扩展路径清单。 + ## Database Branch Workflow In CI The repository currently uses Neon branch automation for pull requests. diff --git a/docs/40-deployment/runtime-orchestration.md b/docs/40-deployment/runtime-orchestration.md index a13a85ad..6649a3c1 100644 --- a/docs/40-deployment/runtime-orchestration.md +++ b/docs/40-deployment/runtime-orchestration.md @@ -74,6 +74,9 @@ same database-owned config semantics through its runtime owner. ### 4. Shutdown must close long-lived runtime resources +`run.py` 的 `drain_runtime_producers()` 是退出和环境应用共用的生产者排空顺序所有者； +Extension 的 active effects 由 Host 自己关闭，不把 Host 锁持有规则暴露给 bootstrap。 + - scheduler admission is paused first; JobManager stops admitting work, cancels active handlers and awaits their terminal database writes without cancelling a close already in progress - remaining tracked callbacks are cancelled and awaited before Sink/Extension resources close - running Sink instances close before Extension teardown, so an external endpoint cannot observe disappearing @@ -90,6 +93,10 @@ validates the wheel while the Peer can still serve work. It then closes admissio pauses the scheduler, drains Jobs, and closes Sinks and running Extensions before applying the wheel to the shared Core interpreter. Python dependencies may change, so Core does not resume in the same process or hot-load the new module. +The pause callback drains requests and Core producers only. The Host closes its running Extensions after that callback +returns, while still owning its runtime lock, and only then applies the wheel. Startup failure, ordinary stop, and +intent-conflict cleanup share the active-effect release order while retaining their different error policies. + The production entrypoint `scripts/container.py web` first waits for the read-only database contract check, then reads this Peer’s enabled exact selections and restores any missing Python distributions through the same Registry resolver, wheel validator, and consumer used by `select`. This pre-start step never changes the selected version or enabled intent, diff --git a/extensions/github/.changes/source-identity.changed.md b/extensions/github/.changes/source-identity.changed.md new file mode 100644 index 00000000..8cdd4f16 --- /dev/null +++ b/extensions/github/.changes/source-identity.changed.md @@ -0,0 +1 @@ +显式保留 GitHub Stars 的已发布 Source 类型身份，后续移动实现时可继续使用现有 Source 配置。 diff --git a/extensions/github/stars.py b/extensions/github/stars.py index 15d97dc0..d33709ec 100644 --- a/extensions/github/stars.py +++ b/extensions/github/stars.py @@ -13,9 +13,14 @@ from .schema import GitHubSourceConfig -class Source(SourceBase[GitHubSourceConfig], config_cls=GitHubSourceConfig): +class Source( + SourceBase[GitHubSourceConfig], + config_cls=GitHubSourceConfig, +): """Synchronize the authenticated GitHub Account's Stars and Lists.""" + source_type = "extensions.github.stars.Source" + async def collect(self, job: JobModel, config: pydantic.BaseModel) -> None: del config source_config = await self.get_config() diff --git a/extensions/mail/.changes/source-identity.changed.md b/extensions/mail/.changes/source-identity.changed.md new file mode 100644 index 00000000..d45e6b50 --- /dev/null +++ b/extensions/mail/.changes/source-identity.changed.md @@ -0,0 +1 @@ +显式保留 Mail 的已发布 Source 类型身份，后续移动实现时可继续使用现有 Source 配置。 diff --git a/extensions/mail/source.py b/extensions/mail/source.py index 52a1343f..fddcad78 100644 --- a/extensions/mail/source.py +++ b/extensions/mail/source.py @@ -54,6 +54,8 @@ class Source( ): """Collect Mail through one configured public protocol access context.""" + source_type = "extensions.mail.source.Source" + async def collect(self, job: JobModel, config: pydantic.BaseModel) -> None: source, setup = await self._load_effective_source() diagnostics: list[dict[str, typing.Any]] = [] diff --git a/extensions/rss/.changes/source-identity.changed.md b/extensions/rss/.changes/source-identity.changed.md new file mode 100644 index 00000000..c6137ba4 --- /dev/null +++ b/extensions/rss/.changes/source-identity.changed.md @@ -0,0 +1 @@ +显式保留 RSS 与 Atom 的已发布 Source 类型身份，后续移动实现时可继续使用现有 Source 配置。 diff --git a/extensions/rss/atom.py b/extensions/rss/atom.py index 4d68ce61..e98d8cdb 100644 --- a/extensions/rss/atom.py +++ b/extensions/rss/atom.py @@ -17,6 +17,8 @@ class Source( ): """Collect one configured Atom feed through the shared feed service.""" + source_type = "extensions.rss.atom.Source" + expected_family = "atom" diff --git a/extensions/rss/rss.py b/extensions/rss/rss.py index 8cef14cb..def814a2 100644 --- a/extensions/rss/rss.py +++ b/extensions/rss/rss.py @@ -17,6 +17,8 @@ class Source( ): """Collect one configured RSS 2.0 feed through the shared feed service.""" + source_type = "extensions.rss.rss.Source" + expected_family = "rss" diff --git a/extensions/telegram/.changes/source-identity.changed.md b/extensions/telegram/.changes/source-identity.changed.md new file mode 100644 index 00000000..2df1b37a --- /dev/null +++ b/extensions/telegram/.changes/source-identity.changed.md @@ -0,0 +1 @@ +显式保留 Telegram 的已发布 Source 类型身份，后续移动实现时可继续使用现有 Source 配置。 diff --git a/extensions/telegram/source.py b/extensions/telegram/source.py index d042a923..bd4927d1 100644 --- a/extensions/telegram/source.py +++ b/extensions/telegram/source.py @@ -94,9 +94,14 @@ def _semantic_text(message: telegram.Message, *, caption: bool = False) -> Block return BlockForm(resolver="core.text.v1", content=value) -class Source(SourceBase[TelegramSourceConfig], config_cls=TelegramSourceConfig): +class Source( + SourceBase[TelegramSourceConfig], + config_cls=TelegramSourceConfig, +): """Collect useful content sent privately by one configured Telegram identity.""" + source_type = "extensions.telegram.source.Source" + async def _pin_bot(self, bot_id: int) -> TelegramSourceState: async with source_uow() as uow: source = await uow.sources.get(self._id, lock=True) diff --git a/extensions/twitter/.changes/source-identity.changed.md b/extensions/twitter/.changes/source-identity.changed.md new file mode 100644 index 00000000..8c45d57d --- /dev/null +++ b/extensions/twitter/.changes/source-identity.changed.md @@ -0,0 +1 @@ +显式保留 Twitter Bookmark 的已发布 Source 类型身份，后续移动实现时可继续使用现有 Source 配置。 diff --git a/extensions/twitter/bookmark.py b/extensions/twitter/bookmark.py index 68be43f1..7c3f38fd 100644 --- a/extensions/twitter/bookmark.py +++ b/extensions/twitter/bookmark.py @@ -61,6 +61,8 @@ class Source( ): """Twitter Bookmark as Source""" + source_type = "extensions.twitter.bookmark.Source" + API_BASE_URL = "https://api.x.com/2" async def collect(self, job: JobModel, config: pydantic.BaseModel) -> None: diff --git a/pyproject.toml b/pyproject.toml index 44f6581b..f237f29f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -96,6 +96,7 @@ format = "ruff format ." typecheck = "pyrefly check --min-severity=warn --output-format=min-text --progress-bar=no" test = "python -m pytest -q" "test:database" = "bash scripts/automation/repository_check.sh database-tests" +"test:extensions:database" = "bash scripts/automation/repository_check.sh extension-database-tests" "test:acceptance" = "python -m pytest -q -m acceptance tests/semantic_retrieval/acceptance" "test:organization-acceptance" = "python -m pytest -q -s tests/organization/acceptance/test_black_box.py" "check:lock" = "python scripts/check_lock.py" diff --git a/run.py b/run.py index 69daf633..24aa4873 100644 --- a/run.py +++ b/run.py @@ -104,17 +104,21 @@ def _request_uvicorn_exit() -> None: async def pause_peer_execution() -> None: - """Drain Core work before a shared Python environment can be changed.""" + """Drain Core producers; the Host then closes its own Extension resources.""" global _PEER_EXECUTION_OPEN async with _REQUEST_CONDITION: _PEER_EXECUTION_OPEN = False await _REQUEST_CONDITION.wait_for(lambda: _ACTIVE_PEER_REQUESTS == 0) + await drain_runtime_producers() + + +async def drain_runtime_producers() -> None: + """Stop work in dependency order before either environment application or exit.""" if scheduler.running: scheduler.pause() await JobManager.shutdown() await drain_scheduler() await SinkManager.shutdown() - await EXTENSION_HOST.close_running_for_environment_application() EXTENSION_HOST.pause_peer_execution = pause_peer_execution @@ -268,11 +272,7 @@ async def lifespan(app: fastapi.FastAPI): bootstrap_task.cancel() with contextlib.suppress(asyncio.CancelledError): await bootstrap_task - if scheduler.running: - scheduler.pause() - await JobManager.shutdown() - await drain_scheduler() - await SinkManager.shutdown() + await drain_runtime_producers() await EXTENSION_HOST.close_running() if scheduler.running: scheduler.shutdown(wait=True) diff --git a/scripts/automation/repository_check.sh b/scripts/automation/repository_check.sh index 1681af1e..bba4300b 100644 --- a/scripts/automation/repository_check.sh +++ b/scripts/automation/repository_check.sh @@ -2,6 +2,18 @@ source "$(dirname "$0")/_common.sh" +extension_database_tests() { + require_env INKCRE_TEST_DATABASE_URL + # Real persistence and protocol journeys; provider inputs are local fixtures. + pdm run python -m pytest -q -rs \ + tests/extensions/memos/integration \ + tests/extensions/rss/integration/test_feed_collection.py \ + tests/extensions/mail/test_source_state.py \ + scripts/tests/twitter/integration \ + scripts/tests/caldav/test_dav_graph.py scripts/tests/caldav/test_semantic_graph.py \ + "$@" +} + case "${1:-}" in database-tests) require_env INKCRE_TEST_DATABASE_URL @@ -10,8 +22,11 @@ case "${1:-}" in tests/agent/integration tests/ai/integration tests/deployment_config/integration \ tests/info_base/integration tests/lexical_retrieval/integration \ tests/organization/integration tests/peer/integration tests/semantic_retrieval/integration \ - tests/extensions/memos/integration tests/extensions/rss/integration/test_feed_collection.py \ tests/migrations/test_memos_attachment_v2_migration.py + extension_database_tests + ;; + extension-database-tests) + extension_database_tests "${@:2}" ;; release-intent) arguments=() @@ -35,5 +50,5 @@ case "${1:-}" in --project "extensions/$extension" --wheel "$wheel" done ;; - *) echo "usage: $0 database-tests|release-intent|extension-wheels" >&2; exit 2 ;; + *) echo "usage: $0 database-tests|extension-database-tests|release-intent|extension-wheels" >&2; exit 2 ;; esac diff --git a/scripts/tests/caldav/test_dav_graph.py b/scripts/tests/caldav/test_dav_graph.py index 82f99a79..aab7bbf5 100644 --- a/scripts/tests/caldav/test_dav_graph.py +++ b/scripts/tests/caldav/test_dav_graph.py @@ -7,15 +7,20 @@ import xml.etree.ElementTree as ET import pytest +import sqlalchemy +import sqlmodel from sqlalchemy import event from sqlalchemy.orm import Session from app.engine import ASYNC_DB_ENGINE +from app.schemas.info_base.block import BlockModel +from app.schemas.info_base.storage import StorageBlobModel from app.persistence.info_base.uow import graph_uow from app.business.info_base.resolver import ResolverManager from extensions.caldav import Extension from extensions.caldav.schema import CALENDAR, GraphContent from tests.extensions.runtime_support import publish_extension +from tests.database import TestSession pytestmark = pytest.mark.skipif( not os.getenv("INKCRE_TEST_DATABASE_URL"), @@ -27,17 +32,45 @@ @pytest.fixture def dav(): - published = publish_extension( - Extension, {"password": PASSWORD}, persist_config=True, raise_server_exceptions=False - ) - with published.client as client: - client.auth = ("inkcre", PASSWORD) - try: - yield client - finally: - assert client.portal is not None - client.portal.call(ASYNC_DB_ENGINE.dispose) - published.unpublish() + # DAV deletion deliberately retains shared graph content. The test fixture + # owns its newly created graph/blob assets even after their DAV publication + # disappears; remove those assets without touching the pre-existing graph. + with TestSession() as session: + existing_blocks = set(session.exec(sqlmodel.select(BlockModel.id)).all()) + existing_blobs = set(session.exec(sqlmodel.select(StorageBlobModel.id)).all()) + try: + published = publish_extension( + Extension, {"password": PASSWORD}, persist_config=True, raise_server_exceptions=False + ) + with published.client as client: + client.auth = ("inkcre", PASSWORD) + try: + yield client + finally: + assert client.portal is not None + client.portal.call(ASYNC_DB_ENGINE.dispose) + published.unpublish() + finally: + with TestSession() as session: + created_blocks = ( + set(session.exec(sqlmodel.select(BlockModel.id)).all()) - existing_blocks + ) + created_blobs = ( + set(session.exec(sqlmodel.select(StorageBlobModel.id)).all()) - existing_blobs + ) + if created_blocks: + session.connection().execute( + sqlalchemy.delete(BlockModel).where( + sqlmodel.col(BlockModel.id).in_(created_blocks) + ) + ) + if created_blobs: + session.connection().execute( + sqlalchemy.delete(StorageBlobModel).where( + sqlmodel.col(StorageBlobModel.id).in_(created_blobs) + ) + ) + session.commit() def calendar(summary="InKCre 测试"): diff --git a/tests/test_builtin_job_contracts.py b/tests/test_builtin_job_contracts.py new file mode 100644 index 00000000..336f06c6 --- /dev/null +++ b/tests/test_builtin_job_contracts.py @@ -0,0 +1,24 @@ +"""The artifact catalog remains usable without importing an application runtime.""" + +import subprocess +import sys + + +def test_catalog_import_needs_no_runtime_environment(): + result = subprocess.run( + [ + sys.executable, + "-c", + """ +import sys +from app.database_contract.profile import BUILTIN_JOB_TYPES +assert BUILTIN_JOB_TYPES +assert not {'sqlmodel', 'sqlalchemy', 'fastapi', 'app.settings', 'app.engine', + 'app.scheduler'} & sys.modules.keys() +""", + ], + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr diff --git a/tests/test_extension_participation_lifecycle_races.py b/tests/test_extension_participation_lifecycle_races.py index 2bb5b7db..53e78dbc 100644 --- a/tests/test_extension_participation_lifecycle_races.py +++ b/tests/test_extension_participation_lifecycle_races.py @@ -7,7 +7,7 @@ import pytest from app.business.extension.errors import ExtensionStateConflictError -from app.business.extension.main import ExtensionHost +from app.business.extension.main import ActiveExtensionEffects, ExtensionHost from app.business.extension.state import ( ExtensionParticipation, ExtensionStore, @@ -96,6 +96,9 @@ def test_late_startup_candidate_is_discarded_after_intent_is_disabled( "version": "1.2.3", "extension_class": candidate, "claim": claim, + "effects": ActiveExtensionEffects( + typing.cast(typing.Any, claim), typing.cast(typing.Any, candidate) + ), }, )() start_observed_enabled = False diff --git a/tests/test_extension_startup_provision.py b/tests/test_extension_startup_provision.py index dc016de9..e0702ee1 100644 --- a/tests/test_extension_startup_provision.py +++ b/tests/test_extension_startup_provision.py @@ -693,3 +693,103 @@ async def scenario(): async_runner.run(scenario()) assert restarted == [True] + + +@pytest.mark.parametrize("operation", ["startup_failure", "disable", "intent_conflict"]) +def test_runtime_effect_cleanup_preserves_each_operation_error_policy( + async_runner, monkeypatch, operation +): + name = "inkcre/cleanup-policy" + descriptor = release(name, "1.0.0") + store = SelectionStore(participation(name, "1.0.0", enabled=False)) + resources = {"bound": False, "published": False, "runtime_owned": True} + fail_close = True + + class Extension: + __extid__ = "cleanup_policy" + + @classmethod + def bind(cls, model): + resources["bound"] = True + + @classmethod + async def on_start_async(cls, app): + resources["published"] = True + if operation == "startup_failure": + raise ValueError("startup error") + if operation == "intent_conflict": + store.selections[0] = store.selections[0].model_copy(update={"enabled": False}) + + @classmethod + async def on_close(cls): + if fail_close: + raise RuntimeError("close error") + + @classmethod + def unpublish(cls): + resources["published"] = False + + @classmethod + def unbind(cls): + resources["bound"] = False + + @classmethod + def release_runtime(cls): + resources["runtime_owned"] = False + + @classmethod + def peer_inbounds(cls): + return () + + class Modules: + def __init__(self, acquired): + pass + + def load(self, base): + return Extension + + def assert_origins(self): + pass + + host = ExtensionHost( + store=typing.cast(ExtensionStore, store), + release_client=ReleaseReader({(name, "1.0.0"): descriptor}), + distribution_consumer=DistributionConsumer(), + ) + monkeypatch.setattr(extension_main, "DistributionModules", Modules) + monkeypatch.setattr( + extension_main.AcquiredDistribution, + "discover", + lambda *_: acquired_distribution(descriptor), + ) + + async def scenario(): + nonlocal fail_close + try: + if operation == "startup_failure": + with pytest.raises(ValueError, match="startup error"): + await host.enable(name, peer_id=PEER_ID, app=extension_main.fastapi.FastAPI()) + elif operation == "intent_conflict": + with pytest.raises(BaseExceptionGroup) as caught: + await host.enable(name, peer_id=PEER_ID, app=extension_main.fastapi.FastAPI()) + assert isinstance( + caught.value.exceptions[0], extension_main.ExtensionStateConflictError + ) + assert str(caught.value.exceptions[1]) == "close error" + else: + await host.enable(name, peer_id=PEER_ID, app=extension_main.fastapi.FastAPI()) + with pytest.raises(RuntimeError, match="close error"): + await host.disable(name, peer_id=PEER_ID) + assert name in host.running + assert store.selections[0].enabled + assert all(resources.values()) + return + assert not any(resources.values()) + assert not host.running + claim = extension_main.ExtensionRuntimeClaim.acquire("cleanup_policy") + claim.release() + finally: + fail_close = False + await host.close_running() + + async_runner.run(scenario()) diff --git a/tests/test_source_identity.py b/tests/test_source_identity.py new file mode 100644 index 00000000..1f0335fc --- /dev/null +++ b/tests/test_source_identity.py @@ -0,0 +1,78 @@ +"""Source code can move without changing persisted execution identity.""" + +import pydantic +import pytest + +from app.business.source import SourceBase, SourceManager +from app.business.source.job import SourceCollectJobHandler +from app.schemas.source import SourceModel, SourceCollectParameters +from app.schemas.job import JobModel + + +class EmptyConfig(pydantic.BaseModel): + pass + + +def test_explicit_identity_survives_class_move_and_executes_existing_source( + async_runner, monkeypatch +): + monkeypatch.setattr(SourceManager, "_SOURCE_CLASSES", {}) + monkeypatch.setattr(SourceManager, "SOURCES", {}) + received = [] + existing_type = "extensions.example.old_module.Source" + + class RenamedSource( + SourceBase[EmptyConfig], + config_cls=EmptyConfig, + ): + source_type = existing_type + + async def collect(self, job, config): + received.append(self._id) + + RenamedSource.__module__ = "extensions.example.new_module" + + async def get(source_id): + return SourceModel(id=source_id, type=existing_type, config={}) + + monkeypatch.setattr(SourceManager, "get", get) + SourceManager.add_source_type(RenamedSource) + async_runner.run( + SourceCollectJobHandler.handle( + JobModel(type=SourceCollectJobHandler.type, parameters={}), + SourceCollectParameters(source=17), + ) + ) + assert received == [17] + assert tuple(SourceManager._SOURCE_CLASSES) == (existing_type,) + + with pytest.raises(ValueError, match="already registered"): + + class ConflictingSource( + SourceBase[EmptyConfig], + config_cls=EmptyConfig, + ): + source_type = existing_type + + async def collect(self, job, config): + pass + + assert SourceManager._SOURCE_CLASSES[existing_type] is RenamedSource + + class ChildWithoutIdentity(RenamedSource, config_cls=EmptyConfig): + pass + + child_type = ChildWithoutIdentity.__module__ + "." + ChildWithoutIdentity.__qualname__ + assert SourceManager._SOURCE_CLASSES[child_type] is ChildWithoutIdentity + assert ChildWithoutIdentity.source_type != existing_type + + +def test_legacy_wheel_omitting_identity_keeps_its_original_registration(monkeypatch): + monkeypatch.setattr(SourceManager, "_SOURCE_CLASSES", {}) + + class LegacySource(SourceBase[EmptyConfig], config_cls=EmptyConfig): + async def collect(self, job, config): + pass + + expected = LegacySource.__module__ + "." + LegacySource.__qualname__ + assert SourceManager._SOURCE_CLASSES == {expected: LegacySource}