From 250d0d8d330afe9592b36deac167075462ddeed0 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Wed, 28 Oct 2026 14:59:41 +0800 Subject: [PATCH 1/3] feat(ops): admit existing managed SQLite databases safely --- docs/en/docs/operate/managed-local-server.md | 85 +++-- docs/zh/docs/operate/managed-local-server.md | 24 +- release/upgrade-support.json | 4 +- .../builtin/persistence/migrations/sqlite.py | 38 +- src/powercontext/ops/database_admission.py | 326 ++++++++++++++++++ src/powercontext/ops/installer.py | 61 +++- src/powercontext/ops/local_policy.py | 8 +- src/powercontext/ops/runtime.py | 13 +- src/powercontext/server/database_admission.py | 144 ++++++++ tests/e2e/test_managed_database_admission.py | 282 +++++++++++++++ tests/test_ops_database_admission.py | 179 ++++++++++ 11 files changed, 1094 insertions(+), 70 deletions(-) create mode 100644 src/powercontext/ops/database_admission.py create mode 100644 src/powercontext/server/database_admission.py create mode 100644 tests/e2e/test_managed_database_admission.py create mode 100644 tests/test_ops_database_admission.py diff --git a/docs/en/docs/operate/managed-local-server.md b/docs/en/docs/operate/managed-local-server.md index 1842df2e17..3102ef9efa 100644 --- a/docs/en/docs/operate/managed-local-server.md +++ b/docs/en/docs/operate/managed-local-server.md @@ -91,42 +91,57 @@ Preflight does not create a configuration file, data directory, or database. ## Database compatibility boundary -Database inventory does not construct a Server runtime, connect to SQLite, initialize tables, change WAL settings, or -apply migrations. It observes the selected file and recovery sidecars and binds that observation to the exact target -Server version. Ops' own package version cannot prove compatibility with the target Server. - -Program-only repair binds the existing database's file identity and target version, allowing normal writes and WAL -checkpointing while resources are prepared. Replacing the database invalidates the plan. This repair observation does -not authorize startup; installation and startup retain strict database and recovery-sidecar observations. - -| Observation | Installer behavior | +Passive inspection inventories the selected SQLite file and recovery sidecars without opening SQLite or starting a +target process. Existing files remain `existing_unknown` in ordinary status. Program-only repair binds the file +identity and target version, allows normal writes and WAL checkpointing during resource preparation, and completes +with the service stopped. Replacing the database invalidates its repair plan. + +Operations that can start the managed service perform a separate admission check: `start`, `restart`, registration, +and configuration changes that actually restart an active service. Ops verifies the owned target resource against its +fixed manifest, then invokes that bundle's Server admission entry with isolated Python (`-I -B`). Windows uses the +manifest's console Python from the same bundle. The child checks its installed package version against the selected +Server version. An older target without this entry remains unsupported; the Ops package version or a previous +successful startup cannot replace target admission. + +The target checks the complete Server schema and indexes, supported revision history, configured domains and +embedding profile, SQLite integrity, original-path maintenance evidence, and unfinished archive restores. It reads a +private copy of the main database and WAL, including committed data not yet checkpointed into main. The original +database is never opened by SQLite during admission; its SHM is neither copied nor changed, and no original sidecars +are created or removed. File identities and modification evidence must remain stable while copying and checking. +Admission performs no initialization, migration, restoration, model call, HTTP startup, or worker startup. + +| Observation | Startup behavior | | --- | --- | -| Database and SQLite recovery sidecars are absent | Classify as `new`; the approved target Server may initialize a new database during startup. | -| An existing SQLite file, including an empty file | Classify as `existing_unknown`; do not automatically start or restart against it. | -| An existing file has an unsupported database header | Classify as `incompatible`; preserve it and block startup. | -| Orphan sidecars, a linked or unverified location, an unreadable file, or a changing observation | Classify as `unknown`; preserve the files and block startup. | - -The current public maintenance checks cover individual domains, not complete Server schema, indexes, persisted task -formats, and all configured capabilities. A Dream or processing check, an Atomic Memory verification, or a migration -bundle reporting `ready` therefore cannot lift the installer gate. A `pc_schema_revision` marker does not establish -whole-Server readiness. The installer does not create one for new databases. - -For existing data without a qualified whole-database check for the fixed target version, keep the service stopped and -preserve the database. Same-version program repair can complete while the database remains `existing_unknown`; that -does not authorize startup. There is no general maintenance command that this installer can recommend to make every -such database compatible. Consult [Database migrations](database-migrations.md) for the actual supported bundle -boundary, and wait for a maintenance release supporting the specific database before attempting a data transition. -Do not repeatedly run a partial migration as an automatic repair. - -This also limits normal operation today: after first startup creates a database, subsequent `start`, `restart`, and -`config_apply` with `restart=true` when it would restart an active service are rejected through the managed entry while -whole-database compatibility remains unknown. An already stopped service stays stopped during configuration changes, -including when `restart=true` is supplied. A previous successful startup or the installation manifest does not bypass -this gate. Inspection, stopping, -configuration changes without restart, and same-version program repair remain available. Repeating `install` for an -owned current installation only reports facts. This implementation therefore does not yet provide a complete daily -restart workflow; qualified whole-database checks are still required. The separate prepared-runtime operator binding -retains its existing contract and is not evidence that the managed database is compatible. +| Database and recovery evidence are absent | `new`; only the approved Server startup may initialize storage. | +| Existing storage passes the exact target's complete check | `existing_ready` for that explicit admission; normal same-version stop/start/restart is supported. | +| Unknown schema, empty existing file, older layout, future revision, unsupported profile, or missing target helper | Block startup and preserve storage. | +| Interrupted maintenance/restore, rollback journal, active maintenance lock, orphan sidecars, unsafe location, or changing source | Block startup and preserve recovery evidence. | + +A revision marker or a partial domain check is insufficient. Required migrations and recovery remain explicit +operator work under [Database migrations](database-migrations.md). `database_migration_required`, +`database_recovery_required`, and `database_configuration_mismatch` identify those boundaries; +`database_compatibility_unknown` means the exact target could not establish compatibility. Repair changes programs +only and does not remove these requirements. + +Admission copies and hashes the complete main file and WAL, then scans the snapshot using Server validators. Each +preflight and apply revalidation repeats this work; it is not an incremental or cached check. The combined main/WAL +limit is **8 GiB**, the temporary filesystem needs their combined size plus **32 MiB** free, and copying plus target +validation has a **120-second** deadline. Oversize, insufficient space, and timeout return +`database_snapshot_too_large`, `database_snapshot_space_insufficient`, and `database_admission_timeout`. Recovery +journals are limited to 4 MiB each. Large or busy databases require an operator-controlled maintenance workflow; +raising a limit does not establish compatibility. Temporary copies are removed after the check. + +A local macOS arm64 / CPython 3.12.15 measurement with 2,000 public Memory records, 4,000 revisions and a 168.4 MiB +SQLite main file took 3.78–3.86 seconds per admission across three runs; copying and hashing took 0.074–0.075 seconds. +Source hashes remained unchanged. This measurement excludes resource-manifest verification and native service +commands and does not qualify other machines, WAL sizes, or databases near the limit. + +Apply repeats admission and compares the approved plan's database, configuration, target-resource and recovery +evidence. Changed evidence rejects the old plan before a service command. The running Server still owns its normal +startup checks. Inspection, stopping, registration removal, and stopped program repair remain available when the +target interpreter is damaged or absent. A stopped service stays stopped +when configuration changes request `restart=true`. The independent prepared-runtime operator binding retains its +existing contract. ## Startup and connection handoff diff --git a/docs/zh/docs/operate/managed-local-server.md b/docs/zh/docs/operate/managed-local-server.md index c966e0d718..30f3d99d42 100644 --- a/docs/zh/docs/operate/managed-local-server.md +++ b/docs/zh/docs/operate/managed-local-server.md @@ -57,22 +57,26 @@ description: 安装或修复固定版本的个人 Server 运行环境，并保 ## 数据库兼容边界 -数据库检查不构造 Server runtime，不连接 SQLite，不初始化表、不修改 WAL 设置，也不执行迁移。它只观察选定文件及恢复侧文件，并将观察结果绑定到精确目标 Server 版本。Ops 自身的包版本不能证明目标 Server 的数据库兼容性。 +普通检查只观察选定的 SQLite 文件及恢复侧文件，不打开 SQLite，也不启动目标进程；状态中的已有文件仍标记为 `existing_unknown`。只修复程序时，计划绑定数据库文件身份和目标版本，允许准备资源期间正常写入及 WAL checkpoint，修复完成后保持服务停止。替换数据库会使修复计划过期。 -只修复程序时，计划绑定已有数据库的文件身份和目标版本，允许准备资源期间正常写入及 WAL checkpoint；替换数据库会使计划过期。这种修复观察不授予启动许可，安装和启动仍使用严格的数据库及恢复侧文件观察。 +可能启动受管服务的操作会另外执行准入检查，包括 `start`、`restart`、服务注册，以及会实际重启运行中服务的配置修改。Ops 先按照固定清单核验自有目标资源，再使用该包中的隔离 Python（`-I -B`）调用 Server 准入入口。Windows 使用同一资源清单中的控制台 Python。子进程核对已安装包版本是否与选定的 Server 版本完全一致。缺少该入口的旧目标仍不受支持；Ops 自身版本和历史启动成功都不能替代目标准入。 -| 观察结果 | 安装器处理 | +目标检查完整 Server schema 与索引、受支持的版本历史、配置对应的领域和 Embedding profile、SQLite 完整性、原数据库身份对应的维护证据，以及未完成的归档恢复。检查读取主库和 WAL 的私有副本，包含尚未 checkpoint 到主库的已提交数据。准入期间 SQLite 不打开原库；不复制或改动原 SHM，也不创建或删除原侧文件。复制和检查期间，源文件身份和修改证据必须保持稳定。准入不初始化、迁移或恢复数据，不调用模型，也不启动 HTTP 或 worker。 + +| 观察结果 | 启动处理 | | --- | --- | -| 数据库及 SQLite 恢复侧文件均不存在 | 分类为 `new`，确认后的目标 Server 可以在启动时初始化新库。 | -| 已有 SQLite 文件，包括空文件 | 分类为 `existing_unknown`，不自动启动或重启。 | -| 已有文件的数据库头不受支持 | 分类为 `incompatible`，保留文件并阻止启动。 | -| 孤立侧文件、链接或未核验位置、不可读文件、观察过程中发生变化 | 分类为 `unknown`，保留文件并阻止启动。 | +| 数据库及恢复证据均不存在 | `new`；只有确认后的 Server 启动可以初始化存储。 | +| 已有存储通过精确目标版本的完整检查 | 本次明确准入为 `existing_ready`，支持正常的同版本停止、启动与重启。 | +| 未知 schema、已有空文件、旧布局、未来版本、不支持的 profile 或缺少目标检查入口 | 阻止启动并保留存储。 | +| 未完成的维护或恢复、回滚日志、正在持有的维护锁、孤立侧文件、不安全路径或源文件发生变化 | 阻止启动并保留恢复证据。 | + +版本标记或局部领域检查不能证明整库可用。所需迁移和恢复仍须按照[数据库迁移](database-migrations.md)显式执行。`database_migration_required`、`database_recovery_required` 和 `database_configuration_mismatch` 分别提示这些边界；`database_compatibility_unknown` 表示精确目标未能证明兼容性。程序修复不会取消这些要求。 -当前公开维护检查只覆盖部分领域，不能证明完整 Server 的 schema、索引、持久任务格式及全部配置能力可用。Dream 或 processing 检查、Atomic Memory 验证、迁移 bundle 的 `ready` 都不能解除安装器的启动阻断。`pc_schema_revision` 标记也不代表整库可用，安装器不会为新库创建该标记。 +准入复制并计算完整主库与 WAL 的摘要，再由 Server 验证器扫描副本。每次预检和 apply 重新核验都会重复这些工作，不是增量检查，也不复用缓存。主库与 WAL 合计上限为 **8 GiB**；临时文件系统需要其合计大小加 **32 MiB** 的可用空间；复制和目标验证共用 **120 秒**时限。超出大小、空间不足或超时分别返回 `database_snapshot_too_large`、`database_snapshot_space_insufficient`、`database_admission_timeout`。每个恢复日志上限为 4 MiB。大型或持续写入的数据库需要由操作者安排维护流程，单纯提高限制不能证明兼容。检查结束后删除临时副本。 -已有数据缺少固定目标版本的合格整库检查时，应保持服务停止并保留数据库。同版本程序修复可以完成，而数据库继续标记为 `existing_unknown`；修好程序不授予启动许可。当前没有一条能让所有这类数据库变得兼容的通用维护命令。请参阅[数据库迁移](database-migrations.md)了解实际支持的 bundle 边界，并等待支持该数据库的维护版本后再执行数据转换。不要把反复执行局部迁移当成自动修复。 +一次本地 macOS arm64 / CPython 3.12.15 测量使用公开 Memory 接口创建 2,000 条记录、4,000 个修订版本，SQLite 主库为 168.4 MiB。三次准入各耗时 3.78–3.86 秒，其中复制及计算摘要耗时 0.074–0.075 秒；源文件摘要保持不变。该测量不包含资源清单核验和原生服务命令，不能证明其他机器、WAL 大小或接近上限的数据库具有同样性能。 -这也限制了当前的日常操作：首次启动创建数据库后，只要整库兼容性仍为未知，受管入口后续的 `start`、`restart`，以及会实际重启运行中服务的 `config_apply(restart=true)` 都会被拒绝。原本停止的服务在配置修改后仍保持停止，即使传入 `restart=true` 也不会启动。历史启动成功或安装清单不能绕过该检查。检查、停止、不重启的配置修改和同版本程序修复仍可执行。对自有当前安装重复执行 `install` 只返回事实。因此，这一实现尚不具备完整的日常重启流程，仍需补齐合格的整库检查。独立的已备妥 runtime 操作者绑定保留既有契约，也不能用它证明受管数据库兼容。 +Apply 会重新执行准入，并比较获批计划中的数据库、配置、目标资源及恢复证据；证据变化会在派发服务命令前拒绝旧计划。实际 Server 启动仍执行自己的启动检查。即使目标解释器损坏或缺失，检查、停止、移除注册和保持停止的程序修复仍可使用。原本停止的服务在配置修改后仍保持停止，即使传入 `restart=true` 也不会启动。独立的已备妥 runtime 操作者绑定保留既有契约。 ## 启动与连接交接 diff --git a/release/upgrade-support.json b/release/upgrade-support.json index 52a02d606e..759a4fd932 100644 --- a/release/upgrade-support.json +++ b/release/upgrade-support.json @@ -3,8 +3,8 @@ "target_version": "1.4.1", "mode": "upgrade-required", "schema_review": { - "persistence_sha256": "99d448c2b9f3a3ef23a5e348a747d9e277c4cca13ad3bee475b26d10faabac98", - "rationale": "The reusable Server migration framework registers actual PowerContext 1.3.0 through s1300 -> s1400 -> s1401 -> s1402 and actual 1.4.0 through the separate immutable r1400 -> s1400 -> s1401 -> s1402 source bridge. Original 1.3 resources and checksums remain byte-identical. Real 1.4.0 SQLite, embedded seekDB and OceanBase structures were captured from the official wheel in all twelve vector/Code configurations; provenance and exact source layouts are packaged with the bridge. The 1.4 bridge preserves current Candidate, Artifact and task data, adds the missing Topic budget/block structure, revokes old lease fences and joins the existing archive/portable-storage path. Empty retained tables align the final schema without inventing 1.3 Candidate history. Source histories retain their identity through subsequent upgrades and recovery; shared head names never authorize adopting unknown layouts. Vector profiles remain bound to historical model, profile ID, dimension, unit normalization and L2 distance, with no provider calls during maintenance. Complete schema, domain, projection, profile and retained-history checks precede startup and partial maintenance. Local SQLite and embedded seekDB startup reject an observable active maintenance lock through a read-only probe. SQLite runtime admission validates the default adjacent journal, rejecting unfinished full or partial maintenance, changed recorded revision checksums and missing required retained tables. Absent journals do not invent retention obligations; custom evidence directories require explicit operator verification and remote coordinator locks remain subject to deployment orchestration. Native read-only readiness uses a consistent snapshot and does not acquire Atomic Memory head locks; normal business transaction isolation is unchanged. Offline projection repairs retain the original locked state comparison and write-side CAS protection. Target locks, backup choices, original recovery evidence and explicit remote acknowledgments remain enforced. The installed-artifact catalog retains twelve 1.3.0 source/backend/vector/Code combinations. Embedded seekDB uses pylibseekdb 1.4.0.post2 or newer with the recovered-transaction lock fix. Main and publication require all twelve configurations, including embedded seekDB full-server-code and full-server-vector-code. No configuration is waived; incomplete, failed or skipped evidence rejects acceptance. The seekdb-code diagnostic selection cannot authorize publication. The registered 1.4.0 bridge remains implemented but is outside the CI upgrade acceptance matrix; passing this gate does not certify that path. The gate preserves source-specific Processing counters and isolates OceanBase source databases. Any failed, skipped or missing automatic case rejects publication. The policy remains upgrade-required until the exact target artifact passes the required automatic matrix and the reviewed support declaration is updated. Captured source structure, local tests and checkpoint replay are distinct from installed-release acceptance, native interruption/observer recovery and backup restoration. The Candidate generation prototype observes actual SQLite/seekDB/OceanBase migration checkpoints outside the production bundle; it does not replace frozen registrations or weaken recovery checks. Existing target schema and data formats, including batched persistence interfaces, remain governed by the shared registered path. The synchronized master adds blank-text validation at Source creation and managed worker logging. It changes neither registered table layout nor migration data encodings; existing stored Source values are not rewritten. The source bridges and frozen history remain unchanged. Compatibility commands identify Server-owned storage before opening a writable legacy profile and delegate source admission to the shared framework, including actual 1.4.0 and rejection of unknown layouts. Partial Processing and Atomic Memory projection maintenance verify the original recorded revision checksums; native admission also validates typed journal identity and its completed position against the database revision. Persisted vector profiles reject unsupported partial maintenance before plans, backups or operation evidence, including when the supplied configuration omits embeddings. These are admission-only changes; frozen migration resources, registered layouts, historical checksums and data transformations remain unchanged. PowerContext 1.4.1 assigns release-based presentation names v1_4_1_001 through v1_4_1_003 to the existing shared upgrade steps; the 1.3.0 and 1.4.0 source baselines remain independently registered. Existing revision IDs, bundle and revision checksums, accepted plan identities and journal schemas are unchanged. New Server revisions use software-version-plus-sequence names directly; software releases without storage changes may reuse the same database revision. Naming does not grant acceptance or replace source registration. Full Server status, plan, apply and verify reject unfinished Processing configuration and Atomic Memory projection journals through the same typed admission check used by partial maintenance and SQLite startup. The owning domain runner can still resume its own journal. This only changes admission; frozen revisions, journal formats, database layouts and data transformations remain unchanged. Revision v1_4_1_004 registers the pc_access_owner_recovery outbox on the shared Server chain for both installed-release sources and all registered vector/Code profiles. It creates an empty table through explicit maintenance for existing managed databases and through the registered bootstrap for fresh databases. Earlier revision resources, fingerprints, checksums and existing domain row encodings remain unchanged. External relationship writers use version-1 Artifact and Candidate intents committed with domain data; replay retains the original deployment, principal, resource or target, and idempotency identity. Stable operation and audit event identities prevent lost acknowledgements from duplicating ownership or audit. Conflicting intents are rejected and unknown historical ownership is not inferred. Unregistered outbox layouts are not adopted merely from a table name. This registration does not establish installed-release or external-Provider upgrade acceptance. The policy remains upgrade-required with the full automatic matrix and publication restrictions described above." + "persistence_sha256": "270800ed36aebbf26bf95f0146187589101e9712b47150420e39e98ca7a17352", + "rationale": "The reusable Server migration framework registers actual PowerContext 1.3.0 through s1300 -> s1400 -> s1401 -> s1402 and actual 1.4.0 through the separate immutable r1400 -> s1400 -> s1401 -> s1402 source bridge. Original 1.3 resources and checksums remain byte-identical. Real 1.4.0 SQLite, embedded seekDB and OceanBase structures were captured from the official wheel in all twelve vector/Code configurations; provenance and exact source layouts are packaged with the bridge. The 1.4 bridge preserves current Candidate, Artifact and task data, adds the missing Topic budget/block structure, revokes old lease fences and joins the existing archive/portable-storage path. Empty retained tables align the final schema without inventing 1.3 Candidate history. Source histories retain their identity through subsequent upgrades and recovery; shared head names never authorize adopting unknown layouts. Vector profiles remain bound to historical model, profile ID, dimension, unit normalization and L2 distance, with no provider calls during maintenance. Complete schema, domain, projection, profile and retained-history checks precede startup and partial maintenance. Local SQLite and embedded seekDB startup reject an observable active maintenance lock through a read-only probe. SQLite runtime admission validates the default adjacent journal, rejecting unfinished full or partial maintenance, changed recorded revision checksums and missing required retained tables. Absent journals do not invent retention obligations; custom evidence directories require explicit operator verification and remote coordinator locks remain subject to deployment orchestration. Native read-only readiness uses a consistent snapshot and does not acquire Atomic Memory head locks; normal business transaction isolation is unchanged. Offline projection repairs retain the original locked state comparison and write-side CAS protection. Target locks, backup choices, original recovery evidence and explicit remote acknowledgments remain enforced. The installed-artifact catalog retains twelve 1.3.0 source/backend/vector/Code combinations. Embedded seekDB uses pylibseekdb 1.4.0.post2 or newer with the recovered-transaction lock fix. Main and publication require all twelve configurations, including embedded seekDB full-server-code and full-server-vector-code. No configuration is waived; incomplete, failed or skipped evidence rejects acceptance. The seekdb-code diagnostic selection cannot authorize publication. The registered 1.4.0 bridge remains implemented but is outside the CI upgrade acceptance matrix; passing this gate does not certify that path. The gate preserves source-specific Processing counters and isolates OceanBase source databases. Any failed, skipped or missing automatic case rejects publication. The policy remains upgrade-required until the exact target artifact passes the required automatic matrix and the reviewed support declaration is updated. Captured source structure, local tests and checkpoint replay are distinct from installed-release acceptance, native interruption/observer recovery and backup restoration. The Candidate generation prototype observes actual SQLite/seekDB/OceanBase migration checkpoints outside the production bundle; it does not replace frozen registrations or weaken recovery checks. Existing target schema and data formats, including batched persistence interfaces, remain governed by the shared registered path. The synchronized master adds blank-text validation at Source creation and managed worker logging. It changes neither registered table layout nor migration data encodings; existing stored Source values are not rewritten. The source bridges and frozen history remain unchanged. Compatibility commands identify Server-owned storage before opening a writable legacy profile and delegate source admission to the shared framework, including actual 1.4.0 and rejection of unknown layouts. Partial Processing and Atomic Memory projection maintenance verify the original recorded revision checksums; native admission also validates typed journal identity and its completed position against the database revision. Persisted vector profiles reject unsupported partial maintenance before plans, backups or operation evidence, including when the supplied configuration omits embeddings. These are admission-only changes; frozen migration resources, registered layouts, historical checksums and data transformations remain unchanged. PowerContext 1.4.1 assigns release-based presentation names v1_4_1_001 through v1_4_1_003 to the existing shared upgrade steps; the 1.3.0 and 1.4.0 source baselines remain independently registered. Existing revision IDs, bundle and revision checksums, accepted plan identities and journal schemas are unchanged. New Server revisions use software-version-plus-sequence names directly; software releases without storage changes may reuse the same database revision. Naming does not grant acceptance or replace source registration. Full Server status, plan, apply and verify reject unfinished Processing configuration and Atomic Memory projection journals through the same typed admission check used by partial maintenance and SQLite startup. The owning domain runner can still resume its own journal. This only changes admission; frozen revisions, journal formats, database layouts and data transformations remain unchanged. Revision v1_4_1_004 registers the pc_access_owner_recovery outbox on the shared Server chain for both installed-release sources and all registered vector/Code profiles. It creates an empty table through explicit maintenance for existing managed databases and through the registered bootstrap for fresh databases. Earlier revision resources, fingerprints, checksums and existing domain row encodings remain unchanged. External relationship writers use version-1 Artifact and Candidate intents committed with domain data; replay retains the original deployment, principal, resource or target, and idempotency identity. Stable operation and audit event identities prevent lost acknowledgements from duplicating ownership or audit. Conflicting intents are rejected and unknown historical ownership is not inferred. Unregistered outbox layouts are not adopted merely from a table name. This registration does not establish installed-release or external-Provider upgrade acceptance. The policy remains upgrade-required with the full automatic matrix and publication restrictions described above. Managed Ops admission validates a private main-file and WAL snapshot through the exact target Server bundle, retaining the original database identity and maintenance evidence. Snapshot admission is read-only and cannot initialize, migrate or recover storage. It adds no schema revision, changes no persisted row or journal format, and does not expand supported upgrade declarations or the installed-release acceptance matrix." }, "cases": [], "automatic_sources": { diff --git a/src/powercontext/builtin/persistence/migrations/sqlite.py b/src/powercontext/builtin/persistence/migrations/sqlite.py index 04bae947a3..86ed34bc7d 100644 --- a/src/powercontext/builtin/persistence/migrations/sqlite.py +++ b/src/powercontext/builtin/persistence/migrations/sqlite.py @@ -202,6 +202,38 @@ def verify(self) -> MigrationResult: raise MigrationError(plan.state, "Inspect and explicitly apply the registered bundle before verification.") return MigrationResult(revision=self.bundle.head, changed=False, readiness_scope=self.bundle.readiness_scope) + def admit_snapshot(self, snapshot: Path) -> MigrationPlan: + """Check a private SQLite snapshot using this target's original recovery identity. + + The caller must copy a stable main file and WAL, exclude SHM, and keep + source identities stable throughout this check. Only the snapshot is + opened by SQLite. This entry cannot initialize, migrate or recover data. + """ + from .connections import BackendIdentity + from .locking import assert_no_local_migration + from .mysql import require_completed_domain_maintenance + from .vector_profile import read_embedding_profile + + if self.bundle.readiness_scope != "server": + raise MigrationError("unsupported_target", "Admission requires the complete Server bundle.") + self.bundle.assert_unchanged() + assert_no_local_migration(self.database) + evidence = self._evidence() + if evidence is not None and evidence["state"] != "complete": + raise MigrationError("recovery_required", "Complete recorded Server maintenance before startup.") + require_completed_domain_maintenance( + self.evidence_path.parent, BackendIdentity("sqlite", self._database_id(), self.database.name) + ) + with self._connect(read_only=True, snapshot=snapshot) as connection: + profile = self.bundle.vector_profile.profile if self.bundle.vector_profile is not None else None + if read_embedding_profile(connection) != profile: + raise MigrationError("embedding_profile_mismatch", "Configuration differs from managed storage.") + plan = self._inspect(connection, backup_policy="skip", service=None, shared_database=False) + if plan.state != "ready": + raise MigrationError("migration_required", "Explicit full Server maintenance is required.") + assert_no_local_migration(self.database) + return plan + def apply( self, *, @@ -304,8 +336,10 @@ def _check_file(self) -> None: raise MigrationError("unsupported_target", "Use a regular SQLite file without hard-link aliases.") @contextmanager - def _connect(self, *, read_only: bool) -> Iterator[Connection]: - uri = self.database.as_uri() + ("?mode=ro" if read_only else "?mode=rwc") + def _connect(self, *, read_only: bool, snapshot: Path | None = None) -> Iterator[Connection]: + if snapshot is not None and not read_only: + raise MigrationError("unsupported_target", "A snapshot is only available for read-only admission.") + uri = (snapshot or self.database).as_uri() + ("?mode=ro" if read_only else "?mode=rwc") def connect() -> sqlite3.Connection: driver = sqlite3.connect(uri, uri=True, isolation_level=None, timeout=0) diff --git a/src/powercontext/ops/database_admission.py b/src/powercontext/ops/database_admission.py new file mode 100644 index 0000000000..28c32be283 --- /dev/null +++ b/src/powercontext/ops/database_admission.py @@ -0,0 +1,326 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Bounded private snapshot transport to the verified target Server's admission entry.""" + +from __future__ import annotations + +import hashlib +import os +import shutil +import stat +import subprocess +import tempfile +import time +from dataclasses import asdict +from pathlib import Path +from typing import Any, cast + +from powercontext.ops.local_policy import ConfigurationSummary, DatabaseClassification, classify_database +from powercontext.ops.protocol import OpsError, canonical, decode, digest +from powercontext.service.environment import protect_directory + +# Admission deliberately performs a complete copy and Server validation. Large +# or busy databases need explicit operator maintenance, not a partial certificate. +MAX_SNAPSHOT_BYTES = 8 * 1024**3 +ADMISSION_TIMEOUT_SECONDS = 120 +_COPY_BLOCK_BYTES = 1024 * 1024 + + +def _identity(value: os.stat_result) -> tuple[int, ...]: + return ( + value.st_dev, + value.st_ino, + value.st_mode, + value.st_nlink, + value.st_size, + value.st_mtime_ns, + value.st_ctime_ns, + ) + + +def _source_files(database: Path) -> dict[str, tuple[int, ...] | None]: + result = {} + for suffix in ("", "-wal", "-shm", "-journal"): + try: + observed = Path(f"{database}{suffix}").lstat() + except FileNotFoundError: + result[suffix] = None + continue + if ( + not stat.S_ISREG(observed.st_mode) + or observed.st_nlink != 1 + or getattr(observed, "st_file_attributes", 0) & 0x400 + ): + raise OpsError("database_location_unverified") + result[suffix] = _identity(observed) + return result + + +def _copy(source: Path, destination: Path, expected: tuple[int, ...], deadline: float) -> str: + checksum = hashlib.sha256() + descriptor = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0)) + with os.fdopen(descriptor, "rb") as input_file, destination.open("xb") as output: + if _identity(os.fstat(input_file.fileno())) != expected: + raise OpsError("database_changed_during_admission") + remaining = expected[4] + while remaining: + if time.monotonic() >= deadline: + raise OpsError("database_admission_timeout") + block = input_file.read(min(_COPY_BLOCK_BYTES, remaining)) + if not block: + raise OpsError("database_changed_during_admission") + checksum.update(block) + output.write(block) + remaining -= len(block) + if input_file.read(1) or _identity(os.fstat(input_file.fileno())) != expected: + raise OpsError("database_changed_during_admission") + return checksum.hexdigest() + + +def _windows_pipe_bytes(descriptor: int) -> int: + # Windows 3.11 does not implement os.set_blocking for pipes. Peek only; + # os.read below consumes the known available bytes without waiting for EOF. + import ctypes + import msvcrt + from ctypes import wintypes + + kernel = getattr(ctypes, "WinDLL")("kernel32", use_last_error=True) # noqa: B009 + peek = kernel.PeekNamedPipe + peek.argtypes = [ + wintypes.HANDLE, + wintypes.LPVOID, + wintypes.DWORD, + wintypes.LPVOID, + wintypes.LPDWORD, + wintypes.LPVOID, + ] + peek.restype = wintypes.BOOL + available = wintypes.DWORD() + if not peek(getattr(msvcrt, "get_osfhandle")(descriptor), None, 0, None, ctypes.byref(available), None): # noqa: B009 + if getattr(ctypes, "get_last_error")() == 109: # noqa: B009 -- ERROR_BROKEN_PIPE + return 0 + raise OSError("admission_pipe_unavailable") + return available.value + + +def _read_child(child: subprocess.Popen[bytes], deadline: float) -> bytes: + if child.stdout is None: + raise OpsError("database_admission_unavailable") + descriptor = child.stdout.fileno() + if os.name != "nt": + os.set_blocking(descriptor, False) + content = bytearray() + exited = False + while time.monotonic() < deadline: + available = _windows_pipe_bytes(descriptor) if os.name == "nt" else 8193 + try: + block = os.read(descriptor, min(available, 8193 - len(content))) if available else None + except BlockingIOError: + block = None + if block: + content.extend(block) + if len(content) > 8192: + raise OpsError("database_admission_unavailable") + continue + if block == b"" or exited: + return bytes(content) + if child.poll() is not None: + # Drain once more to include bytes written between read and poll, + # then stop even if a descendant still holds the output pipe open. + exited = True + continue + time.sleep(0.01) + raise OpsError("database_admission_timeout") + + +def _target_admission( + configuration: ConfigurationSummary, version: str, executable: Path, snapshot: Path, deadline: float +) -> dict[str, Any]: + if configuration.identity is None: + raise OpsError("database_admission_unavailable") + request = { + "protocol_version": 1, + "target_version": version, + "configuration": asdict(configuration.identity), + "data_dir": str(configuration.data_dir), + "database": str(configuration.database_path), + "snapshot": str(snapshot), + } + remaining = deadline - time.monotonic() + if remaining <= 0: + raise OpsError("database_admission_timeout") + # Never inherit Python injection or ambient Server settings. The protected + # configuration and registered data directory are the target's authority. + environment = { + key: value + for key, value in os.environ.items() + if not key.upper().startswith(("PYTHON", "POWERCONTEXT_SERVER_")) and key.upper() != "POWERCONTEXT_HOME" + } + with tempfile.TemporaryFile(dir=snapshot.parent) as request_file: + request_file.write(canonical(request)) + request_file.seek(0) + with subprocess.Popen( # noqa: S603 -- Verified managed-resource console interpreter. + [str(executable), "-I", "-B", "-m", "powercontext.server.database_admission"], + stdin=request_file, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + bufsize=0, + cwd=snapshot.parent, + env=environment, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) as child: + try: + content = _read_child(child, deadline) + child.wait(timeout=max(0, deadline - time.monotonic())) + except BaseException: + child.kill() + child.wait() + raise + if child.returncode != 0: + raise OpsError("database_admission_unavailable") + decoded = decode(content) + if not isinstance(decoded, dict): + raise OpsError("database_admission_unavailable") + result = cast(dict[str, Any], decoded) + if ( + not isinstance(result, dict) + or type(result.get("protocol_version")) is not int + or result["protocol_version"] != 1 + or result.get("state") not in {"ready", "rejected"} + or not isinstance(result.get("reason"), str) + or ( + result["state"] == "ready" + and (result.get("reason") != "server_ready" or not isinstance(result.get("fingerprint"), str)) + ) + ): + raise OpsError("database_admission_unavailable") + return result + + +def _snapshot_admission( + configuration: ConfigurationSummary, + target_version: str, + python_executable: Path, + observed: DatabaseClassification, + evidence: dict[str, object], +) -> dict[str, Any]: + database = configuration.database_path + source = _source_files(database) + evidence["source"] = source + if source[""] is None: + raise OpsError("database_changed_during_admission") + if source["-journal"] is not None: + raise OpsError("database_recovery_required") + copied = { + suffix: identity for suffix, identity in source.items() if suffix in {"", "-wal"} and identity is not None + } + required = sum(identity[4] for identity in copied.values()) + if required > MAX_SNAPSHOT_BYTES: + raise OpsError("database_snapshot_too_large") + deadline = time.monotonic() + ADMISSION_TIMEOUT_SECONDS + with tempfile.TemporaryDirectory(prefix="powercontext-admission-") as temporary: + root = protect_directory(Path(temporary).resolve()) + if shutil.disk_usage(root).free < required + 32 * 1024**2: + raise OpsError("database_snapshot_space_insufficient") + snapshot = root / "database.db" + evidence["content"] = { + suffix: _copy(Path(f"{database}{suffix}"), Path(f"{snapshot}{suffix}"), identity, deadline) + for suffix, identity in copied.items() + } + if _source_files(database) != source: + raise OpsError("database_changed_during_admission") + result = _target_admission(configuration, target_version, python_executable, snapshot, deadline) + if _source_files(database) != source or classify_database(configuration, target_version) != observed: + raise OpsError("database_changed_during_admission") + return result + + +def admit_database( # noqa: C901 -- Preserve ordered fail-closed result classification. + configuration: ConfigurationSummary, target_version: str, *, python_executable: Path +) -> DatabaseClassification: + """Admit only using a verified target; never connect SQLite to the source files. + + The installer must verify the complete target resource before calling this + function. Passive inspect and program repair use ``classify_database`` only. + """ + observed = classify_database(configuration, target_version) + database = configuration.database_path + if observed.classification != "existing_unknown": + if observed.classification == "new" and any( + path.exists() or path.is_symlink() + for path in (Path(f"{database}.pc-migration-state"), Path(f"{database}.pc-migration.lock")) + ): + return DatabaseClassification("unknown", False, observed.fingerprint, "database_recovery_required") + return observed + evidence: dict[str, object] = {"inventory": observed.fingerprint} + + def rejected(reason: str, *, incompatible: bool = False) -> DatabaseClassification: + return DatabaseClassification( + "incompatible" if incompatible else "existing_unknown", False, digest(evidence), reason + ) + + try: + result = _snapshot_admission(configuration, target_version, python_executable, observed, evidence) + if result["state"] == "ready": + evidence["target"] = result["fingerprint"] + return DatabaseClassification("existing_ready", True, digest(evidence), "server_ready") + reason = result["reason"] + if reason in { + "recovery_required", + "migration_locked", + "migration_lock_unsupported", + "checksum_conflict", + "archive_recovery_required", + }: + return rejected("database_recovery_required") + if reason in {"migration_required", "server_migration_unavailable"}: + return rejected("database_migration_required") + if reason in { + "embedding_profile_mismatch", + "unsupported_upgrade_configuration", + "configuration_invalid", + "configuration_changed", + }: + return rejected("database_configuration_mismatch") + if reason in {"incompatible_schema", "invalid_database", "verification_failed", "unknown_revision"}: + return rejected("database_incompatible", incompatible=True) + return rejected("database_admission_unavailable") + except OpsError as error: + return rejected(error.code) + except (subprocess.TimeoutExpired, TimeoutError): + return rejected("database_admission_timeout") + except (OSError, ValueError): + return rejected("database_admission_unavailable") + + +def require_admitted(database: DatabaseClassification) -> None: + if database.allowed_to_start: + return + actionable = { + "database_recovery_required", + "database_migration_required", + "database_configuration_mismatch", + "database_snapshot_too_large", + "database_snapshot_space_insufficient", + "database_admission_timeout", + "database_changed_during_admission", + } + raise OpsError( + database.reason + if database.reason in actionable + else "database_incompatible" + if database.classification == "incompatible" + else "database_compatibility_unknown" + ) diff --git a/src/powercontext/ops/installer.py b/src/powercontext/ops/installer.py index 743056c219..857082b3b1 100644 --- a/src/powercontext/ops/installer.py +++ b/src/powercontext/ops/installer.py @@ -29,6 +29,7 @@ from uuid import uuid4 from powercontext.ops import protocol +from powercontext.ops.database_admission import admit_database, require_admitted from powercontext.ops.local_policy import ( classify_database, default_data_dir, @@ -254,6 +255,28 @@ def _repair_owner(self, ops: Ops, snapshot: dict[str, Any], evidence: dict[str, ops._require_native_settled(original) return owner + def admit_start(self, ops: Ops, snapshot: dict[str, Any]): + """Verify the owned target before delegating database admission to it.""" + evidence = self._installation(ops) or self._pending(ops, snapshot=snapshot) + resources, identity = self._resource_fact(evidence) + if resources != "verified": + raise OpsError("runtime_damaged") + database = admit_database( + self._configuration(ops, snapshot, {}), + self.spec.server_version, + python_executable=self.directory / self.spec.python_console_relative, + ) + require_admitted(database) + return replace( + database, + fingerprint=protocol.digest({ + "database": database.fingerprint, + "resource_digest": self.spec.digest, + "resource_identity": identity, + "catalog": self.catalog.sha256, + }), + ) + def preflight(self, ops: Ops, operation: str, options: dict[str, Any]) -> dict[str, Any]: # noqa: C901 validate_destination(self.spec, self.directory) validate_destination(self.spec, self.layout.programs / "staging" / ("resource-" + "0" * 32)) @@ -296,6 +319,16 @@ def preflight(self, ops: Ops, operation: str, options: dict[str, Any]) -> dict[s and isinstance(maintenance, dict) and maintenance.get("phase") in {"manual_stopped", "start_failed"} ) + if operation == "install" and not current and not preserve_stopped and not verify_running: + if database.classification == "new": + database = admit_database( + configuration, + self.spec.server_version, + python_executable=self.directory / self.spec.python_console_relative, + ) + require_admitted(database) + elif resources == "verified" and database.classification == "existing_unknown": + database = self.admit_start(ops, snapshot) may_start = ( operation == "install" and not current @@ -507,10 +540,16 @@ def checkpoint(stage: str) -> None: return {"outcome": "cancelled", "components": self.components(ops, state)} if plan["effects"]["may_start"] and not plan["installation"]["verify_running"]: configuration = self._configuration(ops, ops.controller.operation_snapshot(), plan["options"]) - if ( - classify_database(configuration, self.spec.server_version).as_dict() - != plan["installation"]["database"] - ): + database = ( + self.admit_start(ops, ops.controller.operation_snapshot()) + if plan["installation"]["database"]["classification"] == "existing_ready" + else admit_database( + configuration, + self.spec.server_version, + python_executable=self.directory / self.spec.python_console_relative, + ) + ) + if database.as_dict() != plan["installation"]["database"]: raise OpsError("plan_stale") with observe_native_commands(lambda event: self._native_checkpoint(ops, record, event)): if plan["installation"]["preserve_stopped"]: @@ -544,10 +583,20 @@ def _revalidate(self, ops: Ops, plan: dict[str, Any]) -> None: if protocol.digest(snapshot) != plan["installation"]["service_before"]: raise OpsError("plan_stale") summary = self._configuration(ops, snapshot, plan["options"]) + database = ( + self.admit_start(ops, snapshot) + if plan["installation"]["database"]["classification"] == "existing_ready" + else admit_database( + summary, + self.spec.server_version, + python_executable=self.directory / self.spec.python_console_relative, + ) + if plan["installation"]["database"]["classification"] == "new" and plan["operation"] == "install" + else classify_database(summary, self.spec.server_version, for_repair=plan["operation"] == "repair") + ) if ( summary.as_dict() != plan["installation"]["configuration"] - or classify_database(summary, self.spec.server_version, for_repair=plan["operation"] == "repair").as_dict() - != plan["installation"]["database"] + or database.as_dict() != plan["installation"]["database"] ): raise OpsError("plan_stale") diff --git a/src/powercontext/ops/local_policy.py b/src/powercontext/ops/local_policy.py index c07d65bce8..c488bf038a 100644 --- a/src/powercontext/ops/local_policy.py +++ b/src/powercontext/ops/local_policy.py @@ -58,9 +58,9 @@ def as_dict(self) -> dict[str, object]: @dataclass(frozen=True) class DatabaseClassification: - """A preflight observation, not a whole-database compatibility certificate.""" + """A passive inventory or an explicit target Server admission result.""" - classification: Literal["new", "existing_unknown", "incompatible", "unknown"] + classification: Literal["new", "existing_ready", "existing_unknown", "incompatible", "unknown"] allowed_to_start: bool fingerprint: str reason: str @@ -197,8 +197,8 @@ def classify_database( # noqa: C901 """Inventory without SQLite connection, initialization, migration, or WAL side effects. Existing data remains unknown even when the Ops version matches the target. - The current public maintenance checks cover individual domains, not the - complete Server schema, indexes, task formats, and configured capabilities. + Only explicit admission by the verified target Server can establish full + schema, domain, recovery and configuration compatibility. Program-only repair binds an existing database's stable file identity, allowing normal writes and checkpointing before the service is stopped. Startup observations retain file sizes, times, and all recovery sidecars. diff --git a/src/powercontext/ops/runtime.py b/src/powercontext/ops/runtime.py index 6e90e6d264..e494b49f6e 100644 --- a/src/powercontext/ops/runtime.py +++ b/src/powercontext/ops/runtime.py @@ -229,7 +229,7 @@ def _assert_target(self, snapshot: dict[str, Any], operation: str) -> None: ): raise OpsError("registration_missing") - def preflight(self, operation: object, options: object) -> dict[str, Any]: # noqa: C901 + def preflight(self, operation: object, options: object) -> dict[str, Any]: selected = protocol.options(operation, options) if operation in {"install", "repair"}: if self.installer is None: @@ -262,16 +262,7 @@ def preflight(self, operation: object, options: object) -> dict[str, Any]: # no ) database = None if self.installer is not None and may_start: - from powercontext.ops.local_policy import classify_database - - configuration = self.installer._configuration(self, snapshot, {}) - database = classify_database(configuration, self.target.package_version) - if not database.allowed_to_start: - raise OpsError( - "database_incompatible" - if database.classification == "incompatible" - else "database_compatibility_unknown" - ) + database = self.installer.admit_start(self, snapshot) result = { "protocol_version": 1, "target_id": self.target.target_id, diff --git a/src/powercontext/server/database_admission.py b/src/powercontext/server/database_admission.py new file mode 100644 index 0000000000..543d742b27 --- /dev/null +++ b/src/powercontext/server/database_admission.py @@ -0,0 +1,144 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Private, versioned managed-Ops entry: admit a snapshot without starting Server.""" + +from __future__ import annotations + +import hashlib +import json +import os +import stat +import sys +from dataclasses import asdict +from importlib.metadata import version +from pathlib import Path +from typing import Any, cast + +from powercontext.builtin.persistence.migrations.models import MigrationError, digest + +_MAX_REQUEST_BYTES = 16 * 1024 +_MAX_EVIDENCE_BYTES = 4 * 1024 * 1024 + + +def _recovery_fingerprint(database: Path) -> str: + """Bind the same original-path journals that full Server admission consumes.""" + root = database.with_name(database.name + ".pc-migration-state") + database_id = digest({"kind": "sqlite", "path": str(database)}) + paths = [root / "maintenance.json"] + [ + root / operation / database_id / "maintenance.json" + for operation in ("processing_configuration", "atomic_projection") + ] + observations = [] + for path in paths: + for parent in path.parents: + if parent == root.parent: + break + try: + item = parent.lstat() + except FileNotFoundError: + continue + if not stat.S_ISDIR(item.st_mode) or getattr(item, "st_file_attributes", 0) & 0x400: + raise MigrationError("recovery_required", "The maintenance location cannot be inspected safely.") + try: + item = path.lstat() + except FileNotFoundError: + observations.append(None) + continue + if ( + not stat.S_ISREG(item.st_mode) + or getattr(item, "st_file_attributes", 0) & 0x400 + or item.st_size > _MAX_EVIDENCE_BYTES + ): + raise MigrationError("recovery_required", "The maintenance evidence cannot be inspected safely.") + with path.open("rb") as source: + before = os.fstat(source.fileno()) + content = source.read(_MAX_EVIDENCE_BYTES + 1) + after = os.fstat(source.fileno()) + identities = [ + (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns) + for value in (item, before, after, path.lstat()) + ] + if len(content) > _MAX_EVIDENCE_BYTES or len(set(identities)) != 1: + raise MigrationError("recovery_required", "The maintenance evidence changed during admission.") + observations.append([identities[0], hashlib.sha256(content).hexdigest()]) + return digest(observations) + + +def _admit(request: object) -> dict[str, object]: + from powercontext.builtin.persistence.migrations.deployment import deployment_runner + from powercontext.builtin.persistence.migrations.sqlite import SQLiteMigrationRunner + from powercontext.server.configuration import server_settings_context + from powercontext.service.environment import load_protected_environment_file + from powercontext.service.model import EnvironmentFileIdentity + + if not isinstance(request, dict): + raise MigrationError("admission_unsupported", "Unsupported admission request.") + request = cast(dict[str, Any], request) + if ( + set(request) != {"protocol_version", "target_version", "configuration", "data_dir", "database", "snapshot"} + or type(request["protocol_version"]) is not int + or request["protocol_version"] != 1 + ): + raise MigrationError("admission_unsupported", "Unsupported admission request.") + if version("powercontext") != request["target_version"]: + raise MigrationError("target_version_mismatch", "The target package version differs from its resource.") + identity = EnvironmentFileIdentity(**request["configuration"]) + loaded = load_protected_environment_file(Path(identity.path), expected=identity) + database, snapshot, data_dir = (Path(request[key]) for key in ("database", "snapshot", "data_dir")) + if ( + any(not path.is_absolute() for path in (database, snapshot, data_dir)) + or snapshot.resolve() == database.resolve() + ): + raise MigrationError("unsupported_target", "Admission requires a separate private snapshot.") + with server_settings_context(environment=loaded.values, data_dir=data_dir) as settings: + runner = deployment_runner(settings) + if not isinstance(runner, SQLiteMigrationRunner) or runner.database != database.resolve(): + raise MigrationError("unsupported_target", "The configured database differs from the admission target.") + before = _recovery_fingerprint(runner.database) + plan = runner.admit_snapshot(snapshot) + if _recovery_fingerprint(runner.database) != before: + raise MigrationError("recovery_required", "The maintenance evidence changed during admission.") + if asdict(load_protected_environment_file(Path(identity.path)).identity) != asdict(identity): + raise MigrationError("configuration_changed", "The configuration changed during admission.") + return { + "state": "ready", + "reason": "server_ready", + "fingerprint": digest({"plan": plan.plan_id, "evidence": before}), + } + + +def main() -> int: + """Read one bounded private request and return safe codes, never exception values.""" + from powercontext.server.configuration import ServerConfigurationError + + try: + content = sys.stdin.buffer.read(_MAX_REQUEST_BYTES + 1) + if len(content) > _MAX_REQUEST_BYTES: + raise ValueError # noqa: TRY301 + result = _admit(json.loads(content)) + except MigrationError as error: + result = {"state": "rejected", "reason": error.code} + except ServerConfigurationError: + result = {"state": "rejected", "reason": "configuration_invalid"} + except Exception: + # Import, dependency, malformed-input and unsupported-target failures + # must not expose configuration, database contents or provider messages. + result = {"state": "rejected", "reason": "admission_unsupported"} + sys.stdout.write(json.dumps({"protocol_version": 1, **result}, sort_keys=True) + "\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/e2e/test_managed_database_admission.py b/tests/e2e/test_managed_database_admission.py new file mode 100644 index 0000000000..c49c3c43f9 --- /dev/null +++ b/tests/e2e/test_managed_database_admission.py @@ -0,0 +1,282 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Real target child, full SQLite Server and public Memory; simulated OS manager/resources.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import shlex +import sqlite3 +import sys +from contextlib import closing +from importlib.metadata import version +from pathlib import Path + +import pytest + +from powercontext.builtin.persistence.migrations.server_readiness import prepare_server_storage +from powercontext.builtin.records import ArtifactWrite +from powercontext.builtin.runtime import BuiltinConfig, open_builtin_contexts +from powercontext.ops import Ops +from powercontext.ops.installer import ManagedInstaller, ManagedLayout +from powercontext.ops.local_policy import inspect_existing_configuration +from powercontext.ops.resources import PreparedResource, ResourceCatalog, validate_owned_tree, verify +from powercontext.server.configuration import server_settings_context +from powercontext.service.controller import ServiceController +from powercontext.service.environment import protect_directory, read_protected_file, write_protected_file +from powercontext.service.model import ManagerState, ServiceTarget +from tests.test_ops import _apply_request, _plan, _request +from tests.test_ops_installer import install +from tests.test_ops_resources import _archive, _spec +from tests.test_service import FakeAdapter, _manager_probe + +pytestmark = pytest.mark.skipif( + os.name != "posix", reason="POSIX test launcher delegates to the actual test interpreter" +) + + +@pytest.fixture +def managed_database(tmp_path, monkeypatch): + for name in tuple(os.environ): + if name.upper() == "POWERCONTEXT_HOME" or name.upper().startswith("POWERCONTEXT_SERVER_"): + monkeypatch.delenv(name) + monkeypatch.setattr("powercontext.ops.resources._current_platform", lambda: ("windows", "x86_64")) + launcher = f'#!/bin/sh\nexec {shlex.quote(sys.executable)} "$@"\n'.encode() + files = {"python/pythonw.exe": launcher, "python/python.exe": launcher, "LICENSE": b"local test fixture"} + spec = _spec( + _archive(files), + server_version=version("powercontext"), + unpacked_size=sum(map(len, files.values())), + files=[ + { + "path": name, + "size": len(content), + "sha256": hashlib.sha256(content).hexdigest(), + "executable": name.endswith(".exe"), + } + for name, content in files.items() + ], + ) + layout = ManagedLayout(*(tmp_path / name for name in ("programs", "config", "state", "cache", "data"))) + installer = ManagedInstaller(ResourceCatalog("b" * 64, (spec,)), spec, layout) + + def prepare(selected, staging_parent, *, cancelled): + directory = protect_directory(staging_parent / "test-resource", create=True) + for name, content in files.items(): + path = directory / name + protect_directory(path.parent, create=True) + path.write_bytes(content) + path.chmod(0o700 if name.endswith(".exe") else 0o600) + verify(selected, directory) + return PreparedResource(directory, selected, validate_owned_tree(directory)) + + monkeypatch.setattr(installer.resources, "prepare", prepare) + target = installer.target() + adapter = FakeAdapter(protect_directory(tmp_path / "manager", create=True)) + controller = ServiceController( + adapter, + target=ServiceTarget(target.python_executable, target.package_version, isolated_runtime=True), + probe=_manager_probe(adapter), + readiness_probe=_manager_probe(adapter), + ) + ops = Ops(target, layout.state, controller=controller, installer=installer) + install(ops) + assert ops.handle(_apply_request(_plan(ops, "stop"), "stop"))["outcome"] == "succeeded" + summary = inspect_existing_configuration(installer.environment, data_dir=layout.data) + with server_settings_context(env_file=installer.environment, data_dir=layout.data) as settings: + prepare_server_storage(settings) + return ops, adapter, installer, summary + + +def _files(database): + return { + suffix: (path.read_bytes(), path.stat().st_mtime_ns, path.stat().st_ctime_ns) + for suffix in ("", "-wal", "-shm", "-journal") + if (path := Path(f"{database}{suffix}")).exists() + } + + +def _runtime(installer): + with server_settings_context(env_file=installer.environment, data_dir=installer.layout.data) as settings: + return BuiltinConfig(database=settings.database) + + +async def _seed(config): + async with open_builtin_contexts(config) as contexts: + await contexts.get("admission-history") + record = await contexts.records.create_artifact( + "admission-history", + "atomic-memory", + ArtifactWrite(content={"kind": "fact", "text": "Original exact fact."}), + ) + memory = contexts.atomic_memory.for_scope("admission-history") + original = await memory.get(record.artifact_id) + await contexts.records.replace_artifact( + "admission-history", + "atomic-memory", + original.ref.artifact_id, + f'"revision:{original.ref.revision}"', + ArtifactWrite(content={"kind": "fact", "text": "Latest exact fact."}), + ) + return original, await memory.get(record.artifact_id) + + +async def _read(config, original, latest): + async with open_builtin_contexts(config) as contexts: + memory = contexts.atomic_memory.for_scope("admission-history") + assert await memory.get(latest.ref.artifact_id) == latest + assert await memory.get(original.ref.artifact_id, revision=original.ref.revision) == original + for item in (original, latest): + stored = await contexts.records.get_artifact_revision( + "admission-history", "atomic-memory", item.ref.artifact_id, item.ref.revision + ) + assert stored.content["text"] == item.artifact.content.text + + +def test_start_restart_and_program_repair_retain_exact_memory_and_configuration(managed_database): + ops, adapter, installer, summary = managed_database + config = _runtime(installer) + original, latest = asyncio.run(_seed(config)) + for operation in ("start", "restart", "config_apply"): + before = _files(summary.database_path) + options = {"restart": True, "dashboard_enabled": False} if operation == "config_apply" else {} + plan = _plan(ops, operation, **options) + assert plan["effects"]["may_start"] + assert _files(summary.database_path) == before + result = ops.handle(_apply_request(plan, operation)) + assert result["outcome"] == "succeeded", result + assert _files(summary.database_path) == before + asyncio.run(_read(config, original, latest)) + selected = Path(adapter.definition.env_file.path) + configuration = read_protected_file(selected) + Path(ops.target.python_executable).unlink() + assert ops.handle(_request("inspect"))["outcome"] == "succeeded" + assert ops.handle(_apply_request(_plan(ops, "stop"), "stop-damaged"))["outcome"] == "succeeded" + repaired = ops.handle(_apply_request(_plan(ops, "repair"), "repair")) + assert repaired["outcome"] == "succeeded", repaired + assert adapter.manager == ManagerState.INACTIVE + assert read_protected_file(selected) == configuration + restarted = ops.handle(_apply_request(_plan(ops, "start"), "start-repaired")) + assert restarted["outcome"] == "succeeded", restarted + asyncio.run(_read(config, original, latest)) + + +def test_wal_only_schema_is_admitted_without_touching_source_sidecars(managed_database, tmp_path): + ops, _adapter, installer, summary = managed_database + # Start with a fresh managed Server in WAL. Keep the first connection open + # so committed schema and records have not been checkpointed into main. + summary.database_path.unlink() + with closing(sqlite3.connect(summary.database_path)) as writer: + writer.execute("PRAGMA journal_mode=WAL") + writer.execute("PRAGMA wal_autocheckpoint=0") + writer.execute("BEGIN") + writer.execute("SELECT name FROM sqlite_master").fetchall() + with server_settings_context(env_file=installer.environment, data_dir=installer.layout.data) as settings: + # The previous fresh bootstrap's complete journal is not an adoption + # authority for a different file; create this fixture independently. + import shutil + + shutil.rmtree(Path(f"{summary.database_path}.pc-migration-state")) + prepare_server_storage(settings) + config = _runtime(installer) + original, latest = asyncio.run(_seed(config)) + assert Path(f"{summary.database_path}-wal").stat().st_size > 0 + isolated_main = tmp_path / "only-main.db" + isolated_main.write_bytes(summary.database_path.read_bytes()) + with closing(sqlite3.connect(isolated_main)) as main: + assert main.execute("SELECT name FROM sqlite_master WHERE name='pc_schema_revision'").fetchone() is None + before = _files(summary.database_path) + plan = _plan(ops, "start") + assert ops.handle(_apply_request(plan, "wal-start"))["outcome"] == "succeeded" + assert _files(summary.database_path) == before + asyncio.run(_read(config, original, latest)) + + +@pytest.mark.parametrize("changed", ["database", "configuration", "evidence", "resource"]) +def test_start_revalidates_database_configuration_evidence_and_target(managed_database, changed): + ops, adapter, installer, summary = managed_database + plan = _plan(ops, "start") + if changed == "database": + asyncio.run(_seed(_runtime(installer))) + elif changed == "configuration": + write_protected_file(installer.environment, installer.environment.read_bytes() + b"\n# revised configuration\n") + elif changed == "evidence": + path = Path(f"{summary.database_path}.pc-migration-state/maintenance.json") + path.write_text(path.read_text() + "\n") + else: + (installer.directory / "LICENSE").write_bytes(b"changed") + before = list(adapter.events) + result = ops.handle(_apply_request(plan, "stale")) + assert result["outcome"] == "rejected", result + assert result["error"]["code"] in {"plan_stale", "runtime_damaged"} + assert adapter.events == before + + +@pytest.mark.parametrize("damage", ["unknown", "future", "schema", "maintenance", "restore", "profile", "journal"]) +def test_full_server_rejects_unsupported_or_unfinished_database(managed_database, damage): + ops, adapter, installer, summary = managed_database + if damage == "maintenance": + path = Path(f"{summary.database_path}.pc-migration-state/maintenance.json") + evidence = json.loads(path.read_text()) + evidence["state"] = "active" + path.write_text(json.dumps(evidence)) + elif damage == "profile": + write_protected_file( + installer.environment, + installer.environment.read_bytes() + + ( + b"POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_MODEL=openai:fake\n" + b"POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_PROFILE_ID=different\n" + b"POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_DIMENSION=3\n" + ), + ) + elif damage == "journal": + Path(f"{summary.database_path}-journal").write_bytes(b"hot rollback journal") + else: + with closing(sqlite3.connect(summary.database_path)) as connection, connection: + if damage == "unknown": + connection.execute("DROP TABLE pc_schema_revision") + connection.execute("CREATE TABLE unfamiliar (value TEXT)") + elif damage == "future": + connection.execute("UPDATE pc_schema_revision SET version_num='future_unknown'") + elif damage == "schema": + connection.execute("DROP INDEX ix_pc_processing_intent_binding_sequence") + else: + connection.execute( + "INSERT INTO pc_portable_restore_receipts " + "(bundle_id, format_version, total_digest, status, record_count, inserted, already_present, projections_ready, updated_at) " + "VALUES ('partial',1,'sha256:digest','authoritative_restored',0,0,0,0,'2026-01-01T00:00:00Z')" + ) + before = _files(summary.database_path) + events = list(adapter.events) + result = ops.handle(_request("preflight", operation="start", options={})) + assert result["outcome"] == "rejected", result + assert ( + result["error"]["code"] + == { + "unknown": "database_compatibility_unknown", + "future": "database_incompatible", + "schema": "database_incompatible", + "maintenance": "database_recovery_required", + "restore": "database_recovery_required", + "profile": "database_configuration_mismatch", + "journal": "database_recovery_required", + }[damage] + ) + assert _files(summary.database_path) == before + assert adapter.events == events diff --git a/tests/test_ops_database_admission.py b/tests/test_ops_database_admission.py new file mode 100644 index 0000000000..cf1f736d29 --- /dev/null +++ b/tests/test_ops_database_admission.py @@ -0,0 +1,179 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Actual bounded admission subprocesses and immutable source files, without native services.""" + +from __future__ import annotations + +import os +import shlex +import signal +import sys +import time +from contextlib import suppress +from importlib.metadata import version +from pathlib import Path + +import pytest + +from powercontext.builtin.persistence.migrations.locking import local_migration_lock +from powercontext.builtin.persistence.migrations.server_readiness import prepare_server_storage +from powercontext.ops import database_admission +from powercontext.ops.database_admission import admit_database +from powercontext.ops.local_policy import inspect_existing_configuration, new_configuration, preview_new_configuration +from powercontext.server.configuration import server_settings_context +from powercontext.service.environment import write_protected_file + + +@pytest.fixture +def database(tmp_path): + data = tmp_path / "data" + environment = tmp_path / "configuration" / "server.env" + write_protected_file(environment, new_configuration(data)) + with server_settings_context(env_file=environment, data_dir=data) as settings: + prepare_server_storage(settings) + return inspect_existing_configuration(environment, data_dir=data) + + +def test_absent_database_is_not_created_or_admitted_by_a_fallback_runtime(tmp_path): + summary = preview_new_configuration(tmp_path / "absent") + result = admit_database(summary, "1.0.0", python_executable=tmp_path / "missing-python") + assert result.allowed_to_start and result.classification == "new" + assert not summary.data_dir.exists() + + +def test_target_metadata_version_must_match_the_selected_resource(database): + before = database.database_path.read_bytes() + result = admit_database(database, "0.0.0", python_executable=Path(sys.executable)) + assert not result.allowed_to_start + assert result.reason == "database_admission_unavailable" + assert database.database_path.read_bytes() == before + + +def test_active_maintenance_blocks_target_admission(database): + with local_migration_lock(database.database_path): + result = admit_database(database, version("powercontext"), python_executable=Path(sys.executable)) + assert not result.allowed_to_start + assert result.reason == "database_recovery_required" + + +def test_snapshot_budget_rejects_before_copying(database, monkeypatch): + monkeypatch.setattr(database_admission, "MAX_SNAPSHOT_BYTES", database.database_path.stat().st_size - 1) + before = database.database_path.read_bytes() + result = admit_database(database, version("powercontext"), python_executable=Path(sys.executable)) + assert not result.allowed_to_start and result.reason == "database_snapshot_too_large" + assert database.database_path.read_bytes() == before + + +def test_source_change_during_copy_cannot_be_certified(database, monkeypatch): + original = database_admission._copy + + def changing(source, destination, expected, deadline): + result = original(source, destination, expected, deadline) + with source.open("ab") as output: + output.write(b"concurrent write") + return result + + monkeypatch.setattr(database_admission, "_copy", changing) + result = admit_database(database, version("powercontext"), python_executable=Path(sys.executable)) + assert not result.allowed_to_start and result.reason == "database_changed_during_admission" + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX failure fixture delegates to the current test interpreter") +@pytest.mark.parametrize("failure", ["missing", "overflow", "timeout", "descendant_exit", "descendant_timeout"]) +def test_target_failures_and_inherited_output_remain_bounded(database, tmp_path, monkeypatch, failure): + marker = tmp_path / "descendant.pid" + statements = [] + if failure.startswith("descendant"): + statements.append( + "import pathlib, subprocess, sys; " + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)']); " + f"pathlib.Path({str(marker)!r}).write_text(str(child.pid))" + ) + if failure == "overflow": + statements.append("import os; os.write(1, b'x' * 65536)") + if failure in {"timeout", "descendant_timeout", "overflow"}: + statements.append("import time; time.sleep(30)") + else: + statements.append("raise SystemExit(1)") + wrapper = tmp_path / "target-python" + wrapper.write_text(f"#!/bin/sh\nexec {shlex.quote(sys.executable)} -I -B -c {shlex.quote('; '.join(statements))}\n") + wrapper.chmod(0o700) + monkeypatch.setattr(database_admission, "ADMISSION_TIMEOUT_SECONDS", 5) + before = database.database_path.read_bytes() + started = time.monotonic() + try: + result = admit_database(database, version("powercontext"), python_executable=wrapper) + assert time.monotonic() - started < 10 + assert not result.allowed_to_start + assert result.reason == ( + "database_admission_timeout" + if failure in {"timeout", "descendant_timeout"} + else "database_admission_unavailable" + ) + assert database.database_path.read_bytes() == before + finally: + if marker.exists(): + with suppress(ProcessLookupError): + os.kill(int(marker.read_text()), signal.SIGTERM) + + +def test_supported_old_server_layout_requires_explicit_migration(database): + import shutil + + from alembic import command + from sqlalchemy import create_engine + + from powercontext.builtin.persistence.migrations.deployment import deployment_runner, server_bundle + + database.database_path.unlink() + shutil.rmtree(Path(f"{database.database_path}.pc-migration-state")) + bundle = server_bundle(source_release="1.4.0") + engine = create_engine(f"sqlite:///{database.database_path}") + try: + with server_settings_context(env_file=Path(database.identity.path), data_dir=database.data_dir) as settings: + configuration = deployment_runner(settings).configuration + with engine.begin() as connection: + config = bundle.config(connection) + config.attributes["configuration"] = configuration + command.upgrade(config, bundle.source_revision) + connection.exec_driver_sql("DROP TABLE pc_schema_revision") + finally: + engine.dispose() + before = database.database_path.read_bytes() + result = admit_database(database, version("powercontext"), python_executable=Path(sys.executable)) + assert not result.allowed_to_start and result.reason == "database_migration_required" + assert database.database_path.read_bytes() == before + + +def test_configured_vector_profile_is_compared_with_stored_declaration(tmp_path): + environment = tmp_path / "configuration" / "server.env" + data = tmp_path / "data" + content = new_configuration(data) + ( + b"POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_MODEL=openai:fixture\n" + b"POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_PROFILE_ID=retained-profile\n" + b"POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_DIMENSION=3\n" + ) + write_protected_file(environment, content) + with server_settings_context(env_file=environment, data_dir=data) as settings: + prepare_server_storage(settings) + summary = inspect_existing_configuration(environment, data_dir=data) + before = summary.database_path.read_bytes() + admitted = admit_database(summary, version("powercontext"), python_executable=Path(sys.executable)) + assert admitted.allowed_to_start, admitted + write_protected_file(environment, content.replace(b"retained-profile", b"different-profile")) + summary = inspect_existing_configuration(environment, data_dir=data) + rejected = admit_database(summary, version("powercontext"), python_executable=Path(sys.executable)) + assert not rejected.allowed_to_start and rejected.reason == "database_configuration_mismatch" + assert summary.database_path.read_bytes() == before From 23d1100675871ea00263acbb799586e2099a8350 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Wed, 28 Oct 2026 15:34:55 +0800 Subject: [PATCH 2/3] fix(ops): normalize admission failures and clarify busy storage --- docs/en/docs/operate/managed-local-server.md | 7 ++++++- docs/zh/docs/operate/managed-local-server.md | 4 +++- src/powercontext/ops/database_admission.py | 9 ++++++--- tests/e2e/test_managed_database_admission.py | 16 +++++++++++++++ tests/test_ops_database_admission.py | 21 ++++++++++++++++++-- 5 files changed, 50 insertions(+), 7 deletions(-) diff --git a/docs/en/docs/operate/managed-local-server.md b/docs/en/docs/operate/managed-local-server.md index 3102ef9efa..50d8c39994 100644 --- a/docs/en/docs/operate/managed-local-server.md +++ b/docs/en/docs/operate/managed-local-server.md @@ -113,7 +113,7 @@ Admission performs no initialization, migration, restoration, model call, HTTP s | Observation | Startup behavior | | --- | --- | | Database and recovery evidence are absent | `new`; only the approved Server startup may initialize storage. | -| Existing storage passes the exact target's complete check | `existing_ready` for that explicit admission; normal same-version stop/start/restart is supported. | +| Existing storage passes the exact target's complete check | `existing_ready` for that explicit admission; same-version startup and restart require unchanged storage through apply. | | Unknown schema, empty existing file, older layout, future revision, unsupported profile, or missing target helper | Block startup and preserve storage. | | Interrupted maintenance/restore, rollback journal, active maintenance lock, orphan sidecars, unsafe location, or changing source | Block startup and preserve recovery evidence. | @@ -131,6 +131,11 @@ validation has a **120-second** deadline. Oversize, insufficient space, and time journals are limited to 4 MiB each. Large or busy databases require an operator-controlled maintenance workflow; raising a limit does not establish compatibility. Temporary copies are removed after the check. +Normal writes also invalidate approved database evidence. If the running Server keeps writing, or data changes +between preflight and apply, stop it through Ops, obtain a fresh `start` plan after writes have stopped, and approve +that plan. A rejected restart leaves the running service unchanged. Admission does not automatically stop writers or +relax the approved database identity; other processes writing the same database must also be quiescent. + A local macOS arm64 / CPython 3.12.15 measurement with 2,000 public Memory records, 4,000 revisions and a 168.4 MiB SQLite main file took 3.78–3.86 seconds per admission across three runs; copying and hashing took 0.074–0.075 seconds. Source hashes remained unchanged. This measurement excludes resource-manifest verification and native service diff --git a/docs/zh/docs/operate/managed-local-server.md b/docs/zh/docs/operate/managed-local-server.md index 30f3d99d42..7907f4e563 100644 --- a/docs/zh/docs/operate/managed-local-server.md +++ b/docs/zh/docs/operate/managed-local-server.md @@ -66,7 +66,7 @@ description: 安装或修复固定版本的个人 Server 运行环境，并保 | 观察结果 | 启动处理 | | --- | --- | | 数据库及恢复证据均不存在 | `new`；只有确认后的 Server 启动可以初始化存储。 | -| 已有存储通过精确目标版本的完整检查 | 本次明确准入为 `existing_ready`，支持正常的同版本停止、启动与重启。 | +| 已有存储通过精确目标版本的完整检查 | 本次明确准入为 `existing_ready`；同版本启动和重启要求存储在 apply 前保持不变。 | | 未知 schema、已有空文件、旧布局、未来版本、不支持的 profile 或缺少目标检查入口 | 阻止启动并保留存储。 | | 未完成的维护或恢复、回滚日志、正在持有的维护锁、孤立侧文件、不安全路径或源文件发生变化 | 阻止启动并保留恢复证据。 | @@ -74,6 +74,8 @@ description: 安装或修复固定版本的个人 Server 运行环境，并保 准入复制并计算完整主库与 WAL 的摘要，再由 Server 验证器扫描副本。每次预检和 apply 重新核验都会重复这些工作，不是增量检查，也不复用缓存。主库与 WAL 合计上限为 **8 GiB**；临时文件系统需要其合计大小加 **32 MiB** 的可用空间；复制和目标验证共用 **120 秒**时限。超出大小、空间不足或超时分别返回 `database_snapshot_too_large`、`database_snapshot_space_insufficient`、`database_admission_timeout`。每个恢复日志上限为 4 MiB。大型或持续写入的数据库需要由操作者安排维护流程，单纯提高限制不能证明兼容。检查结束后删除临时副本。 +正常业务写入也会使已批准的数据库证据过期。如果运行中的 Server 持续写入，或预检与 apply 之间数据发生变化，应先通过 Ops 停止服务，待写入停止后重新获取并批准 `start` 计划。被拒绝的重启不会改变正在运行的服务。准入不会自动停止写入方或放宽已批准的数据库身份；其他进程对同一数据库的写入也必须暂停。 + 一次本地 macOS arm64 / CPython 3.12.15 测量使用公开 Memory 接口创建 2,000 条记录、4,000 个修订版本，SQLite 主库为 168.4 MiB。三次准入各耗时 3.78–3.86 秒，其中复制及计算摘要耗时 0.074–0.075 秒；源文件摘要保持不变。该测量不包含资源清单核验和原生服务命令，不能证明其他机器、WAL 大小或接近上限的数据库具有同样性能。 Apply 会重新执行准入，并比较获批计划中的数据库、配置、目标资源及恢复证据；证据变化会在派发服务命令前拒绝旧计划。实际 Server 启动仍执行自己的启动检查。即使目标解释器损坏或缺失，检查、停止、移除注册和保持停止的程序修复仍可使用。原本停止的服务在配置修改后仍保持停止，即使传入 `restart=true` 也不会启动。独立的已备妥 runtime 操作者绑定保留既有契约。 diff --git a/src/powercontext/ops/database_admission.py b/src/powercontext/ops/database_admission.py index 28c32be283..0f2c2510f2 100644 --- a/src/powercontext/ops/database_admission.py +++ b/src/powercontext/ops/database_admission.py @@ -70,7 +70,10 @@ def _source_files(database: Path) -> dict[str, tuple[int, ...] | None]: def _copy(source: Path, destination: Path, expected: tuple[int, ...], deadline: float) -> str: checksum = hashlib.sha256() - descriptor = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0)) + try: + descriptor = os.open(source, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0)) + except FileNotFoundError as error: + raise OpsError("database_changed_during_admission") from error with os.fdopen(descriptor, "rb") as input_file, destination.open("xb") as output: if _identity(os.fstat(input_file.fileno())) != expected: raise OpsError("database_changed_during_admission") @@ -195,9 +198,9 @@ def _target_admission( raise OpsError("database_admission_unavailable") result = cast(dict[str, Any], decoded) if ( - not isinstance(result, dict) - or type(result.get("protocol_version")) is not int + type(result.get("protocol_version")) is not int or result["protocol_version"] != 1 + or not isinstance(result.get("state"), str) or result.get("state") not in {"ready", "rejected"} or not isinstance(result.get("reason"), str) or ( diff --git a/tests/e2e/test_managed_database_admission.py b/tests/e2e/test_managed_database_admission.py index c49c3c43f9..7d98c67947 100644 --- a/tests/e2e/test_managed_database_admission.py +++ b/tests/e2e/test_managed_database_admission.py @@ -176,6 +176,22 @@ def test_start_restart_and_program_repair_retain_exact_memory_and_configuration( asyncio.run(_read(config, original, latest)) +def test_writes_invalidate_restart_approval_and_stop_then_fresh_start_preserves_them(managed_database): + ops, adapter, installer, _summary = managed_database + assert ops.handle(_apply_request(_plan(ops, "start"), "initial-start"))["outcome"] == "succeeded" + plan = _plan(ops, "restart") + config = _runtime(installer) + original, latest = asyncio.run(_seed(config)) + before = list(adapter.events) + result = ops.handle(_apply_request(plan, "restart-after-write")) + assert result["outcome"] == "rejected" and result["error"]["code"] == "plan_stale" + assert adapter.events == before and adapter.manager == ManagerState.ACTIVE + assert ops.handle(_apply_request(_plan(ops, "stop"), "stop-writer"))["outcome"] == "succeeded" + assert adapter.manager == ManagerState.INACTIVE + assert ops.handle(_apply_request(_plan(ops, "start"), "fresh-start"))["outcome"] == "succeeded" + asyncio.run(_read(config, original, latest)) + + def test_wal_only_schema_is_admitted_without_touching_source_sidecars(managed_database, tmp_path): ops, _adapter, installer, summary = managed_database # Start with a fresh managed Server in WAL. Keep the first connection open diff --git a/tests/test_ops_database_admission.py b/tests/test_ops_database_admission.py index cf1f736d29..d93c607c9c 100644 --- a/tests/test_ops_database_admission.py +++ b/tests/test_ops_database_admission.py @@ -90,8 +90,23 @@ def changing(source, destination, expected, deadline): assert not result.allowed_to_start and result.reason == "database_changed_during_admission" +def test_disappearing_observed_source_requires_a_fresh_admission(database, monkeypatch): + original = database_admission._copy + + def disappearing(source, destination, expected, deadline): + source.unlink() + return original(source, destination, expected, deadline) + + monkeypatch.setattr(database_admission, "_copy", disappearing) + result = admit_database(database, version("powercontext"), python_executable=Path(sys.executable)) + assert not result.allowed_to_start and result.reason == "database_changed_during_admission" + assert not database.database_path.exists() + + @pytest.mark.skipif(os.name != "posix", reason="POSIX failure fixture delegates to the current test interpreter") -@pytest.mark.parametrize("failure", ["missing", "overflow", "timeout", "descendant_exit", "descendant_timeout"]) +@pytest.mark.parametrize( + "failure", ["missing", "malformed", "overflow", "timeout", "descendant_exit", "descendant_timeout"] +) def test_target_failures_and_inherited_output_remain_bounded(database, tmp_path, monkeypatch, failure): marker = tmp_path / "descendant.pid" statements = [] @@ -103,10 +118,12 @@ def test_target_failures_and_inherited_output_remain_bounded(database, tmp_path, ) if failure == "overflow": statements.append("import os; os.write(1, b'x' * 65536)") + if failure == "malformed": + statements.append("import json; print(json.dumps({'protocol_version': 1, 'state': [], 'reason': 'bad'}))") if failure in {"timeout", "descendant_timeout", "overflow"}: statements.append("import time; time.sleep(30)") else: - statements.append("raise SystemExit(1)") + statements.append(f"raise SystemExit({0 if failure == 'malformed' else 1})") wrapper = tmp_path / "target-python" wrapper.write_text(f"#!/bin/sh\nexec {shlex.quote(sys.executable)} -I -B -c {shlex.quote('; '.join(statements))}\n") wrapper.chmod(0o700) From 0803ed7e706563f46ed493004aabb8c4b0eb4da4 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Wed, 28 Oct 2026 16:09:15 +0800 Subject: [PATCH 3/3] test(claude-code): make shared deadline regression deterministic --- tests/claude_code_plugin/test_hook.py | 45 ++++++++++++++++++++++++--- 1 file changed, 40 insertions(+), 5 deletions(-) diff --git a/tests/claude_code_plugin/test_hook.py b/tests/claude_code_plugin/test_hook.py index 3320bb5c4c..86c44cba3c 100644 --- a/tests/claude_code_plugin/test_hook.py +++ b/tests/claude_code_plugin/test_hook.py @@ -21,9 +21,11 @@ import time from collections.abc import Generator from contextlib import contextmanager +from email.message import Message from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import ModuleType from typing import Any +from urllib.error import HTTPError import pytest @@ -760,7 +762,7 @@ def log_message(self, format: str, *args: Any) -> None: # noqa: A002 assert caught.value.code == "invalid_request" -def test_hook_aborts_a_slow_error_response_at_the_shared_deadline( +def test_hook_aborts_a_slow_http_error_response( hook_module: ModuleType, ) -> None: class SlowErrorHandler(BaseHTTPRequestHandler): @@ -782,7 +784,6 @@ def log_message(self, format: str, *args: Any) -> None: # noqa: A002 pass with _serve(SlowErrorHandler) as server_url: - started = time.monotonic() settings = hook_module.ClaudeCodePluginSettings( server_url=server_url, request_timeout_seconds=1.0, @@ -793,9 +794,43 @@ def log_message(self, format: str, *args: Any) -> None: # noqa: A002 "/v1/context/prepare", {}, settings=settings, - deadline=started + 0.1, + deadline=time.monotonic() + 0.1, expected_status=200, ) - elapsed = time.monotonic() - started - assert elapsed < 0.5 + +def test_hook_error_body_reads_share_one_deadline( + hook_module: ModuleType, + monkeypatch: pytest.MonkeyPatch, +) -> None: + elapsed = 0.0 + content = b'{"error":{"code":"invalid_request","message":"slow"}}' + + class SlowErrorBody(io.BytesIO): + def read(self, size: int | None = -1, /) -> bytes: + nonlocal elapsed + chunk = super().read(min(size, 1) if size is not None and size >= 0 else 1) + elapsed += 0.02 + return chunk + + body = SlowErrorBody(content) + error = HTTPError("http://127.0.0.1/v1/context/prepare", 422, "invalid request", Message(), body) + + def open_error(*args: Any, **kwargs: Any) -> None: + raise error + + # Advance only transport time so scheduler pauses cannot masquerade as + # deadline overruns. The separate HTTP test covers urllib's real error path. + monkeypatch.setattr(hook_module, "monotonic", lambda: elapsed) + monkeypatch.setattr(hook_module, "open_bounded", open_error) + with pytest.raises(hook_module._ServerUnavailableError): + hook_module._post_json( + "/v1/context/prepare", + {}, + settings=hook_module.ClaudeCodePluginSettings(request_timeout_seconds=1.0, http_budget_seconds=0.1), + deadline=0.1, + expected_status=200, + ) + + assert elapsed == pytest.approx(0.1) + assert body.tell() < len(content)