feat(docs): define semantic subject contract

This commit is contained in:
2026-08-21 14:55:58 +02:00
parent 4b0737e1cd
commit 925dc33696
6 changed files with 1176 additions and 0 deletions
+1
View File
@@ -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. | | 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. | | 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. | | 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. | | 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. | | 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. | | 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. |
+11
View File
@@ -1621,6 +1621,17 @@ URLs never contain credentials; only credential-envelope references cross the
contract. Provider-specific details belong in sanitized provenance rather than contract. Provider-specific details belong in sanitized provenance rather than
in a shared domain schema. 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.<module_id>` 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 ## Build And Verification
Backend verification from core: Backend verification from core:
+118
View File
@@ -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.<module_id>`. 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.
+47
View File
@@ -61,6 +61,11 @@ from govoplan_core.core.search import (
SearchProvider, SearchProvider,
SearchSourceProvider, 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 ( from govoplan_core.core.tasks import (
RegisteredWorkItemProvider, RegisteredWorkItemProvider,
WorkItemProvider, WorkItemProvider,
@@ -1215,6 +1220,48 @@ def _validate_documentation_extensions(manifest: ModuleManifest) -> None:
"documentation contract version must not be empty" "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() provider_keys: set[str] = set()
for registration in manifest.documentation_configuration_providers: for registration in manifest.documentation_configuration_providers:
if not registration.keys: if not registration.keys:
@@ -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",
]
@@ -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()