from __future__ import annotations from typing import Any, Mapping from urllib.parse import urlsplit from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from govoplan_core.auth import ApiPrincipal, has_scope, require_any_scope from govoplan_core.core.modules import ( DocumentationConfigurationDecision, DocumentationCondition, DocumentationContext, DocumentationLink, DocumentationTopic, DocumentationType, FrontendRoute, ModuleManifest, NavItem, PermissionDefinition, localized_documentation_metadata, user_workflow_scope_condition_issues, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.provider_governance import ( ExternalProviderStateContext, collect_external_provider_states, ) from govoplan_core.core.versioning import ( format_version_range, version_satisfies_range, version_tuple, ) from govoplan_core.db.session import get_database from govoplan_docs.backend.manifest import DOCS_ADMIN_READ_SCOPES, DOCS_READ_SCOPES from govoplan_docs.backend.api.v1.semantic_routes import router as semantic_router from govoplan_docs.backend.sources import ( RegisteredDocumentationSource, build_documentation_source_registry, ) from govoplan_docs.backend.semantic_service import ( list_semantic_entries, prefetch_semantic_revisions, select_locale_entries, semantic_entry_payload, ) router = APIRouter(prefix="/docs", tags=["docs"]) router.include_router(semantic_router) TOPIC_KINDS = ("workflow", "reference", "pattern", "system") _CONFIGURATION_ACTIVE_STATES = frozenset({"enabled", "inherited"}) _CONFIGURATION_STATES = frozenset( {"enabled", "disabled", "inherited", "unavailable"} ) @router.get("/context") def docs_context( request: Request, documentation_type: DocumentationType = Query(default="user", alias="type", pattern="^(admin|user)$"), locale: str | None = Query(default=None, min_length=2, max_length=20), version: str | None = Query(default=None, min_length=1, max_length=40), principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)), ) -> dict[str, Any]: can_read_admin_documentation = _has_any_scope(principal, DOCS_ADMIN_READ_SCOPES) if documentation_type == "admin" and not can_read_admin_documentation: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Administrative documentation requires documentation-administrator authority", ) registry = _registry(request) external_provider_states = _external_provider_states(registry, principal) target_version = version if isinstance(version, str) else None resolved_locale = _preferred_locale(request, locale) route_items = _route_items(registry.manifests(), principal) visible_route_items = [item for item in route_items if item["visible"]] documentation_layers = _documentation_layers( request, registry, principal, documentation_type=documentation_type, locale=resolved_locale, target_version=target_version, ) evidence_sources = ( _documentation_source_summaries( request, registry, principal, documentation_type=documentation_type, locale=resolved_locale, ) if documentation_type == "admin" else [] ) if documentation_type == "admin": catalog = _admin_documentation_catalog( registry, principal, external_provider_states=external_provider_states, route_items=route_items, visible_route_items=visible_route_items, ) else: catalog = _user_documentation_catalog( registry, external_provider_states=external_provider_states, visible_route_items=visible_route_items, documentation_layers=documentation_layers, ) topic_groups = _documentation_topic_groups(documentation_layers) return { "versions": _documentation_version_context( registry, target_version=target_version, ), "actor": _documentation_actor( principal, documentation_type=documentation_type, locale=resolved_locale, can_read_admin=can_read_admin_documentation, ), "summary": _documentation_summary( catalog, documentation_layers=documentation_layers, topic_groups=topic_groups, ), "topic_groups": topic_groups, "layers": _documentation_layer_payload( catalog, documentation_layers=documentation_layers, evidence_sources=evidence_sources, ), } @router.get("/sources") def list_documentation_sources( request: Request, documentation_type: DocumentationType = Query( default="admin", alias="type", pattern="^(admin|user)$", ), locale: str | None = Query(default=None, min_length=2, max_length=20), principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)), ) -> dict[str, Any]: _require_documentation_type_access(principal, documentation_type) registry = _registry(request) items = _documentation_source_items( request, registry, principal, documentation_type=documentation_type, locale=_preferred_locale(request, locale), ) return { "items": [source.item.summary() for source in items], "total": len(items), } @router.get("/sources/{source_id}") def inspect_documentation_source( source_id: str, request: Request, documentation_type: DocumentationType = Query( default="admin", alias="type", pattern="^(admin|user)$", ), locale: str | None = Query(default=None, min_length=2, max_length=20), principal: ApiPrincipal = Depends(require_any_scope(*DOCS_READ_SCOPES)), ) -> dict[str, Any]: _require_documentation_type_access(principal, documentation_type) registry = _registry(request) items = _documentation_source_items( request, registry, principal, documentation_type=documentation_type, locale=_preferred_locale(request, locale), ) source = next((item for item in items if item.item.id == source_id), None) if source is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Documentation source not found", ) return source.item.model_dump(mode="json") def _require_documentation_type_access( principal: ApiPrincipal, documentation_type: DocumentationType, ) -> None: if documentation_type == "admin" and not _has_any_scope( principal, DOCS_ADMIN_READ_SCOPES, ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Administrative documentation requires documentation-administrator authority", ) def _admin_documentation_catalog( registry: PlatformRegistry, principal: ApiPrincipal, *, external_provider_states: Mapping[str, Mapping[str, object]], route_items: list[dict[str, Any]], visible_route_items: list[dict[str, Any]], ) -> dict[str, list[dict[str, Any]]]: return { "modules": [ _module_payload( manifest, technical=True, external_provider_states=external_provider_states, ) for manifest in registry.manifests() ], "permissions": [ _permission_payload(permission, principal) for permission in registry.permissions() ], "visible_routes": visible_route_items, "available_routes": [item for item in route_items if not item["visible"]], "optional_modules": _optional_module_evidence(registry.manifests()), } def _user_documentation_catalog( registry: PlatformRegistry, *, external_provider_states: Mapping[str, Mapping[str, object]], visible_route_items: list[dict[str, Any]], documentation_layers: Mapping[str, list[dict[str, Any]]], ) -> dict[str, list[dict[str, Any]]]: visible_module_ids = { *(str(item["module_id"]) for item in visible_route_items), *( str(topic["source_module_id"]) for topic in _all_documentation_topics(documentation_layers) ), } return { "modules": [ _module_payload( manifest, technical=False, external_provider_states=external_provider_states, ) for manifest in registry.manifests() if manifest.id in visible_module_ids ], "permissions": [], "visible_routes": _user_route_payloads(visible_route_items), "available_routes": [], "optional_modules": [], } def _documentation_actor( principal: ApiPrincipal, *, documentation_type: DocumentationType, locale: str, can_read_admin: bool, ) -> dict[str, Any]: actor: dict[str, Any] = { "documentation_type": documentation_type, "locale": locale, "available_documentation_types": [ "user", *(["admin"] if can_read_admin else []), ], } if documentation_type == "admin": actor.update( tenant_id=principal.tenant_id, user_id=principal.user.id, scope_count=len(principal.scopes), ) return actor def _documentation_version_context( registry: PlatformRegistry, *, target_version: str | None, ) -> dict[str, Any]: manifests = registry.manifests() installed_versions = {manifest.id: manifest.version for manifest in manifests} supported_versions = sorted( { *(manifest.version for manifest in manifests), *( topic.version_min for manifest in manifests for topic in manifest.documentation if topic.version_min ), }, key=version_tuple, reverse=True, ) latest = supported_versions[0] if supported_versions else None if target_version is None: status_name = "installed" elif target_version == latest: status_name = "stable" elif target_version in supported_versions: status_name = "older_supported" else: status_name = "unsupported" return { "mode": "selected" if target_version else "installed", "selected_version": target_version, "status": status_name, "latest_version": latest, "stable_version": latest, "supported_versions": supported_versions, "installed_versions": installed_versions, "fallback_policy": ( "Topics without bounds apply to every version. Bounded topics are " "hidden outside their declared half-open version range." ), } def _documentation_summary( catalog: Mapping[str, list[dict[str, Any]]], *, documentation_layers: Mapping[str, list[dict[str, Any]]], topic_groups: Mapping[str, list[dict[str, Any]]], ) -> dict[str, int]: permissions = catalog["permissions"] return { "module_count": len(catalog["modules"]), "architecture_declared_module_count": sum( 1 for item in catalog["modules"] if item.get("architecture") ), "external_provider_count": sum( int(item.get("external_provider_count") or 0) for item in catalog["modules"] ), "visible_route_count": len(catalog["visible_routes"]), "available_route_count": len(catalog["available_routes"]), "permission_count": len(permissions), "granted_permission_count": sum( 1 for item in permissions if item["granted"] ), "optional_module_count": len(catalog["optional_modules"]), "documentation_topic_count": sum( len(layer) for layer in documentation_layers.values() ), "configured_documentation_topic_count": len( documentation_layers["configured"] ), "workflow_topic_count": len(topic_groups["workflow"]), "reference_topic_count": len(topic_groups["reference"]), "pattern_topic_count": len(topic_groups["pattern"]), "system_topic_count": len(topic_groups["system"]), } def _documentation_layer_payload( catalog: Mapping[str, list[dict[str, Any]]], *, documentation_layers: Mapping[str, list[dict[str, Any]]], evidence_sources: list[dict[str, Any]], ) -> dict[str, dict[str, object]]: permissions = catalog["permissions"] return { "always": { "documentation": documentation_layers["always"], }, "configured": { "modules": catalog["modules"], "routes": catalog["visible_routes"], "permissions": permissions, "documentation": documentation_layers["configured"], }, "available": { "routes": catalog["available_routes"], "permissions": [item for item in permissions if not item["granted"]], "documentation": documentation_layers["available"], }, "evidence": { "optional_modules": catalog["optional_modules"], "sources": evidence_sources, "documentation": documentation_layers["evidence"], }, } def _registry(request: Request) -> PlatformRegistry: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="GovOPlaN module registry is not configured") return registry def _module_payload( manifest: ModuleManifest, *, technical: bool, external_provider_states: Mapping[str, Mapping[str, object]] | None = None, ) -> dict[str, Any]: frontend = manifest.frontend migration = manifest.migration_spec architecture = manifest.architecture.to_dict() if manifest.architecture else None if architecture is not None and not technical: architecture = { "contract_version": architecture["contract_version"], "layer": architecture["layer"], "kind": architecture["kind"], "maturity": architecture["maturity"], "known_limits": architecture["known_limits"], "supported_authority_modes": architecture[ "supported_authority_modes" ], "owned_concepts": architecture["owned_concepts"], "non_owned_concepts": architecture["non_owned_concepts"], "reference_packages": architecture["reference_packages"], "target_tested_providers": architecture[ "target_tested_providers" ], "evidence": [], "documentation": {}, } return { "id": manifest.id, "name": manifest.name, "version": manifest.version if technical else "", "dependencies": list(manifest.dependencies) if technical else [], "optional_dependencies": list(manifest.optional_dependencies) if technical else [], "permission_count": len(manifest.permissions) if technical else 0, "role_template_count": len(manifest.role_templates) if technical else 0, "nav_count": len(manifest.nav_items) + (len(frontend.nav_items) if frontend else 0) if technical else 0, "route_count": (1 if manifest.route_factory else 0) + (len(frontend.routes) if frontend else 0) if technical else 0, "frontend_package": frontend.package_name if frontend and technical else None, "backend_route_contributed": manifest.route_factory is not None if technical else False, "migration_module_id": migration.module_id if migration and technical else None, "capabilities": sorted(manifest.capability_factories) if technical else [], "documentation_count": len(manifest.documentation) if technical else 0, "documentation_provider_count": len(manifest.documentation_providers) if technical else 0, "architecture": architecture, "external_provider_count": len(manifest.external_providers), "external_providers": [ { **( declaration.to_dict() if technical else { "id": declaration.id, "module_id": declaration.module_id, "label": declaration.label, "maturity": declaration.maturity, "operations": list(declaration.operations), "authority_modes": list(declaration.authority_modes), "known_outage_behavior": declaration.behavior.outage, } ), "runtime_state": _documentation_provider_state( (external_provider_states or {}).get(declaration.id), technical=technical, ), } for declaration in manifest.external_providers ], } def _documentation_provider_state( state: Mapping[str, object] | None, *, technical: bool, ) -> dict[str, object] | None: if state is None: return None if technical: return dict(state) return { key: state.get(key) for key in ( "configured", "active", "authority_mode", "authority_modes", "health", "freshness", "conflict", "recovery", "observed_at", ) } def _external_provider_states( registry: PlatformRegistry, principal: ApiPrincipal, ) -> dict[str, dict[str, object]]: registrations = registry.external_provider_state_providers() if not registrations: return {} tenant_id = str(principal.tenant_id or "").strip() or None try: with get_database().session() as session: return collect_external_provider_states( registrations, ExternalProviderStateContext( session=session, tenant_id=tenant_id, principal=principal, ), ) except Exception: # noqa: BLE001 - return sanitized per-provider diagnostics. return collect_external_provider_states( registrations, ExternalProviderStateContext( session=None, tenant_id=tenant_id, principal=principal, ), ) def _user_route_payload(item: Mapping[str, Any]) -> dict[str, Any]: return { "module_id": item["module_id"], "path": item["path"], "label": item["label"], "icon": item["icon"], "section": item["section"], "source": "visible", "component": None, "required_all": [], "required_any": [], "order": item["order"], "visible": True, "reason": "visible", } def _user_route_payloads(items: list[dict[str, Any]]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() for item in items: key = (str(item["module_id"]), str(item["path"])) if key in seen: continue seen.add(key) result.append(_user_route_payload(item)) return result def _permission_payload(permission: PermissionDefinition, principal: ApiPrincipal) -> dict[str, Any]: return { "scope": permission.scope, "label": permission.label, "description": permission.description, "category": permission.category, "level": permission.level, "module_id": permission.module_id, "resource": permission.resource, "action": permission.action, "deprecated": permission.deprecated, "granted": has_scope(principal, permission.scope), } def _route_items(manifests: tuple[ModuleManifest, ...], principal: ApiPrincipal) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] for manifest in manifests: seen: set[tuple[str, str]] = set() for item in manifest.nav_items: key = ("nav", item.path) if key not in seen: seen.add(key) items.append(_nav_item_payload(manifest.id, item, principal, source="nav")) if manifest.frontend is None: continue nav_by_path = {item.path: item for item in manifest.frontend.nav_items} for item in manifest.frontend.nav_items: key = ("frontend_nav", item.path) if key not in seen: seen.add(key) items.append(_nav_item_payload(manifest.id, item, principal, source="frontend_nav")) for route in manifest.frontend.routes: key = ("frontend_route", route.path) if key in seen: continue seen.add(key) items.append(_frontend_route_payload(manifest.id, route, nav_by_path.get(route.path), principal)) return sorted(items, key=lambda item: (int(item["order"]), str(item["path"]), str(item["source"]))) def _nav_item_payload(module_id: str, item: NavItem, principal: ApiPrincipal, *, source: str) -> dict[str, Any]: visible, reason = _visibility(item.required_all, item.required_any, principal) return { "module_id": module_id, "path": item.path, "label": item.label, "icon": item.icon, "section": item.section, "source": source, "component": None, "required_all": list(item.required_all), "required_any": list(item.required_any), "order": item.order, "visible": visible, "reason": reason, } def _frontend_route_payload(module_id: str, route: FrontendRoute, nav_item: NavItem | None, principal: ApiPrincipal) -> dict[str, Any]: visible, reason = _visibility(route.required_all, route.required_any, principal) return { "module_id": module_id, "path": route.path, "label": nav_item.label if nav_item else route.component, "icon": nav_item.icon if nav_item else None, "section": nav_item.section if nav_item else None, "source": "frontend_route", "component": route.component, "required_all": list(route.required_all), "required_any": list(route.required_any), "order": route.order, "visible": visible, "reason": reason, } def _visibility(required_all: tuple[str, ...], required_any: tuple[str, ...], principal: ApiPrincipal) -> tuple[bool, str]: missing_all = [scope for scope in required_all if not has_scope(principal, scope)] any_satisfied = not required_any or any(has_scope(principal, scope) for scope in required_any) if not missing_all and any_satisfied: return True, "visible" reasons: list[str] = [] if missing_all: reasons.append("missing " + ", ".join(missing_all)) if not any_satisfied: reasons.append("requires one of " + ", ".join(required_any)) return False, "; ".join(reasons) def _optional_module_evidence(manifests: tuple[ModuleManifest, ...]) -> list[dict[str, str]]: installed = {manifest.id for manifest in manifests} evidence: list[dict[str, str]] = [] for manifest in manifests: for module_id in manifest.optional_dependencies: evidence.append({ "module_id": module_id, "source_module_id": manifest.id, "status": "installed" if module_id in installed else "not_installed", "reason": f"{manifest.name} declares optional integration with {module_id}.", }) return sorted(evidence, key=lambda item: (item["status"], item["module_id"], item["source_module_id"])) def _documentation_layers( request: Request, registry: PlatformRegistry, principal: ApiPrincipal, *, documentation_type: DocumentationType, locale: str, target_version: str | None = None, ) -> dict[str, list[dict[str, Any]]]: settings = _settings(request) try: with get_database().SessionLocal() as session: return _classify_documentation(registry, principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale, target_version=target_version) except RuntimeError: return _classify_documentation(registry, principal, settings=settings, session=None, documentation_type=documentation_type, locale=locale, target_version=target_version) def _settings(request: Request) -> object | None: direct_settings = getattr(request.app.state, "govoplan_settings", None) if direct_settings is not None: return direct_settings lifecycle = getattr(request.app.state, "govoplan_lifecycle", None) return getattr(lifecycle, "settings", None) def _documentation_source_summaries( request: Request, registry: PlatformRegistry, principal: ApiPrincipal, *, documentation_type: DocumentationType, locale: str, ) -> list[dict[str, Any]]: return [ source.item.summary() for source in _documentation_source_items( request, registry, principal, documentation_type=documentation_type, locale=locale, ) ] def _documentation_source_items( request: Request, registry: PlatformRegistry, principal: ApiPrincipal, *, documentation_type: DocumentationType, locale: str, ) -> list[RegisteredDocumentationSource]: source_registry = build_documentation_source_registry(registry.manifests()) settings = _settings(request) try: with get_database().SessionLocal() as session: return _visible_documentation_sources( source_registry.sources(), registry, principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale, ) except RuntimeError: return _visible_documentation_sources( source_registry.sources(), registry, principal, settings=settings, session=None, documentation_type=documentation_type, locale=locale, ) def _visible_documentation_sources( sources: tuple[RegisteredDocumentationSource, ...], registry: PlatformRegistry, principal: ApiPrincipal, *, settings: object | None, session: object | None, documentation_type: DocumentationType, locale: str, ) -> list[RegisteredDocumentationSource]: installed = {manifest.id for manifest in registry.manifests()} context = DocumentationContext( registry=registry, principal=principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale, ) visible: list[RegisteredDocumentationSource] = [] for source in sources: if documentation_type not in source.documentation_types: continue keys = tuple(dict.fromkeys(( *source.condition.configuration_keys, *([source.configuration_key] if source.configuration_key else []), ))) configuration = _resolve_documentation_configuration( registry, source.item.owner_module_id, keys, context=context, ) active, reason, blockers = _condition_visibility( source.condition, installed, registry, principal, configuration=configuration, ) if blockers["scopes"]: continue state = source.item.state source_configuration = ( configuration.get(source.configuration_key) if source.configuration_key else None ) if state == "configured" and source_configuration is not None: if source_configuration.state == "unavailable": state = "unavailable" reason = source_configuration.reason or reason elif source_configuration.state == "disabled": state = "disabled" reason = source_configuration.reason or reason if state == "configured" and not active: state = ( "unavailable" if any( decision.state == "unavailable" for decision in configuration.values() ) or blockers["modules"] or blockers["capabilities"] else "disabled" ) item = source.item.model_copy(update={ "state": state, "state_reason": None if state == "configured" else reason, }) visible.append(RegisteredDocumentationSource( item=item, condition=source.condition, documentation_types=source.documentation_types, configuration_key=source.configuration_key, )) return visible def _classify_documentation( registry: PlatformRegistry, principal: ApiPrincipal, *, settings: object | None, session: object | None, documentation_type: DocumentationType, locale: str, target_version: str | None = None, ) -> dict[str, list[dict[str, Any]]]: layers: dict[str, list[dict[str, Any]]] = {"always": [], "configured": [], "available": [], "evidence": []} installed = {manifest.id for manifest in registry.manifests()} visible_area_modules = frozenset( str(item["module_id"]) for item in _route_items(registry.manifests(), principal) if item["visible"] ) if documentation_type == "user" else frozenset(installed) visible_runtime_paths = frozenset([ "/settings", # Authenticated shell route, not contributed by a module manifest. *( str(item["path"]) for item in _route_items(registry.manifests(), principal) if item["visible"] ), ]) if documentation_type == "user" else frozenset() context = DocumentationContext( registry=registry, principal=principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale, data={ "target_version": target_version, "installed_versions": { manifest.id: manifest.version for manifest in registry.manifests() }, }, ) topics = _collect_documentation_topics( registry, principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale, target_version=target_version, ) for source_module_id, topic in sorted(topics, key=lambda item: (item[1].order, item[0], item[1].id)): if not _topic_matches_documentation_type(topic, documentation_type): continue module_id = topic.source_module_id or source_module_id manifest = registry.get(module_id) resolved_version = target_version or (manifest.version if manifest else "0") if not version_satisfies_range( resolved_version, version_min=topic.version_min, version_max_exclusive=topic.version_max_exclusive, ): continue configuration_keys = _documentation_configuration_keys(topic) configuration = _resolve_documentation_configuration( registry, topic.source_module_id or source_module_id, configuration_keys, context=context, ) active, reason, blockers = _documentation_visibility( topic, installed, registry, principal, configuration=configuration, ) if documentation_type == "user" and not active: continue target_layer = _documentation_target_layer( topic, active, blockers, configuration=configuration, ) if target_layer not in layers: target_layer = "evidence" layers[target_layer].append(_documentation_topic_payload( source_module_id, topic, active=active, reason=reason, target_layer=target_layer, blockers=blockers, locale=locale, documentation_type=documentation_type, visible_runtime_paths=visible_runtime_paths, configuration=configuration, resolved_version=resolved_version, visible_area_modules=visible_area_modules, )) return layers def _collect_documentation_topics( registry: PlatformRegistry, principal: ApiPrincipal, *, settings: object | None, session: object | None, documentation_type: DocumentationType, locale: str, target_version: str | None = None, ) -> list[tuple[str, DocumentationTopic]]: topics: list[tuple[str, DocumentationTopic]] = [] for manifest in registry.manifests(): for topic in manifest.documentation: if not user_workflow_scope_condition_issues(topic): topics.append((manifest.id, topic)) if not manifest.documentation_providers: continue context = DocumentationContext( registry=registry, principal=principal, settings=settings, session=session, documentation_type=documentation_type, locale=locale, data={ "source_module_id": manifest.id, "target_version": target_version, "installed_versions": { item.id: item.version for item in registry.manifests() }, }, ) for provider in manifest.documentation_providers: try: provided_topics = tuple(provider(context)) except Exception as exc: if documentation_type != "admin": continue provided_topics = ( DocumentationTopic( id=f"{manifest.id}.runtime-documentation-unavailable", title=f"{manifest.name} runtime documentation unavailable", summary="The module registered runtime documentation, but it could not be evaluated for this request.", body="The static module documentation remains available. Check the module API and logs if this persists.", layer="evidence", documentation_types=(documentation_type,), source_module_id=manifest.id, metadata={"error_type": type(exc).__name__}, ), ) for topic in provided_topics: if not user_workflow_scope_condition_issues(topic): topics.append((manifest.id, topic)) if session is not None: try: topics.extend( ("docs", topic) for topic in _semantic_documentation_topics( registry, principal, session=session, locale=locale, ) ) except Exception as exc: if documentation_type == "admin": topics.append( ( "docs", DocumentationTopic( id="docs.semantic-runtime-unavailable", title="Tenant semantic documentation unavailable", summary="Stored tenant semantics could not be projected for this request.", body="Static module documentation remains available. Check the Docs database migration, subject providers, and application logs.", layer="evidence", documentation_types=("admin",), source_module_id="docs", metadata={ "kind": "system", "error_type": type(exc).__name__, }, ), ) ) return topics def _semantic_documentation_topics( registry: PlatformRegistry, principal: ApiPrincipal, *, session: object, locale: str, ) -> tuple[DocumentationTopic, ...]: entries = select_locale_entries( list_semantic_entries(session, principal), locale=locale, ) revisions = prefetch_semantic_revisions(session, principal, entries=entries, editor=False) topics: list[DocumentationTopic] = [] for entry in entries: payload = semantic_entry_payload( session, registry, principal, entry=entry, editor=False, requested_locale=locale, revisions=revisions, ) if payload is None: continue content = payload["content"] resolution = payload["subject_resolution"] subject = resolution.get("subject") if isinstance(resolution, Mapping) else None route = subject.get("route") if isinstance(subject, Mapping) else None route_anchor = ( subject.get("route_anchor") if isinstance(subject, Mapping) else None ) links = [ DocumentationLink( label=str(item["label"]), href=str(item["href"]), kind="runtime" if str(item["href"]).startswith("/") else "external", ) for item in content.get("links", ()) if isinstance(item, Mapping) and item.get("label") and item.get("href") ] if isinstance(route, str) and route.startswith("/"): links.insert( 0, DocumentationLink( label="Open configured subject", href=(f"{route}#{route_anchor}" if route_anchor else route), kind="runtime", ), ) body_parts = [ str(content.get(key) or "").strip() for key in ( "meaning", "body", "intended_use", "non_intended_use", ) ] topics.append( DocumentationTopic( id=f"docs.semantic.{entry.id}", title=str(content.get("title") or "Semantic documentation"), summary=str(content.get("summary") or "Tenant semantic guidance"), body="\n\n".join(part for part in body_parts if part), layer="configured", documentation_types=("admin", "user"), source_module_id="docs", order=200, links=tuple(links), related_modules=(entry.subject_module_id,), metadata={ "kind": "reference", "source_badge": "tenant_semantic", "semantic_entry_id": entry.id, "semantic_subject": payload["subject"], "semantic_subject_stable_key": entry.subject_stable_key, "subject_availability": resolution.get("availability"), "locale": payload["locale"], "requested_locale": payload["requested_locale"], "locale_fallback": payload["locale_fallback"], "lifecycle_state": payload["lifecycle_state"], "pending_draft": payload["pending_draft"], "route": route, "route_anchor": route_anchor, "help_contexts": [ _semantic_help_context(entry, route_anchor=route_anchor), ], }, ) ) return tuple(topics) def _semantic_help_context( entry: object, *, route_anchor: object | None, ) -> str: values = [ "semantic", str(getattr(entry, "subject_module_id")), str(getattr(entry, "subject_kind")), str(getattr(entry, "subject_id")), ] if route_anchor: values.append(str(route_anchor)) return ".".join(values) def _topic_matches_documentation_type(topic: DocumentationTopic, documentation_type: DocumentationType) -> bool: return documentation_type in (topic.documentation_types or ("admin",)) def _documentation_configuration_keys( topic: DocumentationTopic, ) -> tuple[str, ...]: return tuple(dict.fromkeys(( *topic.configuration_keys, *( key for condition in topic.conditions for key in condition.configuration_keys ), ))) def _resolve_documentation_configuration( registry: PlatformRegistry, module_id: str, keys: tuple[str, ...], *, context: DocumentationContext, ) -> dict[str, DocumentationConfigurationDecision]: decisions = { key: DocumentationConfigurationDecision( key=key, state="unavailable", reason="No configuration-state provider is registered.", ) for key in keys } manifest = registry.get(module_id) if manifest is None or not keys: return decisions for registration in manifest.documentation_configuration_providers: requested = tuple(key for key in keys if key in registration.keys) if not requested: continue try: provided = registration.resolve(context, requested) except Exception as exc: for key in requested: decisions[key] = DocumentationConfigurationDecision( key=key, state="unavailable", reason=f"Configuration-state provider failed ({type(exc).__name__}).", ) continue for key in requested: decision = provided.get(key) if ( not isinstance(decision, DocumentationConfigurationDecision) or decision.key != key or decision.state not in _CONFIGURATION_STATES ): decisions[key] = DocumentationConfigurationDecision( key=key, state="unavailable", reason="Configuration-state provider returned an invalid decision.", ) continue decisions[key] = DocumentationConfigurationDecision( key=key, state=decision.state, source=_bounded_configuration_text(decision.source), reason=_bounded_configuration_text(decision.reason), ) return decisions def _bounded_configuration_text(value: object | None) -> str | None: if value is None: return None clean = str(value).strip() return clean[:500] if clean else None def _documentation_configuration_payload( decision: DocumentationConfigurationDecision, ) -> dict[str, str | None]: return { "key": decision.key, "state": decision.state, "source": decision.source, "reason": decision.reason, } def _documentation_visibility( topic: DocumentationTopic, installed: set[str], registry: PlatformRegistry, principal: ApiPrincipal, *, configuration: Mapping[str, DocumentationConfigurationDecision] | None = None, ) -> tuple[bool, str, dict[str, list[str]]]: if not topic.conditions: return True, "documented", { "modules": [], "capabilities": [], "scopes": [], "configuration": [], } reasons: list[str] = [] blockers = { "modules": [], "capabilities": [], "scopes": [], "configuration": [], } for condition in topic.conditions: active, reason, condition_blockers = _condition_visibility( condition, installed, registry, principal, configuration=configuration, ) if active: return True, reason, { "modules": [], "capabilities": [], "scopes": [], "configuration": [], } reasons.append(reason) for key, values in condition_blockers.items(): blockers[key].extend(value for value in values if value not in blockers[key]) return False, "; ".join(reason for reason in reasons if reason) or "documentation conditions are not satisfied", blockers def _condition_visibility( condition: DocumentationCondition, installed: set[str], registry: PlatformRegistry, principal: ApiPrincipal, *, configuration: Mapping[str, DocumentationConfigurationDecision] | None = None, ) -> tuple[bool, str, dict[str, list[str]]]: missing_required_modules = _missing_required_modules(condition, installed) unsatisfied_any_modules = _unsatisfied_any_modules(condition, installed) conflicting_modules = _conflicting_modules(condition, installed) missing_capabilities = _missing_required_capabilities(condition, registry) missing_scopes = _missing_required_scopes(condition, principal) unsatisfied_any_scopes = _unsatisfied_any_scopes(condition, principal) unavailable_configuration = _unavailable_configuration_keys( condition, configuration or {}, ) blockers = _condition_blockers( missing_required_modules=missing_required_modules, unsatisfied_any_modules=unsatisfied_any_modules, conflicting_modules=conflicting_modules, missing_capabilities=missing_capabilities, missing_scopes=missing_scopes, unsatisfied_any_scopes=unsatisfied_any_scopes, unavailable_configuration=unavailable_configuration, ) if _condition_has_no_blockers(blockers): return True, "conditions satisfied", blockers reason = _condition_blocker_reason( missing_required_modules=missing_required_modules, unsatisfied_any_modules=unsatisfied_any_modules, conflicting_modules=conflicting_modules, missing_capabilities=missing_capabilities, missing_scopes=missing_scopes, unsatisfied_any_scopes=unsatisfied_any_scopes, unavailable_configuration=unavailable_configuration, ) return False, reason, blockers def _missing_required_modules(condition: DocumentationCondition, installed: set[str]) -> list[str]: return [module_id for module_id in condition.required_modules if module_id not in installed] def _unsatisfied_any_modules(condition: DocumentationCondition, installed: set[str]) -> list[str]: if not condition.any_modules: return [] if any(module_id in installed for module_id in condition.any_modules): return [] return list(condition.any_modules) def _conflicting_modules(condition: DocumentationCondition, installed: set[str]) -> list[str]: return [module_id for module_id in condition.missing_modules if module_id in installed] def _missing_required_capabilities(condition: DocumentationCondition, registry: PlatformRegistry) -> list[str]: return [name for name in condition.required_capabilities if not registry.has_capability(name)] def _missing_required_scopes(condition: DocumentationCondition, principal: ApiPrincipal) -> list[str]: return [scope for scope in condition.required_scopes if not has_scope(principal, scope)] def _unsatisfied_any_scopes(condition: DocumentationCondition, principal: ApiPrincipal) -> list[str]: if not condition.any_scopes: return [] if any(has_scope(principal, scope) for scope in condition.any_scopes): return [] return list(condition.any_scopes) def _unavailable_configuration_keys( condition: DocumentationCondition, configuration: Mapping[str, DocumentationConfigurationDecision], ) -> list[str]: return [ key for key in condition.configuration_keys if configuration.get( key, DocumentationConfigurationDecision(key=key, state="unavailable"), ).state not in _CONFIGURATION_ACTIVE_STATES ] def _condition_blockers( *, missing_required_modules: list[str], unsatisfied_any_modules: list[str], conflicting_modules: list[str], missing_capabilities: list[str], missing_scopes: list[str], unsatisfied_any_scopes: list[str], unavailable_configuration: list[str], ) -> dict[str, list[str]]: blockers = { "modules": [], "capabilities": [], "scopes": [], "configuration": [], } _extend_unique(blockers["modules"], missing_required_modules) _extend_unique(blockers["modules"], unsatisfied_any_modules) _extend_unique(blockers["modules"], conflicting_modules) _extend_unique(blockers["capabilities"], missing_capabilities) _extend_unique(blockers["scopes"], missing_scopes) _extend_unique(blockers["scopes"], unsatisfied_any_scopes) _extend_unique(blockers["configuration"], unavailable_configuration) return blockers def _extend_unique(target: list[str], values: list[str]) -> None: target.extend(value for value in values if value not in target) def _condition_has_no_blockers(blockers: dict[str, list[str]]) -> bool: return not any(blockers.values()) def _condition_blocker_reason( *, missing_required_modules: list[str], unsatisfied_any_modules: list[str], conflicting_modules: list[str], missing_capabilities: list[str], missing_scopes: list[str], unsatisfied_any_scopes: list[str], unavailable_configuration: list[str], ) -> str: parts: list[str] = [] if missing_required_modules: parts.append("missing modules: " + ", ".join(missing_required_modules)) if unsatisfied_any_modules: parts.append("requires one installed module from: " + ", ".join(unsatisfied_any_modules)) if conflicting_modules: parts.append("not active when installed: " + ", ".join(conflicting_modules)) if missing_capabilities: parts.append("missing capabilities: " + ", ".join(missing_capabilities)) if missing_scopes: parts.append("missing scopes: " + ", ".join(missing_scopes)) if unsatisfied_any_scopes: parts.append("requires one scope from: " + ", ".join(unsatisfied_any_scopes)) if unavailable_configuration: parts.append( "configuration is disabled or unavailable: " + ", ".join(unavailable_configuration) ) return "; ".join(parts) def _documentation_target_layer( topic: DocumentationTopic, active: bool, blockers: dict[str, list[str]], *, configuration: Mapping[str, DocumentationConfigurationDecision] | None = None, ) -> str: if topic.layer == "always": return "always" if active: return topic.layer if blockers["modules"] or blockers["capabilities"]: return "evidence" if any( decision.state == "unavailable" for decision in (configuration or {}).values() ): return "evidence" return "available" def _documentation_topic_payload( source_module_id: str, topic: DocumentationTopic, *, active: bool, reason: str, target_layer: str, blockers: dict[str, list[str]], locale: str, documentation_type: DocumentationType, visible_runtime_paths: frozenset[str], configuration: Mapping[str, DocumentationConfigurationDecision], resolved_version: str, visible_area_modules: frozenset[str] = frozenset(), ) -> dict[str, Any]: module_id = topic.source_module_id or source_module_id translation_locale, translation = _translation_for_locale(topic, locale) structured_translation_locale = _structured_translation_locale(topic, locale) localized_metadata = localized_documentation_metadata( topic, structured_translation_locale ) kind = _documentation_topic_kind(topic) payload = { "id": topic.id, "source_module_id": module_id, "kind": kind, "anchor_id": _documentation_topic_anchor(module_id, topic.id), "title": str(translation.get("title") or topic.title), "summary": str(translation.get("summary") or topic.summary), "body": str(translation.get("body") or topic.body), "layer": topic.layer, "target_layer": target_layer, "documentation_types": list(topic.documentation_types), "active": active, "reason": reason, "blockers": { "modules": sorted(dict.fromkeys(blockers.get("modules", ()))), "capabilities": sorted(dict.fromkeys(blockers.get("capabilities", ()))), "scopes": sorted(dict.fromkeys(blockers.get("scopes", ()))), "configuration": sorted( dict.fromkeys(blockers.get("configuration", ())) ), }, "audience": list(topic.audience), "order": topic.order, "i18n_key": topic.i18n_key or topic.id, "locale": locale, "translation_locale": translation_locale, "structured_translation_locale": structured_translation_locale, "structured_translation_version": topic.structured_translation_version, "version": { "resolved": resolved_version, "minimum": topic.version_min, "maximum_exclusive": topic.version_max_exclusive, "range": format_version_range( version_min=topic.version_min, version_max_exclusive=topic.version_max_exclusive, ), "fallback": ( "unversioned" if topic.version_min is None and topic.version_max_exclusive is None else "matching_range" ), }, "conditions": [_documentation_condition_payload(condition) for condition in topic.conditions], "links": [_documentation_link_payload(link) for link in topic.links], "related_modules": list(topic.related_modules), "area_module_ids": sorted({ module_id, *( area for area in ( *topic.related_modules, *_bounded_string_list(localized_metadata.get("areas"), maximum_items=32, maximum_length=255), ) if area in visible_area_modules ), }), "unlocks": list(topic.unlocks), "configuration_keys": sorted({*topic.configuration_keys, *(key for condition in topic.conditions for key in condition.configuration_keys)}), "configuration_states": [ _documentation_configuration_payload(configuration[key]) for key in sorted(configuration) ], "metadata": localized_metadata, } if documentation_type == "admin": return payload return { "id": payload["id"], "source_module_id": payload["source_module_id"], "area_module_ids": payload["area_module_ids"], "kind": payload["kind"], "anchor_id": payload["anchor_id"], "title": payload["title"], "summary": payload["summary"], "body": payload["body"], "layer": payload["layer"], "target_layer": payload["target_layer"], "documentation_types": ["user"], "active": True, "reason": "documented", "blockers": { "modules": [], "capabilities": [], "scopes": [], "configuration": [], }, "audience": [], "order": payload["order"], "i18n_key": "", "locale": locale, "translation_locale": payload["translation_locale"], "structured_translation_locale": payload[ "structured_translation_locale" ], "structured_translation_version": payload[ "structured_translation_version" ], "conditions": [], "links": [ _documentation_link_payload(link) for link in topic.links if _user_link_allowed(link, visible_runtime_paths=visible_runtime_paths) ], "related_modules": [], "unlocks": list(topic.unlocks), "configuration_keys": [], "configuration_states": [], "metadata": _user_topic_metadata(kind, localized_metadata), } def _user_topic_metadata(kind: str, metadata: Mapping[str, Any]) -> dict[str, Any]: projected: dict[str, Any] = {} scalar_keys = ("kind", "screen", "section", "outcome", "result", "verification", "purpose", "when_used", "user_explanation") list_keys = ("prerequisites", "steps", "current_configuration", "limitations", "related_topic_ids", "help_contexts") id_keys = ("tree_parent_id", "parent_topic_id", "parent_id") for key in scalar_keys: value = _bounded_string(metadata.get(key), maximum=2_000) if value: projected[key] = value for key in list_keys: values = _bounded_string_list(metadata.get(key), maximum_items=64, maximum_length=2_000) if values: projected[key] = values for key in id_keys: value = _bounded_string(metadata.get(key), maximum=255) if value: projected[key] = value tags = _bounded_string_list(metadata.get("tags"), maximum_items=32, maximum_length=80) if tags: projected["tags"] = tags if kind == "reference" and isinstance(metadata.get("fields"), list): fields = [_user_field_metadata(item) for item in metadata["fields"][:64] if isinstance(item, Mapping)] if fields: projected["fields"] = [field for field in fields if field] constraints = _user_constraints(metadata.get("constraints")) if constraints: projected["constraints"] = constraints return projected def _user_field_metadata(field: Mapping[str, Any]) -> dict[str, Any]: allowed = ("field_id", "label", "user_description", "validation") return { key: value for key in allowed if (value := _bounded_string(field.get(key), maximum=2_000)) } def _user_constraints(value: object) -> list[dict[str, Any]]: if not isinstance(value, list): return [] constraints: list[dict[str, Any]] = [] for item in value[:32]: if not isinstance(item, Mapping): continue constraint = { key: text for key in ("id", "label", "description") if (text := _bounded_string(item.get(key), maximum=2_000 if key == "description" else 255)) } values = _bounded_string_list(item.get("values"), maximum_items=64, maximum_length=500) if values: constraint["values"] = values if constraint.get("label") and constraint.get("description"): constraints.append(constraint) return constraints def _bounded_string(value: object, *, maximum: int) -> str: if not isinstance(value, str): return "" clean = value.strip() return clean if len(clean) <= maximum else "" def _bounded_string_list(value: object, *, maximum_items: int, maximum_length: int) -> list[str]: if not isinstance(value, list): return [] return [ clean for item in value[:maximum_items] if (clean := _bounded_string(item, maximum=maximum_length)) ] def _user_link_allowed(link: DocumentationLink, *, visible_runtime_paths: frozenset[str]) -> bool: href = link.href.strip() if link.kind == "runtime": if not href.startswith("/") or href.startswith("//") or "\\" in href: return False try: parsed = urlsplit(href) except ValueError: return False return not parsed.scheme and not parsed.netloc and parsed.path in visible_runtime_paths if link.kind != "public": return False try: parsed = urlsplit(href) except ValueError: return False return parsed.scheme == "https" and bool(parsed.netloc) and not parsed.username and not parsed.password def _has_any_scope(principal: ApiPrincipal, scopes: tuple[str, ...]) -> bool: return any(has_scope(principal, scope) for scope in scopes) def _documentation_topic_kind(topic: DocumentationTopic) -> str: raw_kind = topic.metadata.get("kind") if isinstance(raw_kind, str): normalized = raw_kind.strip().lower().replace("_", "-") if normalized in TOPIC_KINDS: return normalized return "system" def _documentation_topic_anchor(source_module_id: str, topic_id: str) -> str: raw = f"docs-topic-{source_module_id}-{topic_id}" cleaned = "".join(char if char.isalnum() or char in {"_", "-"} else "-" for char in raw) while "--" in cleaned: cleaned = cleaned.replace("--", "-") return cleaned.strip("-").lower() def _documentation_topic_groups(documentation_layers: dict[str, list[dict[str, Any]]]) -> dict[str, list[dict[str, Any]]]: groups: dict[str, list[dict[str, Any]]] = {kind: [] for kind in TOPIC_KINDS} for topic in _all_documentation_topics(documentation_layers): kind = str(topic.get("kind") or "system") groups.setdefault(kind, []).append(topic) return groups def _all_documentation_topics(documentation_layers: dict[str, list[dict[str, Any]]]) -> list[dict[str, Any]]: topics: list[dict[str, Any]] = [] for layer in ("always", "configured", "available", "evidence"): topics.extend(documentation_layers.get(layer, ())) return topics def _translation_for_locale(topic: DocumentationTopic, locale: str) -> tuple[str, Mapping[str, str]]: translations = topic.translations or {} for candidate in _locale_candidates(locale): translation = translations.get(candidate) if translation: return candidate, translation return "source", {} def _structured_translation_locale(topic: DocumentationTopic, locale: str) -> str: for candidate in _locale_candidates(locale): if candidate in topic.structured_translations: return candidate return "source" def _locale_candidates(locale: str) -> tuple[str, ...]: normalized = _normalize_locale(locale) base = normalized.split("-", 1)[0] candidates = [normalized] if base and base != normalized: candidates.append(base) if "en" not in candidates: candidates.append("en") return tuple(candidates) def _preferred_locale(request: Request, explicit_locale: str | None) -> str: if explicit_locale: return _normalize_locale(explicit_locale) accept_language = request.headers.get("accept-language", "") first = accept_language.split(",", 1)[0].split(";", 1)[0].strip() return _normalize_locale(first or "en") def _normalize_locale(value: str) -> str: clean = "".join(char for char in value.strip().replace("_", "-") if char.isalnum() or char == "-") if not clean: return "en" parts = [part for part in clean.split("-") if part] if not parts: return "en" head = parts[0].lower() tail = [part.upper() if len(part) == 2 else part for part in parts[1:]] return "-".join([head, *tail])[:20] def _documentation_condition_payload(condition: DocumentationCondition) -> dict[str, list[str]]: return { "required_modules": list(condition.required_modules), "any_modules": list(condition.any_modules), "missing_modules": list(condition.missing_modules), "required_capabilities": list(condition.required_capabilities), "required_scopes": list(condition.required_scopes), "any_scopes": list(condition.any_scopes), "configuration_keys": list(condition.configuration_keys), } def _documentation_link_payload(link: DocumentationLink) -> dict[str, str]: return { "label": link.label, "href": link.href, "kind": link.kind, }