From 925dc3369651378428d37b4995c30d059d785fa1 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 14:55:58 +0200 Subject: [PATCH] feat(docs): define semantic subject contract --- docs/DOCUMENTATION_MAP.md | 1 + docs/MODULE_ARCHITECTURE.md | 11 + docs/SEMANTIC_DOCUMENTATION_SUBJECTS.md | 118 ++++ src/govoplan_core/core/registry.py | 47 ++ .../core/semantic_documentation.py | 621 ++++++++++++++++++ tests/test_semantic_documentation_contract.py | 378 +++++++++++ 6 files changed, 1176 insertions(+) create mode 100644 docs/SEMANTIC_DOCUMENTATION_SUBJECTS.md create mode 100644 src/govoplan_core/core/semantic_documentation.py create mode 100644 tests/test_semantic_documentation_contract.py diff --git a/docs/DOCUMENTATION_MAP.md b/docs/DOCUMENTATION_MAP.md index 128b70d..4b2af64 100644 --- a/docs/DOCUMENTATION_MAP.md +++ b/docs/DOCUMENTATION_MAP.md @@ -22,6 +22,7 @@ operator, and roadmap pages. | Cross-module information governance adoption | `INFORMATION_GOVERNANCE_ADOPTION.md` | Manifest evidence and enforcement rules for temporal browsing, purpose-aware access, retention, and institutional context. | | Data-subject access and erasure requests | `DATA_SUBJECT_REQUESTS.md` | Provider-owned search and mutation, explicit coverage, governed export, retained evidence, permissions, and idempotent execution. | | Context-sensitive F1 help | `CONTEXTUAL_HELP_CONTRACT.md` | Focus, route, module-manifest documentation contexts, Docs projection, and hosted fallback. | +| Semantic documentation subjects | `SEMANTIC_DOCUMENTATION_SUBJECTS.md` | Stable configured-artifact identity, safe provider discovery, revision review, authorization, and lifecycle semantics. | | German localization and help quality gate | `LOCALIZATION_AND_HELP_QUALITY.md` | German reference locale, new-installation default, catalog completeness, automatic page associations, and explicit-help review priorities. | | Postbox E2EE target architecture | `POSTBOX_E2EE_ARCHITECTURE.md` | Strategic encrypted postbox/mailbox model, key ownership, role mailbox semantics, and retraction limits. | | Shared state, runtime coordination, and recovery | `STATE_AND_RECOVERY_CONTRACT.md` | State profiles, object storage, node registration/drain, fenced leases, migration ordering, and recovery evidence. | diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md index 8807277..d90d979 100644 --- a/docs/MODULE_ARCHITECTURE.md +++ b/docs/MODULE_ARCHITECTURE.md @@ -1621,6 +1621,17 @@ URLs never contain credentials; only credential-envelope references cross the contract. Provider-specific details belong in sanitized provenance rather than in a shared domain schema. +## Semantic Documentation Subject Contract + +Optional modules expose configured artifacts that can be documented through +the module-scoped `documentation.semantic_subjects.` capability. +Core supplies stable tenant-scoped references, typed nested anchors, safe +localized descriptors, revision/fingerprint review signals, and explicit +availability states. Providers remain responsible for authorization and do not +expose configuration payloads or credentials. Docs discovers the capability +and owns authored content; it does not import feature internals. See +`SEMANTIC_DOCUMENTATION_SUBJECTS.md` for the contract and adoption rules. + ## Build And Verification Backend verification from core: diff --git a/docs/SEMANTIC_DOCUMENTATION_SUBJECTS.md b/docs/SEMANTIC_DOCUMENTATION_SUBJECTS.md new file mode 100644 index 0000000..a860dfb --- /dev/null +++ b/docs/SEMANTIC_DOCUMENTATION_SUBJECTS.md @@ -0,0 +1,118 @@ +# Semantic Documentation Subjects + +## Purpose And Ownership + +The semantic-documentation subject contract lets an optional module expose the +configured artifacts that administrators may document: for example a form, a +form field, a workflow, or a workflow state. It is a discovery and resolution +contract, not a second configuration API. + +The module that owns an artifact also owns its subject provider, authorization, +identity, revision, route, and lifecycle semantics. Docs may discover those +providers through Core and attach authored documentation to their stable +references. Docs must not import the feature module, read its tables, or copy +configuration content into a generic index. + +This contract is additive to manifest `DocumentationTopic` contributions and +configured-state `documentation_providers`. Every providing module must retain +static user and administrator documentation baselines. The baselines explain +the feature even when the provider is disabled, unavailable, or has no +configured subjects. + +## Identity And Versioning + +`SemanticDocumentationSubjectReference` identifies a subject with: + +- owning module and tenant; +- a module-defined subject kind and stable identifier; +- an optional typed nested anchor, such as `field/registration-number`; +- the revision and canonical fingerprint observed when documentation was + authored or reviewed. + +The `stable_key` derives only from identity. A rename or configuration revision +therefore does not detach existing documentation. A nested anchor has its own +identity so a field can be documented independently from its form. + +Providers must resolve an old reference as one of: + +- `available`: the observed revision/fingerprint is still current; +- `changed`: the same stable subject has changed and may need review; +- `superseded`: another stable reference replaced it; +- `missing`: the subject was removed or is no longer resolvable; +- `temporarily_unavailable`: the provider cannot currently determine state. + +Absence is not authorization. A provider returns `None` when the principal may +not learn whether a subject exists. Core also rejects cross-tenant list and +resolution requests before calling a provider. + +## Safe Projection + +Descriptors contain only bounded, explicit presentation fields: localized +labels and descriptions, breadcrumbs, a local route, audience, +classification, and required scopes. They must not contain credentials, +personal data, arbitrary provider metadata, configuration payloads, or the +authored documentation itself. Routes are application-local and are still +subject to normal route authorization. + +The fingerprint is a review signal, not a concurrency token or a content hash +that callers may use to reconstruct configuration. Providers should calculate +it from the smallest canonical JSON projection whose semantic changes require +documentation review. Volatile timestamps and secrets must be excluded. + +## Provider Registration + +A provider is registered under its exact module-scoped capability name: + +```python +from govoplan_core.core.modules import CapabilityDocumentation +from govoplan_core.core.semantic_documentation import ( + SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION, + semantic_documentation_subject_capability, +) + +capability = semantic_documentation_subject_capability("forms") + +manifest = ModuleManifest( + id="forms", + # ... + capability_factories={capability: build_semantic_subject_provider}, + capability_documentation={ + capability: CapabilityDocumentation( + label="Form semantic subjects", + summary="Lists authorized configured forms and fields for Docs.", + contract_version=SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION, + documentation_types=("admin", "user"), + ) + }, + documentation=(admin_baseline, user_baseline), +) +``` + +The capability is `documentation.semantic_subjects.`. Registry +validation rejects a mismatched owner, missing capability documentation, a +wrong contract version, or missing static baselines. + +`list_semantic_documentation_subjects` performs authorized, paginated discovery +across installed providers. `resolve_semantic_documentation_subject` targets +one owner without loading another feature module. Providers must apply the +current tenant and principal on every call and must not infer visibility from a +previous list result. + +## Lifecycle And Integration Rules + +- Keep subject and anchor identifiers stable across display-name and route + changes. +- Return `superseded` only with the replacement reference; do not silently + rewrite stored references. +- Return a reason code for missing or temporarily unavailable subjects without + exposing sensitive detail. +- Reauthorize both discovery and resolution. Stored documentation references + confer no access to a live artifact. +- Treat a changed fingerprint as a request for editorial review. It does not + automatically invalidate or publish authored documentation. +- Removing a feature module leaves references resolvable as provider + unavailable. Docs can preserve history without importing the module. + +Forms, Workflow, and later modules should implement their subject providers in +their own repositories. Docs owns the authored semantic-documentation records, +review workflow, and projection UI. diff --git a/src/govoplan_core/core/registry.py b/src/govoplan_core/core/registry.py index 1575187..f8947fd 100644 --- a/src/govoplan_core/core/registry.py +++ b/src/govoplan_core/core/registry.py @@ -61,6 +61,11 @@ from govoplan_core.core.search import ( SearchProvider, SearchSourceProvider, ) +from govoplan_core.core.semantic_documentation import ( + SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX, + SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION, + semantic_documentation_subject_capability, +) from govoplan_core.core.tasks import ( RegisteredWorkItemProvider, WorkItemProvider, @@ -1215,6 +1220,48 @@ def _validate_documentation_extensions(manifest: ModuleManifest) -> None: "documentation contract version must not be empty" ) + semantic_capabilities = tuple( + capability + for capability in manifest.capability_factories + if capability.startswith( + SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX + ) + ) + for capability in semantic_capabilities: + expected = semantic_documentation_subject_capability(manifest.id) + if capability != expected: + raise RegistryError( + f"Module {manifest.id!r} semantic-documentation capability " + f"must be {expected!r}, not {capability!r}" + ) + metadata = manifest.capability_documentation.get(capability) + if metadata is None: + raise RegistryError( + f"Module {manifest.id!r} semantic-documentation capability " + "must declare capability documentation" + ) + if ( + metadata.contract_version + != SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION + ): + raise RegistryError( + f"Module {manifest.id!r} semantic-documentation capability " + f"must declare contract version " + f"{SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION!r}" + ) + static_types = { + documentation_type + for topic in manifest.documentation + for documentation_type in topic.documentation_types + } + missing_types = {"admin", "user"} - static_types + if missing_types: + raise RegistryError( + f"Module {manifest.id!r} semantic-documentation provider must " + "retain static user and administrator documentation baselines; " + f"missing: {', '.join(sorted(missing_types))}" + ) + provider_keys: set[str] = set() for registration in manifest.documentation_configuration_providers: if not registration.keys: diff --git a/src/govoplan_core/core/semantic_documentation.py b/src/govoplan_core/core/semantic_documentation.py new file mode 100644 index 0000000..8f87449 --- /dev/null +++ b/src/govoplan_core/core/semantic_documentation.py @@ -0,0 +1,621 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Literal, Protocol, runtime_checkable + + +SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX = ( + "documentation.semantic_subjects." +) +SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION = "1" + +SemanticDocumentationSubjectAvailability = Literal[ + "available", + "changed", + "superseded", + "missing", + "temporarily_unavailable", +] + +_MODULE_ID_RE = re.compile(r"^[a-z][a-z0-9_]{0,79}$") +_KIND_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,119}$") +_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:@-]{0,254}$") +_LOCALE_RE = re.compile(r"^[A-Za-z]{2,3}(?:-[A-Za-z0-9]{2,8})*$") +_REASON_CODE_RE = re.compile(r"^[a-z][a-z0-9_]{0,79}$") +_SHA256_RE = re.compile(r"^(?:sha256:)?[0-9a-fA-F]{64}$") + + +class SemanticDocumentationContractError(ValueError): + """Raised when a semantic-documentation subject violates the Core contract.""" + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationSubjectAnchor: + kind: str + id: str + + def __post_init__(self) -> None: + _require_match(self.kind, _KIND_RE, "Semantic subject anchor kind") + _require_match(self.id, _IDENTIFIER_RE, "Semantic subject anchor id") + + def to_dict(self) -> dict[str, str]: + return {"kind": self.kind, "id": self.id} + + @classmethod + def from_mapping( + cls, value: Mapping[str, object] + ) -> SemanticDocumentationSubjectAnchor: + _require_keys(value, {"kind", "id"}, "Semantic subject anchor") + return cls(kind=_required_text(value, "kind"), id=_required_text(value, "id")) + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationSubjectReference: + module_id: str + tenant_id: str + subject_kind: str + subject_id: str + anchor: SemanticDocumentationSubjectAnchor | None = None + observed_revision: str | None = None + observed_fingerprint: str | None = None + + def __post_init__(self) -> None: + _require_match(self.module_id, _MODULE_ID_RE, "Semantic subject module id") + _require_match(self.tenant_id, _IDENTIFIER_RE, "Semantic subject tenant id") + _require_match(self.subject_kind, _KIND_RE, "Semantic subject kind") + _require_match(self.subject_id, _IDENTIFIER_RE, "Semantic subject id") + _optional_text(self.observed_revision, "Semantic subject observed revision", 255) + if self.observed_fingerprint is not None and not _SHA256_RE.fullmatch( + self.observed_fingerprint + ): + raise SemanticDocumentationContractError( + "Semantic subject observed fingerprint must be a SHA-256 digest." + ) + + @property + def stable_key(self) -> str: + identity = { + "anchor": self.anchor.to_dict() if self.anchor else None, + "module_id": self.module_id, + "subject_id": self.subject_id, + "subject_kind": self.subject_kind, + "tenant_id": self.tenant_id, + } + encoded = json.dumps( + identity, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + def to_dict(self) -> dict[str, object]: + return { + "module_id": self.module_id, + "tenant_id": self.tenant_id, + "subject_kind": self.subject_kind, + "subject_id": self.subject_id, + "anchor": self.anchor.to_dict() if self.anchor else None, + "observed_revision": self.observed_revision, + "observed_fingerprint": self.observed_fingerprint, + } + + @classmethod + def from_mapping( + cls, value: Mapping[str, object] + ) -> SemanticDocumentationSubjectReference: + _require_keys( + value, + { + "module_id", + "tenant_id", + "subject_kind", + "subject_id", + "anchor", + "observed_revision", + "observed_fingerprint", + }, + "Semantic subject reference", + ) + raw_anchor = value.get("anchor") + if raw_anchor is not None and not isinstance(raw_anchor, Mapping): + raise SemanticDocumentationContractError( + "Semantic subject anchor must be an object." + ) + return cls( + module_id=_required_text(value, "module_id"), + tenant_id=_required_text(value, "tenant_id"), + subject_kind=_required_text(value, "subject_kind"), + subject_id=_required_text(value, "subject_id"), + anchor=( + SemanticDocumentationSubjectAnchor.from_mapping(raw_anchor) + if isinstance(raw_anchor, Mapping) + else None + ), + observed_revision=_mapping_optional_text(value, "observed_revision"), + observed_fingerprint=_mapping_optional_text( + value, "observed_fingerprint" + ), + ) + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationBreadcrumb: + label: str + subject_kind: str + subject_id: str + anchor: SemanticDocumentationSubjectAnchor | None = None + + def __post_init__(self) -> None: + _required_bounded_text(self.label, "Semantic subject breadcrumb label", 300) + _require_match(self.subject_kind, _KIND_RE, "Semantic breadcrumb kind") + _require_match(self.subject_id, _IDENTIFIER_RE, "Semantic breadcrumb id") + + def to_dict(self) -> dict[str, object]: + return { + "label": self.label, + "subject_kind": self.subject_kind, + "subject_id": self.subject_id, + "anchor": self.anchor.to_dict() if self.anchor else None, + } + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationSubjectDescriptor: + reference: SemanticDocumentationSubjectReference + labels: Mapping[str, str] + descriptions: Mapping[str, str] = field(default_factory=dict) + breadcrumbs: tuple[SemanticDocumentationBreadcrumb, ...] = () + route: str | None = None + route_anchor: str | None = None + audience: tuple[str, ...] = () + classification: str = "internal" + required_scopes: tuple[str, ...] = () + + def __post_init__(self) -> None: + if not self.reference.observed_revision: + raise SemanticDocumentationContractError( + "Semantic subject descriptors require a current revision." + ) + if not self.reference.observed_fingerprint: + raise SemanticDocumentationContractError( + "Semantic subject descriptors require a current fingerprint." + ) + _localized_text(self.labels, "Semantic subject labels", required=True, limit=300) + _localized_text( + self.descriptions, + "Semantic subject descriptions", + required=False, + limit=2_000, + ) + if len(self.breadcrumbs) > 32: + raise SemanticDocumentationContractError( + "Semantic subject breadcrumbs are limited to 32 items." + ) + if self.route is not None: + _optional_text(self.route, "Semantic subject route", 2_000) + if not self.route.startswith("/") or self.route.startswith("//"): + raise SemanticDocumentationContractError( + "Semantic subject routes must be local absolute paths." + ) + if self.route_anchor is not None: + _require_match( + self.route_anchor, + _IDENTIFIER_RE, + "Semantic subject route anchor", + ) + _text_tuple(self.audience, "Semantic subject audience", maximum=32) + _required_bounded_text( + self.classification, "Semantic subject classification", 120 + ) + _text_tuple( + self.required_scopes, "Semantic subject required scopes", maximum=64 + ) + + def to_dict(self) -> dict[str, object]: + return { + "reference": self.reference.to_dict(), + "labels": dict(self.labels), + "descriptions": dict(self.descriptions), + "breadcrumbs": [item.to_dict() for item in self.breadcrumbs], + "route": self.route, + "route_anchor": self.route_anchor, + "audience": list(self.audience), + "classification": self.classification, + "required_scopes": list(self.required_scopes), + } + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationSubjectResolution: + requested_reference: SemanticDocumentationSubjectReference + availability: SemanticDocumentationSubjectAvailability + subject: SemanticDocumentationSubjectDescriptor | None = None + superseded_by: SemanticDocumentationSubjectReference | None = None + reason_code: str | None = None + + def __post_init__(self) -> None: + if self.reason_code is not None: + _require_match( + self.reason_code, _REASON_CODE_RE, "Semantic resolution reason code" + ) + if self.availability in {"available", "changed"}: + if self.subject is None: + raise SemanticDocumentationContractError( + f"Semantic subject {self.availability} resolutions require a descriptor." + ) + if ( + self.subject.reference.stable_key + != self.requested_reference.stable_key + ): + raise SemanticDocumentationContractError( + "Semantic subject resolution changed the requested identity." + ) + changed = _reference_changed( + self.requested_reference, self.subject.reference + ) + if self.availability == "available" and changed: + raise SemanticDocumentationContractError( + "Changed semantic subjects must use the changed availability." + ) + if self.availability == "changed" and not changed: + raise SemanticDocumentationContractError( + "Changed semantic subject resolutions require a revision or fingerprint change." + ) + elif self.subject is not None: + raise SemanticDocumentationContractError( + f"Semantic subject {self.availability} resolutions cannot include a descriptor." + ) + if self.availability == "superseded": + if self.superseded_by is None: + raise SemanticDocumentationContractError( + "Superseded semantic subjects require a replacement reference." + ) + elif self.superseded_by is not None: + raise SemanticDocumentationContractError( + "Only superseded semantic subjects may declare a replacement." + ) + if self.availability in {"missing", "temporarily_unavailable"} and not self.reason_code: + raise SemanticDocumentationContractError( + f"Semantic subject {self.availability} resolutions require a reason code." + ) + + def to_dict(self) -> dict[str, object]: + return { + "requested_reference": self.requested_reference.to_dict(), + "availability": self.availability, + "subject": self.subject.to_dict() if self.subject else None, + "superseded_by": ( + self.superseded_by.to_dict() if self.superseded_by else None + ), + "reason_code": self.reason_code, + } + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationSubjectQuery: + tenant_id: str + query: str = "" + subject_kinds: tuple[str, ...] = () + limit: int = 50 + cursor: str | None = None + + def __post_init__(self) -> None: + _require_match(self.tenant_id, _IDENTIFIER_RE, "Semantic query tenant id") + if not isinstance(self.query, str) or len(self.query) > 300: + raise SemanticDocumentationContractError( + "Semantic subject query must be text of at most 300 characters." + ) + if self.query: + _required_bounded_text(self.query, "Semantic subject query", 300) + if not 1 <= self.limit <= 200: + raise SemanticDocumentationContractError( + "Semantic subject query limit must be between 1 and 200." + ) + _text_tuple(self.subject_kinds, "Semantic query subject kinds", maximum=100) + for kind in self.subject_kinds: + _require_match(kind, _KIND_RE, "Semantic query subject kind") + _optional_text(self.cursor, "Semantic query cursor", 1_000) + + +@dataclass(frozen=True, slots=True) +class SemanticDocumentationSubjectPage: + subjects: tuple[SemanticDocumentationSubjectDescriptor, ...] = () + next_cursor: str | None = None + has_more: bool = False + + def __post_init__(self) -> None: + keys = tuple(item.reference.stable_key for item in self.subjects) + if len(keys) != len(set(keys)): + raise SemanticDocumentationContractError( + "Semantic subject pages cannot contain duplicate identities." + ) + _optional_text(self.next_cursor, "Semantic subject page cursor", 1_000) + if self.has_more and not self.next_cursor: + raise SemanticDocumentationContractError( + "Semantic subject pages with more results require a cursor." + ) + + +@runtime_checkable +class SemanticDocumentationSubjectProvider(Protocol): + provider_id: str + module_id: str + contract_version: str + + def list_subjects( + self, + session: object, + principal: object, + *, + request: SemanticDocumentationSubjectQuery, + ) -> SemanticDocumentationSubjectPage: ... + + def resolve_subject( + self, + session: object, + principal: object, + *, + reference: SemanticDocumentationSubjectReference, + ) -> SemanticDocumentationSubjectResolution | None: + """Return None when the principal may not know whether a subject exists.""" + + +def semantic_documentation_subject_capability(module_id: str) -> str: + _require_match(module_id, _MODULE_ID_RE, "Semantic subject module id") + return f"{SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX}{module_id}" + + +def semantic_documentation_subject_provider_names( + registry: object | None, +) -> tuple[str, ...]: + if registry is None or not hasattr(registry, "capability_names"): + return () + return tuple( + str(name) + for name in registry.capability_names() + if str(name).startswith(SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX) + ) + + +def semantic_documentation_subject_providers( + registry: object | None, +) -> tuple[tuple[str, SemanticDocumentationSubjectProvider], ...]: + if registry is None or not hasattr(registry, "capability"): + return () + providers: list[tuple[str, SemanticDocumentationSubjectProvider]] = [] + for capability_name in semantic_documentation_subject_provider_names(registry): + module_id = capability_name.removeprefix( + SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX + ) + provider = registry.capability(capability_name) + if not isinstance(provider, SemanticDocumentationSubjectProvider): + raise TypeError( + f"Invalid semantic-documentation provider capability: {capability_name}" + ) + if provider.module_id != module_id: + raise SemanticDocumentationContractError( + f"Semantic provider module {provider.module_id!r} does not match " + f"capability {capability_name!r}." + ) + if ( + provider.contract_version + != SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION + ): + raise SemanticDocumentationContractError( + f"Unsupported semantic-documentation provider contract: " + f"{provider.contract_version!r}." + ) + providers.append((module_id, provider)) + return tuple(providers) + + +def list_semantic_documentation_subjects( + registry: object | None, + session: object, + principal: object, + *, + request: SemanticDocumentationSubjectQuery, +) -> tuple[tuple[str, SemanticDocumentationSubjectPage], ...]: + if _principal_tenant_id(principal) != request.tenant_id: + return () + pages: list[tuple[str, SemanticDocumentationSubjectPage]] = [] + for module_id, provider in semantic_documentation_subject_providers(registry): + page = provider.list_subjects( + session, + principal, + request=request, + ) + if any( + subject.reference.module_id != module_id + or subject.reference.tenant_id != request.tenant_id + for subject in page.subjects + ): + raise SemanticDocumentationContractError( + f"Semantic provider {module_id!r} returned a foreign subject." + ) + pages.append((module_id, page)) + return tuple(pages) + + +def resolve_semantic_documentation_subject( + registry: object | None, + session: object, + principal: object, + *, + reference: SemanticDocumentationSubjectReference, +) -> SemanticDocumentationSubjectResolution | None: + if _principal_tenant_id(principal) != reference.tenant_id: + return None + capability_name = semantic_documentation_subject_capability(reference.module_id) + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(capability_name) + ): + return SemanticDocumentationSubjectResolution( + requested_reference=reference, + availability="temporarily_unavailable", + reason_code="provider_unavailable", + ) + provider = registry.capability(capability_name) + if not isinstance(provider, SemanticDocumentationSubjectProvider): + raise TypeError( + f"Invalid semantic-documentation provider capability: {capability_name}" + ) + if ( + provider.module_id != reference.module_id + or provider.contract_version + != SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION + ): + raise SemanticDocumentationContractError( + f"Semantic provider {capability_name!r} does not match the Core contract." + ) + result = provider.resolve_subject( + session, + principal, + reference=reference, + ) + if result is None: + return None + if result.requested_reference != reference: + raise SemanticDocumentationContractError( + "Semantic provider returned a resolution for another reference." + ) + return result + + +def semantic_documentation_fingerprint(value: object) -> str: + try: + encoded = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise SemanticDocumentationContractError( + "Semantic fingerprint input must be canonical JSON data." + ) from exc + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + +def _reference_changed( + requested: SemanticDocumentationSubjectReference, + current: SemanticDocumentationSubjectReference, +) -> bool: + comparisons = ( + (requested.observed_revision, current.observed_revision), + (requested.observed_fingerprint, current.observed_fingerprint), + ) + return any(expected is not None and expected != actual for expected, actual in comparisons) + + +def _principal_tenant_id(principal: object) -> str: + return str(getattr(principal, "tenant_id", "") or "") + + +def _localized_text( + values: Mapping[str, str], + label: str, + *, + required: bool, + limit: int, +) -> None: + if required and not values: + raise SemanticDocumentationContractError(f"{label} are required.") + if len(values) > 20: + raise SemanticDocumentationContractError(f"{label} are limited to 20 locales.") + for locale, value in values.items(): + if not _LOCALE_RE.fullmatch(str(locale)): + raise SemanticDocumentationContractError( + f"{label} contain an invalid locale: {locale!r}." + ) + _required_bounded_text(value, f"{label} value", limit) + + +def _text_tuple(values: Sequence[str], label: str, *, maximum: int) -> None: + if len(values) > maximum: + raise SemanticDocumentationContractError( + f"{label} are limited to {maximum} items." + ) + normalized = tuple(str(value).strip() for value in values) + if any(not value or len(value) > 255 for value in normalized): + raise SemanticDocumentationContractError( + f"{label} must contain non-empty bounded text." + ) + if len(normalized) != len(set(normalized)): + raise SemanticDocumentationContractError(f"{label} must be unique.") + + +def _require_match(value: str, pattern: re.Pattern[str], label: str) -> None: + if not isinstance(value, str) or not pattern.fullmatch(value): + raise SemanticDocumentationContractError(f"{label} is invalid.") + + +def _required_bounded_text(value: str, label: str, limit: int) -> None: + if not isinstance(value, str) or not value.strip() or len(value) > limit: + raise SemanticDocumentationContractError( + f"{label} must be non-empty and at most {limit} characters." + ) + if any(ord(character) < 32 and character not in "\n\t" for character in value): + raise SemanticDocumentationContractError(f"{label} contains control characters.") + + +def _optional_text(value: str | None, label: str, limit: int) -> None: + if value is not None: + _required_bounded_text(value, label, limit) + + +def _require_keys( + value: Mapping[str, object], allowed: set[str], label: str +) -> None: + unexpected = sorted(str(key) for key in value if str(key) not in allowed) + if unexpected: + raise SemanticDocumentationContractError( + f"{label} contains unsupported fields: {', '.join(unexpected)}." + ) + + +def _required_text(value: Mapping[str, object], key: str) -> str: + result = value.get(key) + if not isinstance(result, str) or not result.strip(): + raise SemanticDocumentationContractError( + f"Semantic subject field {key} is required." + ) + return result + + +def _mapping_optional_text(value: Mapping[str, object], key: str) -> str | None: + result = value.get(key) + if result is None: + return None + if not isinstance(result, str): + raise SemanticDocumentationContractError( + f"Semantic subject field {key} must be text." + ) + return result + + +__all__ = [ + "SEMANTIC_DOCUMENTATION_SUBJECT_CAPABILITY_PREFIX", + "SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION", + "SemanticDocumentationBreadcrumb", + "SemanticDocumentationContractError", + "SemanticDocumentationSubjectAnchor", + "SemanticDocumentationSubjectAvailability", + "SemanticDocumentationSubjectDescriptor", + "SemanticDocumentationSubjectPage", + "SemanticDocumentationSubjectProvider", + "SemanticDocumentationSubjectQuery", + "SemanticDocumentationSubjectReference", + "SemanticDocumentationSubjectResolution", + "list_semantic_documentation_subjects", + "resolve_semantic_documentation_subject", + "semantic_documentation_fingerprint", + "semantic_documentation_subject_capability", + "semantic_documentation_subject_provider_names", + "semantic_documentation_subject_providers", +] diff --git a/tests/test_semantic_documentation_contract.py b/tests/test_semantic_documentation_contract.py new file mode 100644 index 0000000..00892e3 --- /dev/null +++ b/tests/test_semantic_documentation_contract.py @@ -0,0 +1,378 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace + +from govoplan_core.core.modules import ( + CapabilityDocumentation, + DocumentationTopic, + ModuleContext, + ModuleManifest, +) +from govoplan_core.core.registry import PlatformRegistry, RegistryError +from govoplan_core.core.semantic_documentation import ( + SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION, + SemanticDocumentationBreadcrumb, + SemanticDocumentationContractError, + SemanticDocumentationSubjectAnchor, + SemanticDocumentationSubjectDescriptor, + SemanticDocumentationSubjectPage, + SemanticDocumentationSubjectQuery, + SemanticDocumentationSubjectReference, + SemanticDocumentationSubjectResolution, + list_semantic_documentation_subjects, + resolve_semantic_documentation_subject, + semantic_documentation_fingerprint, + semantic_documentation_subject_capability, + semantic_documentation_subject_provider_names, + semantic_documentation_subject_providers, +) + + +def reference( + *, + tenant_id: str = "tenant-1", + revision: str | None = "4", + fingerprint: str | None = None, + anchor: SemanticDocumentationSubjectAnchor | None = None, +) -> SemanticDocumentationSubjectReference: + return SemanticDocumentationSubjectReference( + module_id="forms", + tenant_id=tenant_id, + subject_kind="form", + subject_id="permit-application", + anchor=anchor, + observed_revision=revision, + observed_fingerprint=fingerprint + or semantic_documentation_fingerprint({"revision": revision}), + ) + + +def descriptor( + *, + tenant_id: str = "tenant-1", + revision: str = "4", + fingerprint: str | None = None, + label: str = "Permit application", + anchor: SemanticDocumentationSubjectAnchor | None = None, +) -> SemanticDocumentationSubjectDescriptor: + return SemanticDocumentationSubjectDescriptor( + reference=reference( + tenant_id=tenant_id, + revision=revision, + fingerprint=fingerprint, + anchor=anchor, + ), + labels={"de": "Antrag auf Parkerlaubnis", "en": label}, + descriptions={"de": "Konfiguriertes Antragsformular."}, + breadcrumbs=( + SemanticDocumentationBreadcrumb( + label="Forms", + subject_kind="form", + subject_id="permit-application", + ), + ), + route="/forms/permit-application", + route_anchor=anchor.id if anchor else None, + audience=("case_worker",), + classification="internal", + required_scopes=("forms:definition:read",), + ) + + +class Provider: + provider_id = "forms.semantic_subjects" + module_id = "forms" + contract_version = SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION + + def __init__(self) -> None: + self.list_calls = 0 + self.resolve_calls = 0 + + def list_subjects(self, _session, principal, *, request): + self.list_calls += 1 + if getattr(principal, "account_id", "") == "denied": + return SemanticDocumentationSubjectPage() + return SemanticDocumentationSubjectPage(subjects=(descriptor(),)) + + def resolve_subject(self, _session, principal, *, reference): + self.resolve_calls += 1 + if getattr(principal, "account_id", "") == "denied": + return None + current = descriptor() + availability = ( + "changed" + if reference.observed_fingerprint + != current.reference.observed_fingerprint + else "available" + ) + return SemanticDocumentationSubjectResolution( + requested_reference=reference, + availability=availability, + subject=current, + ) + + +class SemanticDocumentationContractTests(unittest.TestCase): + def test_reference_identity_is_stable_across_labels_and_revisions(self) -> None: + original = descriptor(label="Old label") + renamed = descriptor(revision="5", label="New label") + + self.assertEqual( + original.reference.stable_key, + renamed.reference.stable_key, + ) + self.assertNotEqual( + original.reference.observed_fingerprint, + renamed.reference.observed_fingerprint, + ) + + def test_nested_anchor_is_typed_and_changes_subject_identity(self) -> None: + form = reference() + field = reference( + anchor=SemanticDocumentationSubjectAnchor( + kind="field", + id="vehicle-registration", + ) + ) + + self.assertNotEqual(form.stable_key, field.stable_key) + restored = SemanticDocumentationSubjectReference.from_mapping( + field.to_dict() + ) + self.assertEqual(restored, field) + + def test_reference_mapping_rejects_unbounded_provider_metadata(self) -> None: + payload = reference().to_dict() + payload["credentials"] = {"token": "must-not-cross"} + + with self.assertRaisesRegex( + SemanticDocumentationContractError, + "unsupported fields: credentials", + ): + SemanticDocumentationSubjectReference.from_mapping(payload) + + def test_descriptors_reject_remote_routes_and_missing_localization(self) -> None: + with self.assertRaisesRegex( + SemanticDocumentationContractError, + "local absolute paths", + ): + SemanticDocumentationSubjectDescriptor( + reference=reference(), + labels={"de": "Formular"}, + route="https://provider.example/form", + ) + with self.assertRaisesRegex( + SemanticDocumentationContractError, + "labels are required", + ): + SemanticDocumentationSubjectDescriptor( + reference=reference(), + labels={}, + ) + + def test_resolution_distinguishes_change_supersession_and_absence(self) -> None: + old = reference( + revision="3", + fingerprint=semantic_documentation_fingerprint({"revision": "3"}), + ) + current = descriptor() + changed = SemanticDocumentationSubjectResolution( + requested_reference=old, + availability="changed", + subject=current, + ) + superseded = SemanticDocumentationSubjectResolution( + requested_reference=old, + availability="superseded", + superseded_by=SemanticDocumentationSubjectReference( + module_id="forms", + tenant_id="tenant-1", + subject_kind="form", + subject_id="permit-application-v2", + ), + ) + missing = SemanticDocumentationSubjectResolution( + requested_reference=old, + availability="missing", + reason_code="subject_removed", + ) + + self.assertEqual(changed.availability, "changed") + self.assertEqual(superseded.superseded_by.subject_id, "permit-application-v2") + self.assertEqual(missing.reason_code, "subject_removed") + with self.assertRaisesRegex( + SemanticDocumentationContractError, + "require a revision or fingerprint change", + ): + SemanticDocumentationSubjectResolution( + requested_reference=current.reference, + availability="changed", + subject=current, + ) + + def test_subject_pages_reject_duplicate_stable_identities(self) -> None: + with self.assertRaisesRegex( + SemanticDocumentationContractError, + "duplicate identities", + ): + SemanticDocumentationSubjectPage( + subjects=(descriptor(), descriptor(label="Renamed")), + ) + + def test_registry_discovers_optional_providers_without_direct_imports(self) -> None: + provider = Provider() + registry = provider_registry(provider) + + self.assertEqual( + semantic_documentation_subject_provider_names(registry), + ("documentation.semantic_subjects.forms",), + ) + self.assertEqual( + semantic_documentation_subject_providers(registry), + (("forms", provider),), + ) + + def test_listing_and_resolution_are_tenant_and_principal_aware(self) -> None: + provider = Provider() + registry = provider_registry(provider) + allowed = SimpleNamespace(tenant_id="tenant-1", account_id="account-1") + wrong_tenant = SimpleNamespace( + tenant_id="tenant-2", account_id="account-1" + ) + denied = SimpleNamespace(tenant_id="tenant-1", account_id="denied") + + pages = list_semantic_documentation_subjects( + registry, + object(), + allowed, + request=SemanticDocumentationSubjectQuery(tenant_id="tenant-1"), + ) + self.assertEqual(len(pages), 1) + self.assertEqual(pages[0][1].subjects[0].reference.tenant_id, "tenant-1") + self.assertEqual( + list_semantic_documentation_subjects( + registry, + object(), + wrong_tenant, + request=SemanticDocumentationSubjectQuery(tenant_id="tenant-1"), + ), + (), + ) + self.assertIsNone( + resolve_semantic_documentation_subject( + registry, + object(), + denied, + reference=reference(), + ) + ) + self.assertEqual(provider.list_calls, 1) + self.assertEqual(provider.resolve_calls, 1) + + def test_missing_optional_module_resolves_without_loading_feature_code(self) -> None: + principal = SimpleNamespace(tenant_id="tenant-1", account_id="account-1") + result = resolve_semantic_documentation_subject( + PlatformRegistry(), + object(), + principal, + reference=reference(), + ) + + self.assertEqual(result.availability, "temporarily_unavailable") + self.assertEqual(result.reason_code, "provider_unavailable") + + def test_manifest_requires_exact_capability_docs_and_static_baselines(self) -> None: + provider = Provider() + capability = semantic_documentation_subject_capability("forms") + missing_baseline = PlatformRegistry() + missing_baseline.register( + ModuleManifest( + id="forms", + name="Forms", + version="1.0.0", + capability_factories={capability: lambda _context: provider}, + capability_documentation={ + capability: CapabilityDocumentation( + label="Form semantics", + summary="Lists documentation-safe configured form subjects.", + contract_version=( + SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION + ), + ) + }, + documentation=(static_topic("admin"),), + ) + ) + + with self.assertRaisesRegex(RegistryError, "static user and administrator"): + missing_baseline.validate() + + wrong_capability = PlatformRegistry() + wrong_capability.register( + ModuleManifest( + id="forms", + name="Forms", + version="1.0.0", + capability_factories={ + semantic_documentation_subject_capability("workflow"): ( + lambda _context: provider + ) + }, + ) + ) + with self.assertRaisesRegex(RegistryError, "must be"): + wrong_capability.validate() + + def test_fingerprint_is_canonical_and_rejects_non_json_values(self) -> None: + self.assertEqual( + semantic_documentation_fingerprint({"b": 2, "a": 1}), + semantic_documentation_fingerprint({"a": 1, "b": 2}), + ) + with self.assertRaisesRegex( + SemanticDocumentationContractError, + "canonical JSON", + ): + semantic_documentation_fingerprint({"unsafe": object()}) + + +def static_topic(documentation_type: str) -> DocumentationTopic: + return DocumentationTopic( + id=f"forms.semantic.{documentation_type}", + title="Form semantics", + summary="Static form semantics baseline.", + documentation_types=(documentation_type,), # type: ignore[arg-type] + ) + + +def provider_registry(provider: Provider) -> PlatformRegistry: + capability = semantic_documentation_subject_capability("forms") + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="forms", + name="Forms", + version="1.0.0", + capability_factories={capability: lambda _context: provider}, + capability_documentation={ + capability: CapabilityDocumentation( + label="Form semantic subjects", + summary="Lists safe configured forms and fields for Docs.", + contract_version=( + SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION + ), + documentation_types=("admin", "user"), + ) + }, + documentation=(static_topic("admin"), static_topic("user")), + ) + ) + registry.validate() + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + return registry + + +if __name__ == "__main__": + unittest.main()