From 2098c93e897666d62250ddedb15a6558d00c930c Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Thu, 22 Oct 2026 13:16:20 +0800 Subject: [PATCH 1/4] feat(authz): recover external owner writes from durable intents Persist trusted owner and candidate attestation intents in the business transaction, then replay them outside it before returning success. Retain stable audit identity across retries, process restarts and lost acknowledgements. Expose bounded, content-free operator inventory without inferring historical owners. Constraint: Preserve immutable owners, exact candidate targets and independent read grants; retain same-database atomic writes and domain-specific security requirements. Scope-risk: Adds a Server-owned recovery table for external-store compositions. Artifact recovery requires an event-idempotent audit store; custom assemblies without commit hooks retain their existing retry guarantees. Tested: Windows ownership/Access regressions 129 passed; Linux recovery/document tests 27 passed; changed-file Ruff/format/pre-commit checks and full Linux Ty/Python 3.11 passed. SQLite first-SAVEPOINT rollback regression fails without the transaction guard. --- docs/en/docs/workflows/scopes-and-access.md | 42 ++ docs/en/rfcs/1396_handoff_access_control.md | 46 +- docs/zh/docs/workflows/scopes-and-access.md | 39 ++ docs/zh/rfcs/1396_handoff_access_control.md | 36 +- src/powercontext/server/app.py | 39 +- src/powercontext/server/authz/__init__.py | 2 + .../server/authz/owner_recovery.py | 538 ++++++++++++++++++ src/powercontext/server/authz/repository.py | 36 ++ src/powercontext/server/authz/service.py | 8 + src/powercontext/server/factory.py | 39 +- tests/e2e/test_owner_recovery_http.py | 268 +++++++++ tests/test_owner_recovery.py | 330 +++++++++++ 12 files changed, 1385 insertions(+), 38 deletions(-) create mode 100644 src/powercontext/server/authz/owner_recovery.py create mode 100644 tests/e2e/test_owner_recovery_http.py create mode 100644 tests/test_owner_recovery.py diff --git a/docs/en/docs/workflows/scopes-and-access.md b/docs/en/docs/workflows/scopes-and-access.md index b96f5ef907..d077b5a7e0 100644 --- a/docs/en/docs/workflows/scopes-and-access.md +++ b/docs/en/docs/workflows/scopes-and-access.md @@ -57,3 +57,45 @@ target's tags follows that target's write permission. Use [Tags](manage-artifact Teams use the API, MCP, or host integrations with their own identities and permissions. The Dashboard is limited to personal and demonstration deployments, with optional static token authentication. It is not a team RBAC interface; leave `POWERCONTEXT_SERVER_DASHBOARD_ENABLED=false` in team deployments. + +## Recover external ownership writes + +When `create_server_app` composes the built-in Runtime with relationship/audit stores outside its business transaction, +supported creation hooks persist ownership intents in `pc_access_owner_recovery`. A failed delivery returns 503 even +when the domain object has committed. Keep the original database: the Server retries pending intents after restart. +Do not repeat a non-idempotent create blindly, assign the retrying user as owner, or edit an intent's payload. + +An operator with access to the embedded application can inspect metadata and explicitly retry an accepted intent: + +```python +recovery = app.state.owner_recovery +if recovery is not None: + pending = await recovery.pending(scope_id=scope_id, limit=100) + for operation in pending: + await recovery.replay(operation.operation_id) + + after = None + while True: + page = await recovery.inventory_gaps(scope_id, kind="artifact", after=after, limit=100) + # Export page.items only to the trusted operator's destination. + after = page.after + if after is None: + break +``` + +`pending` also accepts an `after` operation ID for pagination. Use `kind="candidate"` to inventory proposed-owner +attestations. Inventory returns identity metadata, never Artifact or Candidate bodies. Its `after` cursor follows +scanned identities, so an empty `items` page may still have a continuation. `unknown` means no trusted intent exists; +`delivered_relation_missing` means a completed receipt no longer has its relation; `legacy_selector_required` means a +Memory collection needs an explicit entry inventory. These require an operator policy based on trusted original +evidence; replay does not accept a new owner or restore completed receipts whose external relation was deleted. +An existing independent read grant remains valid and does not fill the ownership gap. + +External Artifact recovery requires event-idempotent `append_audit_once`; missing capability fails before the domain +commit. Relationship writers must enforce their existing immutable owner/key and exact Candidate target contracts. +Conflicts remain pending for investigation. Completed receipts are retained. Keep them with the business database's +backup and retention policy. No public recovery HTTP/MCP endpoint or CLI is provided. + +The recovery tests use two separate real SQLite databases with injected delivery/acknowledgement failures, process +lifecycle restart, and concurrent replay. They do not establish acceptance for a remote relationship service, +OceanBase/MySQL, multiple Server nodes, or custom Runtime assemblies without commit hooks. diff --git a/docs/en/rfcs/1396_handoff_access_control.md b/docs/en/rfcs/1396_handoff_access_control.md index 6e8f5c7616..e05238fcfd 100644 --- a/docs/en/rfcs/1396_handoff_access_control.md +++ b/docs/en/rfcs/1396_handoff_access_control.md @@ -620,16 +620,16 @@ logical resource, one `PrincipalRef`, trusted creation time, policy revision, an contains no Artifact Revision. The owner relation is immutable; creating it again is idempotent only for the same owner and key, and a different owner returns a conflict. Ownership transfer is not part of this RFC. -In enforced mode, Artifact authorization fails closed with `artifact_owner_pending` until this relation exists. New +In enforced mode, a write requiring ownership establishment fails closed with `artifact_owner_pending` until this relation exists. New Memory entries and first Handoff commits are owned by the creating Principal. A new Experience or Skill Candidate records a Server-side proposed-owner attestation; approval establishes that Principal as owner. A Candidate targeting an existing identity must retain its existing owner. Cross-Scope publication establishes the publisher as owner of the new target identity. -The first version assumes that a deployment enables enforced mode before it persists its first Artifact. It does not -backfill or infer owners for a catalog populated while access control was disabled, and it exposes no general owner -repair workflow. Switching such a catalog to enforced mode without a separate operator migration leaves those -Artifacts unavailable by design. +Enable enforced mode before persisting the first Artifact when possible. The Server does not backfill or infer owners +for a catalog populated while access control was disabled. Historical records without trusted ownership evidence remain +unknown and require an explicit operator policy; current administrator status does not establish their original owner. +An existing independent read grant continues to authorize reads without becoming ownership. When the canonical relational repository owns relationships and audit in the business database, Artifact creation, first Handoff commit, cross-Scope publication, and Experience/Skill Candidate creation commit their owner relation or proposed-owner @@ -637,13 +637,35 @@ attestation in the business transaction. This includes generated, uploaded, impo An ownership failure rolls back the associated write; scheduled incubation also retains its Source cursor for retry. Atomic Memory, Profile generation, Dream, and supervised worker writes retain their domain-owned transaction hooks. -External relationship writers and separate audit stores cannot join that transaction. If domain persistence succeeds -but owner establishment fails in those compositions, the request still fails closed; only a business flow that -explicitly supports idempotent replay may repair the relation on retry. Local existing ownerless records are not attributed -to the current caller or reconstructed from content. They remain unavailable until an operator migration can establish -the original identity from trusted persisted evidence. Local publication replay and Handoff no-op replay report -`artifact_owner_pending` for those records without creating an owner. A general transactional outbox and operator -recovery procedure are future work outside this RFC. +External relationship writers and separate audit stores cannot join that transaction. The composed Server persists +`pc_access_owner_recovery` in the business database for creation paths with domain commit hooks: base Artifact creation, +first Handoff commit, publication, Candidate proposal/generation/upload/import, scheduled incubation, and Profile flush. +The immutable intent commits alongside the domain object and retains only the trusted initiating Principal type/ID, +exact logical resource or Candidate identity and target, idempotency key, and data-minimized audit context. It contains +no Artifact/Source bodies, model output, tokens, or credentials. + +The Server delivers accepted intents after the business transaction and before returning success. A delivery failure +returns 503 and leaves the intent durable. A background task retries bounded pages every five seconds, advancing past +failed entries. Restart and concurrent replay preserve the original identity; a different owner/key/target conflicts +rather than transferring ownership. Artifact recovery requires the external audit store's +`append_audit_once(AccessAuditEvent)` capability: identical event ID and fields return the original event, while changed +fields conflict. The canonical relational audit store implements it. If absent, the commit hook fails before the domain +commit. Candidate intents retain the immutable proposed-owner receipt; they do not establish ownership of an existing +target. Completed recovery receipts are retained and must not be independently purged while their identities remain. + +Operators can inspect `app.state.owner_recovery.pending()`, page through `inventory_gaps(scope_id)`, and call +`replay(operation_id)` from trusted embedding code; there is no owner-assignment HTTP/MCP endpoint or recovery CLI. +Inventory reads only head identities and classifies pending intents, unknown history, completed receipts whose +relations are missing, and legacy Memory collections requiring entry selectors. Replay accepts an existing intent ID, +never a caller-supplied Principal. Unknown history is not replayable. Local publication and Handoff no-op replay do not +attribute historical ownerless records to the current caller. Existing grants retain their independent read semantics. + +Candidate approval already retains its original proposed-owner attestation and approved result; retrying the same +terminal approval can establish that proposed owner without approving twice. Custom Runtime/RecordService assemblies +without transactional commit hooks retain their explicit business-retry guarantees and do not gain durable recovery +merely by injecting an external writer. Atomic Memory, Dream, and supervised worker security retain their domain-owned +transaction requirements. See [operator recovery](../docs/workflows/scopes-and-access.md#recover-external-ownership-writes) +for inspection and validation limits. ### Built-in relational persistence diff --git a/docs/zh/docs/workflows/scopes-and-access.md b/docs/zh/docs/workflows/scopes-and-access.md index dc9069a5b9..dadad49a9a 100644 --- a/docs/zh/docs/workflows/scopes-and-access.md +++ b/docs/zh/docs/workflows/scopes-and-access.md @@ -49,3 +49,42 @@ Scope 选择项目上下文边界。其不透明 ID 用于标识数据，不负 团队通过 API、MCP 或宿主集成使用各自的身份与权限。Dashboard 仅用于个人和演示部署，可选用静态 token 鉴权， 不作为 RBAC 团队入口；团队部署保持 `POWERCONTEXT_SERVER_DASHBOARD_ENABLED=false`。 + +## 恢复外部所有权写入 + +`create_server_app` 将内置 Runtime 与业务事务之外的 relationship/audit store 组合时， +受支持的创建回调在 `pc_access_owner_recovery` 保存所有权 intent。交付失败返回 503，此时业务对象可能已提交。 +保留原业务库，Server 重启后会继续重试 pending intent。不要盲目重复非幂等创建、把重试者指定为 owner，或修改 intent payload。 + +能访问嵌入应用的运营者可以查看元数据，并显式重试已有 intent： + +```python +recovery = app.state.owner_recovery +if recovery is not None: + pending = await recovery.pending(scope_id=scope_id, limit=100) + for operation in pending: + await recovery.replay(operation.operation_id) + + after = None + while True: + page = await recovery.inventory_gaps(scope_id, kind="artifact", after=after, limit=100) + # 仅向可信运营目标导出 page.items。 + after = page.after + if after is None: + break +``` + +`pending` 同样支持用 operation ID 作为 `after` 分页。使用 `kind="candidate"` 盘点拟定归属凭证。 +清单只返回 identity 元数据，不返回 Artifact 或 Candidate 正文。`after` 跟随扫描过的 identity， +因此 `items` 为空时也可能有下一页。`unknown` 表示没有可信 intent；`delivered_relation_missing` 表示完成 receipt +对应的外部 relation 已缺失；`legacy_selector_required` 表示旧 Memory collection 需要显式 entry 清单。 +这些情况需要基于可信原始证据确定 operator policy；重放不接受新 owner，也不会恢复已完成 receipt 对应的被删除 relation。 +已有独立读取授权仍有效，但不会填补所有权缺口。 + +外部 Artifact 恢复要求支持按事件幂等的 `append_audit_once`；缺少该能力会在业务提交前失败。 +Relationship writer 必须遵循原有不可变 owner/key 以及精确 Candidate target 契约。 +冲突保持 pending，供运营者调查。完成 receipt 持续保留，应纳入业务库备份和保留策略。 +没有新增公共 recovery HTTP/MCP 接口或 CLI。 + +恢复测试使用两个独立真实 SQLite 数据库，注入交付/确认失败，验证应用生命周期重启与并发重放。 +这不等于远程权限服务、OceanBase/MySQL、多 Server 节点或不具备 commit hook 的自组装 Runtime 已通过验收。 diff --git a/docs/zh/rfcs/1396_handoff_access_control.md b/docs/zh/rfcs/1396_handoff_access_control.md index 83a554f264..8419205241 100644 --- a/docs/zh/rfcs/1396_handoff_access_control.md +++ b/docs/zh/rfcs/1396_handoff_access_control.md @@ -579,25 +579,43 @@ Server 把 ownership 与普通 `AccessBinding` 分开保存。`ArtifactOwnerRela `PrincipalRef`、可信创建时间、policy revision 和 idempotency key，刻意不包含 Artifact Revision。Owner relation 不可变；只有相同 owner 和 key 的重复建立才是幂等操作，不同 owner 返回 conflict。本 RFC 不定义 ownership transfer。 -在 enforced mode 下，owner relation 建立前的 Artifact authorization 以 `artifact_owner_pending` fail closed。新 Memory +在 enforced mode 下，要求建立所有权的写入在 owner relation 建立前以 `artifact_owner_pending` fail closed。新 Memory Entry 和首次 Handoff commit 由创建者拥有。新 Experience/Skill Candidate 记录 Server-side proposed-owner attestation， 批准后再建立 ownership；以已有 identity 为 target 的 Candidate 必须保留原 owner。跨 Scope publication 的新 target identity 归 publisher 所有。 -首版假设 deployment 在持久化第一个 Artifact 前就已启用 enforced mode。它不会为 access control disabled 期间已写入的 -catalog 回填或推断 owner，也不提供通用 owner repair 流程。未经过独立 operator migration 就把这类 catalog 切换到 -enforced mode 时，其中的 Artifact 按设计保持不可用。 +建议在持久化第一个 Artifact 前启用 enforced mode。Server 不会为 access control disabled 期间已写入的 catalog +回填或推断 owner。没有可信归属证据的历史记录保持 unknown，需要显式 operator policy；当前管理员身份不能证明历史 owner。 +独立的读取授权仍可允许读取，但不会因此成为所有权。 标准关系型 repository 同时负责 relationship 和 audit，且与业务共用数据库时，Artifact 创建、首次 Handoff commit、跨 Scope publication，以及 Experience/Skill Candidate 创建，会在业务事务内提交 owner relation 或 proposed-owner attestation。 这包括生成、上传、导入和定时处理产生的 Candidate。所有权写入失败时，相关业务写入一起回滚；定时 incubation 的 Source cursor 也保持不变，供后续重试。Atomic Memory、Profile 生成、Dream 和 supervisor worker 保留各自领域已有的事务回调。 -外部 relationship writer 和独立 audit store 无法加入该事务。这类组合中，若 domain persistence 已成功但 owner -establishment 失败，请求仍 fail closed；只有明确支持幂等重放的业务流程才能通过重试修复 relation。同库已有的 owner 缺失记录 -不会被归给当前调用者，也不会根据内容重建 owner；它们保持不可用，直到独立 operator migration 能根据可信持久化证据确认 -原始身份。同库 publication 重放和 Handoff no-op 重放会对这类记录返回 `artifact_owner_pending`，不会创建 owner。 -通用 transactional outbox 和 operator recovery 流程属于本 RFC 之外的未来工作。 +外部 relationship writer 和独立 audit store 无法加入业务事务。标准组合的 Server 为具备 domain commit hook 的创建路径， +在业务库保存 `pc_access_owner_recovery`：包括基础 Artifact 创建、首次 Handoff commit、publication、Candidate +提交/生成/上传/导入、定时 incubation 和 Profile flush。不可变 intent 与业务对象同事务提交，仅保存可信发起者的 +Principal type/ID、精确逻辑 resource 或 Candidate identity/target、幂等键和最小化审计上下文； +不保存 Artifact/Source 正文、模型输出、token 或凭据。 + +Server 在业务事务结束后交付 intent，全部成功后才返回成功。交付失败返回 503，intent 持久保留。 +后台任务每五秒重试有界分页，并越过失败项。重启或并发重放保留原始身份；不同 owner/key/target 返回冲突，不转移所有权。 +Artifact 恢复要求外部 audit store 提供 `append_audit_once(AccessAuditEvent)`：相同 event ID 和字段返回原事件， +同 ID 不同字段返回冲突；标准关系型 audit store 已实现该能力。缺少该能力时，commit hook 在业务提交前失败回滚。 +Candidate intent 保留不可变拟定归属凭证，不会建立已有 target 的新所有权。完成的恢复 receipt 持续保留， +关联 identity 尚存在时不能独立清理这些 receipt。 + +运营者可通过可信嵌入代码调用 `app.state.owner_recovery.pending()`、分页 `inventory_gaps(scope_id)` 和 +`replay(operation_id)`；没有新增 owner 赋权 HTTP/MCP 接口或 recovery CLI。清单仅读取 head identity， +区分 pending、unknown、已完成 receipt 对应 relation 缺失，以及需要 entry selector 的旧 Memory collection。 +重放只接受已有 intent ID，不接受调用者指定 Principal；unknown 历史不能重放。同库 publication 和 Handoff no-op +重放不会把历史缺失 owner 归给当前调用者；已有独立授权保留原来的读取语义。 + +Candidate approval 已保存原拟定归属凭证及 approved result；对同一终态 approval 的重试可恢复拟定 owner，不会再次批准。 +不具备事务 commit hook 的自组装 Runtime/RecordService 保留各自明确的业务重试保证， +不会仅因注入外部 writer 就获得持久恢复能力。Atomic Memory、Dream 和 supervisor worker security 保留各自领域的事务要求。 +操作方法和验证边界见[外部所有权写入恢复](../docs/workflows/scopes-and-access.md#恢复外部所有权写入)。 ### 内置关系型存储 diff --git a/src/powercontext/server/app.py b/src/powercontext/server/app.py index fa165f2362..c94f680e95 100644 --- a/src/powercontext/server/app.py +++ b/src/powercontext/server/app.py @@ -842,6 +842,12 @@ ROLE_SUBJECT_TYPES, HandoffReceiptIdentity, ) +from powercontext.server.authz.owner_recovery import ( + flush_owner_writes, + owner_recovery_available, + owner_recovery_session, + owner_write_access, +) from powercontext.server.authz.profiles import ARTIFACT_FAMILY_PROFILES, artifact_family_profile from powercontext.server.authz.service import _require_allowed from powercontext.server.context import ( @@ -1365,6 +1371,7 @@ def create_app( app.state.readiness_probe = readiness_probe app.state.access_control = access_control app.state.authentication_provider = authentication_provider + app.state.owner_recovery = None app.state.access_mode = ( ("disabled" if access_control is None else access_control.mode) if access_mode is None else access_mode ) @@ -2215,9 +2222,11 @@ async def on_commit(connection, publication): nonlocal owner_committed if access is None: return + if principal is None: + raise AccessIdentityRequiredError target = publication.target if target.artifact.family != "topic-memory": - await access.with_connection(connection).establish_artifact_owner( + await owner_write_access(access, connection).establish_artifact_owner( ResourceRef.artifact( target.scope_id, family=target.artifact.family, artifact_id=target.artifact.artifact_id ), @@ -2238,6 +2247,7 @@ async def on_commit(connection, publication): ), **({"on_commit": on_commit} if access is not None else {}), ) + await flush_owner_writes() if access is None and result.target.artifact.family != "topic-memory": await _establish_created_owner( http_request, @@ -2482,8 +2492,10 @@ async def authorize_snapshot(current): async def on_commit(connection, artifact, candidate): if access is None: return - bound = access.with_connection(connection) - if artifact is not None and await bound.artifact_owner(resource) is None: + if principal is None: + raise AccessIdentityRequiredError + bound = owner_write_access(access, connection) + if artifact is not None and await access.for_read(connection).artifact_owner(resource) is None: await bound.establish_artifact_owner( resource, principal, @@ -3398,8 +3410,10 @@ async def on_commit(connection, artifact): nonlocal committed if access is None: return + if principal is None: + raise AccessIdentityRequiredError if artifact.revision == 1: - await access.with_connection(connection).establish_artifact_owner( + await owner_write_access(access, connection).establish_artifact_owner( ResourceRef.artifact(request.scope_id, family="handoff", artifact_id=artifact.artifact_id), principal, idempotency_key=f"handoff-owner:{request.scope_id}:{artifact.artifact_id}", @@ -3411,6 +3425,7 @@ async def on_commit(connection, artifact): mapping.runtime_prepared_handoff(request.handoff), **({"on_commit": on_commit} if access is not None else {}), ) + await flush_owner_writes() if access is None and result.revision == 1: await _establish_created_owner( http_request, @@ -4180,6 +4195,7 @@ async def _establish_base_artifact_owners( result: RuntimeArtifactCreated, ) -> None: if _transactional_owner_access(request) is not None: + await flush_owner_writes() return if result.family in {"topic-memory", "atomic-memory"}: return @@ -4254,6 +4270,7 @@ async def _attest_candidate_owner( target: ArtifactRef | None, ) -> None: if _transactional_owner_access(request) is not None: + await flush_owner_writes() return access = access_control_for_mode( request.app.state.access_control, @@ -4287,7 +4304,10 @@ def _transactional_owner_access(request: Request) -> AccessControlService | None access is not None and isinstance(runtime, BuiltinRuntime) and runtime.owner_write_database is not None - and access.supports_write_transaction(runtime.owner_write_database.engine) + and ( + access.supports_write_transaction(runtime.owner_write_database.engine) + or owner_recovery_available(access, runtime.owner_write_database) + ) ): return access return None @@ -4302,7 +4322,7 @@ def _artifact_owner_commit(request: Request, scope_id: str, family: str) -> dict async def on_commit(connection, artifact): resource = ResourceRef.artifact(scope_id, family=family, artifact_id=artifact.artifact_id) - await access.with_connection(connection).establish_artifact_owner( + await owner_write_access(access, connection).establish_artifact_owner( resource, principal, idempotency_key=_base_owner_idempotency_key(resource), context=context ) @@ -4317,7 +4337,7 @@ def _candidate_owner_commit(request: Request, scope_id: str) -> dict[str, Any]: async def on_commit(connection, candidate): target = candidate.target - await access.with_connection(connection).attest_candidate_owner( + await owner_write_access(access, connection).attest_candidate_owner( scope_id=scope_id, candidate_id=candidate.candidate_id, family=candidate.family, @@ -5445,7 +5465,10 @@ async def observed_endpoint(*args: Any, **kwargs: Any) -> _ResponseT | Response: started_at = perf_counter() span = _start_application_span(app, operation) try: - async with _current_scope(app, operation, args, kwargs): + async with ( + owner_recovery_session(app.state.owner_recovery), + _current_scope(app, operation, args, kwargs), + ): result = await endpoint(*args, **kwargs) except asyncio.CancelledError: _observe_application(app, operation, "cancelled", started_at) diff --git a/src/powercontext/server/authz/__init__.py b/src/powercontext/server/authz/__init__.py index 02320b77b3..f5adf83c04 100644 --- a/src/powercontext/server/authz/__init__.py +++ b/src/powercontext/server/authz/__init__.py @@ -63,6 +63,7 @@ BindingSearchRequest, BuiltinAuthorizationProvider, CreateBinding, + IdempotentAccessAuditStore, RelationshipReader, RelationshipStore, RelationshipWriter, @@ -132,6 +133,7 @@ def __getattr__(name: str) -> object: "CreateBinding", "GroupRef", "HandoffReceiptIdentity", + "IdempotentAccessAuditStore", "MemoryEntrySelector", "PrincipalRef", "RelationshipReader", diff --git a/src/powercontext/server/authz/owner_recovery.py b/src/powercontext/server/authz/owner_recovery.py new file mode 100644 index 0000000000..89bcc0fd1b --- /dev/null +++ b/src/powercontext/server/authz/owner_recovery.py @@ -0,0 +1,538 @@ +# 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. + +"""Persist trusted ownership intents alongside domain commits, then replay externally.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager, suppress +from contextvars import ContextVar +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from hashlib import sha256 +from typing import Literal, cast +from uuid import NAMESPACE_URL, uuid5 + +from pydantic import TypeAdapter, ValidationError +from sqlalchemy import Boolean, Column, MetaData, Table, Text, UniqueConstraint, insert, select, tuple_, update +from sqlalchemy.engine import RowMapping +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncConnection +from sqlalchemy.schema import CreateTable + +from powercontext.builtin.persistence.database import AsyncDatabase +from powercontext.builtin.persistence.tables import ARTIFACT_HEADS_TABLE, CANDIDATE_HEADS_TABLE, identity_string +from powercontext.server.authz.errors import ( + AccessConflictError, + AccessInvalidRequestError, + AccessUnavailableError, +) +from powercontext.server.authz.models import ( + AccessAction, + AccessAuditEvent, + AccessResourceType, + AccessRole, + ArtifactOwnerRelation, + CandidateOwnerAttestation, + PrincipalRef, + ResourceRef, +) +from powercontext.server.authz.profiles import ARTIFACT_FAMILY_PROFILES +from powercontext.server.authz.service import ( + AccessAuditContext, + AccessControlService, + IdempotentAccessAuditStore, + _access_call, + _validate_idempotency_key, +) + +_METADATA = MetaData() +OWNER_RECOVERY_TABLE = Table( + "pc_access_owner_recovery", + _METADATA, + Column("operation_id", identity_string(64), primary_key=True), + Column("deployment_id", identity_string(128), nullable=False), + Column("scope_id", identity_string(256), nullable=False), + Column("completed", Boolean, nullable=False), + Column("payload", Text, nullable=False), + # Render the pending-page index inside CREATE TABLE so concurrent startup + # never races a separate CREATE INDEX after the table becomes visible. + UniqueConstraint("deployment_id", "completed", "operation_id", name="uq_pc_access_owner_recovery_pending"), +) + + +@dataclass(frozen=True, slots=True) +class _ArtifactIntent: + relation: ArtifactOwnerRelation + context: AccessAuditContext + kind: Literal["artifact"] = "artifact" + schema: Literal[1] = 1 + + +@dataclass(frozen=True, slots=True) +class _CandidateIntent: + attestation: CandidateOwnerAttestation + kind: Literal["candidate"] = "candidate" + schema: Literal[1] = 1 + + +_Intent = _ArtifactIntent | _CandidateIntent +_CODEC = TypeAdapter(_Intent) + + +@dataclass(frozen=True, slots=True) +class PendingOwnerWrite: + """Operator metadata; no Artifact, Candidate, or Source content is included.""" + + operation_id: str + scope_id: str + kind: Literal["artifact", "candidate"] + principal: PrincipalRef + resource: ResourceRef | None + candidate_id: str | None + target: ResourceRef | None + + +@dataclass(frozen=True, slots=True) +class OwnerGap: + """An operator-visible identity, never an inferred ownership assignment.""" + + kind: Literal["artifact", "candidate"] + scope_id: str + family: str + identity: str + reason: Literal["pending", "unknown", "delivered_relation_missing", "legacy_selector_required"] + operation_id: str | None + + +@dataclass(frozen=True, slots=True) +class OwnerGapPage: + items: tuple[OwnerGap, ...] + after: tuple[str, str] | None + + +@dataclass(slots=True) +class _RecoverySession: + recovery: OwnerRecovery + operation_ids: list[str] + + +_SESSION: ContextVar[_RecoverySession | None] = ContextVar("powercontext_owner_recovery", default=None) + + +class OwnerRecovery: + """An outbox owned by one primary database and one configured deployment. + + The relationship writer must honor its existing same-owner/key idempotency + contract. Artifact recovery additionally requires event-idempotent audit. + """ + + def __init__(self, database: AsyncDatabase, access: AccessControlService) -> None: + self.database = database + self.access = access + + async def initialize(self) -> None: + async with self.database.transaction() as connection: + await connection.execute(CreateTable(OWNER_RECOVERY_TABLE, if_not_exists=True)) + + def for_transaction(self, connection: AsyncConnection) -> OwnerCommit: + if connection.engine is not self.database.engine: + raise AccessUnavailableError("owner_recovery_database_mismatch") + return OwnerCommit(self, connection) + + async def pending( + self, *, scope_id: str | None = None, after: str | None = None, limit: int = 100 + ) -> tuple[PendingOwnerWrite, ...]: + if not 1 <= limit <= 500: + raise AccessInvalidRequestError("limit") + statement = select(OWNER_RECOVERY_TABLE).where( + OWNER_RECOVERY_TABLE.c.deployment_id == self.access.deployment_id, + OWNER_RECOVERY_TABLE.c.completed.is_(False), + ) + if scope_id is not None: + statement = statement.where(OWNER_RECOVERY_TABLE.c.scope_id == scope_id) + if after is not None: + statement = statement.where(OWNER_RECOVERY_TABLE.c.operation_id > after) + statement = statement.order_by(OWNER_RECOVERY_TABLE.c.operation_id).limit(limit) + async with self.database.transaction() as connection: + rows = (await connection.execute(statement)).mappings().all() + return tuple(_pending(str(row["operation_id"]), self._decode_row(row)) for row in rows) + + async def inventory_gaps( + self, + scope_id: str, + *, + kind: Literal["artifact", "candidate"] = "artifact", + after: tuple[str, str] | None = None, + limit: int = 100, + ) -> OwnerGapPage: + """Inspect metadata in bounded pages; unknown history grants no authority.""" + if kind not in {"artifact", "candidate"}: + raise AccessInvalidRequestError("owner-recovery-kind") + if not 1 <= limit <= 500: + raise AccessInvalidRequestError("limit") + table = ARTIFACT_HEADS_TABLE if kind == "artifact" else CANDIDATE_HEADS_TABLE + identity = table.c.artifact_id if kind == "artifact" else table.c.candidate_id + families = tuple(family for family, profile in ARTIFACT_FAMILY_PROFILES.items() if profile.enabled) + statement = select(table.c.family, identity).where(table.c.scope_id == scope_id, table.c.family.in_(families)) + if kind == "candidate": + statement = statement.where(table.c.candidate_kind == "artifact") + if after is not None: + statement = statement.where(tuple_(table.c.family, identity) > after) + statement = statement.order_by(table.c.family, identity).limit(limit + 1) + async with self.database.transaction() as connection: + rows = (await connection.execute(statement)).all() + items = [] + for family, object_id in rows[:limit]: + resource = ResourceRef.artifact(scope_id, family=family, artifact_id=object_id) + if kind == "artifact" and ARTIFACT_FAMILY_PROFILES[family].share_unit == "memory_entry": + items.append(OwnerGap(kind, scope_id, family, object_id, "legacy_selector_required", None)) + continue + owner = ( + await self.access.artifact_owner(resource) + if kind == "artifact" + else await self.access.candidate_owner(scope_id, object_id) + ) + if owner is not None: + continue + key = resource.key if kind == "artifact" else f"{scope_id}\0{object_id}" + operation_id = _key(self.access.deployment_id, kind, key) + async with self.database.transaction() as connection: + completed = await connection.scalar( + select(OWNER_RECOVERY_TABLE.c.completed).where( + OWNER_RECOVERY_TABLE.c.operation_id == operation_id, + OWNER_RECOVERY_TABLE.c.deployment_id == self.access.deployment_id, + ) + ) + reason = "unknown" if completed is None else "delivered_relation_missing" if completed else "pending" + items.append( + OwnerGap(kind, scope_id, family, object_id, reason, None if completed is None else operation_id) + ) + next_after = (str(rows[limit - 1][0]), str(rows[limit - 1][1])) if len(rows) > limit else None + return OwnerGapPage(tuple(items), next_after) + + async def replay(self, operation_id: str) -> None: + """Complete an accepted intent without accepting a caller-supplied owner.""" + async with self.database.transaction() as connection: + row = ( + ( + await connection.execute( + select(OWNER_RECOVERY_TABLE).where( + OWNER_RECOVERY_TABLE.c.operation_id == operation_id, + OWNER_RECOVERY_TABLE.c.deployment_id == self.access.deployment_id, + ) + ) + ) + .mappings() + .one_or_none() + ) + if row is None or row["completed"]: + return + intent = self._decode_row(row) + # No primary transaction is held across external relationship/audit I/O. + if isinstance(intent, _ArtifactIntent): + await self._deliver_artifact(operation_id, intent) + else: + await self._deliver_candidate(intent) + async with self.database.transaction() as connection: + await connection.execute( + update(OWNER_RECOVERY_TABLE) + .where( + OWNER_RECOVERY_TABLE.c.operation_id == operation_id, + OWNER_RECOVERY_TABLE.c.deployment_id == self.access.deployment_id, + OWNER_RECOVERY_TABLE.c.payload == row["payload"], + ) + .values(completed=True) + ) + + def _decode_row(self, row: RowMapping) -> _Intent: + intent = _decode(str(row["payload"])) + if ( + row["deployment_id"] != self.access.deployment_id + or row["operation_id"] != _operation_id(self.access.deployment_id, intent) + or row["scope_id"] != _scope(intent) + ): + raise AccessUnavailableError("owner_recovery_payload_invalid") + return intent + + @asynccontextmanager + async def monitor(self) -> AsyncIterator[None]: + """Retry bounded pages, advancing past unavailable or conflicting intents.""" + + async def run() -> None: + after = None + while True: + await asyncio.sleep(5) + try: + pending = await self.pending(after=after) + except Exception: # noqa: S112 + # Leave durable work pending; do not emit external error values. + continue + for operation in pending: + try: + await self.replay(operation.operation_id) + except Exception: # noqa: S112 + # Explicit replay reports errors; this worker retains the intent. + continue + after = pending[-1].operation_id if len(pending) == 100 else None + + task = asyncio.create_task(run(), name="powercontext-owner-recovery") + try: + yield + finally: + task.cancel() + with suppress(asyncio.CancelledError): + await task + + async def _deliver_artifact(self, operation_id: str, intent: _ArtifactIntent) -> None: + audit = _idempotent_audit(self.access) + writer = self.access.relationships + if writer is None: + raise AccessUnavailableError("relationship_management_unavailable") + try: + established = await _access_call(writer.establish_artifact_owner(intent.relation)) + except AccessConflictError: + # Concurrent idempotent inserts may race at the remote unique key. + established = await self.access.artifact_owner(intent.relation.resource) + if established is None or not _same_relation(established, intent.relation): + raise AccessConflictError("artifact-owner") + context = intent.context + event = AccessAuditEvent( + cursor=None, + event_id=str(uuid5(NAMESPACE_URL, f"powercontext:owner-recovery:{operation_id}")), + occurred_at=intent.relation.established_at, + request_id=context.request_id, + transport=context.transport, + operation=context.operation, + principal=intent.relation.owner, + actor=context.actor, + action=AccessAction.ARTIFACT_WRITE, + resource=established.resource, + allowed=True, + reason_code="artifact-owner-established", + policy_revision=established.policy_revision, + target=established.owner, + role=AccessRole.ARTIFACT_OWNER, + ) + recorded = await _access_call(audit.append_audit_once(event)) + if replace(recorded, cursor=None) != event: + raise AccessConflictError("owner-recovery-audit") + + async def _deliver_candidate(self, intent: _CandidateIntent) -> None: + writer = self.access.relationships + if writer is None: + raise AccessUnavailableError("relationship_management_unavailable") + attestation = intent.attestation + try: + established = await _access_call(writer.attest_candidate_owner(attestation)) + except AccessConflictError: + established = await self.access.candidate_owner(attestation.scope_id, attestation.candidate_id) + if established != attestation: + raise AccessConflictError("candidate-owner") + + async def _stage(self, connection: AsyncConnection, intent: _Intent) -> str: + operation_id = _operation_id(self.access.deployment_id, intent) + if connection.dialect.name == "sqlite": + # Legacy sqlite3 does not BEGIN on SELECT or SAVEPOINT. Start a + # native write transaction before a nested insert can be released. + await connection.execute( + update(OWNER_RECOVERY_TABLE) + .where(OWNER_RECOVERY_TABLE.c.operation_id == operation_id) + .values(completed=OWNER_RECOVERY_TABLE.c.completed) + ) + statement = ( + select(OWNER_RECOVERY_TABLE.c.payload) + .where(OWNER_RECOVERY_TABLE.c.operation_id == operation_id) + .with_for_update() + ) + existing = await connection.scalar(statement) + if existing is not None: + _require_same_intent(_decode(existing), intent) + _track(self, operation_id) + return operation_id + try: + async with connection.begin_nested(): + await connection.execute( + insert(OWNER_RECOVERY_TABLE).values( + operation_id=operation_id, + deployment_id=self.access.deployment_id, + scope_id=_scope(intent), + completed=False, + payload=_CODEC.dump_json(intent).decode(), + ) + ) + except IntegrityError: + existing = await connection.scalar(statement) + if existing is None: + raise + _require_same_intent(_decode(existing), intent) + _track(self, operation_id) + return operation_id + + +@asynccontextmanager +async def owner_recovery_session(recovery: OwnerRecovery | None) -> AsyncIterator[None]: + """Deliver this operation's staged intents before its response succeeds.""" + if recovery is None: + yield + return + token = _SESSION.set(_RecoverySession(recovery, [])) + try: + yield + await flush_owner_writes() + finally: + _SESSION.reset(token) + + +async def flush_owner_writes() -> None: + session = _SESSION.get() + if session is not None: + while session.operation_ids: + await session.recovery.replay(session.operation_ids[0]) + session.operation_ids.pop(0) + + +def owner_recovery_available(access: AccessControlService, database: AsyncDatabase) -> bool: + session = _SESSION.get() + return session is not None and session.recovery.access is access and session.recovery.database is database + + +def owner_write_access(access: AccessControlService, connection: AsyncConnection) -> AccessControlService | OwnerCommit: + if access.supports_write_transaction(connection.engine): + return access.with_connection(connection) + session = _SESSION.get() + if session is None or session.recovery.access is not access: + raise AccessUnavailableError("owner_recovery_unavailable") + return session.recovery.for_transaction(connection) + + +class OwnerCommit: + """The two ownership operations that may be staged inside a domain commit.""" + + def __init__(self, recovery: OwnerRecovery, connection: AsyncConnection) -> None: + self.recovery = recovery + self.connection = connection + + async def establish_artifact_owner( + self, resource: ResourceRef, owner: PrincipalRef, *, idempotency_key: str, context: AccessAuditContext + ) -> str: + if resource.type is not AccessResourceType.ARTIFACT: + raise AccessInvalidRequestError("artifact-identity") + _validate_idempotency_key(idempotency_key) + _idempotent_audit(self.recovery.access) + relation = ArtifactOwnerRelation( + resource=resource, + owner=_identity(owner), + established_at=datetime.now(UTC), + policy_revision="pending", + idempotency_key=idempotency_key, + ) + context = replace(context, actor=None if context.actor is None else _identity(context.actor), subject_groups=()) + return await self.recovery._stage(self.connection, _ArtifactIntent(relation, context)) + + async def attest_candidate_owner( + self, + *, + scope_id: str, + candidate_id: str, + family: str, + proposed_owner: PrincipalRef, + target: ResourceRef | None, + idempotency_key: str, + ) -> str: + _validate_idempotency_key(idempotency_key) + attestation = CandidateOwnerAttestation( + scope_id, candidate_id, family, _identity(proposed_owner), target, idempotency_key + ) + return await self.recovery._stage(self.connection, _CandidateIntent(attestation)) + + +def _idempotent_audit(access: AccessControlService) -> IdempotentAccessAuditStore: + if not callable(getattr(access.audit, "append_audit_once", None)): + raise AccessUnavailableError("owner_recovery_audit_idempotency_unavailable") + return cast("IdempotentAccessAuditStore", access.audit) + + +def _track(recovery: OwnerRecovery, operation_id: str) -> None: + session = _SESSION.get() + if session is not None and session.recovery is recovery and operation_id not in session.operation_ids: + session.operation_ids.append(operation_id) + + +def _identity(principal: PrincipalRef) -> PrincipalRef: + return PrincipalRef(type=principal.type, id=principal.id) + + +def _same_relation(left: ArtifactOwnerRelation, right: ArtifactOwnerRelation) -> bool: + return ( + left.resource == right.resource + and _identity(left.owner) == _identity(right.owner) + and left.idempotency_key == right.idempotency_key + ) + + +def _require_same_intent(left: _Intent, right: _Intent) -> None: + if isinstance(left, _ArtifactIntent) and isinstance(right, _ArtifactIntent): + same = _same_relation(left.relation, right.relation) + else: + same = isinstance(left, _CandidateIntent) and isinstance(right, _CandidateIntent) and left == right + if not same: + raise AccessConflictError("owner-recovery-intent") + + +def _scope(intent: _Intent) -> str: + if isinstance(intent, _CandidateIntent): + return intent.attestation.scope_id + if intent.relation.resource.scope_id is None: + raise AccessInvalidRequestError("artifact-identity") + return intent.relation.resource.scope_id + + +def _operation_id(deployment_id: str, intent: _Intent) -> str: + key = ( + intent.relation.resource.key + if isinstance(intent, _ArtifactIntent) + else f"{intent.attestation.scope_id}\0{intent.attestation.candidate_id}" + ) + return _key(deployment_id, intent.kind, key) + + +def _key(deployment_id: str, kind: str, logical_key: str) -> str: + return sha256(f"{deployment_id}\0{kind}\0{logical_key}".encode()).hexdigest() + + +def _decode(payload: str) -> _Intent: + try: + return _CODEC.validate_json(payload) + except ValidationError as error: + raise AccessUnavailableError("owner_recovery_payload_invalid") from error + + +def _pending(operation_id: str, intent: _Intent) -> PendingOwnerWrite: + if isinstance(intent, _ArtifactIntent): + return PendingOwnerWrite( + operation_id, _scope(intent), "artifact", intent.relation.owner, intent.relation.resource, None, None + ) + attestation = intent.attestation + return PendingOwnerWrite( + operation_id, + attestation.scope_id, + "candidate", + attestation.proposed_owner, + None, + attestation.candidate_id, + attestation.target, + ) diff --git a/src/powercontext/server/authz/repository.py b/src/powercontext/server/authz/repository.py index d600e0c849..4067369e89 100644 --- a/src/powercontext/server/authz/repository.py +++ b/src/powercontext/server/authz/repository.py @@ -897,6 +897,42 @@ async def append_audits(self, events: Sequence[AccessAuditEvent]) -> None: async with self._database.connection(self._bound_connection) as connection: await connection.execute(insert(ACCESS_AUDIT_EVENTS_TABLE), [_audit_row(event) for event in events]) + async def append_audit_once(self, event: AccessAuditEvent, /) -> AccessAuditEvent: + """Keep a recovery event idempotent across a lost acknowledgement.""" + async with self._database.connection(self._bound_connection) as connection: + if connection.dialect.name == "sqlite": + # A released first SAVEPOINT must not commit a caller's transaction. + await connection.execute( + update(ACCESS_AUDIT_EVENTS_TABLE) + .where(ACCESS_AUDIT_EVENTS_TABLE.c.event_id == event.event_id) + .values(event_id=ACCESS_AUDIT_EVENTS_TABLE.c.event_id) + ) + try: + async with connection.begin_nested(): + result = await connection.execute(insert(ACCESS_AUDIT_EVENTS_TABLE).values(_audit_row(event))) + key = result.inserted_primary_key + if key is None or key[0] is None: + raise RuntimeError("Access audit insert did not return a cursor") # noqa: TRY003 + return replace(event, cursor=int(key[0])) + except IntegrityError: + row = ( + ( + await connection.execute( + select(ACCESS_AUDIT_EVENTS_TABLE) + .where(ACCESS_AUDIT_EVENTS_TABLE.c.event_id == event.event_id) + .with_for_update() + ) + ) + .mappings() + .one_or_none() + ) + if row is None: + raise + existing = _decode_audit(row) + if replace(existing, cursor=None) != replace(event, cursor=None): + raise AccessConflictError("owner-recovery-audit") from None + return existing + async def list_audit( self, *, diff --git a/src/powercontext/server/authz/service.py b/src/powercontext/server/authz/service.py index d49e9dbb97..332ba9a537 100644 --- a/src/powercontext/server/authz/service.py +++ b/src/powercontext/server/authz/service.py @@ -332,6 +332,14 @@ async def list_audit( ) -> tuple[AccessAuditEvent, ...]: ... +class IdempotentAccessAuditStore(AccessAuditStore, Protocol): + """Optional recovery capability: one immutable event per stable event_id.""" + + async def append_audit_once(self, event: AccessAuditEvent, /) -> AccessAuditEvent: + """Return an identical replay, rejecting a reused ID with different fields.""" + ... + + class AccessRepository(RelationshipReader, RelationshipWriter, AccessAuditStore, Protocol): """Read requirements used by the built-in Provider.""" diff --git a/src/powercontext/server/factory.py b/src/powercontext/server/factory.py index af56ad2ee7..5abf82da06 100644 --- a/src/powercontext/server/factory.py +++ b/src/powercontext/server/factory.py @@ -74,6 +74,7 @@ access_control_for_mode, ) from powercontext.server.authz.composition import open_builtin_access_control +from powercontext.server.authz.owner_recovery import OwnerRecovery, owner_recovery_session, owner_write_access from powercontext.server.authz.repository import RelationalAccessRepository from powercontext.server.context import current_principal, current_request_id from powercontext.server.cursor_secret import resolve_cursor_secret @@ -256,6 +257,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: ), ) ) + if ( + active_access_control is not None + and runtime.owner_write_database is not None + and not active_access_control.supports_write_transaction(runtime.owner_write_database.engine) + ): + recovery = OwnerRecovery(runtime.owner_write_database, active_access_control) + await recovery.initialize() + app.state.owner_recovery = recovery + await resources.enter_async_context(recovery.monitor()) identity_repository = ServerIdentityRepository(runtime.primary_database) await identity_repository.initialize() server_id = await identity_repository.load_or_create() @@ -303,6 +313,7 @@ def current_capabilities() -> Capabilities: app.state.capability_provider = None app.state.access_control = configured_access_control app.state.authentication_provider = configured_authentication + app.state.owner_recovery = None app.state.server_info = None app.state.capabilities = Capabilities( source_types=[], @@ -457,7 +468,7 @@ async def incubate_experience(scope_id: str, runtime: BuiltinRuntime) -> Experie await access.require(principal, AccessAction.SCOPE_CONTRIBUTE, ResourceRef.scope(scope_id), context=context) async def on_commit(connection, candidates): - bound = access.with_connection(connection) + bound = owner_write_access(access, connection) for candidate in candidates: await bound.attest_candidate_owner( scope_id=scope_id, @@ -474,10 +485,14 @@ async def on_commit(connection, candidates): idempotency_key=f"background-candidate-owner:{scope_id}:{candidate.candidate_id}", ) - if runtime.owner_write_database is not None and access.supports_write_transaction( - runtime.owner_write_database.engine - ): - return await runtime.experience.for_scope(scope_id).incubate(on_commit=on_commit) + if runtime.owner_write_database is not None: + recovery = ( + None + if access.supports_write_transaction(runtime.owner_write_database.engine) + else OwnerRecovery(runtime.owner_write_database, access) + ) + async with owner_recovery_session(recovery): + return await runtime.experience.for_scope(scope_id).incubate(on_commit=on_commit) result = await runtime.experience.for_scope(scope_id).incubate() for candidate_id in result.candidate_ids: await access.attest_candidate_owner( @@ -514,8 +529,8 @@ async def authorize_snapshot(current): await access.require(principal, AccessAction.ARTIFACT_WRITE, resource, context=context) async def on_commit(connection, artifact, candidate): - bound = access.with_connection(connection) - if artifact is not None and await bound.artifact_owner(resource) is None: + bound = owner_write_access(access, connection) + if artifact is not None and await access.for_read(connection).artifact_owner(resource) is None: await bound.establish_artifact_owner( resource, principal, @@ -532,9 +547,15 @@ async def on_commit(connection, artifact, candidate): idempotency_key=f"candidate-owner:{scope_id}:{candidate.candidate_id}", ) - return await profiles.flush( - scope_id, high_watermark=high, authorize_snapshot=authorize_snapshot, on_commit=on_commit + recovery = ( + None + if access.supports_write_transaction(profiles.database.engine) + else OwnerRecovery(profiles.database, access) ) + async with owner_recovery_session(recovery): + return await profiles.flush( + scope_id, high_watermark=high, authorize_snapshot=authorize_snapshot, on_commit=on_commit + ) return run diff --git a/tests/e2e/test_owner_recovery_http.py b/tests/e2e/test_owner_recovery_http.py new file mode 100644 index 0000000000..0c84afb83f --- /dev/null +++ b/tests/e2e/test_owner_recovery_http.py @@ -0,0 +1,268 @@ +# 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. + +"""HTTP writes recover through an independent, fault-injected relationship store.""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from typing import cast + +import httpx + +from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.records import ArtifactWrite +from powercontext.builtin.runtime import RuntimeConfig +from powercontext.server.authentication import AuthenticationResult, ProviderReadiness +from powercontext.server.authz import AccessUnavailableError, PrincipalRef, RelationshipStore, ResourceRef +from powercontext.server.authz.composition import open_builtin_access_control +from powercontext.server.factory import create_server_app +from powercontext.server.settings import AccessControlConfig, McpConfig, MetricsConfig, ServerSettings + +ALICE = PrincipalRef(type="user", id="alice") + + +class _Authentication: + async def authenticate(self, request): + identity = request.headers.get("authorization", "Bearer alice").removeprefix("Bearer ") + return AuthenticationResult(subject=PrincipalRef(type="user", id=identity)) + + async def readiness(self): + return ProviderReadiness(ready=True) + + +class _InterruptedRelationships: + def __init__(self, repository): + self.repository = repository + self.unavailable = False + + def __getattr__(self, name): + return getattr(self.repository, name) + + async def establish_artifact_owner(self, relation): + if self.unavailable: + raise AccessUnavailableError("external_owner_unavailable") + return await self.repository.establish_artifact_owner(relation) + + async def attest_candidate_owner(self, attestation): + if self.unavailable: + raise AccessUnavailableError("external_owner_unavailable") + return await self.repository.attest_candidate_owner(attestation) + + +@asynccontextmanager +async def _server(tmp_path, *, interrupted=False): + primary = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'domain.db'}") + external = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'relationships.db'}") + async with open_builtin_access_control( + external, bootstrap_administrators=(ALICE,), deployment_id="external-test" + ) as access: + writer = _InterruptedRelationships(access.relationships) + access.relationships = cast("RelationshipStore", writer) + app = create_server_app( + settings=ServerSettings( + database=primary, + runtime=RuntimeConfig(artifact_processing_families=()), + access=AccessControlConfig(mode="enforced", deployment_id="external-test"), + mcp=McpConfig(enabled=False), + metrics=MetricsConfig(enabled=False), + ), + access_control=access, + authentication_provider=_Authentication(), + scheduler_path=tmp_path / "scheduler.db", + ) + async with ( + app.router.lifespan_context(app), + httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client, + ): + writer.unavailable = interrupted + yield app, client, access + + +async def _scope(client): + response = await client.post( + "/v1/scopes", + json={"title": "External ownership", "summary": "Recovery", "idempotency_key": "external-scope"}, + ) + assert response.status_code == 201, response.text + return response.json()["scope_id"] + + +def test_http_creation_and_restart_recover_the_original_owner(tmp_path): + async def scenario(): + async with _server(tmp_path, interrupted=True) as (app, client, access): + scope_id = await _scope(client) + response = await client.post( + f"/v1/scopes/{scope_id}/artifacts", + json={"family": "profile", "content": {"content": "# Original private profile"}}, + ) + assert response.status_code == 503, response.text + pending = await app.state.owner_recovery.pending() + assert len(pending) == 1 and pending[0].principal == ALICE + resource = ResourceRef.artifact(scope_id, family="profile", artifact_id="profile") + assert await access.artifact_owner(resource) is None + records = await app.state.application.records.for_scope(scope_id).logical_artifacts() + assert len(records) == 1 + async with _server(tmp_path) as (app, client, access): + # No request Principal is bound while the durable background task repairs. + async with asyncio.timeout(12): + while await app.state.owner_recovery.pending(): # noqa: ASYNC110 - poll durable state, no process-local event + await asyncio.sleep(0.05) + assert await app.state.owner_recovery.pending() == () + relation = await access.artifact_owner(resource) + assert relation is not None and relation.owner == ALICE + restored = await client.get(f"/v1/scopes/{scope_id}/artifacts/profile/profile") + assert restored.status_code == 200, restored.text + assert restored.json()["content"]["content"] == "# Original private profile\n" + denied = await client.get( + f"/v1/scopes/{scope_id}/artifacts/profile/profile", headers={"Authorization": "Bearer bob"} + ) + assert denied.status_code == 403, denied.text + records = await app.state.application.records.for_scope(scope_id).logical_artifacts() + assert len(records) == 1 + audit = await access.audit.list_audit(resource=resource, limit=100) + owners = [record for record in audit if record.reason_code == "artifact-owner-established"] + assert len(owners) == 1 and owners[0].principal == ALICE + + asyncio.run(scenario()) + + +def test_candidate_http_failure_retains_a_replayable_proposer_attestation(tmp_path): + async def scenario(): + async with _server(tmp_path, interrupted=True) as (app, client, access): + scope_id = await _scope(client) + source = await client.post( + f"/v1/scopes/{scope_id}/sources", json={"content": "A verified recovery observation."} + ) + assert source.status_code == 201, source.text + response = await client.post( + "/v1/experience/propose", + json={ + "scope_id": scope_id, + "proposal": { + "situation": "An external owner write fails.", + "action": "Persist a trusted recovery intent.", + "outcome": "The original identity can be restored.", + "lesson": "Do not infer ownership from the retrying caller.", + }, + "source_refs": [{"name": "content", "source_id": source.json()["source_id"]}], + "artifact_refs": [], + }, + ) + assert response.status_code == 503, response.text + pending = await app.state.owner_recovery.pending() + assert len(pending) == 1 and pending[0].kind == "candidate" + candidate_id = pending[0].candidate_id + assert candidate_id is not None + assert await access.candidate_owner(scope_id, candidate_id) is None + async with _server(tmp_path) as (app, client, access): + pending = await app.state.owner_recovery.pending() + await app.state.owner_recovery.replay(pending[0].operation_id) + attestation = await access.candidate_owner(scope_id, candidate_id) + assert attestation is not None and attestation.proposed_owner == ALICE + response = await client.post( + "/v1/candidates/get", json={"scope_id": scope_id, "candidate_id": candidate_id} + ) + assert response.status_code == 200, response.text + assert response.json()["candidate_id"] == candidate_id + + asyncio.run(scenario()) + + +def test_unknown_historical_owner_inventory_never_returns_content_or_assigns_a_caller(tmp_path): + async def scenario(): + async with _server(tmp_path) as (app, client, access): + scope_id = await _scope(client) + records = app.state.application.records.for_scope(scope_id) + await records.create_artifact("profile", ArtifactWrite(content={"content": "# HISTORICAL_PRIVATE_CONTENT"})) + page = await app.state.owner_recovery.inventory_gaps(scope_id) + assert len(page.items) == 1 and page.after is None + gap = page.items[0] + assert gap.identity == "profile" and gap.reason == "unknown" and gap.operation_id is None + assert "HISTORICAL_PRIVATE_CONTENT" not in repr(page) + assert await app.state.owner_recovery.pending() == () + # An explicit read grant remains valid without inferring ownership. + response = await client.get( + f"/v1/scopes/{scope_id}/artifacts/profile/profile", + headers={"Authorization": "Bearer bob"}, + ) + assert response.status_code == 403, response.text + assert "HISTORICAL_PRIVATE_CONTENT" not in response.text + assert ( + await access.artifact_owner(ResourceRef.artifact(scope_id, family="profile", artifact_id="profile")) + is None + ) + + asyncio.run(scenario()) + + +def test_inventory_paginates_metadata_without_loading_historical_bodies(tmp_path): + async def scenario(): + async with _server(tmp_path) as (app, client, _access): + scope_id = await _scope(client) + records = app.state.application.records.for_scope(scope_id) + created = [] + for index in range(3): + result = await records.create_artifact( + "experience", + ArtifactWrite( + content={ + "situation": f"Private historical observation {index}", + "action": "Observe the prior work.", + "outcome": "The creator was not recorded.", + "lesson": "Do not guess the owner.", + } + ), + ) + created.append(result.artifact_id) + after = None + identities = [] + while True: + page = await app.state.owner_recovery.inventory_gaps(scope_id, after=after, limit=1) + assert len(page.items) == 1 and page.items[0].reason == "unknown" + assert "Private historical observation" not in repr(page) + identities.append(page.items[0].identity) + if page.after is None: + break + after = page.after + assert sorted(identities) == sorted(created) + + asyncio.run(scenario()) + + +def test_non_idempotent_external_audit_fails_before_the_domain_commit(tmp_path): + class AuditWithoutReplay: + append_audit_once = None + + def __init__(self, audit): + self.audit = audit + + def __getattr__(self, name): + return getattr(self.audit, name) + + async def scenario(): + async with _server(tmp_path) as (app, client, access): + scope_id = await _scope(client) + access.audit = AuditWithoutReplay(access.audit) + response = await client.post( + f"/v1/scopes/{scope_id}/artifacts", + json={"family": "profile", "content": {"content": "# Rejected before commit"}}, + ) + assert response.status_code == 503, response.text + assert await app.state.owner_recovery.pending() == () + records = await app.state.application.records.for_scope(scope_id).logical_artifacts() + assert records == () + + asyncio.run(scenario()) diff --git a/tests/test_owner_recovery.py b/tests/test_owner_recovery.py new file mode 100644 index 0000000000..2f5c0e0a06 --- /dev/null +++ b/tests/test_owner_recovery.py @@ -0,0 +1,330 @@ +# 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. + +"""Recovery uses a separate real relationship database and the original identity.""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from dataclasses import replace + +import pytest +from sqlalchemy import Column, MetaData, String, Table, event, insert, select, update + +from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile +from powercontext.server.authz import ( + AccessAuditContext, + AccessConflictError, + AccessUnavailableError, + PrincipalRef, + ResourceRef, +) +from powercontext.server.authz.composition import open_builtin_access_control +from powercontext.server.authz.owner_recovery import OWNER_RECOVERY_TABLE, OwnerRecovery + +ALICE = PrincipalRef(type="user", id="alice") +BOB = PrincipalRef(type="user", id="bob") +RESOURCE = ResourceRef.artifact("scope", family="profile", artifact_id="profile") +CONTEXT = AccessAuditContext(transport="http", operation="create_artifact", request_id="original-request") +DOMAIN_FAILURE = "domain failed" +PRIVATE_AUDIT_FAILURE = "private audit backend detail" +ACK_FAILURE = "primary acknowledgement failed" +METADATA = MetaData() +OBJECTS = Table("test_domain_objects", METADATA, Column("identity", String(128), primary_key=True)) + + +@asynccontextmanager +async def _stores(tmp_path): + primary = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'primary.db'}") + external = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'external.db'}") + async with ( + SQLiteProfile.open(primary, tables=(OBJECTS,)) as profile, + open_builtin_access_control(external, deployment_id="recovery-test") as access, + ): + recovery = OwnerRecovery(profile.database, access) + await recovery.initialize() + yield profile.database, access, recovery + + +async def _create(database, recovery): + async with database.transaction() as connection: + await connection.execute(insert(OBJECTS).values(identity="profile")) + operation = await recovery.for_transaction(connection).establish_artifact_owner( + RESOURCE, ALICE, idempotency_key="original-owner", context=CONTEXT + ) + return operation + + +def test_domain_rollback_also_discards_recovery_intent(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + with pytest.raises(RuntimeError, match="domain failed"): + async with database.transaction() as connection: + await connection.execute(insert(OBJECTS).values(identity="profile")) + await recovery.for_transaction(connection).establish_artifact_owner( + RESOURCE, ALICE, idempotency_key="original-owner", context=CONTEXT + ) + raise RuntimeError(DOMAIN_FAILURE) + async with database.transaction() as connection: + assert (await connection.execute(select(OBJECTS))).all() == [] + assert await recovery.pending() == () + assert await access.artifact_owner(RESOURCE) is None + + asyncio.run(scenario()) + + +def test_staging_before_the_first_domain_write_does_not_commit_early(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, _access, recovery): + with pytest.raises(RuntimeError, match="domain failed"): + async with database.transaction() as connection: + await recovery.for_transaction(connection).establish_artifact_owner( + RESOURCE, ALICE, idempotency_key="original-owner", context=CONTEXT + ) + raise RuntimeError(DOMAIN_FAILURE) + assert await recovery.pending() == () + + asyncio.run(scenario()) + + +def test_recovery_survives_reopen_without_the_retrying_principal(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + await _create(database, recovery) + assert await access.artifact_owner(RESOURCE) is None + pending = await recovery.pending() + assert len(pending) == 1 + operation_id = pending[0].operation_id + async with _stores(tmp_path) as (database, access, recovery): + await recovery.replay(operation_id) + relation = await access.artifact_owner(RESOURCE) + assert relation is not None and relation.owner == ALICE + assert relation.idempotency_key == "original-owner" + assert await recovery.pending() == () + async with database.transaction() as connection: + assert len((await connection.execute(select(OBJECTS))).all()) == 1 + audits = await access.audit.list_audit(resource=RESOURCE) + assert len(audits) == 1 + assert audits[0].principal == ALICE + assert audits[0].request_id == "original-request" + await recovery.replay(operation_id) + assert len(await access.audit.list_audit(resource=RESOURCE)) == 1 + + asyncio.run(scenario()) + + +def test_audit_failure_after_external_owner_commit_is_recoverable(tmp_path, monkeypatch): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + await _create(database, recovery) + operation_id = (await recovery.pending())[0].operation_id + original = access.audit.append_audit_once + + async def unavailable(_event): + raise RuntimeError(PRIVATE_AUDIT_FAILURE) + + monkeypatch.setattr(access.audit, "append_audit_once", unavailable) + with pytest.raises(Exception, match="unavailable"): + await recovery.replay(operation_id) + relation = await access.artifact_owner(RESOURCE) + assert relation is not None and relation.owner == ALICE + assert len(await recovery.pending()) == 1 + monkeypatch.setattr(access.audit, "append_audit_once", original) + restarted = OwnerRecovery(database, access) + await restarted.replay(operation_id) + assert await restarted.pending() == () + assert len(await access.audit.list_audit(resource=RESOURCE)) == 1 + + asyncio.run(scenario()) + + +def test_lost_ack_replays_the_same_audit_event_once(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + await _create(database, recovery) + operation_id = (await recovery.pending())[0].operation_id + + def lose_ack(_connection, _cursor, statement, _parameters, _context, _many): + if statement.startswith("UPDATE pc_access_owner_recovery"): + raise RuntimeError(ACK_FAILURE) + + event.listen(database.engine.sync_engine, "before_cursor_execute", lose_ack) + try: + with pytest.raises(RuntimeError, match="acknowledgement"): + await recovery.replay(operation_id) + finally: + event.remove(database.engine.sync_engine, "before_cursor_execute", lose_ack) + assert len(await recovery.pending()) == 1 + first = await access.audit.list_audit(resource=RESOURCE) + assert len(first) == 1 + await recovery.replay(operation_id) + assert await access.audit.list_audit(resource=RESOURCE) == first + assert await recovery.pending() == () + + asyncio.run(scenario()) + + +def test_existing_owner_conflict_is_not_transferred_or_acknowledged(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + await _create(database, recovery) + await access.establish_artifact_owner(RESOURCE, BOB, idempotency_key="conflicting-owner", context=CONTEXT) + operation_id = (await recovery.pending())[0].operation_id + with pytest.raises(AccessConflictError): + await recovery.replay(operation_id) + relation = await access.artifact_owner(RESOURCE) + assert relation is not None and relation.owner == BOB + assert len(await recovery.pending()) == 1 + assert len(await access.audit.list_audit(resource=RESOURCE)) == 1 + + asyncio.run(scenario()) + + +def test_concurrent_recovery_keeps_one_owner_and_one_audit(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + await _create(database, recovery) + operation_id = (await recovery.pending())[0].operation_id + second = OwnerRecovery(database, access) + await asyncio.gather(recovery.replay(operation_id), second.replay(operation_id)) + assert await recovery.pending() == () + relation = await access.artifact_owner(RESOURCE) + assert relation is not None and relation.owner == ALICE + assert len(await access.audit.list_audit(resource=RESOURCE)) == 1 + + asyncio.run(scenario()) + + +def test_candidate_recovery_retains_exact_existing_target(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + async with database.transaction() as connection: + await connection.execute(insert(OBJECTS).values(identity="candidate")) + await recovery.for_transaction(connection).attest_candidate_owner( + scope_id="scope", + candidate_id="candidate", + family="profile", + proposed_owner=ALICE, + target=RESOURCE, + idempotency_key="candidate-owner", + ) + operation_id = (await recovery.pending())[0].operation_id + await recovery.replay(operation_id) + attestation = await access.candidate_owner("scope", "candidate") + assert attestation is not None + assert attestation.target == RESOURCE and attestation.proposed_owner == ALICE + assert await access.artifact_owner(RESOURCE) is None + + asyncio.run(scenario()) + + +def test_retry_cannot_replace_the_staged_principal_or_target(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, _access, recovery): + await _create(database, recovery) + with pytest.raises(AccessConflictError): + async with database.transaction() as connection: + await recovery.for_transaction(connection).establish_artifact_owner( + RESOURCE, BOB, idempotency_key="original-owner", context=CONTEXT + ) + await recovery.replay((await recovery.pending())[0].operation_id) + + asyncio.run(scenario()) + + +def test_normal_retry_and_recovery_preserve_the_original_relation(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + operation_id = await _create(database, recovery) + + async def retry(): + async with database.transaction() as connection: + repeated = await recovery.for_transaction(connection).establish_artifact_owner( + RESOURCE, ALICE, idempotency_key="original-owner", context=CONTEXT + ) + assert repeated == operation_id + await recovery.replay(repeated) + + await asyncio.gather(recovery.replay(operation_id), retry()) + assert await recovery.pending() == () + assert (await access.artifact_owner(RESOURCE)).owner == ALICE + assert len(await access.audit.list_audit(resource=RESOURCE)) == 1 + + asyncio.run(scenario()) + + +def test_candidate_retry_cannot_change_the_target(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + async with database.transaction() as connection: + operation_id = await recovery.for_transaction(connection).attest_candidate_owner( + scope_id="scope", + candidate_id="candidate", + family="profile", + proposed_owner=ALICE, + target=RESOURCE, + idempotency_key="candidate-owner", + ) + with pytest.raises(AccessConflictError): + async with database.transaction() as connection: + await recovery.for_transaction(connection).attest_candidate_owner( + scope_id="scope", + candidate_id="candidate", + family="profile", + proposed_owner=ALICE, + target=None, + idempotency_key="candidate-owner", + ) + await recovery.replay(operation_id) + assert (await access.candidate_owner("scope", "candidate")).target == RESOURCE + + asyncio.run(scenario()) + + +def test_corrupted_recovery_identity_cannot_write_an_external_owner(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + operation_id = await _create(database, recovery) + async with database.transaction() as connection: + await connection.execute( + update(OWNER_RECOVERY_TABLE) + .where(OWNER_RECOVERY_TABLE.c.operation_id == operation_id) + .values(scope_id="another-scope") + ) + with pytest.raises(AccessUnavailableError) as rejected: + await recovery.replay(operation_id) + assert rejected.value.code == "owner_recovery_payload_invalid" + assert await access.artifact_owner(RESOURCE) is None + assert await access.audit.list_audit(resource=RESOURCE) == () + + asyncio.run(scenario()) + + +def test_audit_replay_rejects_changed_fields_and_preserves_transaction_rollback(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + await recovery.replay(await _create(database, recovery)) + original = (await access.audit.list_audit(resource=RESOURCE))[0] + assert await access.audit.append_audit_once(replace(original, cursor=None)) == original + with pytest.raises(AccessConflictError): + await access.audit.append_audit_once(replace(original, principal=BOB)) + with pytest.raises(RuntimeError, match="domain failed"): + async with access.audit._database.transaction() as connection: + bound = access.audit.with_connection(connection) + await bound.append_audit_once(replace(original, cursor=None, event_id="rolled-back-event")) + raise RuntimeError(DOMAIN_FAILURE) + assert await access.audit.list_audit(resource=RESOURCE) == (original,) + + asyncio.run(scenario()) From 702232342c8abf8bf3a6c1a94233577833e5ef79 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Thu, 22 Oct 2026 13:29:56 +0800 Subject: [PATCH 2/4] fix(authz): isolate damaged intents during background recovery Read pending operation IDs before decoding each intent so malformed metadata cannot starve unrelated recovery work. Preserve fail-closed explicit replay. Exercise provider disappearance with the real built-in Access service rather than an incomplete startup-only test double. Constraint: Keep original ownership and all provider-disappearance denial assertions; never repair a damaged intent by guessing an identity. Scope-risk: Changes only recovery polling and the startup test fixture. Tested: Damaged-page regression fails before the fix; focused recovery 19 passed, Windows ownership/Access regressions 130 passed, Server suite 58 passed. --- .../server/authz/owner_recovery.py | 16 ++++++++--- tests/test_owner_recovery.py | 27 +++++++++++++++++++ tests/test_server.py | 7 +---- 3 files changed, 41 insertions(+), 9 deletions(-) diff --git a/src/powercontext/server/authz/owner_recovery.py b/src/powercontext/server/authz/owner_recovery.py index 89bcc0fd1b..5a57f58c5c 100644 --- a/src/powercontext/server/authz/owner_recovery.py +++ b/src/powercontext/server/authz/owner_recovery.py @@ -277,17 +277,27 @@ async def run() -> None: while True: await asyncio.sleep(5) try: - pending = await self.pending(after=after) + # Decode each intent during replay so one damaged payload + # cannot prevent delivery of every other item on its page. + statement = select(OWNER_RECOVERY_TABLE.c.operation_id).where( + OWNER_RECOVERY_TABLE.c.deployment_id == self.access.deployment_id, + OWNER_RECOVERY_TABLE.c.completed.is_(False), + ) + if after is not None: + statement = statement.where(OWNER_RECOVERY_TABLE.c.operation_id > after) + statement = statement.order_by(OWNER_RECOVERY_TABLE.c.operation_id).limit(100) + async with self.database.transaction() as connection: + pending = (await connection.execute(statement)).scalars().all() except Exception: # noqa: S112 # Leave durable work pending; do not emit external error values. continue for operation in pending: try: - await self.replay(operation.operation_id) + await self.replay(operation) except Exception: # noqa: S112 # Explicit replay reports errors; this worker retains the intent. continue - after = pending[-1].operation_id if len(pending) == 100 else None + after = pending[-1] if len(pending) == 100 else None task = asyncio.create_task(run(), name="powercontext-owner-recovery") try: diff --git a/tests/test_owner_recovery.py b/tests/test_owner_recovery.py index 2f5c0e0a06..7971e48112 100644 --- a/tests/test_owner_recovery.py +++ b/tests/test_owner_recovery.py @@ -328,3 +328,30 @@ async def scenario(): assert await access.audit.list_audit(resource=RESOURCE) == (original,) asyncio.run(scenario()) + + +def test_corrupted_intent_does_not_block_background_recovery_of_other_objects(tmp_path): + async def scenario(): + async with _stores(tmp_path) as (database, access, recovery): + damaged_id = await _create(database, recovery) + other = ResourceRef.artifact("scope", family="experience", artifact_id="second") + async with database.transaction() as connection: + await connection.execute(insert(OBJECTS).values(identity="second")) + await recovery.for_transaction(connection).establish_artifact_owner( + other, ALICE, idempotency_key="second-owner", context=CONTEXT + ) + await connection.execute( + update(OWNER_RECOVERY_TABLE) + .where(OWNER_RECOVERY_TABLE.c.operation_id == damaged_id) + .values(scope_id="damaged-scope") + ) + async with recovery.monitor(), asyncio.timeout(12): + while await access.artifact_owner(other) is None: # noqa: ASYNC110 - observe persisted external state + await asyncio.sleep(0.05) + assert await access.artifact_owner(RESOURCE) is None + audits = await access.audit.list_audit(resource=other) + assert len(audits) == 1 and audits[0].principal == ALICE + with pytest.raises(AccessUnavailableError): + await recovery.replay(damaged_id) + + asyncio.run(scenario()) diff --git a/tests/test_server.py b/tests/test_server.py index f017dedb8d..577f07043e 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -68,11 +68,6 @@ _ACCESS_FAMILIES = "atomic-memory:enabled,experience:enabled,handoff:enabled,memory:enabled,profile:enabled,prompt:enabled,skill:enabled" -class _StartupOnlyAccessControl: - async def committed_receipt_identity(self, _scope_id: str, _source_id: str, /) -> None: - return None - - def _access_readiness_checks( *, mode: str = "disabled", @@ -680,11 +675,11 @@ def test_enforced_mode_fails_closed_if_the_authorization_provider_disappears(tmp database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'runtime.db'}"), mcp=McpConfig(enabled=False), ), - access_control=cast(AccessControlService, _StartupOnlyAccessControl()), ) headers = {"Authorization": "Bearer server-secret"} with TestClient(app) as client: + assert isinstance(app.state.access_control, AccessControlService) app.state.access_control = None readiness = client.get("/health/ready") protected = ( From 8ccd0c75d5dd2f0a53ab0d3de5fcc1ca580f693c Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Thu, 22 Oct 2026 18:09:01 +0800 Subject: [PATCH 3/4] fix(authz): finish accepted owner intents on domain retries Replay the persisted Artifact intent when publication or Handoff returns an idempotent result without invoking its commit hook. Keep the original Principal and audit context and retain failure until delivery finishes. Wait for durable recovery completion before cancelling the monitor test. Constraint: Never create or restage ownership from the retrying identity. Scope-risk: HTTP publication/Handoff retries with external ownership and audit stores. Tested: Both new HTTP cases fail before the fix; Windows recovery/HTTP 21 passed; Linux recovery/HTTP/access regressions 86 passed; changed-file hooks and full Linux Ty/Python 3.11 passed. --- src/powercontext/server/app.py | 18 ++-- .../server/authz/owner_recovery.py | 6 +- tests/e2e/test_owner_recovery_http.py | 95 +++++++++++++++++++ tests/test_owner_recovery.py | 2 +- 4 files changed, 109 insertions(+), 12 deletions(-) diff --git a/src/powercontext/server/app.py b/src/powercontext/server/app.py index c94f680e95..b1f7169d5e 100644 --- a/src/powercontext/server/app.py +++ b/src/powercontext/server/app.py @@ -2261,16 +2261,13 @@ async def on_commit(connection, publication): ) elif access is not None and not owner_committed and result.target.artifact.family != "topic-memory": target = result.target - if ( - await access.artifact_owner( - ResourceRef.artifact( - target.scope_id, - family=target.artifact.family, - artifact_id=target.artifact.artifact_id, - ) - ) - is None - ): + resource = ResourceRef.artifact( + target.scope_id, + family=target.artifact.family, + artifact_id=target.artifact.artifact_id, + ) + await flush_owner_writes(artifact=resource) + if await access.artifact_owner(resource) is None: raise AccessUnavailableError("artifact_owner_pending") return TransportArtifactPublication.model_validate(result.model_dump(mode="json")) @@ -3439,6 +3436,7 @@ async def on_commit(connection, artifact): ) elif access is not None and not committed: resource = ResourceRef.artifact(request.scope_id, family="handoff", artifact_id=result.artifact_id) + await flush_owner_writes(artifact=resource) if await access.artifact_owner(resource) is None: raise AccessUnavailableError("artifact_owner_pending") return mapping.committed_handoff_response(result) diff --git a/src/powercontext/server/authz/owner_recovery.py b/src/powercontext/server/authz/owner_recovery.py index 5a57f58c5c..4b029b0b1c 100644 --- a/src/powercontext/server/authz/owner_recovery.py +++ b/src/powercontext/server/authz/owner_recovery.py @@ -407,12 +407,16 @@ async def owner_recovery_session(recovery: OwnerRecovery | None) -> AsyncIterato _SESSION.reset(token) -async def flush_owner_writes() -> None: +async def flush_owner_writes(*, artifact: ResourceRef | None = None) -> None: session = _SESSION.get() if session is not None: while session.operation_ids: await session.recovery.replay(session.operation_ids[0]) session.operation_ids.pop(0) + if artifact is not None: + # Idempotent domain retries can return before their commit hook. + # Replay only the original persisted intent, never the retry identity. + await session.recovery.replay(_key(session.recovery.access.deployment_id, "artifact", artifact.key)) def owner_recovery_available(access: AccessControlService, database: AsyncDatabase) -> bool: diff --git a/tests/e2e/test_owner_recovery_http.py b/tests/e2e/test_owner_recovery_http.py index 0c84afb83f..1b06975025 100644 --- a/tests/e2e/test_owner_recovery_http.py +++ b/tests/e2e/test_owner_recovery_http.py @@ -21,6 +21,7 @@ from typing import cast import httpx +import pytest from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.records import ArtifactWrite @@ -181,6 +182,100 @@ async def scenario(): asyncio.run(scenario()) +@pytest.mark.parametrize("operation", ["publication", "handoff"]) +def test_idempotent_retry_requires_the_original_owner_audit_to_complete(tmp_path, monkeypatch, operation): + async def scenario(): + async with _server(tmp_path) as (app, client, access): + scope_id = await _scope(client) + source = await client.post(f"/v1/scopes/{scope_id}/sources", json={"content": "Verified task state."}) + assert source.status_code == 201, source.text + statement = { + "text": "Continue the verified task.", + "citations": [ + {"kind": "source", "source_ref": {"name": "content", "source_id": source.json()["source_id"]}} + ], + } + content = { + "schema": "powercontext.handoff.v1", + "objective": "Continue verified work.", + "state": [statement], + "disposition": "continuable", + "next_action": statement, + "omissions": [], + } + destination = scope_id + path = "/v1/handoff/commit" + payload = { + "scope_id": scope_id, + "handoff": { + "schema": "powercontext.prepared-handoff.v1", + "scope_id": scope_id, + "base": None, + "content": content, + }, + } + if operation == "publication": + created = await client.post( + f"/v1/scopes/{scope_id}/artifacts", json={"family": "handoff", "content": content} + ) + assert created.status_code == 201, created.text + target = await client.post( + "/v1/scopes", + json={"title": "Publication target", "summary": "Audit retry", "idempotency_key": "audit-target"}, + ) + assert target.status_code == 201, target.text + destination = target.json()["scope_id"] + path = "/v1/artifact-publications" + payload = { + "source": { + "scope_id": scope_id, + "artifact": { + "family": "handoff", + "artifact_id": "handoff", + "revision": created.json()["revision"], + }, + }, + "target_scope_id": destination, + "idempotency_key": "audit-publication", + } + + async def unavailable(_event): + raise AccessUnavailableError("external_audit_unavailable") + + with monkeypatch.context() as patch: + patch.setattr(access.audit, "append_audit_once", unavailable) + failed = await client.post( + path, json=payload, headers={"X-PowerContext-Request-ID": "accepted-owner-request"} + ) + assert failed.status_code == 503, failed.text + original_request_id = failed.headers["X-PowerContext-Request-ID"] + pending = await app.state.owner_recovery.pending() + assert len(pending) == 1 + resource = pending[0].resource + assert resource is not None + assert (await access.artifact_owner(resource)).owner == ALICE + retried = await client.post( + path, json=payload, headers={"X-PowerContext-Request-ID": "retry-owner-request"} + ) + assert retried.status_code == 503, retried.text + assert await app.state.owner_recovery.pending() == pending + audit = await access.audit.list_audit(resource=resource, limit=100) + assert not [item for item in audit if item.reason_code == "artifact-owner-established"] + retried = await client.post(path, json=payload) + assert retried.is_success, retried.text + assert await app.state.owner_recovery.pending() == () + audit = await access.audit.list_audit(resource=resource, limit=100) + owners = [item for item in audit if item.reason_code == "artifact-owner-established"] + assert len(owners) == 1 + assert owners[0].principal == ALICE and owners[0].request_id == original_request_id + assert len(await app.state.application.records.for_scope(destination).logical_artifacts()) == 1 + assert (await client.post(path, json=payload)).json() == retried.json() + audit = await access.audit.list_audit(resource=resource, limit=100) + assert [item for item in audit if item.reason_code == "artifact-owner-established"] == owners + + asyncio.run(scenario()) + + def test_unknown_historical_owner_inventory_never_returns_content_or_assigns_a_caller(tmp_path): async def scenario(): async with _server(tmp_path) as (app, client, access): diff --git a/tests/test_owner_recovery.py b/tests/test_owner_recovery.py index 7971e48112..519fe1680e 100644 --- a/tests/test_owner_recovery.py +++ b/tests/test_owner_recovery.py @@ -346,7 +346,7 @@ async def scenario(): .values(scope_id="damaged-scope") ) async with recovery.monitor(), asyncio.timeout(12): - while await access.artifact_owner(other) is None: # noqa: ASYNC110 - observe persisted external state + while await recovery.pending(scope_id="scope"): # noqa: ASYNC110 - wait for durable delivery completion await asyncio.sleep(0.05) assert await access.artifact_owner(RESOURCE) is None audits = await access.audit.list_audit(resource=other) From e8c79f06db7b6028f7bf3abcc7982611b9e0efa9 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Fri, 23 Oct 2026 05:33:04 +0800 Subject: [PATCH 4/4] fix(authz): replay ownership on Profile flush retries Complete the original persisted Profile or Candidate ownership intent before HTTP and scheduled flush retries return noop or review_pending. Keep the accepted identity and audit context when external delivery is still failing. Constraint: replay only existing durable intents; never stage ownership from retry identity or infer ownership for historical records. Scope-risk: external ownership delivery on Profile flush response paths. Tested: focused recovery, authorization and Profile HTTP suites (95 passed); HTTP/background snapshot authorization (2 passed); full Python 3.11 ty check; scoped prek hooks. The three new HTTP cases failed before the source fix. Refs: #2084, #2187 --- src/powercontext/server/app.py | 4 ++ .../server/authz/owner_recovery.py | 7 +- src/powercontext/server/factory.py | 14 +++- tests/e2e/test_owner_recovery_http.py | 68 +++++++++++++++++++ 4 files changed, 90 insertions(+), 3 deletions(-) diff --git a/src/powercontext/server/app.py b/src/powercontext/server/app.py index b1f7169d5e..dd5381d435 100644 --- a/src/powercontext/server/app.py +++ b/src/powercontext/server/app.py @@ -2512,6 +2512,10 @@ async def on_commit(connection, artifact, candidate): result = await application.profiles.flush( request.scope_id, authorize_snapshot=authorize_snapshot, on_commit=on_commit ) + await flush_owner_writes( + artifact=resource, + candidate=None if result.candidate_id is None else (request.scope_id, result.candidate_id), + ) return FlushProfileResponse.model_validate(result.model_dump(mode="json")) diff --git a/src/powercontext/server/authz/owner_recovery.py b/src/powercontext/server/authz/owner_recovery.py index 4b029b0b1c..f6b2c35025 100644 --- a/src/powercontext/server/authz/owner_recovery.py +++ b/src/powercontext/server/authz/owner_recovery.py @@ -407,7 +407,7 @@ async def owner_recovery_session(recovery: OwnerRecovery | None) -> AsyncIterato _SESSION.reset(token) -async def flush_owner_writes(*, artifact: ResourceRef | None = None) -> None: +async def flush_owner_writes(*, artifact: ResourceRef | None = None, candidate: tuple[str, str] | None = None) -> None: session = _SESSION.get() if session is not None: while session.operation_ids: @@ -417,6 +417,11 @@ async def flush_owner_writes(*, artifact: ResourceRef | None = None) -> None: # Idempotent domain retries can return before their commit hook. # Replay only the original persisted intent, never the retry identity. await session.recovery.replay(_key(session.recovery.access.deployment_id, "artifact", artifact.key)) + if candidate is not None: + scope_id, candidate_id = candidate + await session.recovery.replay( + _key(session.recovery.access.deployment_id, "candidate", f"{scope_id}\0{candidate_id}") + ) def owner_recovery_available(access: AccessControlService, database: AsyncDatabase) -> bool: diff --git a/src/powercontext/server/factory.py b/src/powercontext/server/factory.py index 5abf82da06..cbc29e135c 100644 --- a/src/powercontext/server/factory.py +++ b/src/powercontext/server/factory.py @@ -74,7 +74,12 @@ access_control_for_mode, ) from powercontext.server.authz.composition import open_builtin_access_control -from powercontext.server.authz.owner_recovery import OwnerRecovery, owner_recovery_session, owner_write_access +from powercontext.server.authz.owner_recovery import ( + OwnerRecovery, + flush_owner_writes, + owner_recovery_session, + owner_write_access, +) from powercontext.server.authz.repository import RelationalAccessRepository from powercontext.server.context import current_principal, current_request_id from powercontext.server.cursor_secret import resolve_cursor_secret @@ -553,9 +558,14 @@ async def on_commit(connection, artifact, candidate): else OwnerRecovery(profiles.database, access) ) async with owner_recovery_session(recovery): - return await profiles.flush( + result = await profiles.flush( scope_id, high_watermark=high, authorize_snapshot=authorize_snapshot, on_commit=on_commit ) + await flush_owner_writes( + artifact=resource, + candidate=None if result.candidate_id is None else (scope_id, result.candidate_id), + ) + return result return run diff --git a/tests/e2e/test_owner_recovery_http.py b/tests/e2e/test_owner_recovery_http.py index 1b06975025..09d8efdcc4 100644 --- a/tests/e2e/test_owner_recovery_http.py +++ b/tests/e2e/test_owner_recovery_http.py @@ -276,6 +276,74 @@ async def unavailable(_event): asyncio.run(scenario()) +@pytest.mark.parametrize( + ("activation_mode", "failure"), + [("automatic", "relationship"), ("automatic", "audit"), ("review_required", "relationship")], +) +def test_profile_flush_retry_completes_the_original_ownership_intent(tmp_path, monkeypatch, activation_mode, failure): + class Generator: + async def generate(self, _value): + return "# Profile\n\n- Preserve the original owner." + + async def scenario(): + async with _server(tmp_path) as (app, client, access): + scope_id = await _scope(client) + policy = await client.put( + f"/v1/scopes/{scope_id}/profile-policy", + json={"generation_enabled": True, "activation_mode": activation_mode, "expected_version": 0}, + ) + assert policy.status_code == 200, policy.text + source = await client.post( + f"/v1/scopes/{scope_id}/sources", json={"content": "Keep the accepted ownership identity."} + ) + assert source.status_code == 201, source.text + app.state.application.profiles.generator = Generator() + + async def unavailable(_event): + raise AccessUnavailableError("external_audit_unavailable") + + with monkeypatch.context() as patch: + if failure == "audit": + patch.setattr(access.audit, "append_audit_once", unavailable) + else: + patch.setattr(access.relationships, "unavailable", True) + failed = await client.post( + "/v1/profile/flush", + json={"scope_id": scope_id}, + headers={"X-PowerContext-Request-ID": "accepted-profile-owner"}, + ) + assert failed.status_code == 503, failed.text + original_request_id = failed.headers["X-PowerContext-Request-ID"] + pending = await app.state.owner_recovery.pending(scope_id=scope_id) + assert len(pending) == 1 and pending[0].principal == ALICE + retried = await client.post("/v1/profile/flush", json={"scope_id": scope_id}) + assert retried.status_code == 503, retried.text + assert await app.state.owner_recovery.pending(scope_id=scope_id) == pending + retried = await client.post("/v1/profile/flush", json={"scope_id": scope_id}) + assert retried.status_code == 200, retried.text + assert await app.state.owner_recovery.pending(scope_id=scope_id) == () + if activation_mode == "review_required": + assert retried.json()["status"] == "review_pending" + assert retried.json()["candidate_id"] == pending[0].candidate_id + attestation = await access.candidate_owner(scope_id, pending[0].candidate_id) + assert attestation is not None and attestation.proposed_owner == ALICE + else: + assert retried.json()["status"] == "noop" + resource = ResourceRef.artifact(scope_id, family="profile", artifact_id="profile") + relation = await access.artifact_owner(resource) + assert relation is not None and relation.owner == ALICE + audit = await access.audit.list_audit(resource=resource, limit=100) + owners = [item for item in audit if item.reason_code == "artifact-owner-established"] + assert len(owners) == 1 and owners[0].request_id == original_request_id + records = await app.state.application.records.for_scope(scope_id).logical_artifacts() + assert len(records) == 1 + repeated = await client.post("/v1/profile/flush", json={"scope_id": scope_id}) + assert repeated.json() == retried.json() + assert await app.state.owner_recovery.pending(scope_id=scope_id) == () + + asyncio.run(scenario()) + + def test_unknown_historical_owner_inventory_never_returns_content_or_assigns_a_caller(tmp_path): async def scenario(): async with _server(tmp_path) as (app, client, access):