feat(runtime-core-py): 增加异步持久化与可取消启动 (#38)

* feat(runtime-core-py): 增加异步持久化和可取消启动

* docs(task): 关联 SDK 与 Core 迁移 PR

Author: xiaoland

.changes/runtime-core-py/0.1.4.md

@@ -0,0 +1,5 @@
+## 0.1.4 - 2026-09-17
+
+### Added
+
+- 新增配置、状态变更及启动的异步 Host 能力；复用 publication 作用域，在启动失败或取消时撤销本次资源，并保留旧同步 Host 入口。

docs/30-unit-tdd/core-python-runtime.md

@@ -32,3 +32,28 @@ published fresh.
 `DistributionModules` uses standard entry-point loading and verifies that every loaded module in the canonical
 `extensions.<name>` package resolves to a file owned by the acquired Distribution. It does not delete, restore or shadow
 `sys.modules`, and it does not create a second import cache.
+
+## 异步 Host 持久化能力
+
+Runtime 0.1.4 增加 `update_config_async`、`get_state_async`、`mutate_state_async`、
+`mutate_config_and_state_async` 和 `on_start_async`。它们只调用 Host model 对应的 async capability；
+不会调用旧同步方法、在线程中运行数据库代码，或创建数据库 session。旧同步入口继续供旧 Host 使用，
+Host 必须明确选择与其持久化模型一致的入口。现有 `ExtensionManager` 仍使用旧同步 Host 合同；
+这些新增入口供 Core 自己的 async Host adapter 采用，不代表旧 Manager 已整体异步化。
+
+Host model 需要提供 `update_config_async`、`read_state_async`、`mutate_state_async`、
+`mutate_config_and_state_async` 与 `update_config_schema_async`。配置和 schema 更新返回刷新后的
+Host model，状态读取返回 persisted JSON。变更方法必须在 Host 的事务内调用同步 transform，
+提交成功后才能返回；事务、行锁及并发控制仍由 Host 拥有。Runtime 恢复 typed 输入后保留 transform
+产出的 typed 结果，避免内部传递时再次恢复。`get_config()` 仍只恢复已绑定 model 的 config 投影。
+
+`on_start_async` 复用同步启动的 publication scope，先发布资源，再 await
+`SourceManager.sync_source_types_async()` 和 Host schema 持久化。普通 Source/Resolver class 注册
+仍是同步的进程内操作；Source catalog 写入必须 await。启动完成后才报告 `runtime_active()`；同一个
+Extension 启动期间再次启动会失败。任何失败或取消都会撤销本次 routes、Peer inbounds 和 public claims，
+随后允许重新启动。进程内已 import 的 decoder/type 注册继续保留，不模拟 module unload。
+
+这一变化不改变 Registry、installed、enabled 或 public-route 合同，也不让旧 wheel 自动兼容新的
+Core async-only API。采用方必须分别声明 Runtime 包版本与 Core Host SDK 版本窗口，并验证真实
+Host 的事务实现。Runtime 层的 scripted 验证可证明 await、typed round trip 和 publication 清理，
+不能替代 Core 对 PostgreSQL 提交、回滚及行锁的验证。

pdm.lock

@@ -3,6 +3,12 @@
 This changelog records notable changes to this independently released Python
 package. It is generated by [Changie](https://changie.dev/).
 
+## 0.1.4 - 2026-09-17
+
+### Added
+
+- 新增配置、状态变更及启动的异步 Host 能力；复用 publication 作用域，在启动失败或取消时撤销本次资源，并保留旧同步 Host 入口。
+
 ## 0.1.3 - 2026-09-14
 
 ### Fixed

runtimes/core-py/CHANGELOG.md

@@ -1,6 +1,6 @@
 [project]
 name = "inkcre-extension-runtime-core-py"
-version = "0.1.3"
+version = "0.1.4"
 description = "InKCre Core Python Extension Host Runtime"
 requires-python = ">=3.12,<3.14"
 dependencies = [

runtimes/core-py/pyproject.toml

@@ -3,11 +3,16 @@
 from __future__ import annotations
 
 import typing
+from collections.abc import Iterator
+from contextlib import contextmanager
 
 import pydantic
 
 from .errors import ExtensionLifecycleError, translate_host_model_error
 
+if typing.TYPE_CHECKING:
+    from .publication import ExtensionPublication
+
 
 class EmptyConfig(pydantic.BaseModel): ...
 
@@ -123,9 +128,102 @@ def mutate(
             translate_host_model_error(error)
         return typing.cast(tuple[ConfigT, StateT], result)
 
+    @classmethod
+    async def update_config_async(cls, value: dict[str, typing.Any] | ConfigT) -> ConfigT:
+        """Persist validated configuration through the Host async capability."""
+        config = (
+            value
+            if isinstance(value, cls.__configcls__)
+            else cls.__configcls__.model_validate(value)
+        )
+        try:
+            cls.__model__ = await cls._model().update_config_async(config.model_dump(mode="json"))
+        except Exception as error:
+            translate_host_model_error(error)
+        return config
+
+    @classmethod
+    async def get_state_async(cls) -> StateT:
+        """Read fresh state through the Host async capability."""
+        try:
+            state = await cls._model().read_state_async()
+        except Exception as error:
+            translate_host_model_error(error)
+        return cls.__statecls__.model_validate(state)
+
+    @classmethod
+    async def mutate_state_async(cls, transform: typing.Callable[[StateT], StateT]) -> StateT:
+        """Await one committed state mutation; transform runs synchronously under the Host lock."""
+        result: StateT | None = None
+
+        def mutate(raw: dict[str, typing.Any]) -> dict[str, typing.Any]:
+            nonlocal result
+            updated = transform(cls.__statecls__.model_validate(raw))
+            if not isinstance(updated, cls.__statecls__):
+                raise TypeError("Extension state transform returned the wrong model")
+            result = updated
+            return updated.model_dump(mode="json")
+
+        try:
+            await cls._model().mutate_state_async(mutate)
+        except Exception as error:
+            translate_host_model_error(error)
+        # The Host invokes the transform under its transaction and commits its
+        # returned JSON unchanged. Keep the typed result rather than revalidating it.
+        return typing.cast(StateT, result)
+
+    @classmethod
+    async def mutate_config_and_state_async(
+        cls, transform: typing.Callable[[ConfigT, StateT], tuple[ConfigT, StateT]]
+    ) -> tuple[ConfigT, StateT]:
+        """Await one atomic config/state mutation with a synchronous typed transform."""
+        result: tuple[ConfigT, StateT] | None = None
+
+        def mutate(
+            config: dict[str, typing.Any], state: dict[str, typing.Any]
+        ) -> tuple[dict[str, typing.Any], dict[str, typing.Any]]:
+            nonlocal result
+            new_config, new_state = transform(
+                cls.__configcls__.model_validate(config), cls.__statecls__.model_validate(state)
+            )
+            if not isinstance(new_config, cls.__configcls__) or not isinstance(
+                new_state, cls.__statecls__
+            ):
+                raise TypeError("Extension config/state transform returned the wrong models")
+            result = new_config, new_state
+            return new_config.model_dump(mode="json"), new_state.model_dump(mode="json")
+
+        try:
+            await cls._model().mutate_config_and_state_async(mutate)
+        except Exception as error:
+            translate_host_model_error(error)
+        return typing.cast(tuple[ConfigT, StateT], result)
+
     @classmethod
     def on_start(cls, app: typing.Any) -> None:
-        """Publish concrete Core contributions; config reads restore types on use."""
+        """Publish contributions using the synchronous Host persistence contract."""
+        with cls._publication_scope(app) as publication:
+            publication.activate_source_types()
+            try:
+                cls.__model__ = cls._model().update_config_schema(dict(cls.__configschema__))
+            except Exception as error:
+                translate_host_model_error(error)
+
+    @classmethod
+    async def on_start_async(cls, app: typing.Any) -> None:
+        """Publish contributions and await catalog/schema persistence before becoming active."""
+        with cls._publication_scope(app) as publication:
+            await publication.activate_source_types_async()
+            try:
+                cls.__model__ = await cls._model().update_config_schema_async(
+                    dict(cls.__configschema__)
+                )
+            except Exception as error:
+                translate_host_model_error(error)
+
+    @classmethod
+    @contextmanager
+    def _publication_scope(cls, app: typing.Any) -> Iterator[ExtensionPublication]:
         import fastapi
         from app.business.peer import PeerManager
 
@@ -134,9 +232,12 @@ def on_start(cls, app: typing.Any) -> None:
             PublicHTTPRouteClaim,
         )
 
-        if cls.runtime_active():
-            raise ExtensionLifecycleError(f"Extension {cls.__extid__} is already active")
+        if cls.runtime_active() or cls.__dict__.get("__runtime_starting__", False):
+            raise ExtensionLifecycleError(
+                f"Extension {cls.__extid__} is already active or starting"
+            )
         publication = ExtensionPublication(app)
+        cls.__runtime_starting__ = True
         try:
             router = fastapi.APIRouter(
                 prefix=f"/{cls.__extid__}", dependencies=cls.api_dependencies()
@@ -154,15 +255,15 @@ def on_start(cls, app: typing.Any) -> None:
             publication.public_http_claim = PublicHTTPRouteClaim.acquire(
                 cls.__extid__, cls.public_http_routes(), publication.routes
             )
-            publication.activate_source_types()
-            try:
-                cls.__model__ = cls._model().update_config_schema(dict(cls.__configschema__))
-            except Exception as error:
-                translate_host_model_error(error)
-        except Exception:
+            yield publication
+        except BaseException:
+            # Cancellation during Host persistence must revoke already-published effects.
             publication.withdraw()
             raise
-        cls.__runtime_publication__ = publication
+        else:
+            cls.__runtime_publication__ = publication
+        finally:
+            del cls.__runtime_starting__
 
     @classmethod
     def api_dependencies(cls) -> list[typing.Any]:

runtimes/core-py/src/inkcre_extension_runtime_core_py/base.py

@@ -172,6 +172,10 @@ def activate_source_types(self) -> None:
         """Synchronize the process-monotonic Source catalog."""
         SourceManager.sync_source_types()
 
+    async def activate_source_types_async(self) -> None:
+        """Synchronize the catalog through the async Host capability."""
+        await SourceManager.sync_source_types_async()
+
     def withdraw(self) -> None:
         """Withdraw this instance's exact active effects."""
         if not self.active:

runtimes/core-py/src/inkcre_extension_runtime_core_py/publication.py

@@ -0,0 +1,52 @@
+# Core 数据库迁移所需的 Runtime SDK 异步能力
+
+## 父任务与授权
+
+本 packet 仅拥有 ext-reg 的上游 SDK 实现 slice。唯一线性计划由 core-py 的
+`tasks/async-database-boundaries/plan.md` 拥有；不得在此建立另一条全量迁移计划。
+Sir 已授权修改 ext-reg、提交、推送与创建 PR，明确禁止 Agent 合并。
+
+上游交付 PR：https://github.com/InKCre/ext-reg/pull/38（实现提交 `5fc4ad7`）。
+父任务 draft PR：https://github.com/InKCre/core-py/pull/105。当前等待本 PR 的检查、外部合并和正式 artifact，
+不把本地 wheel 构建成功当作依赖已经交付。
+
+从 origin/main `006759a` 建立独立分支 `feat/runtime-async-persistence`，worktree 为
+`/Volumes/WorkSSD/Development/InKCre/.worktrees/ext-reg-async-persistence`。既有 ext-reg checkout 未改动。
+
+## 实现与交付边界
+
+Runtime 0.1.4 为 ExtensionBase 增加配置、状态及启动的 async API，绑定 Host 的 awaitable
+持久化能力；旧 API 及现有 SDK ExtensionManager 保留同步合同。一个 publication scope
+同时服务两种启动入口，在失败或取消时撤销本次活动资源，阻止未完成时重复启动。
+
+调查确认 Source catalog activation 也调用数据库。异步启动调用新
+SourceManager.sync_source_types_async；Core 采用 SDK 时先提供这个 capability，再切换 Host。
+SDK 不创建数据库连接、不持有 session、不引入 fallback 或 worker 包装。Host 负责事务和锁。
+
+版本及 changelog 通过 Changie 生成，PDM lock 只将本地 workspace SDK 版本由原有陈旧的
+0.1.2 更新为 0.1.4。Registry 服务、shared contracts、web runtime 和 Toolkit 源码未变。
+
+`.github/workflows/packages-release.yml` 仅在 main 发布正式 artifact。Agent 不得合并，故此
+slice 完成到可审阅 PR 后，Core 等待外部合并和 0.1.4 artifact；不会用本地 wheel URL 代替正式依赖。
+父任务及本 packet 此时保持打开。
+
+## 验证与限度
+
+- Runtime 及仓库 Python lint、format、pyright 通过；pnpm contracts:check、format:check、type-check 通过。
+- `pdm build --project runtimes/core-py --dest dist/runtime-core-py` 成功构建 wheel/sdist。
+- `pdm run python tasks/async-runtime-persistence/probe.py` 通过；另将构建的 wheel 正常安装到独立
+  Python 3.12 环境，使用 Core 当前 FastAPI 0.139.2 执行同一 probe，也通过。源码环境为 Python 3.13。
+- probe 是本任务的 public SDK boundary 实验，不是新常驻 CI suite。它使用真实 FastAPI/HTTP transport，
+  用内存 Host adapter 隔离持久化 owner，验证 typed mutation、两个启动 await 点的取消、失败撤销、
+  重新启动和旧同步启动兼容。它不证明 PostgreSQL 事务；该证据仍由 Core 的真实数据库验收拥有。
+- 本机 Node 为 26，pnpm 对项目要求的 Node 22 给出警告；上述 JS 检查通过。正式 CI 使用 Node 22。
+- 本机没有执行 Registry 的 PostgreSQL 全量 gate，SDK 变更未触及该数据库；仓库 CI 使用一次性 PostgreSQL
+  执行完整 pnpm check。不能把本地 package checks 写成全仓集成通过。
+
+## 已复现的既有问题
+
+当前 FastAPI include_router 会在 app.routes 中保存 _IncludedRouter，而旧 SDK 的 PublicHTTPRouteClaim
+只检查顶层 route.path。单独使用已安装 SDK 0.1.3 + FastAPI 0.139.2 的旧 on_start 已复现
+`Public Extension routes were not published: [('GET', '/probe/callback')]`。因此 probe 的成功启动不声明
+public callback；本次覆盖 routes/inbounds 的取消撤销，没有伪造 public-claim 成功证据。该问题不由
+数据库异步化引入，未扩展本次源码范围；采用方的 callback 验收仍受其影响。