fix(files): align adaptive tasks with live access

docs/FILES_HANDBOOK.md

@@ -56,6 +56,14 @@ The main domain objects are:
 The Files page is available at `/files`. Actions appear only when the current
 principal has the required permission and resource access.
 
+The configured Help Center projects these sections as independently authorized
+tasks, so upload, ZIP import, organization, download, sharing, and deletion do
+not disappear merely because an unrelated permission is absent. It states the
+deployment's actual upload limits. External import appears as a task only when
+the actor has the required permissions and at least one actor-visible profile
+is eligible for the current fail-closed browse/import path; endpoint, path, and
+item policy are still enforced when the operation runs.
+
 ### Choose a space
 
 Every file user sees **My files**. Group spaces are added for active groups of
@@ -727,14 +735,14 @@ returning different content or credentials.
 
 | Area | Implemented now | Planned or explicitly outside the current boundary |
 | --- | --- | --- |
-| Managed storage | Local durable-root backend, fallback read roots, tenant blob deduplication, checksums | Operational S3 after pinned SDK transport; automated integrity/orphan reconciliation |
+| Managed storage | Local durable-root backend, fallback read roots, tenant blob deduplication, checksums | Operational S3 after pinned SDK transport ([#34](https://git.add-ideas.de/add-ideas/govoplan-files/issues/34)); automated integrity/orphan reconciliation ([#36](https://git.add-ideas.de/add-ideas/govoplan-files/issues/36)) |
 | Upload | Bounded direct upload, drag-and-drop UI, ZIP spool/extract limits, explicit conflicts | Malware scanning, quotas, type policy, resumable/chunked upload |
 | Organization | Folders, bulk rename preview/apply, move/copy, drag-and-drop, ZIP download, pattern resolution | General file-history UI and user-driven append-version/restore |
-| Sharing | User/group/tenant/campaign grant or permission update through API; campaign linkage display | Share revocation endpoint and complete general share-management UI |
-| Deletion/retention | Soft-delete assets/folders/spaces; immediate audited connector-secret scrubbing | File restore API, hard purge, retention policy, legal hold, blob GC |
+| Sharing | User/group/tenant/campaign grant or permission update through API; campaign linkage display | Share revocation/expiry and complete general share-management UI ([#37](https://git.add-ideas.de/add-ideas/govoplan-files/issues/37)) |
+| Deletion/retention | Soft-delete assets/folders/spaces; immediate audited connector-secret scrubbing | File restore API, hard purge, retention policy, legal hold, and blob GC ([#38](https://git.add-ideas.de/add-ideas/govoplan-files/issues/38)) |
 | Connector governance | Scoped profiles/credentials/policies, effective source explanation, separate credentials, linked user/group spaces | Provider-owned external secret lifecycle; API `secret_ref` remains rejected |
 | HTTP connectors | Pinned, bounded, no-redirect Seafile and WebDAV/Nextcloud browse/import/manual sync | Background/folder sync, remote mutation, long-running transfer workers |
-| SMB and S3 connectors | Provider descriptors and browse/import logic | Live access only after SDK connections plus DFS referrals/redirects are validated and pinned |
+| SMB and S3 connectors | Provider descriptors and browse/import logic | Live access only after SDK connections plus DFS referrals/redirects are validated and pinned ([SMB #35](https://git.add-ideas.de/add-ideas/govoplan-files/issues/35), [S3 #34](https://git.add-ideas.de/add-ideas/govoplan-files/issues/34)) |
 | Other providers | Reserved SharePoint/OneDrive keys and NFS/local descriptors | Graph/OAuth/provider paging, NFS deployment integration, DMS connectors |
 | Connector spaces | User/group link, browse, manual selected-file sync, edit/disable/delete | Background sync, remote writes/deletes, full conflict-reporting jobs |
 | Profile capabilities | Stored and displayed | Enforce capability flags as an independent operation gate |

src/govoplan_files/backend/documentation.py

@@ -0,0 +1,461 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.campaigns import (
+    CAPABILITY_CAMPAIGNS_ACCESS,
+    CampaignAccessProvider,
+)
+from govoplan_core.core.modules import (
+    DocumentationCondition,
+    DocumentationContext,
+    DocumentationLink,
+    DocumentationTopic,
+)
+from govoplan_files.backend.storage.archives import ZIP_UPLOAD_MAX_FILES
+from govoplan_files.backend.storage.access import user_group_ids
+from govoplan_files.backend.storage.connector_visibility import (
+    connector_profile_usable_for_import,
+    visible_connector_profiles_for_actor,
+)
+
+
+_DEFAULT_UPLOAD_MAX_BYTES = 50 * 1024 * 1024
+_DEFAULT_ZIP_MAX_BYTES = 250 * 1024 * 1024
+_FILES_READ_SCOPE = "files:file:read"
+_FILES_UPLOAD_SCOPE = "files:file:upload"
+
+
+def documentation_topics(
+    context: DocumentationContext,
+) -> tuple[DocumentationTopic, ...]:
+    if context.documentation_type != "user":
+        return ()
+
+    upload_limit = _configured_positive_int(
+        context.settings,
+        "file_upload_max_bytes",
+        default=_DEFAULT_UPLOAD_MAX_BYTES,
+    )
+    zip_limit = _configured_positive_int(
+        context.settings,
+        "file_upload_zip_max_bytes",
+        default=_DEFAULT_ZIP_MAX_BYTES,
+    )
+    topics: list[DocumentationTopic] = []
+    if upload_limit is not None:
+        topics.append(_upload_topic(upload_limit))
+    if upload_limit is not None and zip_limit is not None:
+        topics.append(_zip_topic(upload_limit, zip_limit))
+    topics.append(_connector_import_topic(context))
+    return tuple(topics)
+
+
+def _upload_topic(max_bytes: int) -> DocumentationTopic:
+    limit = _format_byte_limit(max_bytes)
+    return DocumentationTopic(
+        id="files.workflow.upload-managed-files",
+        title="Upload managed files",
+        summary=f"Upload files to a personal or accessible group space; each uploaded file may contain at most {limit}.",
+        body=(
+            f"The current deployment accepts at most {limit} for each ordinary upload. "
+            "A name conflict is never resolved silently: reject stops the upload, rename chooses a copy name, and overwrite retires the old asset before creating a new one."
+        ),
+        layer="configured",
+        documentation_types=("user",),
+        audience=("file_user", "file_manager", "process_participant"),
+        order=39,
+        conditions=(
+            DocumentationCondition(
+                required_modules=("files",),
+                required_scopes=(_FILES_READ_SCOPE, _FILES_UPLOAD_SCOPE),
+            ),
+        ),
+        links=(
+            DocumentationLink(label="Files", href="/files", kind="runtime"),
+            DocumentationLink(
+                label="Files handbook",
+                href="govoplan-files/docs/FILES_HANDBOOK.md",
+                kind="repository",
+            ),
+        ),
+        related_modules=("campaigns",),
+        unlocks=(
+            "Users and connected processes can place governed content in managed storage.",
+        ),
+        source_module_id="files",
+        metadata={
+            "kind": "workflow",
+            "route": "/files",
+            "screen": "Files",
+            "help_contexts": ["files.list"],
+            "prerequisites": [
+                "You may view and upload managed files.",
+                "The destination personal or group space grants this account write access; upload permission alone does not grant access to every space.",
+            ],
+            "steps": [
+                "Open Files and choose My files or an accessible group space.",
+                "Open the intended destination folder and choose Upload, or drag files into the file list.",
+                "For every name conflict, explicitly reject, rename, overwrite, or skip the affected item.",
+                "Wait for the upload to finish, then open the resulting file details.",
+            ],
+            "outcome": "Each accepted file is stored as a governed managed asset in the selected space.",
+            "verification": "Confirm the owner, logical path, size, checksum, and current version in Files.",
+            "constraints": [
+                {
+                    "id": "ordinary-upload-size",
+                    "label": "Maximum size per file",
+                    "description": f"The current safe upload limit is {limit} per file.",
+                    "values": [limit],
+                },
+            ],
+            "related_topic_ids": [
+                "files.workflow.upload-and-unpack-zip",
+                "files.workflow.organize-managed-files",
+                "files.workflow.find-and-download-files",
+            ],
+        },
+    )
+
+
+def _zip_topic(max_file_bytes: int, max_zip_bytes: int) -> DocumentationTopic:
+    member_limit = _format_byte_limit(max_file_bytes)
+    archive_limit = _format_byte_limit(max_zip_bytes)
+    return DocumentationTopic(
+        id="files.workflow.upload-and-unpack-zip",
+        title="Upload and unpack a ZIP archive",
+        summary=(
+            f"Safely unpack up to {ZIP_UPLOAD_MAX_FILES:,} files from a ZIP whose request and extracted total are each limited to {archive_limit}."
+        ),
+        body=(
+            f"The current deployment limits the ZIP request and actual extracted total to {archive_limit}, each member to {member_limit}, "
+            f"and the archive to {ZIP_UPLOAD_MAX_FILES:,} non-directory members. Encrypted archives and unsafe member paths are rejected. "
+            "Actual extracted bytes are counted instead of trusting ZIP headers."
+        ),
+        layer="configured",
+        documentation_types=("user",),
+        audience=("file_user", "file_manager", "process_participant"),
+        order=40,
+        conditions=(
+            DocumentationCondition(
+                required_modules=("files",),
+                required_scopes=(_FILES_READ_SCOPE, _FILES_UPLOAD_SCOPE),
+            ),
+        ),
+        links=(
+            DocumentationLink(label="Files", href="/files", kind="runtime"),
+            DocumentationLink(
+                label="Files handbook",
+                href="govoplan-files/docs/FILES_HANDBOOK.md",
+                kind="repository",
+            ),
+        ),
+        unlocks=(
+            "A bounded archive can become governed managed files without trusting archive paths or size declarations.",
+        ),
+        source_module_id="files",
+        metadata={
+            "kind": "workflow",
+            "route": "/files",
+            "screen": "Files",
+            "help_contexts": ["files.list"],
+            "prerequisites": [
+                "You may view and upload managed files.",
+                "The archive is not encrypted and fits the current configured limits.",
+                "The destination personal or group space grants this account write access.",
+            ],
+            "steps": [
+                "Open Files and choose the managed destination space and folder.",
+                "Enable Unpack ZIP uploads, then choose or drag the ZIP archive.",
+                "Resolve every destination conflict explicitly.",
+                "Wait for extraction and finalization to finish before leaving the page.",
+            ],
+            "outcome": "Accepted archive members are stored as separate governed managed assets below the selected folder.",
+            "verification": "Confirm the expected member paths and inspect representative file sizes, checksums, and versions.",
+            "constraints": [
+                {
+                    "id": "zip-request-and-total",
+                    "label": "Maximum ZIP request and extracted total",
+                    "description": f"Both the compressed request and the actual extracted total are limited to {archive_limit}.",
+                    "values": [archive_limit],
+                },
+                {
+                    "id": "zip-member-size",
+                    "label": "Maximum extracted member size",
+                    "description": f"Each extracted file is limited to {member_limit}.",
+                    "values": [member_limit],
+                },
+                {
+                    "id": "zip-member-count",
+                    "label": "Maximum file count",
+                    "description": f"A ZIP may contain at most {ZIP_UPLOAD_MAX_FILES:,} non-directory members.",
+                    "values": [f"{ZIP_UPLOAD_MAX_FILES:,} files"],
+                },
+            ],
+            "related_topic_ids": [
+                "files.workflow.upload-managed-files",
+                "files.workflow.organize-managed-files",
+                "files.workflow.find-and-download-files",
+            ],
+        },
+    )
+
+
+def _connector_import_topic(context: DocumentationContext) -> DocumentationTopic:
+    principal = context.principal
+    if not _has_all_scopes(principal, (_FILES_READ_SCOPE, _FILES_UPLOAD_SCOPE)):
+        return _connector_import_limitation(
+            "Connector import is not available to this account because both permission to view Files and permission to upload managed files are required."
+        )
+
+    tenant_id = _safe_text_attribute(principal, "tenant_id")
+    user_id = _safe_text_attribute(getattr(principal, "user", None), "id")
+    session = context.session
+    if not tenant_id or not user_id or not isinstance(session, Session):
+        return _connector_import_limitation(
+            "Connector import availability could not be safely evaluated for this request. Try again, or ask a Files administrator to verify an actor-visible connection."
+        )
+
+    try:
+        member_group_ids = _actor_group_ids(
+            session,
+            principal=principal,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            include_admin_groups=False,
+        )
+        connector_group_ids = (
+            _actor_group_ids(
+                session,
+                principal=principal,
+                tenant_id=tenant_id,
+                user_id=user_id,
+                include_admin_groups=True,
+            )
+            if _has_all_scopes(principal, ("files:file:admin",))
+            else member_group_ids
+        )
+        profiles = visible_connector_profiles_for_actor(
+            session,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            group_ids=connector_group_ids,
+            settings=context.settings,
+            campaign_visible=_campaign_visibility(
+                context,
+                session=session,
+                tenant_id=tenant_id,
+                user_id=user_id,
+                group_ids=member_group_ids,
+            ),
+            include_effective_policy=True,
+        )
+        usable_profiles = tuple(
+            profile
+            for profile in profiles
+            if connector_profile_usable_for_import(profile)
+        )
+    except Exception:
+        return _connector_import_limitation(
+            "Connector import availability could not be safely evaluated for this request. Try again, or ask a Files administrator to verify an actor-visible connection."
+        )
+    if not usable_profiles:
+        return _connector_import_limitation(
+            "No connection visible to this account is currently eligible to offer an enabled, credential-ready, policy-allowed browse/import path through a pinning-safe provider. Ask a Files administrator to configure or authorize one."
+        )
+
+    return DocumentationTopic(
+        id="files.workflow.import-managed-snapshot",
+        title="Import an external file as a governed snapshot",
+        summary="Browse an authorized connection read-only and import one selected file into managed storage as a frozen, traceable snapshot.",
+        body=(
+            "At least one connection visible to this account is currently eligible to offer the governed browse/import path. "
+            "The selected remote path and item are re-authorized when used, and browse, import, and sync never mutate the remote source. "
+            "The managed snapshot stays unchanged until an explicit manual sync."
+        ),
+        layer="configured",
+        documentation_types=("user",),
+        audience=("file_user", "campaign_manager", "report_author"),
+        order=41,
+        conditions=(
+            DocumentationCondition(
+                required_modules=("files",),
+                required_scopes=(_FILES_READ_SCOPE, _FILES_UPLOAD_SCOPE),
+            ),
+        ),
+        links=(
+            DocumentationLink(label="Files", href="/files", kind="runtime"),
+            DocumentationLink(
+                label="Files handbook",
+                href="govoplan-files/docs/FILES_HANDBOOK.md",
+                kind="repository",
+            ),
+        ),
+        related_modules=("campaigns",),
+        unlocks=(
+            "Campaigns, reports, and workflows can consume a managed snapshot with stable source evidence.",
+        ),
+        source_module_id="files",
+        metadata={
+            "kind": "workflow",
+            "route": "/files",
+            "screen": "Files",
+            "help_contexts": ["files.list", "files.connector-import"],
+            "prerequisites": [
+                "You may view and upload managed files.",
+                "At least one enabled, credential-ready, policy-allowed connection using a pinning-safe provider is visible to this account.",
+                "The selected remote path and item must pass their operation-time policy checks.",
+                "The managed destination space grants this account write access.",
+            ],
+            "steps": [
+                "Open Files and choose a managed destination space.",
+                "Choose Sync from connection, select an available connection, and browse to the permitted remote file.",
+                "Import the selected file and resolve any destination conflict explicitly.",
+                "Review the managed file's source and current-version details before using it in another task.",
+            ],
+            "current_configuration": [
+                "At least one actor-visible connection is eligible to offer a safe browse/import operation; the selected endpoint, path, and item are still checked when used.",
+                "Every selected remote path and item is re-authorized at operation time.",
+            ],
+            "outcome": "The external content is a tenant-managed snapshot with a checksum, exact version, and recorded source context.",
+            "verification": "Reopen the managed file and confirm its recorded source context, source revision when available, checksum, and current version.",
+            "related_topic_ids": [
+                "files.governed-connectors-and-provenance",
+                "files.reference.integrity-recovery-and-fail-closed-transports",
+                "files.reference.snapshot-provenance-and-capabilities",
+            ],
+        },
+    )
+
+
+def _connector_import_limitation(message: str) -> DocumentationTopic:
+    return DocumentationTopic(
+        id="files.connector-import-unavailable",
+        title="External file import is not currently available",
+        summary=message,
+        body=(
+            f"{message} Files never exposes connection endpoints, storage paths, credential references, or raw connector policies in user documentation."
+        ),
+        layer="available",
+        documentation_types=("user",),
+        order=41,
+        conditions=(
+            DocumentationCondition(
+                required_modules=("files",),
+                any_scopes=(_FILES_READ_SCOPE, _FILES_UPLOAD_SCOPE),
+            ),
+        ),
+        links=(DocumentationLink(label="Files", href="/files", kind="runtime"),),
+        source_module_id="files",
+        metadata={
+            "kind": "reference",
+            "screen": "Files",
+            "help_contexts": ["files.list", "files.connector-import"],
+            "limitations": [message],
+        },
+    )
+
+
+def _configured_positive_int(
+    settings: object | None, name: str, *, default: int
+) -> int | None:
+    raw_value = getattr(settings, name, default) if settings is not None else default
+    try:
+        value = int(raw_value)
+    except (TypeError, ValueError):
+        return None
+    return value if value > 0 else None
+
+
+def _format_byte_limit(value: int) -> str:
+    units = ((1024**3, "GiB"), (1024**2, "MiB"), (1024, "KiB"))
+    for divisor, label in units:
+        if value % divisor == 0:
+            return f"{value // divisor:,} {label} ({value:,} bytes)"
+    return f"{value:,} bytes"
+
+
+def _has_all_scopes(principal: object | None, scopes: tuple[str, ...]) -> bool:
+    checker = getattr(principal, "has", None)
+    if callable(checker):
+        try:
+            return all(bool(checker(scope)) for scope in scopes)
+        except Exception:
+            return False
+    granted = {str(scope) for scope in getattr(principal, "scopes", ())}
+    return all(scope in granted for scope in scopes)
+
+
+def _safe_text_attribute(value: object | None, name: str) -> str:
+    try:
+        result = getattr(value, name, "")
+    except Exception:
+        return ""
+    return str(result or "")
+
+
+def _principal_group_ids(principal: object) -> tuple[str, ...]:
+    try:
+        values = getattr(principal, "group_ids", ())
+    except Exception:
+        return ()
+    return tuple(str(value) for value in values if str(value))
+
+
+def _actor_group_ids(
+    session: Session,
+    *,
+    principal: object,
+    tenant_id: str,
+    user_id: str,
+    include_admin_groups: bool,
+) -> tuple[str, ...]:
+    try:
+        values = user_group_ids(
+            session,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            include_admin_groups=include_admin_groups,
+        )
+    except Exception:
+        return _principal_group_ids(principal)
+    return tuple(str(value) for value in values if str(value))
+
+
+def _campaign_visibility(
+    context: DocumentationContext,
+    *,
+    session: Session,
+    tenant_id: str,
+    user_id: str,
+    group_ids: tuple[str, ...],
+):
+    principal = context.principal
+    registry = context.registry
+    if not _has_all_scopes(principal, ("campaigns:campaign:read",)):
+        return None
+    try:
+        if not registry.has_capability(CAPABILITY_CAMPAIGNS_ACCESS):
+            return None
+        capability = registry.require_capability(CAPABILITY_CAMPAIGNS_ACCESS)
+    except Exception:
+        return None
+    if not isinstance(capability, CampaignAccessProvider):
+        return None
+
+    def campaign_visible(campaign_id: str) -> bool:
+        try:
+            return capability.campaign_exists(
+                session, tenant_id=tenant_id, campaign_id=campaign_id
+            ) and capability.can_read_campaign(
+                session,
+                tenant_id=tenant_id,
+                campaign_id=campaign_id,
+                user_id=user_id,
+                group_ids=group_ids,
+                tenant_admin=_has_all_scopes(principal, ("tenant:*",)),
+            )
+        except Exception:
+            return False
+
+    return campaign_visible

src/govoplan_files/backend/manifest.py

@@ -25,6 +25,7 @@ from govoplan_core.core.modules import (
 from govoplan_core.db.base import Base
 from govoplan_files.backend.change_tracking import register_files_change_tracking
 from govoplan_files.backend.db import models as file_models  # noqa: F401 - populate Files ORM metadata
+from govoplan_files.backend.documentation import documentation_topics
 
 register_files_change_tracking()
 
@@ -97,7 +98,7 @@ PERMISSIONS = (
     _permission("files:file:download", "Download files", "Download managed files and generated archives."),
     _permission("files:file:upload", "Upload files", "Upload new managed file versions."),
     _permission("files:file:organize", "Organize files", "Create folders, rename, move or copy managed files."),
-    _permission("files:file:share", "Share files", "Grant or revoke managed file shares."),
+    _permission("files:file:share", "Share files", "Grant or update managed file shares; revocation is not available yet."),
     _permission("files:file:delete", "Delete files", "Delete or hide managed files and folders where policy allows it."),
     _permission("files:file:admin", "Administer file spaces", "Administer all file spaces in the tenant."),
 )
@@ -195,58 +196,199 @@ manifest = ModuleManifest(
     ),
     documentation=(
         DocumentationTopic(
-            id="files.workflow.import-managed-snapshot",
-            title="Import an external file as a governed snapshot",
-            summary="Browse an authorized connection read-only, import one selected file into managed storage, and use the frozen version in another GovOPlaN task.",
+            id="files.workflow.organize-managed-files",
+            title="Organize managed files and folders",
+            summary="Create folders and rename, move, or copy accessible managed content with explicit conflict handling.",
             body=(
-                "Choose a visible file connection in Files, browse only the permitted remote path, and import the selected content into your personal or group space. "
-                "The managed copy records source provenance and remains unchanged until an explicit manual sync. Browse, import, and sync never mutate the remote source."
+                "Organization stays inside governed personal or group spaces. Moves preserve the asset identity, while copies create new assets and versions that reuse immutable blob bytes. "
+                "Every target conflict must be rejected, renamed, overwritten, or skipped explicitly."
             ),
             layer="configured",
             documentation_types=("user",),
-            audience=("file_user", "campaign_manager", "report_author"),
-            order=40,
+            audience=("file_user", "file_manager", "process_participant"),
+            order=42,
             conditions=(
                 DocumentationCondition(
                     required_modules=("files",),
-                    any_scopes=("files:file:read", "files:file:upload"),
+                    required_scopes=("files:file:read", "files:file:organize"),
                 ),
             ),
             links=(
                 DocumentationLink(label="Files", href="/files", kind="runtime"),
-                DocumentationLink(label="Visible connector profiles", href="/api/v1/files/connectors/profiles", kind="api"),
-                DocumentationLink(
-                    label="Connector import API",
-                    href="/api/v1/files/connectors/profiles/{profile_id}/import",
-                    kind="api",
-                ),
+                DocumentationLink(label="Move or copy API", href="/api/v1/files/transfer", kind="api"),
                 DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
             ),
-            related_modules=("campaigns",),
-            unlocks=("Campaigns, reports, and workflows can consume a managed snapshot with stable source evidence.",),
+            unlocks=("Managed content can be placed at stable logical paths without bypassing space access.",),
             metadata={
                 "kind": "workflow",
                 "route": "/files",
                 "screen": "Files",
+                "help_contexts": ["files.list"],
                 "prerequisites": [
-                    "Files is installed and you may read and upload files.",
-                    "An administrator has configured a connector profile visible in your current user, group, tenant, or campaign scope.",
+                    "You may view and organize managed files.",
+                    "You have write or owner access to every source item and destination space used by the operation; the global organize permission alone does not grant resource access.",
                 ],
                 "steps": [
-                    "Open Files and choose a managed destination space.",
-                    "Choose Sync from connection, select a visible profile, and browse to the permitted remote file.",
-                    "Import or sync the selected file and resolve any destination conflict explicitly.",
-                    "Review the managed file's source and current-version details before using it in another task.",
+                    "Open Files and select the personal or group space to organize.",
+                    "Create the required destination folders or select the files and folders to rename, move, or copy.",
+                    "Choose the destination and resolve each target conflict explicitly.",
+                    "Apply the operation and reopen the destination folder.",
                 ],
-                "outcome": "The external content is a tenant-managed snapshot with a checksum, exact version, and recorded source context.",
-                "verification": "Reopen the managed file and confirm its connector/provider, remote identity or path, source revision when available, checksum, and current version.",
+                "outcome": "The selected content has the intended governed owner and logical path.",
+                "verification": "Confirm each resulting path and owner; for a move, also confirm the old path is gone, and for a copy, confirm the source remains.",
                 "related_topic_ids": [
-                    "files.governed-connectors-and-provenance",
-                    "files.reference.integrity-recovery-and-fail-closed-transports",
+                    "files.workflow.upload-managed-files",
+                    "files.workflow.find-and-download-files",
+                    "files.workflow.delete-managed-files",
+                ],
+            },
+        ),
+        DocumentationTopic(
+            id="files.workflow.find-and-download-files",
+            title="Find and download managed files",
+            summary="Search accessible managed content and download a current file version or a ZIP archive of a selection.",
+            body=(
+                "Files can be sorted and searched by logical path or name pattern. A download always uses the accessible current version; a multi-file selection can be generated as a temporary ZIP archive. "
+                "There is no dedicated content-preview service yet."
+            ),
+            layer="configured",
+            documentation_types=("user",),
+            audience=("file_user", "file_manager", "process_participant"),
+            order=43,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("files",),
+                    required_scopes=("files:file:read", "files:file:download"),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
+            ),
+            unlocks=("Authorized users can retrieve governed current versions without gaining organization or sharing authority.",),
+            metadata={
+                "kind": "workflow",
+                "route": "/files",
+                "screen": "Files",
+                "help_contexts": ["files.list"],
+                "prerequisites": ["You may view and download managed files in the relevant space."],
+                "steps": [
+                    "Open Files and select the relevant personal or group space.",
+                    "Navigate folders, sort the list, or use a path/name pattern to find the intended content.",
+                    "Review the displayed owner, path, size, checksum, and version details.",
+                    "Download one current file version, or select several files and choose Download ZIP.",
+                ],
+                "outcome": "The authorized current file bytes or generated archive are downloaded to the local device.",
+                "verification": "Confirm the downloaded names and, where integrity matters, compare the file bytes with the displayed checksum.",
+                "related_topic_ids": [
+                    "files.workflow.organize-managed-files",
                     "files.reference.snapshot-provenance-and-capabilities",
                 ],
             },
         ),
+        DocumentationTopic(
+            id="files.workflow.share-managed-files",
+            title="Grant or update access to managed files",
+            summary="Grant or update read, write, or manage access through a supporting workflow or API client without changing ownership.",
+            body=(
+                "The Files service can grant or update a share for a user, group, tenant, or campaign. The Files page does not yet provide a general share editor, and the API has no share-revocation route. "
+                "Do not use this task when access must later be revoked until that explicit capability is implemented."
+            ),
+            layer="available",
+            documentation_types=("user",),
+            audience=("file_manager", "process_participant"),
+            order=44,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("files",),
+                    required_scopes=("files:file:read", "files:file:share"),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(label="Grant or update a share", href="/api/v1/files/{file_id}/shares", kind="api"),
+                DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
+            ),
+            related_modules=("campaigns",),
+            unlocks=("A supporting process can grant governed file access without changing file ownership.",),
+            metadata={
+                "kind": "workflow",
+                "route": "/files",
+                "screen": "Files",
+                "help_contexts": ["files.list"],
+                "prerequisites": [
+                    "You may view and share the managed file and have write/manage access to that specific asset; the global share permission alone does not grant resource access.",
+                    "A supporting workflow or API client is available; the Files page has no general share editor yet.",
+                    "The process does not require share revocation, because no revocation route exists yet.",
+                ],
+                "steps": [
+                    "Open Files and verify the file owner, logical path, current version, and checksum.",
+                    "Through the supporting workflow, identify the intended user, group, tenant, or campaign and choose read, write, or manage access.",
+                    "Grant or update the share without changing file ownership.",
+                    "Have the intended recipient verify access and an unrelated account verify denial.",
+                ],
+                "limitations": [
+                    "The Files page has no general share editor.",
+                    "Shares can be granted or updated, but not revoked through the current API.",
+                ],
+                "outcome": "The requested access is granted or updated while the managed asset keeps its owner.",
+                "verification": "Test one intended and one denied access path. Do not claim that the share can later be revoked.",
+                "related_topic_ids": [
+                    "files.workflow.find-and-download-files",
+                    "files.assurance.process-and-release-readiness",
+                ],
+            },
+        ),
+        DocumentationTopic(
+            id="files.workflow.delete-managed-files",
+            title="Delete managed files and folders",
+            summary="Soft-delete accessible managed files or a folder tree where current policy allows it.",
+            body=(
+                "Deletion hides the selected managed assets rather than hard-purging their stored evidence. Folder deletion is recursive by default and includes child folders and files; a non-recursive request fails for a non-empty folder. "
+                "There is no self-service restore or hard-purge workflow today."
+            ),
+            layer="configured",
+            documentation_types=("user",),
+            audience=("file_user", "file_manager", "process_participant"),
+            order=45,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("files",),
+                    required_scopes=("files:file:read", "files:file:delete"),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
+            ),
+            unlocks=("Authorized users can remove obsolete content from active file views while preserving the current soft-delete boundary.",),
+            metadata={
+                "kind": "workflow",
+                "route": "/files",
+                "screen": "Files",
+                "help_contexts": ["files.list"],
+                "prerequisites": [
+                    "You may view and delete the selected managed content and have write or owner access to every affected asset or folder.",
+                    "You have reviewed the complete folder tree when deleting recursively.",
+                ],
+                "steps": [
+                    "Open Files and select the files or folder to delete.",
+                    "Review the selection and, for a folder, all content below it.",
+                    "Confirm the delete action.",
+                    "Refresh or reopen the space and verify that the selected paths are no longer active.",
+                ],
+                "limitations": [
+                    "Deletion is soft deletion, not a hard purge.",
+                    "There is no self-service restore or hard-purge workflow.",
+                ],
+                "outcome": "The selected content is hidden from active Files views under the current soft-delete model.",
+                "verification": "Confirm the deleted paths no longer appear in the active space; do not treat the action as physical erasure.",
+                "related_topic_ids": [
+                    "files.workflow.organize-managed-files",
+                    "files.assurance.process-and-release-readiness",
+                ],
+            },
+        ),
         DocumentationTopic(
             id="files.governed-connectors-and-provenance",
             title="Govern file connections and credential deletion",
@@ -258,7 +400,7 @@ manifest = ModuleManifest(
             layer="configured",
             documentation_types=("admin",),
             audience=("file_admin", "tenant_admin", "system_admin", "security_auditor"),
-            order=41,
+            order=50,
             conditions=(
                 DocumentationCondition(
                     required_modules=("files",),
@@ -315,7 +457,7 @@ manifest = ModuleManifest(
             layer="configured",
             documentation_types=("admin",),
             audience=("operator", "system_admin", "security_auditor"),
-            order=42,
+            order=51,
             conditions=(
                 DocumentationCondition(
                     required_modules=("files",),
@@ -369,7 +511,7 @@ manifest = ModuleManifest(
             layer="available",
             documentation_types=("admin", "user"),
             audience=("module_integrator", "file_admin", "campaign_admin", "process_designer"),
-            order=43,
+            order=52,
             conditions=(
                 DocumentationCondition(
                     required_modules=("files",),
@@ -397,7 +539,83 @@ manifest = ModuleManifest(
                 ],
             },
         ),
+        DocumentationTopic(
+            id="files.assurance.process-and-release-readiness",
+            title="Assure a Files-backed process and release",
+            summary="Check a process against the implemented Files boundary, exercise permitted and denied paths, and retain evidence before approving a release or operational use.",
+            body=(
+                "A process owner must distinguish implemented controls from planned capabilities before relying on Files. A release is not ready until all package and manifest versions align and representative authorization, upload limits, conflict handling, download, deletion, connector, and recovery paths have been exercised. "
+                "Current limitations include no general share-management UI or share revocation, no self-service restore or hard purge, no enforced retention or legal hold, and no dedicated canonical audit event for every ordinary Files mutation. Record these limitations in the process assessment instead of treating soft deletion or change-sequence entries as stronger evidence."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("process_owner", "release_manager", "file_admin", "operator", "security_auditor"),
+            order=53,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("files",),
+                    any_scopes=(
+                        "files:file:read",
+                        "files:file:admin",
+                        "admin:settings:read",
+                        "system:settings:read",
+                        "system:audit:read",
+                    ),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(label="Files API", href="/api/v1/files", kind="api"),
+                DocumentationLink(label="Connector provider status", href="/api/v1/files/connectors/providers", kind="api"),
+                DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
+            ),
+            related_modules=("campaigns", "audit", "ops"),
+            unlocks=("Process owners and release managers can approve Files use against explicit controls, evidence, and known gaps.",),
+            configuration_keys=(
+                "FILE_STORAGE_BACKEND",
+                "FILE_STORAGE_LOCAL_ROOT",
+                "FILE_UPLOAD_MAX_BYTES",
+                "FILE_UPLOAD_ZIP_MAX_BYTES",
+                "MASTER_KEY_B64",
+                "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS",
+            ),
+            metadata={
+                "kind": "workflow",
+                "route": "/files",
+                "screen": "Files process and release assurance",
+                "help_contexts": ["files.list"],
+                "prerequisites": [
+                    "A named process owner has defined the intended users, data classification, retention expectations, and integrations.",
+                    "A candidate release is installed on a clean database and an upgrade copy with aligned Python, root package, WebUI package, and module-manifest versions.",
+                    "Representative permitted and denied accounts, bounded test files, and a coordinated database/blob/key recovery set are available.",
+                ],
+                "steps": [
+                    "Compare the process requirements with the handbook's implemented/planned boundary and record every unsupported requirement or compensating control.",
+                    "Confirm version alignment, apply migrations on clean and upgrade databases, and run the repository plus meta-repository security and static-analysis gates.",
+                    "Exercise allowed and denied personal, group, and share access with representative accounts.",
+                    "Exercise bounded upload and ZIP handling, every required conflict strategy, organization, download, and soft deletion.",
+                    "Where connectors are configured, verify policy explanation and one pinned HTTP provider; verify live S3 and SMB access still fails closed.",
+                    "Restore a coordinated database/blob/key backup and compare representative downloaded bytes with their recorded SHA-256 checksums.",
+                    "Record the tested versions, results, known limitations, evidence locations, residual risks, owner, and approval decision.",
+                ],
+                "outcome": "The process or release has an explicit approval record tied to aligned versions, representative evidence, known limitations, and owned residual risks.",
+                "verification": "A reviewer can reproduce the recorded allow/deny, integrity, connector, and recovery checks and can trace each unmet requirement to a documented limitation or accepted compensating control.",
+                "related_topic_ids": [
+                    "files.workflow.upload-managed-files",
+                    "files.workflow.upload-and-unpack-zip",
+                    "files.workflow.organize-managed-files",
+                    "files.workflow.find-and-download-files",
+                    "files.workflow.share-managed-files",
+                    "files.workflow.delete-managed-files",
+                    "files.workflow.import-managed-snapshot",
+                    "files.governed-connectors-and-provenance",
+                    "files.reference.integrity-recovery-and-fail-closed-transports",
+                    "files.reference.snapshot-provenance-and-capabilities",
+                ],
+            },
+        ),
     ),
+    documentation_providers=(documentation_topics,),
     migration_spec=MigrationSpec(
         module_id="files",
         metadata=Base.metadata,

src/govoplan_files/backend/router.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import json
 import os
 import tempfile
-from dataclasses import replace
 from datetime import datetime
 from io import BytesIO
 from typing import Any, Literal
@@ -131,11 +130,11 @@ from govoplan_files.backend.storage.connector_profile_store import (
     connector_profile_from_row,
     create_connector_profile_row,
     get_connector_profile_row,
-    list_database_connector_profiles,
     update_connector_profile_row,
 )
-from govoplan_files.backend.storage.connector_profiles import ConnectorProfile, connector_profiles_from_settings
+from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
 from govoplan_files.backend.storage.connector_providers import connector_provider_descriptors
+from govoplan_files.backend.storage.connector_visibility import visible_connector_profiles_for_actor
 from govoplan_files.backend.storage.connector_spaces import (
     connector_space_owner_id,
     create_connector_space,
@@ -455,35 +454,6 @@ def _enforce_connector_policy(source_provenance_json: str | None, connector_poli
     ensure_connector_policy_allows(request, sources)
 
 
-def _connector_profile_visible(
-    session: Session,
-    principal: ApiPrincipal,
-    profile: ConnectorProfile,
-    *,
-    campaign_id: str | None,
-    group_ids: set[str],
-) -> bool:
-    if profile.scope_type == "system":
-        return True
-    if profile.scope_type == "tenant":
-        return profile.scope_id == principal.tenant_id
-    if profile.scope_type == "user":
-        return profile.scope_id == principal.user.id
-    if profile.scope_type == "group":
-        return bool(profile.scope_id and profile.scope_id in group_ids)
-    if profile.scope_type == "campaign":
-        if not profile.scope_id:
-            return False
-        if campaign_id and profile.scope_id != campaign_id:
-            return False
-        try:
-            _ensure_campaign_file_access(session, principal, profile.scope_id)
-        except HTTPException:
-            return False
-        return True
-    return False
-
-
 def _visible_connector_profiles(
     session: Session,
     principal: ApiPrincipal,
@@ -494,30 +464,33 @@ def _visible_connector_profiles(
     include_admin_scopes: bool = False,
     include_effective_policy: bool = True,
 ) -> list[ConnectorProfile]:
-    provider_norm = provider.strip().casefold() if provider else None
-    group_ids = set(user_group_ids(session, tenant_id=principal.tenant_id, user_id=principal.user.id, include_admin_groups=_is_admin(principal)))
-    database_profiles = list_database_connector_profiles(
+    group_ids = user_group_ids(
         session,
         tenant_id=principal.tenant_id,
-        include_disabled=include_disabled,
+        user_id=principal.user.id,
+        include_admin_groups=_is_admin(principal),
+    )
+
+    def campaign_visible(profile_campaign_id: str) -> bool:
+        try:
+            _ensure_campaign_file_access(session, principal, profile_campaign_id)
+        except HTTPException:
+            return False
+        return True
+
+    return visible_connector_profiles_for_actor(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=principal.user.id,
+        group_ids=group_ids,
+        settings=settings,
+        provider=provider,
+        campaign_id=campaign_id,
+        campaign_visible=campaign_visible,
+        include_disabled=include_disabled,
+        include_admin_scopes=include_admin_scopes,
+        include_effective_policy=include_effective_policy,
     )
-    env_profiles = connector_profiles_from_settings(settings)
-    profiles_by_id: dict[str, ConnectorProfile] = {}
-    for profile in [*database_profiles, *env_profiles]:
-        if profile.id not in profiles_by_id:
-            profiles_by_id[profile.id] = profile
-    profiles = list(profiles_by_id.values())
-    visible: list[ConnectorProfile] = []
-    for profile in profiles:
-        if not (profile.enabled or include_disabled):
-            continue
-        if provider_norm is not None and profile.provider != provider_norm:
-            continue
-        admin_scope_visible = include_admin_scopes and profile.scope_type in {"user", "group", "campaign"}
-        if not admin_scope_visible and not _connector_profile_visible(session, principal, profile, campaign_id=campaign_id, group_ids=group_ids):
-            continue
-        visible.append(_with_effective_connector_policy(session, principal, profile) if include_effective_policy else profile)
-    return visible
 
 
 def _webdav_discovery_candidates(payload: FileConnectorDiscoveryRequest) -> list[str]:
@@ -641,18 +614,6 @@ def _visible_connector_profile(
     raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connector profile not found")
 
 
-def _with_effective_connector_policy(session: Session, principal: ApiPrincipal, profile: ConnectorProfile) -> ConnectorProfile:
-    sources = effective_connector_policy_sources(
-        session,
-        tenant_id=principal.tenant_id,
-        scope_type=profile.scope_type,
-        scope_id=profile.scope_id,
-    )
-    if not sources:
-        return profile
-    return replace(profile, policy_sources=tuple([*sources, *profile.policy_sources]))
-
-
 def _require_connector_profile_write(principal: ApiPrincipal, scope_type: str) -> None:
     clean_scope = scope_type.strip().casefold()
     if clean_scope == "system":

src/govoplan_files/backend/storage/archives.py

@@ -17,6 +17,7 @@ from govoplan_files.backend.storage.paths import filename_from_path, normalize_f
 
 
 _ZIP_READ_CHUNK_SIZE = 1024 * 1024
+ZIP_UPLOAD_MAX_FILES = 1000
 
 
 def _read_zip_member(
@@ -76,7 +77,7 @@ def extract_zip_upload(
     conflict_resolutions: Iterable[FileConflictResolution] | None = None,
     metadata: dict[str, Any] | None = None,
     is_admin: bool = False,
-    max_files: int = 1000,
+    max_files: int = ZIP_UPLOAD_MAX_FILES,
     max_file_bytes: int = 50 * 1024 * 1024,
     max_total_bytes: int = 250 * 1024 * 1024,
 ) -> list[UploadedStoredFile]:

src/govoplan_files/backend/storage/connector_profile_store.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from typing import Any
 
 from sqlalchemy.orm import Session
@@ -20,7 +20,26 @@ def list_database_connector_profiles(
     *,
     tenant_id: str,
     include_disabled: bool = False,
+    row_visible: Callable[[FileConnectorProfile], bool] | None = None,
 ) -> list[ConnectorProfile]:
+    profiles, _profile_ids = select_database_connector_profiles(
+        session,
+        tenant_id=tenant_id,
+        include_disabled=include_disabled,
+        row_visible=row_visible,
+    )
+    return profiles
+
+
+def select_database_connector_profiles(
+    session: Session,
+    *,
+    tenant_id: str,
+    include_disabled: bool = False,
+    row_visible: Callable[[FileConnectorProfile], bool] | None = None,
+) -> tuple[list[ConnectorProfile], set[str]]:
+    """Return visible profiles and every database id that shadows settings."""
+
     query = session.query(FileConnectorProfile).filter(
         (FileConnectorProfile.scope_type == "system")
         | (FileConnectorProfile.tenant_id == tenant_id)
@@ -28,9 +47,15 @@ def list_database_connector_profiles(
     if not include_disabled:
         query = query.filter(FileConnectorProfile.enabled.is_(True))
     rows = query.order_by(FileConnectorProfile.scope_type.asc(), FileConnectorProfile.label.asc()).all()
+    profile_ids = {row.id for row in rows}
+    if row_visible is not None:
+        rows = [row for row in rows if row_visible(row)]
     credential_ids = {_clean(row.credential_profile_id) for row in rows if _clean(row.credential_profile_id)}
     credentials = credential_rows_by_id(session, tenant_id=tenant_id, credential_ids={item for item in credential_ids if item}, include_disabled=include_disabled)
-    return [connector_profile_from_row(row, credential_row=credentials.get(row.credential_profile_id or "")) for row in rows]
+    return (
+        [connector_profile_from_row(row, credential_row=credentials.get(row.credential_profile_id or "")) for row in rows],
+        profile_ids,
+    )
 
 
 def list_connector_profile_rows(
@@ -188,18 +213,80 @@ def update_connector_profile_row(
     clear_password: bool = False,
     clear_token: bool = False,
 ) -> FileConnectorProfile:
+    _validate_profile_update_references(
+        row,
+        password_env=password_env,
+        token_env=token_env,
+        secret_ref=secret_ref,
+        metadata=metadata,
+    )
+    _update_profile_connection_fields(
+        row,
+        label=label,
+        provider=provider,
+        endpoint_url=endpoint_url,
+        base_path=base_path,
+        enabled=enabled,
+        credential_profile_id=credential_profile_id,
+        credential_mode=credential_mode,
+    )
+    _update_profile_credential_fields(
+        row,
+        username=username,
+        password=password,
+        token=token,
+        password_env=password_env,
+        token_env=token_env,
+        secret_ref=secret_ref,
+        clear_password=clear_password,
+        clear_token=clear_token,
+    )
+    _update_profile_governance_fields(
+        row,
+        capabilities=capabilities,
+        policy=policy,
+        metadata=metadata,
+    )
+    row.updated_by_user_id = user_id
+    session.add(row)
+    session.flush()
+    return row
+
+
+def _validate_profile_update_references(
+    row: FileConnectorProfile,
+    *,
+    password_env: str | None,
+    token_env: str | None,
+    secret_ref: str | None,
+    metadata: Mapping[str, Any] | None,
+) -> None:
     reject_api_controlled_deployment_references(
         password_env=password_env,
         token_env=token_env,
         secret_ref=secret_ref,
         metadata=metadata,
     )
-    if secret_ref is not None and _clean(secret_ref) != _clean(row.secret_ref):
-        if _clean(row.secret_ref):
-            raise FileStorageError(
-                "An existing external secret reference cannot be replaced or cleared until Files can prove "
-                "provider ownership and confirm provider-side deletion"
-            )
+    if secret_ref is None or _clean(secret_ref) == _clean(row.secret_ref):
+        return
+    if _clean(row.secret_ref):
+        raise FileStorageError(
+            "An existing external secret reference cannot be replaced or cleared until Files can prove "
+            "provider ownership and confirm provider-side deletion"
+        )
+
+
+def _update_profile_connection_fields(
+    row: FileConnectorProfile,
+    *,
+    label: str | None,
+    provider: str | None,
+    endpoint_url: str | None,
+    base_path: str | None,
+    enabled: bool | None,
+    credential_profile_id: str | None,
+    credential_mode: str | None,
+) -> None:
     if label is not None:
         row.label = _normalize_label(label)
     if provider is not None:
@@ -214,6 +301,20 @@ def update_connector_profile_row(
         row.credential_profile_id = _clean(credential_profile_id)
     if credential_mode is not None:
         row.credential_mode = _normalize_credential_mode(credential_mode)
+
+
+def _update_profile_credential_fields(
+    row: FileConnectorProfile,
+    *,
+    username: str | None,
+    password: str | None,
+    token: str | None,
+    password_env: str | None,
+    token_env: str | None,
+    secret_ref: str | None,
+    clear_password: bool,
+    clear_token: bool,
+) -> None:
     if username is not None:
         row.username = _clean(username)
     if password is not None:
@@ -230,16 +331,21 @@ def update_connector_profile_row(
         row.token_env = _clean(token_env)
     if secret_ref is not None:
         row.secret_ref = _clean(secret_ref)
+
+
+def _update_profile_governance_fields(
+    row: FileConnectorProfile,
+    *,
+    capabilities: list[str] | None,
+    policy: Mapping[str, Any] | None,
+    metadata: Mapping[str, Any] | None,
+) -> None:
     if capabilities is not None:
         row.capabilities = _string_list(capabilities)
     if policy is not None:
         row.policy = dict(policy)
     if metadata is not None:
         row.metadata_ = dict(metadata)
-    row.updated_by_user_id = user_id
-    session.add(row)
-    session.flush()
-    return row
 
 
 def _normalize_scope(*, tenant_id: str, scope_type: str, scope_id: str | None) -> tuple[str, str | None, str | None]:

src/govoplan_files/backend/storage/connector_visibility.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable
+from dataclasses import replace
+
+from sqlalchemy.orm import Session
+
+from govoplan_files.backend.storage.connector_deployment import (
+    connector_effective_endpoint_url,
+)
+from govoplan_files.backend.storage.connector_profile_store import (
+    select_database_connector_profiles,
+)
+from govoplan_files.backend.storage.connector_profiles import (
+    ConnectorProfile,
+    connector_profiles_from_settings,
+)
+from govoplan_files.backend.storage.connector_policy import (
+    ConnectorAccessRequest,
+    connector_policy_decision,
+)
+from govoplan_files.backend.storage.connector_policy_store import (
+    effective_connector_policy_sources,
+)
+from govoplan_files.backend.storage.connector_providers import (
+    connector_provider_descriptors,
+)
+
+
+CampaignVisibility = Callable[[str], bool]
+
+
+def visible_connector_profiles_for_actor(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    group_ids: Iterable[str],
+    settings: object | None,
+    provider: str | None = None,
+    campaign_id: str | None = None,
+    campaign_visible: CampaignVisibility | None = None,
+    include_disabled: bool = False,
+    include_admin_scopes: bool = False,
+    include_effective_policy: bool = True,
+) -> list[ConnectorProfile]:
+    """Return the same actor-filtered connector set used by API and docs surfaces."""
+
+    provider_norm = provider.strip().casefold() if provider else None
+    actor_group_ids = {str(group_id) for group_id in group_ids if str(group_id)}
+    database_profiles, database_profile_ids = select_database_connector_profiles(
+        session,
+        tenant_id=tenant_id,
+        include_disabled=include_disabled,
+        row_visible=lambda row: (
+            include_admin_scopes
+            and row.scope_type in {"user", "group", "campaign"}
+        ) or _scope_visible_to_actor(
+            scope_type=row.scope_type,
+            scope_id=row.scope_id,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            group_ids=actor_group_ids,
+            campaign_id=campaign_id,
+            campaign_visible=campaign_visible,
+        ),
+    )
+    configured_profiles = connector_profiles_from_settings(settings)
+    profiles_by_id: dict[str, ConnectorProfile] = {}
+    for profile in database_profiles:
+        if profile.id not in profiles_by_id:
+            profiles_by_id[profile.id] = profile
+    for profile in configured_profiles:
+        if profile.id not in database_profile_ids and profile.id not in profiles_by_id:
+            profiles_by_id[profile.id] = profile
+
+    visible: list[ConnectorProfile] = []
+    for profile in profiles_by_id.values():
+        if not (profile.enabled or include_disabled):
+            continue
+        if provider_norm is not None and profile.provider != provider_norm:
+            continue
+        admin_scope_visible = include_admin_scopes and profile.scope_type in {
+            "user",
+            "group",
+            "campaign",
+        }
+        if not admin_scope_visible and not _profile_visible_to_actor(
+            profile,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            group_ids=actor_group_ids,
+            campaign_id=campaign_id,
+            campaign_visible=campaign_visible,
+        ):
+            continue
+        visible.append(
+            _with_effective_connector_policy(
+                session, tenant_id=tenant_id, profile=profile
+            )
+            if include_effective_policy
+            else profile
+        )
+    return visible
+
+
+def connector_profile_usable_for_import(profile: ConnectorProfile) -> bool:
+    """Whether a visible profile can safely offer the current live browse/import task."""
+
+    effective_endpoint_url = connector_effective_endpoint_url(
+        provider=profile.provider,
+        endpoint_url=profile.endpoint_url,
+        metadata=profile.metadata,
+    )
+    if (
+        not profile.enabled
+        or not effective_endpoint_url
+        or not profile.credentials_configured
+        or bool(profile.secret_ref)
+    ):
+        return False
+    descriptor = next(
+        (
+            item
+            for item in connector_provider_descriptors()
+            if item.provider == profile.provider
+        ),
+        None,
+    )
+    if descriptor is None:
+        return False
+    if not (
+        descriptor.implemented
+        and descriptor.installed
+        and descriptor.browse_supported
+        and descriptor.import_supported
+    ):
+        return False
+    # These SDK paths intentionally fail closed until every connection peer and
+    # SDK-managed redirect/referral can be pinned and revalidated.
+    if profile.provider in {"s3", "smb"}:
+        return False
+
+    # Prove that the initial root browse performed by the current Files UI is
+    # policy-allowed. A later selected remote path/item is checked again.
+    return connector_policy_decision(
+        ConnectorAccessRequest(
+            connector_id=profile.id,
+            credential_id=profile.credential_profile_id,
+            provider=profile.provider,
+            external_path="",
+            external_url=effective_endpoint_url,
+            operation="import",
+        ),
+        profile.policy_sources,
+    ).allowed
+
+
+def _profile_visible_to_actor(
+    profile: ConnectorProfile,
+    *,
+    tenant_id: str,
+    user_id: str,
+    group_ids: set[str],
+    campaign_id: str | None,
+    campaign_visible: CampaignVisibility | None,
+) -> bool:
+    return _scope_visible_to_actor(
+        scope_type=profile.scope_type,
+        scope_id=profile.scope_id,
+        tenant_id=tenant_id,
+        user_id=user_id,
+        group_ids=group_ids,
+        campaign_id=campaign_id,
+        campaign_visible=campaign_visible,
+    )
+
+
+def _scope_visible_to_actor(
+    *,
+    scope_type: str,
+    scope_id: str | None,
+    tenant_id: str,
+    user_id: str,
+    group_ids: set[str],
+    campaign_id: str | None,
+    campaign_visible: CampaignVisibility | None,
+) -> bool:
+    if scope_type == "system":
+        return True
+    if scope_type == "tenant":
+        return scope_id == tenant_id
+    if scope_type == "user":
+        return scope_id == user_id
+    if scope_type == "group":
+        return bool(scope_id and scope_id in group_ids)
+    if scope_type != "campaign" or not scope_id:
+        return False
+    if campaign_id and scope_id != campaign_id:
+        return False
+    return bool(campaign_visible and campaign_visible(scope_id))
+
+
+def _with_effective_connector_policy(
+    session: Session,
+    *,
+    tenant_id: str,
+    profile: ConnectorProfile,
+) -> ConnectorProfile:
+    sources = effective_connector_policy_sources(
+        session,
+        tenant_id=tenant_id,
+        scope_type=profile.scope_type,
+        scope_id=profile.scope_id,
+    )
+    if not sources:
+        return profile
+    return replace(profile, policy_sources=tuple([*sources, *profile.policy_sources]))

tests/test_connector_credential_deletion.py

@@ -169,6 +169,10 @@ class ConnectorCredentialDeletionTests(unittest.TestCase):
             "govoplan_files.backend.storage.connector_credential_deletion.audit_event"
         ) as audit:
             plan.destroy_data_executor(self.session, "files")
+        # The retirement executor deliberately enlists secret scrubbing, audit
+        # writes, and DDL in the caller-owned transaction.  Commit that unit of
+        # work before inspecting through a separate Engine connection.
+        self.session.commit()
 
         self.assertEqual(2, audit.call_count)
         self.assertEqual(
@@ -193,6 +197,7 @@ class ConnectorCredentialDeletionTests(unittest.TestCase):
             "govoplan_files.backend.storage.connector_credential_deletion.audit_event"
         ) as audit:
             plan.destroy_data_executor(self.session, "files")
+        self.session.commit()
 
         audit.assert_called_once()
         details = audit.call_args.kwargs["details"]

tests/test_connector_profile_store.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db import models as access_models  # noqa: F401 - resolve Files user foreign keys
+from govoplan_core.db.base import Base
+from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
+from govoplan_files.backend.db.models import FileConnectorProfile
+from govoplan_files.backend.storage.common import FileStorageError
+from govoplan_files.backend.storage.connector_profile_store import (
+    list_database_connector_profiles,
+    update_connector_profile_row,
+)
+
+
+class ConnectorProfileStoreUpdateTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine, tables=[FileConnectorProfile.__table__])
+        self.session = sessionmaker(bind=self.engine)()
+        self.row = FileConnectorProfile(
+            id="profile-1",
+            tenant_id="tenant-1",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            label="Original profile",
+            provider="webdav",
+            endpoint_url="https://dav.example.test/original",
+            base_path="original",
+            enabled=True,
+            credential_mode="basic",
+            username="original-user",
+            password_encrypted=encrypt_secret("original-password"),
+            token_encrypted=encrypt_secret("original-token"),
+            capabilities=["browse"],
+            policy={"allow": {"providers": ["webdav"]}},
+            metadata_={"original": True},
+            created_by_user_id=None,
+            updated_by_user_id=None,
+        )
+        self.session.add(self.row)
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(bind=self.engine)
+        self.engine.dispose()
+
+    def test_update_preserves_normalization_and_flush_only_transaction_boundary(self) -> None:
+        updated = update_connector_profile_row(
+            self.session,
+            self.row,
+            user_id="user-2",
+            label="  Updated profile  ",
+            provider="NEXTCLOUD",
+            endpoint_url="  https://cloud.example.test/dav  ",
+            base_path="  shared/reports  ",
+            enabled=False,
+            credential_profile_id="  credential-2  ",
+            credential_mode="TOKEN",
+            username="  updated-user  ",
+            password="updated-password",
+            token="updated-token",
+            capabilities=["browse", " import ", ""],
+            policy={"deny": {"external_paths": ["private"]}},
+            metadata={"department": "reports"},
+        )
+
+        self.assertIs(self.row, updated)
+        self.assertEqual("Updated profile", updated.label)
+        self.assertEqual("nextcloud", updated.provider)
+        self.assertEqual("https://cloud.example.test/dav", updated.endpoint_url)
+        self.assertEqual("shared/reports", updated.base_path)
+        self.assertFalse(updated.enabled)
+        self.assertEqual("credential-2", updated.credential_profile_id)
+        self.assertEqual("token", updated.credential_mode)
+        self.assertEqual("updated-user", updated.username)
+        self.assertEqual("updated-password", decrypt_secret(updated.password_encrypted))
+        self.assertEqual("updated-token", decrypt_secret(updated.token_encrypted))
+        self.assertEqual(["browse", "import"], updated.capabilities)
+        self.assertEqual({"deny": {"external_paths": ["private"]}}, updated.policy)
+        self.assertEqual({"department": "reports"}, updated.metadata_)
+        self.assertEqual("user-2", updated.updated_by_user_id)
+
+        self.session.rollback()
+        persisted = self.session.get(FileConnectorProfile, self.row.id)
+        assert persisted is not None
+        self.assertEqual("Original profile", persisted.label)
+        self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted))
+        self.assertTrue(persisted.enabled)
+
+    def test_secret_replacement_wins_over_clear_while_clear_removes_an_omitted_token(self) -> None:
+        update_connector_profile_row(
+            self.session,
+            self.row,
+            user_id="user-2",
+            password="replacement-password",
+            clear_password=True,
+            clear_token=True,
+        )
+
+        self.assertEqual("replacement-password", decrypt_secret(self.row.password_encrypted))
+        self.assertIsNone(self.row.token_encrypted)
+        self.assertEqual("https://dav.example.test/original", self.row.endpoint_url)
+        self.assertEqual("original-user", self.row.username)
+
+    def test_legacy_external_secret_reference_cannot_be_cleared_or_partially_mutate_the_row(self) -> None:
+        self.row.secret_ref = "vault:tenant-1:files:profile"
+        self.session.commit()
+
+        with self.assertRaisesRegex(FileStorageError, "provider-side deletion"):
+            update_connector_profile_row(
+                self.session,
+                self.row,
+                user_id="user-2",
+                label="Must not be applied",
+                secret_ref="",
+            )
+
+        self.assertEqual("Original profile", self.row.label)
+        self.assertEqual("vault:tenant-1:files:profile", self.row.secret_ref)
+
+    def test_flush_failure_remains_rollback_safe(self) -> None:
+        with patch.object(self.session, "flush", side_effect=RuntimeError("database unavailable")), self.assertRaisesRegex(
+            RuntimeError,
+            "database unavailable",
+        ):
+            update_connector_profile_row(
+                self.session,
+                self.row,
+                user_id="user-2",
+                label="Uncommitted profile",
+                password="uncommitted-password",
+            )
+
+        self.session.rollback()
+        persisted = self.session.get(FileConnectorProfile, self.row.id)
+        assert persisted is not None
+        self.assertEqual("Original profile", persisted.label)
+        self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted))
+
+    def test_visibility_filter_runs_before_connector_secrets_are_decrypted(self) -> None:
+        invisible = FileConnectorProfile(
+            id="invisible-profile",
+            tenant_id="tenant-1",
+            scope_type="user",
+            scope_id="another-user",
+            label="Invisible profile",
+            provider="webdav",
+            endpoint_url="https://invisible.example.test",
+            enabled=True,
+            credential_mode="basic",
+            password_encrypted="not-a-valid-encrypted-secret",
+            capabilities=["browse"],
+            policy={},
+            metadata_={},
+        )
+        self.session.add(invisible)
+        self.session.commit()
+
+        profiles = list_database_connector_profiles(
+            self.session,
+            tenant_id="tenant-1",
+            row_visible=lambda row: row.id == self.row.id,
+        )
+
+        self.assertEqual([self.row.id], [profile.id for profile in profiles])
+        self.assertEqual("original-password", profiles[0].password_value)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_connector_visibility.py

@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+import unittest
+from dataclasses import replace
+from unittest.mock import patch
+
+from govoplan_files.backend.storage.connector_policy import ConnectorPolicySource
+from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
+from govoplan_files.backend.storage.connector_visibility import (
+    connector_profile_usable_for_import,
+    visible_connector_profiles_for_actor,
+)
+
+
+def _profile(
+    profile_id: str,
+    *,
+    scope_type: str = "system",
+    scope_id: str | None = None,
+    provider: str = "webdav",
+    enabled: bool = True,
+) -> ConnectorProfile:
+    return ConnectorProfile(
+        id=profile_id,
+        label=profile_id,
+        provider=provider,
+        scope_type=scope_type,
+        scope_id=scope_id,
+        endpoint_url=f"https://{profile_id}.example.invalid",
+        enabled=enabled,
+        credential_mode="anonymous",
+    )
+
+
+class ConnectorVisibilityTests(unittest.TestCase):
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources",
+        return_value=[],
+    )
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings"
+    )
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
+    )
+    def test_profiles_are_filtered_to_actor_scopes_and_database_definition_wins(
+        self,
+        database_profiles,
+        configured_profiles,
+        _policy_sources,
+    ) -> None:
+        profiles = [
+            _profile("system"),
+            _profile("tenant", scope_type="tenant", scope_id="tenant-1"),
+            _profile("other-tenant", scope_type="tenant", scope_id="tenant-2"),
+            _profile("user", scope_type="user", scope_id="user-1"),
+            _profile("other-user", scope_type="user", scope_id="user-2"),
+            _profile("group", scope_type="group", scope_id="group-1"),
+            _profile("campaign", scope_type="campaign", scope_id="campaign-1"),
+            _profile("disabled", enabled=False),
+            _profile("duplicate", provider="webdav"),
+        ]
+        database_profiles.return_value = (profiles, {profile.id for profile in profiles})
+        configured_profiles.return_value = [_profile("duplicate", provider="nextcloud")]
+
+        visible = visible_connector_profiles_for_actor(
+            object(),  # type: ignore[arg-type]
+            tenant_id="tenant-1",
+            user_id="user-1",
+            group_ids={"group-1"},
+            settings=object(),
+            campaign_visible=lambda campaign_id: campaign_id == "campaign-1",
+        )
+
+        self.assertEqual(
+            {"system", "tenant", "user", "group", "campaign", "duplicate"},
+            {profile.id for profile in visible},
+        )
+        duplicate = next(profile for profile in visible if profile.id == "duplicate")
+        self.assertEqual("webdav", duplicate.provider)
+
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources",
+        return_value=[],
+    )
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings",
+        return_value=[],
+    )
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
+    )
+    def test_campaign_and_admin_visibility_remain_explicit(
+        self,
+        database_profiles,
+        _configured_profiles,
+        _policy_sources,
+    ) -> None:
+        profiles = [
+            _profile("campaign-a", scope_type="campaign", scope_id="campaign-a"),
+            _profile("campaign-b", scope_type="campaign", scope_id="campaign-b"),
+            _profile("other-user", scope_type="user", scope_id="user-2"),
+        ]
+        database_profiles.return_value = (profiles, {profile.id for profile in profiles})
+
+        campaign_only = visible_connector_profiles_for_actor(
+            object(),  # type: ignore[arg-type]
+            tenant_id="tenant-1",
+            user_id="user-1",
+            group_ids=(),
+            settings=None,
+            campaign_id="campaign-a",
+            campaign_visible=lambda _campaign_id: True,
+        )
+        self.assertEqual(["campaign-a"], [profile.id for profile in campaign_only])
+
+        admin = visible_connector_profiles_for_actor(
+            object(),  # type: ignore[arg-type]
+            tenant_id="tenant-1",
+            user_id="user-1",
+            group_ids=(),
+            settings=None,
+            campaign_visible=None,
+            include_admin_scopes=True,
+        )
+        self.assertEqual(
+            {"campaign-a", "campaign-b", "other-user"},
+            {profile.id for profile in admin},
+        )
+
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings",
+        return_value=[],
+    )
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
+    )
+    @patch(
+        "govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources"
+    )
+    def test_effective_policy_is_attached_without_mutating_profile(
+        self,
+        policy_sources,
+        database_profiles,
+        _configured_profiles,
+    ) -> None:
+        profile = _profile("profile")
+        source = ConnectorPolicySource(
+            scope_type="system",
+            label="System",
+            policy={"allow": {"providers": ["webdav"]}},
+        )
+        database_profiles.return_value = ([profile], {profile.id})
+        policy_sources.return_value = [source]
+
+        visible = visible_connector_profiles_for_actor(
+            object(),  # type: ignore[arg-type]
+            tenant_id="tenant-1",
+            user_id="user-1",
+            group_ids=(),
+            settings=None,
+        )
+
+        self.assertEqual((), profile.policy_sources)
+        self.assertEqual((source,), visible[0].policy_sources)
+
+    def test_import_usability_requires_safe_provider_credentials_endpoint_and_identity_policy(
+        self,
+    ) -> None:
+        self.assertTrue(connector_profile_usable_for_import(_profile("webdav")))
+        self.assertFalse(
+            connector_profile_usable_for_import(_profile("s3", provider="s3"))
+        )
+        self.assertFalse(
+            connector_profile_usable_for_import(_profile("smb", provider="smb"))
+        )
+        self.assertFalse(
+            connector_profile_usable_for_import(
+                _profile("sharepoint", provider="sharepoint")
+            )
+        )
+        self.assertFalse(
+            connector_profile_usable_for_import(
+                ConnectorProfile(
+                    id="no-endpoint",
+                    label="No endpoint",
+                    provider="webdav",
+                    credential_mode="anonymous",
+                )
+            )
+        )
+        self.assertFalse(
+            connector_profile_usable_for_import(
+                ConnectorProfile(
+                    id="no-credentials",
+                    label="No credentials",
+                    provider="webdav",
+                    endpoint_url="https://files.example.invalid",
+                    credential_mode="basic",
+                )
+            )
+        )
+        self.assertTrue(
+            connector_profile_usable_for_import(
+                ConnectorProfile(
+                    id="metadata-endpoint",
+                    label="Metadata endpoint",
+                    provider="webdav",
+                    credential_mode="anonymous",
+                    metadata={"webdav_endpoint_url": "https://files.example.invalid"},
+                )
+            )
+        )
+        self.assertFalse(
+            connector_profile_usable_for_import(
+                ConnectorProfile(
+                    id="unresolved-secret-reference",
+                    label="Unresolved secret reference",
+                    provider="webdav",
+                    endpoint_url="https://files.example.invalid",
+                    credential_mode="basic",
+                    secret_ref="vault://files/webdav",
+                )
+            )
+        )
+
+        denied = replace(
+            _profile("denied"),
+            policy_sources=(
+                ConnectorPolicySource(
+                    scope_type="system",
+                    label="System",
+                    policy={"deny": {"providers": ["webdav"]}},
+                ),
+            ),
+        )
+        self.assertFalse(connector_profile_usable_for_import(denied))
+
+        path_limited = replace(
+            _profile("path-limited"),
+            policy_sources=(
+                ConnectorPolicySource(
+                    scope_type="system",
+                    label="System",
+                    policy={"allow": {"external_paths": ["/approved/*"]}},
+                ),
+            ),
+        )
+        self.assertFalse(connector_profile_usable_for_import(path_limited))
+
+        endpoint_allowed = replace(
+            _profile("endpoint-allowed"),
+            policy_sources=(
+                ConnectorPolicySource(
+                    scope_type="system",
+                    label="System",
+                    policy={"allow": {"external_urls": ["https://*.example.invalid"]}},
+                ),
+            ),
+        )
+        self.assertTrue(connector_profile_usable_for_import(endpoint_allowed))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_documentation.py

@@ -0,0 +1,241 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.modules import DocumentationContext
+from govoplan_files.backend.documentation import documentation_topics
+from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
+
+
+class _Principal:
+    tenant_id = "tenant-1"
+    user = SimpleNamespace(id="user-1")
+    group_ids = frozenset({"group-1"})
+
+    def __init__(self, scopes: set[str]) -> None:
+        self.scopes = frozenset(scopes)
+
+    def has(self, scope: str) -> bool:
+        return scope in self.scopes
+
+
+class FilesRuntimeDocumentationTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.session = Session()
+
+    def tearDown(self) -> None:
+        self.session.close()
+
+    def context(
+        self,
+        scopes: set[str],
+        *,
+        settings: object | None = None,
+        documentation_type: str = "user",
+    ) -> DocumentationContext:
+        return DocumentationContext(
+            registry=object(),
+            principal=_Principal(scopes),
+            settings=settings,
+            session=self.session,
+            documentation_type=documentation_type,  # type: ignore[arg-type]
+        )
+
+    def test_upload_tasks_state_exact_current_safe_limits(self) -> None:
+        settings = SimpleNamespace(
+            file_upload_max_bytes=7 * 1024 * 1024,
+            file_upload_zip_max_bytes=19 * 1024 * 1024,
+        )
+        with patch(
+            "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",
+            return_value=[],
+        ):
+            topics = {
+                topic.id: topic
+                for topic in documentation_topics(
+                    self.context(
+                        {"files:file:read", "files:file:upload"}, settings=settings
+                    )
+                )
+            }
+
+        upload = topics["files.workflow.upload-managed-files"]
+        self.assertIn("7 MiB (7,340,032 bytes)", upload.body)
+        self.assertEqual(["files.list"], upload.metadata["help_contexts"])
+        self.assertEqual(
+            {"files:file:read", "files:file:upload"},
+            set(upload.conditions[0].required_scopes),
+        )
+
+        archive = topics["files.workflow.upload-and-unpack-zip"]
+        self.assertIn("19 MiB (19,922,944 bytes)", archive.body)
+        self.assertIn("7 MiB (7,340,032 bytes)", archive.body)
+        self.assertIn("1,000", archive.body)
+        self.assertIn("Actual extracted bytes are counted", archive.body)
+
+    def test_connector_task_requires_authority_and_a_visible_usable_profile(
+        self,
+    ) -> None:
+        usable = ConnectorProfile(
+            id="private-profile-id",
+            label="Private profile label",
+            provider="webdav",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            endpoint_url="https://private.example.invalid/secret-root",
+            base_path="/secret-root",
+            credential_mode="anonymous",
+        )
+        context = self.context({"files:file:read", "files:file:upload"})
+        with patch(
+            "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",
+            return_value=[usable],
+        ) as visible:
+            topics = {topic.id: topic for topic in documentation_topics(context)}
+
+        self.assertIn("files.workflow.import-managed-snapshot", topics)
+        self.assertNotIn("files.connector-import-unavailable", topics)
+        task = topics["files.workflow.import-managed-snapshot"]
+        self.assertEqual("configured", task.layer)
+        self.assertEqual(
+            ["files.list", "files.connector-import"], task.metadata["help_contexts"]
+        )
+        self.assertIn("re-authorized", task.body)
+        self.assertNotIn("private-profile-id", repr(task))
+        self.assertNotIn("private.example.invalid", repr(task))
+        self.assertNotIn("secret-root", repr(task))
+        visible.assert_called_once()
+        self.assertEqual(("group-1",), tuple(visible.call_args.kwargs["group_ids"]))
+
+        without_authority = {
+            topic.id: topic
+            for topic in documentation_topics(self.context({"files:file:read"}))
+        }
+        self.assertNotIn("files.workflow.import-managed-snapshot", without_authority)
+        self.assertIn(
+            "both permission to view Files and permission to upload",
+            without_authority["files.connector-import-unavailable"].body,
+        )
+
+    def test_connector_limitation_is_precise_but_never_exposes_profile_internals(
+        self,
+    ) -> None:
+        blocked = ConnectorProfile(
+            id="sensitive-profile-id",
+            label="Sensitive label",
+            provider="s3",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            endpoint_url="https://internal.example.invalid/private",
+            base_path="/classified",
+            credential_mode="basic",
+            password_value="credential-secret",
+        )
+        with patch(
+            "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",
+            return_value=[blocked],
+        ):
+            topics = {
+                topic.id: topic
+                for topic in documentation_topics(
+                    self.context({"files:file:read", "files:file:upload"})
+                )
+            }
+
+        self.assertNotIn("files.workflow.import-managed-snapshot", topics)
+        limitation = topics["files.connector-import-unavailable"]
+        self.assertEqual("available", limitation.layer)
+        self.assertIn("pinning-safe provider", limitation.body)
+        payload = repr(limitation)
+        for secret in (
+            "sensitive-profile-id",
+            "Sensitive label",
+            "internal.example.invalid",
+            "classified",
+            "credential-secret",
+        ):
+            self.assertNotIn(secret, payload)
+
+    def test_files_admin_connector_groups_do_not_widen_campaign_membership(self) -> None:
+        usable = ConnectorProfile(
+            id="group-profile",
+            label="Group profile",
+            provider="webdav",
+            scope_type="group",
+            scope_id="admin-visible-group",
+            endpoint_url="https://files.example.invalid",
+            credential_mode="anonymous",
+        )
+
+        def groups_for_actor(_session, *, include_admin_groups, **_kwargs):
+            return ["member-group", "admin-visible-group"] if include_admin_groups else ["member-group"]
+
+        with (
+            patch(
+                "govoplan_files.backend.documentation.user_group_ids",
+                side_effect=groups_for_actor,
+            ),
+            patch(
+                "govoplan_files.backend.documentation._campaign_visibility",
+                return_value=None,
+            ) as campaign_visibility,
+            patch(
+                "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",
+                return_value=[usable],
+            ) as visible,
+        ):
+            topics = {
+                topic.id: topic
+                for topic in documentation_topics(
+                    self.context(
+                        {
+                            "files:file:read",
+                            "files:file:upload",
+                            "files:file:admin",
+                        }
+                    )
+                )
+            }
+
+        self.assertIn("files.workflow.import-managed-snapshot", topics)
+        self.assertEqual(
+            ("member-group", "admin-visible-group"),
+            tuple(visible.call_args.kwargs["group_ids"]),
+        )
+        self.assertEqual(
+            ("member-group",),
+            tuple(campaign_visibility.call_args.kwargs["group_ids"]),
+        )
+
+    def test_connector_evaluation_errors_fail_closed_without_error_details(
+        self,
+    ) -> None:
+        with patch(
+            "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",
+            side_effect=RuntimeError("private endpoint /root credential-id"),
+        ):
+            topics = {
+                topic.id: topic
+                for topic in documentation_topics(
+                    self.context({"files:file:read", "files:file:upload"})
+                )
+            }
+
+        self.assertNotIn("files.workflow.import-managed-snapshot", topics)
+        limitation = topics["files.connector-import-unavailable"]
+        self.assertIn("could not be safely evaluated", limitation.body)
+        self.assertNotIn("private endpoint", repr(limitation))
+        self.assertNotIn("credential-id", repr(limitation))
+
+    def test_runtime_provider_only_emits_user_documentation(self) -> None:
+        self.assertEqual(
+            (), documentation_topics(self.context(set(), documentation_type="admin"))
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_manifest_documentation.py

@@ -3,11 +3,21 @@ from __future__ import annotations
 import unittest
 
 
-TOPIC_IDS = {
-    "files.workflow.import-managed-snapshot",
+STATIC_TOPIC_IDS = {
+    "files.workflow.organize-managed-files",
+    "files.workflow.find-and-download-files",
+    "files.workflow.share-managed-files",
+    "files.workflow.delete-managed-files",
     "files.governed-connectors-and-provenance",
     "files.reference.integrity-recovery-and-fail-closed-transports",
     "files.reference.snapshot-provenance-and-capabilities",
+    "files.assurance.process-and-release-readiness",
+}
+RUNTIME_TOPIC_IDS = {
+    "files.workflow.upload-managed-files",
+    "files.workflow.upload-and-unpack-zip",
+    "files.workflow.import-managed-snapshot",
+    "files.connector-import-unavailable",
 }
 HANDBOOK_HREF = "govoplan-files/docs/FILES_HANDBOOK.md"
 
@@ -17,38 +27,98 @@ class FilesManifestDocumentationTests(unittest.TestCase):
     def setUpClass(cls) -> None:
         from govoplan_files.backend.manifest import manifest
 
+        cls.manifest = manifest
         cls.topics = {topic.id: topic for topic in manifest.documentation}
 
     def topic(self, topic_id: str):
         return self.topics[topic_id]
 
-    def test_adaptive_topics_have_role_scope_module_and_link_contracts(self) -> None:
-        self.assertEqual(TOPIC_IDS, set(self.topics))
-        self.assertEqual({"admin", "user"}, {item for topic in self.topics.values() for item in topic.documentation_types})
+    def test_static_topics_have_role_scope_module_and_link_contracts(self) -> None:
+        self.assertEqual(STATIC_TOPIC_IDS, set(self.topics))
+        self.assertEqual(
+            {"admin", "user"},
+            {
+                item
+                for topic in self.topics.values()
+                for item in topic.documentation_types
+            },
+        )
 
         for topic in self.topics.values():
             with self.subTest(topic=topic.id):
                 self.assertTrue(topic.audience)
                 self.assertTrue(topic.conditions)
-                self.assertTrue(any("files" in condition.required_modules for condition in topic.conditions))
-                self.assertTrue(any(condition.any_scopes or condition.required_scopes for condition in topic.conditions))
+                self.assertTrue(
+                    any(
+                        "files" in condition.required_modules
+                        for condition in topic.conditions
+                    )
+                )
+                self.assertTrue(
+                    any(
+                        condition.any_scopes or condition.required_scopes
+                        for condition in topic.conditions
+                    )
+                )
                 self.assertIn("runtime", {link.kind for link in topic.links})
-                self.assertIn("api", {link.kind for link in topic.links})
-                self.assertIn(HANDBOOK_HREF, {link.href for link in topic.links if link.kind == "repository"})
+                self.assertIn(
+                    HANDBOOK_HREF,
+                    {link.href for link in topic.links if link.kind == "repository"},
+                )
 
-    def test_import_topic_is_a_complete_user_workflow(self) -> None:
-        topic = self.topic("files.workflow.import-managed-snapshot")
+        self.assertIn(
+            "documentation_topics",
+            {provider.__name__ for provider in self.manifest.documentation_providers},
+        )
 
-        self.assertEqual(("user",), topic.documentation_types)
-        self.assertEqual("workflow", topic.metadata["kind"])
-        self.assertEqual("/files", topic.metadata["route"])
-        self.assertEqual("Files", topic.metadata["screen"])
-        for key in ("prerequisites", "steps", "outcome", "verification"):
-            self.assertTrue(topic.metadata[key])
-        self.assertIn("never mutate the remote source", topic.body)
-        self.assertIn("/api/v1/files/connectors/profiles/{profile_id}/import", {link.href for link in topic.links})
+    def test_user_tasks_are_independently_authorized_and_contextual(self) -> None:
+        expected_scopes = {
+            "files.workflow.organize-managed-files": {
+                "files:file:read",
+                "files:file:organize",
+            },
+            "files.workflow.find-and-download-files": {
+                "files:file:read",
+                "files:file:download",
+            },
+            "files.workflow.share-managed-files": {
+                "files:file:read",
+                "files:file:share",
+            },
+            "files.workflow.delete-managed-files": {
+                "files:file:read",
+                "files:file:delete",
+            },
+        }
+        for topic_id, scopes in expected_scopes.items():
+            with self.subTest(topic=topic_id):
+                topic = self.topic(topic_id)
+                self.assertEqual(("user",), topic.documentation_types)
+                self.assertEqual("workflow", topic.metadata["kind"])
+                self.assertEqual(scopes, set(topic.conditions[0].required_scopes))
+                self.assertEqual(["files.list"], topic.metadata["help_contexts"])
+                for key in ("prerequisites", "steps", "outcome", "verification"):
+                    self.assertTrue(topic.metadata[key])
 
-    def test_admin_topic_covers_policy_redaction_and_atomic_credential_deletion(self) -> None:
+    def test_share_and_delete_tasks_state_current_boundaries(self) -> None:
+        share = self.topic("files.workflow.share-managed-files")
+        self.assertEqual("available", share.layer)
+        self.assertIn("does not yet provide a general share editor", share.body)
+        self.assertIn("no share-revocation route", share.body)
+        self.assertIn(
+            "/api/v1/files/{file_id}/shares", {link.href for link in share.links}
+        )
+
+        delete = self.topic("files.workflow.delete-managed-files")
+        self.assertIn("Soft-delete", delete.summary)
+        self.assertIn("no self-service restore or hard-purge", delete.body)
+        self.assertTrue(
+            any("not a hard purge" in item for item in delete.metadata["limitations"])
+        )
+
+    def test_admin_topic_covers_policy_redaction_and_atomic_credential_deletion(
+        self,
+    ) -> None:
         topic = self.topic("files.governed-connectors-and-provenance")
 
         self.assertEqual(("admin",), topic.documentation_types)
@@ -58,12 +128,24 @@ class FilesManifestDocumentationTests(unittest.TestCase):
         self.assertIn("deny rules win", topic.body)
         self.assertIn("redact secret values", topic.body)
         self.assertIn("same transaction", topic.body)
-        self.assertTrue(any("non-owned external references" in item for item in topic.metadata["security_invariants"]))
-        self.assertIn("/api/v1/files/connectors/policies/tenant", {link.href for link in topic.links})
-        self.assertIn("/api/v1/files/connectors/credentials", {link.href for link in topic.links})
+        self.assertTrue(
+            any(
+                "non-owned external references" in item
+                for item in topic.metadata["security_invariants"]
+            )
+        )
+        self.assertIn(
+            "/api/v1/files/connectors/policies/tenant",
+            {link.href for link in topic.links},
+        )
+        self.assertIn(
+            "/api/v1/files/connectors/credentials", {link.href for link in topic.links}
+        )
 
     def test_operator_topic_covers_recovery_and_fail_closed_s3_smb(self) -> None:
-        topic = self.topic("files.reference.integrity-recovery-and-fail-closed-transports")
+        topic = self.topic(
+            "files.reference.integrity-recovery-and-fail-closed-transports"
+        )
 
         self.assertEqual(("admin",), topic.documentation_types)
         self.assertEqual("reference", topic.metadata["kind"])
@@ -74,7 +156,9 @@ class FilesManifestDocumentationTests(unittest.TestCase):
         self.assertIn("does not remove backend blob objects", topic.body)
         self.assertIn("MASTER_KEY_B64", topic.metadata["recovery_unit"])
         self.assertTrue(topic.metadata["verification"])
-        self.assertIn("GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS", topic.configuration_keys)
+        self.assertIn(
+            "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS", topic.configuration_keys
+        )
 
     def test_integrator_topic_exposes_capability_and_provenance_boundary(self) -> None:
         topic = self.topic("files.reference.snapshot-provenance-and-capabilities")
@@ -89,11 +173,37 @@ class FilesManifestDocumentationTests(unittest.TestCase):
         self.assertIn("exact asset, version, blob, checksum", topic.body)
         self.assertIn("collaboration", topic.body)
 
+    def test_process_and_release_assurance_exposes_gates_evidence_and_limits(
+        self,
+    ) -> None:
+        topic = self.topic("files.assurance.process-and-release-readiness")
+
+        self.assertEqual(("admin", "user"), topic.documentation_types)
+        self.assertEqual("workflow", topic.metadata["kind"])
+        self.assertEqual(["files.list"], topic.metadata["help_contexts"])
+        self.assertIn("process_owner", topic.audience)
+        self.assertIn("release_manager", topic.audience)
+        self.assertIn("versions align", topic.body)
+        self.assertIn("no general share-management UI or share revocation", topic.body)
+        self.assertIn("no enforced retention or legal hold", topic.body)
+        for key in ("prerequisites", "steps", "outcome", "verification"):
+            self.assertTrue(topic.metadata[key])
+        self.assertTrue(
+            any("meta-repository security" in step for step in topic.metadata["steps"])
+        )
+        self.assertTrue(any("fails closed" in step for step in topic.metadata["steps"]))
+        self.assertTrue(any("SHA-256" in step for step in topic.metadata["steps"]))
+
     def test_internal_related_topic_ids_resolve(self) -> None:
+        known_ids = STATIC_TOPIC_IDS | RUNTIME_TOPIC_IDS
         for topic in self.topics.values():
             for related_id in topic.metadata.get("related_topic_ids", []):
                 if related_id.startswith("files."):
-                    self.assertIn(related_id, TOPIC_IDS, f"{topic.id} refers to missing topic {related_id}")
+                    self.assertIn(
+                        related_id,
+                        known_ids,
+                        f"{topic.id} refers to missing topic {related_id}",
+                    )
 
 
 if __name__ == "__main__":