feat: govern file lifecycle and connector writes

src/govoplan_files/backend/change_tracking.py

@@ -56,6 +56,10 @@ def _record_asset_change(session: OrmSession, asset: FileAsset) -> None:
             "filename",
             "description",
             "deleted_at",
+            "retained_until",
+            "legal_hold",
+            "lifecycle_revision",
+            "lifecycle_reason",
             "metadata_",
         ),
     )
@@ -79,6 +83,9 @@ def _record_asset_change(session: OrmSession, asset: FileAsset) -> None:
             "previous_path": previous_value(asset, "display_path"),
             "filename": asset.filename,
             "deleted_at": _isoformat(asset.deleted_at),
+            "retained_until": _isoformat(asset.retained_until),
+            "legal_hold": asset.legal_hold,
+            "lifecycle_revision": asset.lifecycle_revision,
         },
     )
 

src/govoplan_files/backend/db/models.py

@@ -214,6 +214,16 @@ class FileAsset(Base, TimestampMixin):
     deleted_at: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True, index=True
     )
+    retained_until: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    legal_hold: Mapped[bool] = mapped_column(
+        Boolean, default=False, nullable=False, index=True
+    )
+    lifecycle_revision: Mapped[int] = mapped_column(
+        Integer, default=1, nullable=False
+    )
+    lifecycle_reason: Mapped[str | None] = mapped_column(String(500), nullable=True)
     metadata_: Mapped[dict[str, Any] | None] = mapped_column(
         "metadata", JSON, nullable=True
     )

src/govoplan_files/backend/manifest.py

@@ -177,6 +177,26 @@ PERMISSIONS = (
         "Delete files",
         "Delete or hide managed files and folders where policy allows it.",
     ),
+    _permission(
+        "files:file:restore",
+        "Restore files",
+        "Restore soft-deleted managed files, folders, and connector spaces.",
+    ),
+    _permission(
+        "files:file:retention",
+        "Govern file retention",
+        "Set managed-file retention deadlines and legal holds.",
+    ),
+    _permission(
+        "files:file:purge",
+        "Purge files",
+        "Irreversibly purge eligible file records and unreferenced managed blobs.",
+    ),
+    _permission(
+        "files:connector:write",
+        "Write connected files",
+        "Write managed file bytes to explicitly writable remote connector spaces.",
+    ),
     _permission(
         "files:file:admin",
         "Administer file spaces",
@@ -196,6 +216,8 @@ ROLE_TEMPLATES = (
             "files:file:organize",
             "files:file:share",
             "files:file:delete",
+            "files:file:restore",
+            "files:file:retention",
         ),
     ),
     RoleTemplate(
@@ -370,7 +392,7 @@ REMOTE_STORAGE_PROVIDER = ExternalProviderDeclaration(
     module_id="files",
     label="Remote file storage mirror",
     maturity="synchronize",
-    operations=("discover", "search", "read", "synchronize", "preview"),
+    operations=("discover", "search", "read", "synchronize", "preview", "write"),
     objects=(
         ProviderObjectDeclaration(
             object_type="remote_folder",
@@ -394,17 +416,18 @@ REMOTE_STORAGE_PROVIDER = ExternalProviderDeclaration(
         idempotency="Source profile, remote object reference, and revision/digest suppress duplicate managed versions.",
         retry="Operators repeat bounded browse/import after a classified transport failure; effects are not blindly retried.",
         timeout_seconds=30,
-        conflicts="Managed-file conflict policy is explicit; the current provider never mutates the remote source.",
-        outcome_unknown="An interrupted download is discarded unless its complete digest and managed version commit are confirmed.",
+        conflicts="Managed-file conflicts are explicit. S3 write-back requires a conditional create or a matching expected revision and never overwrites blindly.",
+        outcome_unknown="Interrupted downloads are discarded unless their managed version commits. An uncertain S3 write remains outcome-unknown or recovery-required until provider request/content markers are reconciled.",
         outcome_unknown_supported=True,
         evidence="Managed versions retain connector profile, remote object identity, source revision, digest, and acquisition time.",
         audit_event_types=(
             "files.connector.accessed",
             "files.connector.imported",
             "files.connector.synced",
+            "files.connector.written",
         ),
-        correction="A later acquisition creates a new managed version and preserves prior provenance.",
-        rollback="Remote reads require no remote rollback; incomplete local objects are reconciled as orphans.",
+        correction="A later acquisition creates a new managed version and preserves prior provenance; a later S3 correction is another explicitly revision-guarded write.",
+        rollback="Remote reads require no remote rollback. S3 writes use forward recovery because provider effects cannot join the Files database transaction.",
         compensation="A wrongly imported managed version can be retired under Files policy without deleting the source.",
         reconciliation="Re-read source metadata and digest, then compare the committed managed version and object-store inventory.",
         outage="Previously imported managed versions remain available while remote spaces report unknown or stale state.",
@@ -1041,7 +1064,7 @@ manifest = ModuleManifest(
             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."
+                "Authorized restoration reactivates the same asset, versions, provenance, folder tree, or connector-space link when its path or label is still free. Irreversible purge is a separate administrator workflow with retention and legal-hold checks."
             ),
             layer="configured",
             documentation_types=("user",),
@@ -1081,7 +1104,7 @@ manifest = ModuleManifest(
                 ],
                 "limitations": [
                     "Deletion is soft deletion, not a hard purge.",
-                    "There is no self-service restore or hard-purge workflow.",
+                    "Restore fails closed when an active file, folder, or connector space already uses the path or label.",
                 ],
                 "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.",
@@ -1091,6 +1114,72 @@ manifest = ModuleManifest(
                 ],
             },
         ),
+        DocumentationTopic(
+            id="files.workflow.restore-retain-and-purge",
+            title="Restore, retain, hold, and irreversibly purge managed files",
+            summary="Restore soft-deleted resources and govern irreversible erasure with retention, legal-hold, preview, approval, and recovery evidence.",
+            body=(
+                "Restore keeps the original asset identity, versions, blob references, and connector provenance. A retention administrator can set an explicit retained-until time or legal hold with a reason and revision check. Hard purge requires the dedicated purge permission, a current immutable preview hash, a caller idempotency key, the literal PURGE confirmation, and an approval reference. Active retention, legal hold, active shares, Campaign evidence, and Form evidence block purge. Purge removes eligible database records first and releases blobs; a separate bounded garbage-collection action rechecks every FileVersion reference under the same distributed blob fence used by uploads before deleting bytes and metadata. Every irreversible action is visible through the Core recovery ledger and audit evidence."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("file_manager", "file_admin", "records_manager", "operator"),
+            order=46,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("files",),
+                    any_scopes=(
+                        "files:file:restore",
+                        "files:file:retention",
+                        "files:file:purge",
+                    ),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(
+                    label="Purge preview API",
+                    href="/api/v1/files/purge/preview",
+                    kind="api",
+                ),
+                DocumentationLink(
+                    label="Files handbook",
+                    href="govoplan-files/docs/FILES_HANDBOOK.md",
+                    kind="repository",
+                ),
+            ),
+            related_modules=("audit", "ops", "campaigns", "forms-runtime"),
+            unlocks=(
+                "Authorized operators can distinguish reversible removal from policy-governed physical erasure.",
+            ),
+            metadata={
+                "kind": "workflow",
+                "route": "/files",
+                "screen": "Files lifecycle",
+                "help_contexts": ["files.list"],
+                "prerequisites": [
+                    "The actor has the distinct restore, retention, or purge permission required by the intended action.",
+                    "A purge target is already soft-deleted and all dependent evidence or shares have been lawfully resolved.",
+                ],
+                "steps": [
+                    "Set or review retention and legal-hold state using the displayed lifecycle revision.",
+                    "For restoration, verify the original path or label is free and restore the same resource.",
+                    "For erasure, create a bounded purge preview and resolve every named blocker.",
+                    "Execute the matching preview with an idempotency key, PURGE confirmation, and approval reference.",
+                    "Run bounded blob garbage collection and inspect any recovery-required or outcome-unknown operation in Ops.",
+                ],
+                "limitations": [
+                    "Automatic time-based purge scheduling is not included; an authorized actor starts preview, execute, and blob collection.",
+                    "A protected blob's Encryption envelope follows the Encryption module's own retention and key-custody policy.",
+                ],
+                "outcome": "Eligible metadata and unreferenced bytes are erased with separately reviewable policy, audit, and recovery evidence.",
+                "verification": "Verify the asset/version rows are absent, the blob remains while referenced, unreferenced bytes are absent after GC, and every operation has a valid recovery evidence chain.",
+                "related_topic_ids": [
+                    "files.workflow.delete-managed-files",
+                    "files.reference.integrity-recovery-and-fail-closed-transports",
+                ],
+            },
+        ),
         DocumentationTopic(
             id="files.governed-connectors-and-provenance",
             title="Govern file connections and credential deletion",
@@ -1098,7 +1187,7 @@ manifest = ModuleManifest(
             body=(
                 "System, tenant, and one user/group/campaign leaf form the effective policy chain: deny rules win and every configured allow rule must match. "
                 "Responses redact secret values and deployment references. Deleting a database-managed credential or profile immediately scrubs Files-owned encrypted material and private metadata in the same transaction as a non-secret audit event; dependent profiles are disabled, while legacy non-owned references are only detached and audited. "
-                "Removing a connector space is a separate owner-authorized operation: it retires only the local virtual-space link and leaves provider content, imported managed files and shares, profiles, credentials, and remote references untouched. Intrinsic user and group managed spaces cannot be removed."
+                "Removing a connector space is a separate owner-authorized operation: it retires only the local virtual-space link and leaves provider content, imported managed files and shares, profiles, credentials, and remote references untouched. Intrinsic user and group managed spaces cannot be removed. Connector spaces remain read-only by default. An administrator may explicitly enable two-way mode only for an S3 profile carrying the write capability; each write requires separate authority, inherited path policy, a conditional create or expected revision, and durable recovery evidence. Automatic remote deletion, rename, and move propagation remain disabled."
             ),
             layer="configured",
             documentation_types=("admin",),
@@ -1169,6 +1258,7 @@ manifest = ModuleManifest(
                     "Deletion and destructive retirement scrub Files-owned encrypted connector material before completion and emit non-secret audit evidence.",
                     "Legacy non-owned external references are detached and audited, never sent to an arbitrary provider delete operation.",
                     "Connector-space removal is local and soft; it never claims to delete remote or previously imported managed content.",
+                    "Two-way connector mode is explicit and S3-only; it never enables automatic remote delete, rename, move, or ACL propagation.",
                 ],
                 "related_topic_ids": [
                     "files.workflow.import-managed-snapshot",
@@ -1183,7 +1273,7 @@ manifest = ModuleManifest(
             summary="Back up database evidence, blob ciphertext, and Encryption custody as one recovery unit, and pin every SDK-managed connector peer.",
             body=(
                 "Local durable storage is the operational baseline. Recover Files from a coordinated database/blob snapshot with the matching Encryption tables and original deployment master key, then run the bounded resumable integrity scan from Administration and verify representative protected and unprotected access paths. Each scan batch and finding action requires the revision shown to the operator, so a stale screen cannot recheck or delete after concurrent reconciliation. Protected scans verify stored ciphertext before decryption and then verify plaintext semantic evidence. On PostgreSQL, managed blob creation/repair and applied orphan cleanup commit lease-fenced Core recovery intent before object effects; success, compensation, and forward completion require independent database and object checks, while mismatch is quarantined and unresolved work remains visible in Ops. Development SQLite instead records blob intent in the caller transaction to avoid a second-writer deadlock, uses a process-local fence, verifies after commit, and reconstructs durable compensation evidence after handled rollback. A hard process loss before the SQLite caller commits can therefore leave an unrecorded object; SQLite is not a production recovery profile and operators must run an integrity scan after such a loss. Missing or mismatched blobs are quarantined; orphan objects are reported before dry-run-first, explicitly authorized cleanup. "
-                "S3 connector pools pin every retry, redirect, discovered endpoint, and provider alias while retaining the configured TLS authority; outbound proxies and ambient credential discovery are disabled. SMB initial connections, reconnects, aliases, and DFS referrals use a Files-owned pinned transport and cache. Both apply the deployment private-network policy immediately before each socket opens and fail closed if an SDK no longer exposes the verified transport seam. Installer-owned Garage storage is supported only at the exact deployment service endpoint with its explicit trust marker. Destructive module retirement drops database tables but does not remove backend blob objects."
+                "Hard purge records irreversible intent before deleting database evidence, and blob garbage collection separately rechecks references under the shared blob lease before deleting a managed object. S3 connector write-back records digest-only forward-recovery intent before a conditional provider effect; request and content markers prove success, while mismatches remain visible in Ops and fence later writers. S3 connector pools pin every retry, redirect, discovered endpoint, and provider alias while retaining the configured TLS authority; outbound proxies and ambient credential discovery are disabled. SMB initial connections, reconnects, aliases, and DFS referrals use a Files-owned pinned transport and cache. Both apply the deployment private-network policy immediately before each socket opens and fail closed if an SDK no longer exposes the verified transport seam. Installer-owned Garage storage is supported only at the exact deployment service endpoint with its explicit trust marker. Destructive module retirement drops database tables but does not remove backend blob objects."
             ),
             layer="configured",
             documentation_types=("admin",),
@@ -1337,7 +1427,7 @@ manifest = ModuleManifest(
             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 self-service restore or hard purge, no enforced retention or legal hold, and no dedicated canonical audit event for every ordinary Files mutation. Share grant, change, expiry, and revocation operations do emit dedicated audit records. Record remaining limitations in the process assessment instead of treating soft deletion or change-sequence entries as stronger evidence."
+                "Restore, retention, legal hold, governed hard purge, blob garbage collection, share lifecycle, and explicit S3 write-back now emit dedicated policy, recovery, or audit evidence. Ordinary organization mutations still rely primarily on the Files change sequence rather than a dedicated canonical audit event. Record remaining limitations in the process assessment instead of treating one evidence stream as another."
             ),
             layer="configured",
             documentation_types=("admin", "user"),
@@ -1584,7 +1674,7 @@ manifest = ModuleManifest(
         documentation_ref="docs/FILES_HANDBOOK.md",
         test_ref="tests/test_storage_backends.py",
         known_limits=(
-            "Target-environment multi-node recovery drills and writable remote connector effects are not reference-ready; hard purge and legal hold remain unimplemented.",
+            "Target-environment multi-node recovery drills remain deployment evidence; two-way connector writes are currently limited to explicit conditional S3 writes, with automatic remote delete, rename, move, and ACL propagation disabled.",
         ),
         supported_authority_modes=(
             "native_authoritative",

src/govoplan_files/backend/migrations/versions/a2b3c4d5e6f9_file_asset_lifecycle.py

@@ -0,0 +1,52 @@
+"""govern managed file retention and legal holds
+
+Revision ID: a2b3c4d5e6f9
+Revises: f1a2b3c4d5e7
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "a2b3c4d5e6f9"
+down_revision = "f1a2b3c4d5e7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("file_assets") as batch_op:
+        batch_op.add_column(
+            sa.Column("retained_until", sa.DateTime(timezone=True), nullable=True)
+        )
+        batch_op.add_column(
+            sa.Column(
+                "legal_hold", sa.Boolean(), nullable=False, server_default=sa.false()
+            )
+        )
+        batch_op.add_column(
+            sa.Column(
+                "lifecycle_revision", sa.Integer(), nullable=False, server_default="1"
+            )
+        )
+        batch_op.add_column(
+            sa.Column("lifecycle_reason", sa.String(length=500), nullable=True)
+        )
+        batch_op.create_index(
+            op.f("ix_file_assets_retained_until"), ["retained_until"], unique=False
+        )
+        batch_op.create_index(
+            op.f("ix_file_assets_legal_hold"), ["legal_hold"], unique=False
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("file_assets") as batch_op:
+        batch_op.drop_index(op.f("ix_file_assets_legal_hold"))
+        batch_op.drop_index(op.f("ix_file_assets_retained_until"))
+        batch_op.drop_column("lifecycle_reason")
+        batch_op.drop_column("lifecycle_revision")
+        batch_op.drop_column("legal_hold")
+        batch_op.drop_column("retained_until")

src/govoplan_files/backend/route_support.py

@@ -1039,6 +1039,12 @@ def _asset_response(
         created_at=asset.created_at.isoformat(),
         updated_at=asset.updated_at.isoformat(),
         deleted_at=asset.deleted_at.isoformat() if asset.deleted_at else None,
+        retained_until=(
+            asset.retained_until.isoformat() if asset.retained_until else None
+        ),
+        legal_hold=asset.legal_hold,
+        lifecycle_revision=asset.lifecycle_revision,
+        lifecycle_reason=asset.lifecycle_reason,
         audit_relevant=asset_is_audit_relevant(session, asset),
         metadata=metadata,
         source_provenance=source_provenance_from_metadata(metadata),

src/govoplan_files/backend/router.py

@@ -13,6 +13,7 @@ from govoplan_files.backend.routes.connector_settings import (
 from govoplan_files.backend.routes.folders import router as folders_router
 from govoplan_files.backend.routes.form_evidence import router as form_evidence_router
 from govoplan_files.backend.routes.integrity import router as integrity_router
+from govoplan_files.backend.routes.lifecycle import router as lifecycle_router
 from govoplan_files.backend.routes.listing import router as listing_router
 from govoplan_files.backend.routes.shares import router as shares_router
 from govoplan_files.backend.routes.spaces import router as spaces_router
@@ -26,6 +27,7 @@ for workflow_router in (
     folders_router,
     form_evidence_router,
     integrity_router,
+    lifecycle_router,
     listing_router,
     uploads_router,
     connector_settings_router,

src/govoplan_files/backend/routes/connector_io.py

@@ -5,11 +5,14 @@ from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal, require_any_scope, require_scope
+from govoplan_core.audit.logging import audit_from_principal
 from govoplan_files.backend.schemas import (
     FileConnectorBrowseItem,
     FileConnectorBrowseResponse,
     FileConnectorImportRequest,
     FileConnectorSyncResponse,
+    FileConnectorWriteRequest,
+    FileConnectorWriteResponse,
     FileUploadResponse,
 )
 from govoplan_core.db.session import get_session
@@ -35,8 +38,14 @@ from govoplan_files.backend.storage.connector_policy import (
 )
 from govoplan_files.backend.storage.files import (
     create_file_asset,
+    get_asset_for_user,
+    read_asset_bytes,
     sync_file_asset_from_source,
 )
+from govoplan_files.backend.storage.connector_spaces import (
+    get_connector_space_for_user,
+)
+from govoplan_files.backend.storage.connector_writes import write_connector_file
 
 
 from govoplan_files.backend.route_support import (
@@ -55,6 +64,112 @@ from govoplan_files.backend.route_support import (
 router = APIRouter(prefix="/files", tags=["files"])
 
 
+@router.post(
+    "/connector-spaces/{space_id}/write-back",
+    response_model=FileConnectorWriteResponse,
+)
+def write_back_connector_file(
+    space_id: str,
+    payload: FileConnectorWriteRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:connector:write")),
+):
+    try:
+        space = get_connector_space_for_user(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            space_id=space_id,
+            is_admin=_is_admin(principal),
+        )
+        if space.read_only:
+            raise FileStorageError("This connector space is read-only")
+        profile = _visible_connector_profile(
+            session, principal, space.connector_profile_id
+        )
+        requested_path = normalize_connector_browse_path(payload.remote_path)
+        remote_path = normalize_connector_browse_path(
+            "/".join(part for part in (space.remote_path, requested_path) if part)
+        )
+        decision = connector_policy_decision(
+            ConnectorAccessRequest(
+                connector_id=profile.id,
+                credential_id=profile.credential_profile_id,
+                provider=profile.provider,
+                external_path=remote_path,
+                external_url=connector_effective_endpoint_url(
+                    provider=profile.provider,
+                    endpoint_url=profile.endpoint_url,
+                    metadata=profile.metadata,
+                ),
+                operation="write",
+            ),
+            profile.policy_sources,
+        )
+        if not decision.allowed:
+            raise ConnectorPolicyDenied(decision)
+        asset = get_asset_for_user(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            asset_id=payload.file_id,
+            require_write=True,
+            is_admin=_is_admin(principal),
+        )
+        data, version, blob = read_asset_bytes(session, asset)
+        connector_library_id = space.library_id
+        content_type = blob.content_type
+        # Close the read snapshot before the independent recovery transaction
+        # records authority for the external effect. This avoids upgrading an
+        # older SQLite read snapshot after the ledger commit.
+        session.commit()
+        result = write_connector_file(
+            profile,
+            tenant_id=principal.tenant_id,
+            library_id=connector_library_id,
+            remote_path=remote_path,
+            data=data,
+            content_type=content_type,
+            expected_revision=payload.expected_revision,
+            idempotency_key=payload.idempotency_key,
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="files.connector.written",
+            object_type="file",
+            object_id=asset.id,
+            details={
+                "file_version_id": version.id,
+                "file_blob_id": blob.id,
+                "checksum_sha256": blob.checksum_sha256,
+                "connector_space_id": space.id,
+                "connector_profile_id": profile.id,
+                "remote_path": remote_path,
+                "recovery_operation_id": result.recovery_operation_id,
+                "recovery_status": result.status,
+                "revision": result.revision,
+            },
+        )
+        session.commit()
+        return FileConnectorWriteResponse(
+            recovery_operation_id=result.recovery_operation_id,
+            status=result.status,
+            replayed=result.replayed,
+            provider=result.provider,
+            remote_path=result.remote_path,
+            revision=result.revision,
+            checksum_sha256=result.checksum_sha256,
+            size_bytes=result.size_bytes,
+        )
+    except ConnectorPolicyDenied as exc:
+        session.rollback()
+        raise _connector_policy_error(exc) from exc
+    except (FileStorageError, ConnectorBrowseError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
 @router.post(
     "/connectors/profiles/{profile_id}/import", response_model=FileUploadResponse
 )

src/govoplan_files/backend/routes/lifecycle.py

@@ -0,0 +1,297 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal, require_scope
+from govoplan_core.audit.logging import audit_from_principal
+from govoplan_core.db.session import get_session
+from govoplan_files.backend.route_support import (
+    _asset_response,
+    _connector_space_response,
+    _http_error,
+    _is_admin,
+)
+from govoplan_files.backend.schemas import (
+    FileAssetResponse,
+    FileBlobGcRequest,
+    FileBlobGcResponse,
+    FileConnectorSpaceResponse,
+    FileFolderDeleteRequest,
+    FileFolderRestoreResponse,
+    FileLifecycleUpdateRequest,
+    FilePurgeExecuteRequest,
+    FilePurgePreviewItem,
+    FilePurgePreviewRequest,
+    FilePurgePreviewResponse,
+    FilePurgeResponse,
+    FileRestoreResponse,
+)
+from govoplan_files.backend.storage.common import FileStorageError
+from govoplan_files.backend.storage.lifecycle import (
+    execute_asset_purge,
+    garbage_collect_unreferenced_blobs,
+    get_asset_for_lifecycle,
+    preview_asset_purge,
+    restore_asset,
+    restore_connector_space,
+    restore_folder,
+    set_asset_lifecycle,
+)
+
+
+router = APIRouter(prefix="/files", tags=["files-lifecycle"])
+
+
+@router.patch("/{file_id}/lifecycle", response_model=FileAssetResponse)
+def update_file_lifecycle(
+    file_id: str,
+    payload: FileLifecycleUpdateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:retention")),
+):
+    try:
+        asset = get_asset_for_lifecycle(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            asset_id=file_id,
+            is_admin=_is_admin(principal),
+        )
+        set_asset_lifecycle(
+            session,
+            asset,
+            retained_until=payload.retained_until,
+            legal_hold=payload.legal_hold,
+            reason=payload.reason,
+            expected_revision=payload.expected_revision,
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="files.lifecycle.updated",
+            object_type="file",
+            object_id=asset.id,
+            details={
+                "retained_until": (
+                    asset.retained_until.isoformat() if asset.retained_until else None
+                ),
+                "legal_hold": asset.legal_hold,
+                "lifecycle_revision": asset.lifecycle_revision,
+                "reason": asset.lifecycle_reason,
+            },
+        )
+        session.commit()
+        return _asset_response(session, asset, include_shares=True)
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc, not_found=True) from exc
+
+
+@router.post("/assets/{file_id}/restore", response_model=FileRestoreResponse)
+def restore_file(
+    file_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:restore")),
+):
+    try:
+        asset = get_asset_for_lifecycle(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            asset_id=file_id,
+            is_admin=_is_admin(principal),
+        )
+        changed = restore_asset(session, asset)
+        if changed:
+            audit_from_principal(
+                session,
+                principal,
+                action="files.file.restored",
+                object_type="file",
+                object_id=asset.id,
+                details={"display_path": asset.display_path},
+            )
+        session.commit()
+        return FileRestoreResponse(restored_count=1 if changed else 0)
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc, not_found=True) from exc
+
+
+@router.post("/folders/restore", response_model=FileFolderRestoreResponse)
+def restore_file_folder(
+    payload: FileFolderDeleteRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:restore")),
+):
+    try:
+        restored_folders, restored_files = restore_folder(
+            session,
+            tenant_id=principal.tenant_id,
+            owner_type=payload.owner_type,
+            owner_id=payload.owner_id,
+            user_id=principal.user.id,
+            path=payload.path,
+            recursive=payload.recursive,
+            is_admin=_is_admin(principal),
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="files.folder.restored",
+            object_type="file_folder",
+            object_id=payload.path,
+            details={
+                "restored_folders": restored_folders,
+                "restored_files": restored_files,
+            },
+        )
+        session.commit()
+        return FileFolderRestoreResponse(
+            restored_folders=restored_folders, restored_files=restored_files
+        )
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc, not_found=True) from exc
+
+
+@router.post(
+    "/connector-spaces/{space_id}/restore",
+    response_model=FileConnectorSpaceResponse,
+)
+def restore_file_connector_space(
+    space_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:restore")),
+):
+    try:
+        space = restore_connector_space(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            space_id=space_id,
+            is_admin=_is_admin(principal),
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="files.connector_space.restored",
+            object_type="file_connector_space",
+            object_id=space.id,
+            details={"label": space.label},
+        )
+        session.commit()
+        return _connector_space_response(space)
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc, not_found=True) from exc
+
+
+@router.post("/purge/preview", response_model=FilePurgePreviewResponse)
+def preview_file_purge(
+    payload: FilePurgePreviewRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:purge")),
+):
+    try:
+        preview = preview_asset_purge(
+            session, tenant_id=principal.tenant_id, file_ids=payload.file_ids
+        )
+        return FilePurgePreviewResponse(
+            preview_sha256=preview.preview_sha256,
+            eligible=preview.eligible,
+            items=[
+                FilePurgePreviewItem(
+                    file_id=item.file_id,
+                    filename=item.filename,
+                    lifecycle_revision=item.lifecycle_revision,
+                    deleted_at=(item.deleted_at.isoformat() if item.deleted_at else None),
+                    retained_until=(
+                        item.retained_until.isoformat() if item.retained_until else None
+                    ),
+                    legal_hold=item.legal_hold,
+                    blockers=list(item.blockers),
+                    blob_ids=list(item.blob_ids),
+                )
+                for item in preview.items
+            ],
+        )
+    except FileStorageError as exc:
+        raise _http_error(exc) from exc
+
+
+@router.post("/purge/execute", response_model=FilePurgeResponse)
+def execute_file_purge(
+    payload: FilePurgeExecuteRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:purge")),
+):
+    try:
+        result = execute_asset_purge(
+            session,
+            tenant_id=principal.tenant_id,
+            file_ids=payload.file_ids,
+            preview_sha256=payload.preview_sha256,
+            idempotency_key=payload.idempotency_key,
+            approval_reference=payload.approval_reference,
+            before_commit=lambda assets: audit_from_principal(
+                session,
+                principal,
+                action="files.files.purged",
+                object_type="file_batch",
+                object_id=payload.preview_sha256,
+                details={
+                    "file_ids": [asset.id for asset in assets],
+                    "approval_reference": payload.approval_reference,
+                    "preview_sha256": payload.preview_sha256,
+                },
+            ),
+        )
+        return FilePurgeResponse(
+            recovery_operation_id=result.recovery_operation_id,
+            status=result.status,
+            replayed=result.replayed,
+            purged_files=result.purged_files,
+            released_blobs=result.released_blobs,
+        )
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.post("/purge/blobs", response_model=FileBlobGcResponse)
+def garbage_collect_file_blobs(
+    payload: FileBlobGcRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:purge")),
+):
+    try:
+        result = garbage_collect_unreferenced_blobs(
+            session,
+            tenant_id=principal.tenant_id,
+            limit=payload.limit,
+            approval_reference=payload.approval_reference,
+        )
+        audit_from_principal(
+            session,
+            principal,
+            action="files.blobs.garbage_collected",
+            object_type="file_blob_batch",
+            object_id=principal.tenant_id,
+            details={
+                "inspected_blobs": result.inspected_blobs,
+                "deleted_blobs": result.deleted_blobs,
+                "unresolved_operation_ids": list(result.unresolved_operation_ids),
+                "approval_reference": payload.approval_reference,
+            },
+            commit=True,
+        )
+        return FileBlobGcResponse(
+            inspected_blobs=result.inspected_blobs,
+            deleted_blobs=result.deleted_blobs,
+            unresolved_operation_ids=list(result.unresolved_operation_ids),
+        )
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc) from exc

src/govoplan_files/backend/routes/spaces.py

@@ -27,6 +27,7 @@ from govoplan_files.backend.storage.connector_spaces import (
     list_connector_spaces_for_user,
     soft_delete_connector_space,
     update_connector_space,
+    validate_connector_space_write_mode,
 )
 from govoplan_files.backend.storage.connector_policy import (
     ConnectorPolicyDenied,
@@ -145,6 +146,7 @@ def create_file_connector_space(
         )
         if not decision.allowed:
             raise ConnectorPolicyDenied(decision)
+        validate_connector_space_write_mode(profile, read_only=payload.read_only)
         space = create_connector_space(
             session,
             tenant_id=principal.tenant_id,
@@ -156,6 +158,7 @@ def create_file_connector_space(
             library_id=payload.library_id,
             remote_path=payload.remote_path,
             sync_mode=payload.sync_mode,
+            read_only=payload.read_only,
             metadata=payload.metadata,
             is_admin=_is_admin(principal),
         )
@@ -201,7 +204,11 @@ def update_file_connector_space(
     except FileStorageError as exc:
         raise _http_error(exc, not_found=True) from exc
     try:
-        if payload.library_id is not None or payload.remote_path is not None:
+        if (
+            payload.library_id is not None
+            or payload.remote_path is not None
+            or payload.read_only is not None
+        ):
             profile = _visible_connector_profile(
                 session, principal, space.connector_profile_id
             )
@@ -217,6 +224,14 @@ def update_file_connector_space(
             )
             if not decision.allowed:
                 raise ConnectorPolicyDenied(decision)
+            validate_connector_space_write_mode(
+                profile,
+                read_only=(
+                    payload.read_only
+                    if payload.read_only is not None
+                    else space.read_only
+                ),
+            )
         update_connector_space(
             session,
             space,
@@ -225,6 +240,7 @@ def update_file_connector_space(
             library_id=payload.library_id,
             remote_path=payload.remote_path,
             sync_mode=payload.sync_mode,
+            read_only=payload.read_only,
             is_active=payload.is_active,
             metadata=payload.metadata,
             is_admin=_is_admin(principal),

src/govoplan_files/backend/schemas.py

@@ -37,6 +37,7 @@ class FileConnectorSpaceCreateRequest(BaseModel):
     library_id: str | None = None
     remote_path: str = ""
     sync_mode: Literal["manual"] = "manual"
+    read_only: bool = True
     metadata: dict[str, Any] = Field(default_factory=dict)
 
 
@@ -45,6 +46,7 @@ class FileConnectorSpaceUpdateRequest(BaseModel):
     library_id: str | None = None
     remote_path: str | None = None
     sync_mode: Literal["manual"] | None = None
+    read_only: bool | None = None
     is_active: bool | None = None
     metadata: dict[str, Any] | None = None
 
@@ -192,6 +194,10 @@ class FileAssetResponse(BaseModel):
     created_at: str
     updated_at: str
     deleted_at: str | None = None
+    retained_until: str | None = None
+    legal_hold: bool = False
+    lifecycle_revision: int = 1
+    lifecycle_reason: str | None = None
     audit_relevant: bool = False
     metadata: dict[str, Any] | None = None
     source_provenance: FileSourceProvenance | None = None
@@ -477,6 +483,7 @@ class FileConnectorProviderResponse(BaseModel):
     installed: bool
     browse_supported: bool
     import_supported: bool
+    write_supported: bool = False
     optional_dependency: str | None = None
     permission_model: str
     sync_strategy: str
@@ -535,6 +542,24 @@ class FileConnectorSyncResponse(BaseModel):
     current_version_id: str
 
 
+class FileConnectorWriteRequest(BaseModel):
+    file_id: str
+    remote_path: str
+    idempotency_key: str = Field(min_length=8, max_length=160)
+    expected_revision: str | None = Field(default=None, max_length=500)
+
+
+class FileConnectorWriteResponse(BaseModel):
+    recovery_operation_id: str
+    status: str
+    replayed: bool = False
+    provider: str
+    remote_path: str
+    revision: str | None = None
+    checksum_sha256: str
+    size_bytes: int
+
+
 class BulkDeleteRequest(BaseModel):
     file_ids: list[str]
 
@@ -543,6 +568,69 @@ class BulkDeleteResponse(BaseModel):
     deleted_count: int
 
 
+class FileLifecycleUpdateRequest(BaseModel):
+    retained_until: datetime | None = None
+    legal_hold: bool = False
+    reason: str = Field(min_length=1, max_length=500)
+    expected_revision: int = Field(ge=1)
+
+
+class FileRestoreResponse(BaseModel):
+    restored_count: int
+
+
+class FileFolderRestoreResponse(BaseModel):
+    restored_folders: int
+    restored_files: int
+
+
+class FilePurgePreviewRequest(BaseModel):
+    file_ids: list[str] = Field(min_length=1, max_length=100)
+
+
+class FilePurgePreviewItem(BaseModel):
+    file_id: str
+    filename: str
+    lifecycle_revision: int
+    deleted_at: str | None = None
+    retained_until: str | None = None
+    legal_hold: bool = False
+    blockers: list[str] = Field(default_factory=list)
+    blob_ids: list[str] = Field(default_factory=list)
+
+
+class FilePurgePreviewResponse(BaseModel):
+    preview_sha256: str
+    eligible: bool
+    items: list[FilePurgePreviewItem]
+
+
+class FilePurgeExecuteRequest(FilePurgePreviewRequest):
+    preview_sha256: str = Field(min_length=64, max_length=64)
+    idempotency_key: str = Field(min_length=8, max_length=160)
+    approval_reference: str = Field(min_length=3, max_length=1000)
+    confirmation: Literal["PURGE"]
+
+
+class FilePurgeResponse(BaseModel):
+    recovery_operation_id: str
+    status: str
+    replayed: bool = False
+    purged_files: int = 0
+    released_blobs: int = 0
+
+
+class FileBlobGcRequest(BaseModel):
+    limit: int = Field(default=25, ge=1, le=100)
+    approval_reference: str = Field(min_length=3, max_length=1000)
+
+
+class FileBlobGcResponse(BaseModel):
+    inspected_blobs: int
+    deleted_blobs: int
+    unresolved_operation_ids: list[str] = Field(default_factory=list)
+
+
 class ConflictResolutionRequest(BaseModel):
     target_path: str
     action: Literal["overwrite", "rename", "skip"]

src/govoplan_files/backend/storage/connector_providers.py

@@ -32,6 +32,7 @@ class ConnectorProviderDescriptor:
             "installed": self.installed,
             "browse_supported": self.browse_supported,
             "import_supported": self.import_supported,
+            "write_supported": self.provider == "s3",
             "optional_dependency": self.optional_dependency,
             "permission_model": self.permission_model,
             "sync_strategy": self.sync_strategy,

src/govoplan_files/backend/storage/connector_spaces.py

@@ -14,6 +14,7 @@ from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
 
 
 SYNC_MODES = {"manual"}
+WRITABLE_PROVIDERS = {"s3"}
 
 
 def connector_space_owner_id(space: FileConnectorSpace) -> str:
@@ -40,6 +41,7 @@ def create_connector_space(
     ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)
     label = _normalize_label(label)
     sync_mode = _normalize_sync_mode(sync_mode)
+    validate_connector_space_write_mode(profile, read_only=read_only)
     remote_path = normalize_connector_browse_path(remote_path)
     library_id = _clean_optional(library_id)
 
@@ -145,6 +147,7 @@ def update_connector_space(
     library_id: str | None = None,
     remote_path: str | None = None,
     sync_mode: str | None = None,
+    read_only: bool | None = None,
     is_active: bool | None = None,
     metadata: Mapping[str, Any] | None = None,
     is_admin: bool = False,
@@ -178,6 +181,8 @@ def update_connector_space(
         space.remote_path = normalize_connector_browse_path(remote_path)
     if sync_mode is not None:
         space.sync_mode = _normalize_sync_mode(sync_mode)
+    if read_only is not None:
+        space.read_only = bool(read_only)
     if is_active is not None:
         space.is_active = bool(is_active)
     if metadata is not None:
@@ -187,6 +192,17 @@ def update_connector_space(
     return space
 
 
+def validate_connector_space_write_mode(
+    profile: ConnectorProfile, *, read_only: bool
+) -> None:
+    if read_only:
+        return
+    if profile.provider not in WRITABLE_PROVIDERS or "write" not in profile.capabilities:
+        raise FileStorageError(
+            "Two-way mode requires an S3 connection with the write capability enabled"
+        )
+
+
 def soft_delete_connector_space(
     session: Session,
     space: FileConnectorSpace,

src/govoplan_files/backend/storage/connector_writes.py

@@ -0,0 +1,438 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+from typing import Any, Mapping
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryPlan,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+from govoplan_core.db.session import get_database
+from govoplan_files.backend.storage.common import FileStorageError
+from govoplan_files.backend.storage.connector_browse import (
+    ConnectorBrowseError,
+    ConnectorBrowseUnsupported,
+    _clean,
+    _s3_bucket,
+    _s3_client,
+    _s3_object_key,
+    normalize_connector_browse_path,
+)
+from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
+
+
+@dataclass(frozen=True, slots=True)
+class ConnectorWriteResult:
+    recovery_operation_id: str
+    status: str
+    replayed: bool
+    provider: str
+    remote_path: str
+    revision: str | None
+    checksum_sha256: str
+    size_bytes: int
+
+
+@dataclass(frozen=True, slots=True)
+class _HeadProbe:
+    observed: Mapping[str, Any] | None
+    verified: bool
+    exception_type: str | None = None
+
+
+def write_connector_file(
+    profile: ConnectorProfile,
+    *,
+    tenant_id: str,
+    library_id: str | None,
+    remote_path: str,
+    data: bytes,
+    content_type: str | None,
+    expected_revision: str | None,
+    idempotency_key: str,
+) -> ConnectorWriteResult:
+    if profile.provider != "s3" or "write" not in profile.capabilities:
+        raise FileStorageError(
+            "Remote writes require an S3 connection with the write capability enabled"
+        )
+    try:
+        normalized_path = normalize_connector_browse_path(remote_path)
+    except ConnectorBrowseError as exc:
+        raise FileStorageError(str(exc)) from exc
+    if not normalized_path:
+        raise FileStorageError("Remote write requires an object path")
+    try:
+        bucket = _s3_bucket(profile, library_id)
+        key = _s3_object_key(profile, normalized_path)
+    except ConnectorBrowseError as exc:
+        raise FileStorageError(str(exc)) from exc
+    if not bucket or not key:
+        raise FileStorageError("S3 remote write requires a bucket and object path")
+    checksum = hashlib.sha256(data).hexdigest()
+    target_digest = hashlib.sha256(f"{bucket}:{key}".encode("utf-8")).hexdigest()
+    request = {
+        "tenant_id": tenant_id,
+        "connector_profile_id": profile.id,
+        "provider": profile.provider,
+        "remote_target_sha256": target_digest,
+        "content_sha256": checksum,
+        "size_bytes": len(data),
+        "expected_revision": expected_revision,
+    }
+    try:
+        started = begin_durable_recovery_operation(
+            get_database().SessionLocal,
+            identity=process_runtime_identity(),
+            module_id="files",
+            operation_type="connector-s3-write",
+            idempotency_key=f"files-connector-write:{idempotency_key}",
+            request=request,
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.FORWARD_RECOVERY,
+                preconditions=(
+                    "the connector space is explicitly configured for two-way writes",
+                    "the connection and inherited policy allow the exact remote path",
+                    "a conditional create or expected remote revision prevents blind overwrite",
+                ),
+                forward_recovery_steps=(
+                    "inspect provider metadata for the request and content digests",
+                    "resolve success only for an exact digest match or retry after verified absence",
+                ),
+                verification_steps=(
+                    "read S3 object metadata after the write",
+                    "match the recorded request and content digests without downloading content",
+                ),
+            ),
+            precondition_evidence={
+                "remote_target_sha256": target_digest,
+                "content_sha256": checksum,
+                "size_bytes": len(data),
+                "conditional_write": True,
+            },
+            lease_resource_key=f"files:connector:{tenant_id}:{profile.id}:{target_digest[:40]}",
+            lease_ttl_seconds=15 * 60,
+            resource_type="file_connector_object",
+            resource_id=target_digest,
+            metadata={
+                "resources": ["postgresql", "s3-connector"],
+                "provider": "s3",
+                "connector_profile_id": profile.id,
+            },
+            block_unresolved_resource=True,
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        raise FileStorageError(
+            "This remote object is owned by another write or unresolved recovery operation"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError, ValueError) as exc:
+        raise FileStorageError(
+            "The Files recovery ledger is unavailable; the remote object was not changed"
+        ) from exc
+    if started.replayed or started.operation is None:
+        return ConnectorWriteResult(
+            recovery_operation_id=started.operation_id,
+            status=started.status,
+            replayed=True,
+            provider="s3",
+            remote_path=normalized_path,
+            revision=None,
+            checksum_sha256=checksum,
+            size_bytes=len(data),
+        )
+    operation = started.operation
+    try:
+        client = _s3_client(profile)
+    except (ConnectorBrowseError, ConnectorBrowseUnsupported) as exc:
+        operation.reject(
+            summary="The S3 client was unavailable before the remote effect",
+            evidence=_verified(
+                {"external_effect_started": False}, exception_type=type(exc).__name__
+            ),
+        )
+        raise FileStorageError(str(exc)) from exc
+    try:
+        try:
+            before = _head_object(client, bucket=bucket, key=key)
+        except Exception as exc:
+            operation.reject(
+                summary="Remote preconditions could not be inspected before the effect",
+                evidence=_verified(
+                    {"external_effect_started": False},
+                    exception_type=type(exc).__name__,
+                ),
+            )
+            raise FileStorageError("S3 remote revision lookup failed") from exc
+        expected = _normalize_revision(expected_revision)
+        if before is not None and expected is None:
+            operation.reject(
+                summary="A blind remote overwrite was rejected",
+                evidence=_verified(
+                    {"remote_object_exists": True, "expected_revision_supplied": False}
+                ),
+            )
+            raise FileStorageError(
+                "The remote object already exists; reload it and supply its expected revision"
+            )
+        if expected is not None and (
+            before is None or expected not in _observed_revisions(before)
+        ):
+            operation.reject(
+                summary="The expected remote revision did not match",
+                evidence=_verified(
+                    {
+                        "remote_object_exists": before is not None,
+                        "expected_revision_matches": False,
+                    }
+                ),
+            )
+            raise FileStorageError("The remote object changed; reload before writing")
+        params: dict[str, object] = {
+            "Bucket": bucket,
+            "Key": key,
+            "Body": data,
+            "Metadata": {
+                "govoplan-sha256": checksum,
+                "govoplan-request-id": started.operation_id,
+            },
+        }
+        if content_type:
+            params["ContentType"] = content_type
+        if before is None:
+            params["IfNoneMatch"] = "*"
+        else:
+            params["IfMatch"] = str(before.get("ETag") or expected_revision or "")
+        try:
+            client.put_object(**params)
+        except Exception as exc:
+            probe = _probe_head_object(client, bucket=bucket, key=key)
+            if probe.verified and _matches_effect(
+                probe.observed,
+                checksum=checksum,
+                operation_id=started.operation_id,
+            ):
+                operation.succeed(
+                    evidence=_success_evidence(
+                        probe.observed,
+                        checksum=checksum,
+                        operation_id=started.operation_id,
+                    )
+                )
+                return _result(
+                    started.operation_id,
+                    normalized_path,
+                    checksum,
+                    len(data),
+                    probe.observed,
+                )
+            operation.unresolved(
+                status=RecoveryStatus.OUTCOME_UNKNOWN,
+                summary="The S3 write outcome requires reconciliation",
+                evidence={
+                    "remote_target_sha256": target_digest,
+                    "observed": _public_observation(probe),
+                    "exception_type": type(exc).__name__,
+                },
+                failure_summary="Inspect the remote object metadata before retrying this path",
+            )
+            return ConnectorWriteResult(
+                recovery_operation_id=started.operation_id,
+                status=RecoveryStatus.OUTCOME_UNKNOWN.value,
+                replayed=False,
+                provider="s3",
+                remote_path=normalized_path,
+                revision=_revision(probe.observed),
+                checksum_sha256=checksum,
+                size_bytes=len(data),
+            )
+        probe = _probe_head_object(client, bucket=bucket, key=key)
+        if not probe.verified:
+            operation.unresolved(
+                status=RecoveryStatus.OUTCOME_UNKNOWN,
+                summary="The S3 write returned but provider evidence could not be queried",
+                evidence={
+                    "remote_target_sha256": target_digest,
+                    "observed": _public_observation(probe),
+                },
+                failure_summary="Inspect the remote object metadata before retrying this path",
+            )
+            return ConnectorWriteResult(
+                recovery_operation_id=started.operation_id,
+                status=RecoveryStatus.OUTCOME_UNKNOWN.value,
+                replayed=False,
+                provider="s3",
+                remote_path=normalized_path,
+                revision=None,
+                checksum_sha256=checksum,
+                size_bytes=len(data),
+            )
+        observed = probe.observed
+        if not _matches_effect(
+            observed, checksum=checksum, operation_id=started.operation_id
+        ):
+            operation.unresolved(
+                status=RecoveryStatus.RECOVERY_REQUIRED,
+                summary="The S3 write returned but exact provider evidence did not match",
+                evidence={
+                    "remote_target_sha256": target_digest,
+                    "observed": _public_observation(probe),
+                },
+                failure_summary="Reconcile the remote object before another write",
+            )
+            return ConnectorWriteResult(
+                recovery_operation_id=started.operation_id,
+                status=RecoveryStatus.RECOVERY_REQUIRED.value,
+                replayed=False,
+                provider="s3",
+                remote_path=normalized_path,
+                revision=_revision(observed),
+                checksum_sha256=checksum,
+                size_bytes=len(data),
+            )
+        operation.succeed(
+            evidence=_success_evidence(
+                observed, checksum=checksum, operation_id=started.operation_id
+            )
+        )
+        return _result(
+            started.operation_id,
+            normalized_path,
+            checksum,
+            len(data),
+            observed,
+        )
+    finally:
+        close = getattr(client, "close", None)
+        if callable(close):
+            close()
+
+
+def _head_object(client: Any, *, bucket: str, key: str) -> Mapping[str, Any] | None:
+    try:
+        response = client.head_object(Bucket=bucket, Key=key)
+    except Exception as exc:
+        if _is_not_found(exc):
+            return None
+        raise
+    if not isinstance(response, Mapping):
+        raise FileStorageError("S3 connector returned invalid object metadata")
+    return response
+
+
+def _probe_head_object(client: Any, *, bucket: str, key: str) -> _HeadProbe:
+    try:
+        return _HeadProbe(
+            observed=_head_object(client, bucket=bucket, key=key), verified=True
+        )
+    except Exception as exc:
+        return _HeadProbe(
+            observed=None,
+            verified=False,
+            exception_type=type(exc).__name__,
+        )
+
+
+def _is_not_found(exc: Exception) -> bool:
+    response = getattr(exc, "response", None)
+    if not isinstance(response, Mapping):
+        return False
+    error = response.get("Error")
+    code = error.get("Code") if isinstance(error, Mapping) else None
+    status = response.get("ResponseMetadata")
+    http_status = status.get("HTTPStatusCode") if isinstance(status, Mapping) else None
+    return str(code).casefold() in {"404", "nosuchkey", "notfound"} or http_status == 404
+
+
+def _metadata(observed: Mapping[str, Any] | None) -> Mapping[str, Any]:
+    value = observed.get("Metadata") if observed else None
+    return value if isinstance(value, Mapping) else {}
+
+
+def _matches_effect(
+    observed: Mapping[str, Any] | None, *, checksum: str, operation_id: str
+) -> bool:
+    metadata = _metadata(observed)
+    return bool(
+        observed is not None
+        and _clean(metadata.get("govoplan-sha256")) == checksum
+        and _clean(metadata.get("govoplan-request-id")) == operation_id
+    )
+
+
+def _revision(observed: Mapping[str, Any] | None) -> str | None:
+    if observed is None:
+        return None
+    return _clean(observed.get("VersionId") or observed.get("ETag"))
+
+
+def _normalize_revision(value: str | None) -> str | None:
+    cleaned = _clean(value)
+    return cleaned.strip('"') if cleaned else None
+
+
+def _observed_revisions(observed: Mapping[str, Any]) -> set[str]:
+    return {
+        normalized
+        for value in (observed.get("VersionId"), observed.get("ETag"))
+        if (normalized := _normalize_revision(_clean(value))) is not None
+    }
+
+
+def _success_evidence(
+    observed: Mapping[str, Any] | None, *, checksum: str, operation_id: str
+) -> dict[str, object]:
+    return _verified(
+        {
+            "remote_object_present": observed is not None,
+            "content_digest_matches": _clean(_metadata(observed).get("govoplan-sha256")) == checksum,
+            "request_marker_matches": _clean(_metadata(observed).get("govoplan-request-id")) == operation_id,
+        },
+        revision=_revision(observed),
+        content_sha256=checksum,
+    )
+
+
+def _public_observation(probe: _HeadProbe) -> dict[str, object]:
+    observed = probe.observed
+    return {
+        "probe_verified": probe.verified,
+        "probe_exception_type": probe.exception_type,
+        "present": observed is not None,
+        "revision": _revision(observed),
+        "has_content_digest": bool(_clean(_metadata(observed).get("govoplan-sha256"))),
+        "has_request_marker": bool(_clean(_metadata(observed).get("govoplan-request-id"))),
+    }
+
+
+def _verified(checks: dict[str, object], **details: object) -> dict[str, object]:
+    return {"verified": True, "checks": checks, **details}
+
+
+def _result(
+    operation_id: str,
+    remote_path: str,
+    checksum: str,
+    size_bytes: int,
+    observed: Mapping[str, Any] | None,
+) -> ConnectorWriteResult:
+    return ConnectorWriteResult(
+        recovery_operation_id=operation_id,
+        status=RecoveryStatus.SUCCEEDED.value,
+        replayed=False,
+        provider="s3",
+        remote_path=remote_path,
+        revision=_revision(observed),
+        checksum_sha256=checksum,
+        size_bytes=size_bytes,
+    )
+
+
+__all__ = ["ConnectorWriteResult", "write_connector_file"]

src/govoplan_files/backend/storage/files.py

@@ -84,6 +84,7 @@ def _get_or_create_blob(
     blob = (
         session.query(FileBlob)
         .filter(FileBlob.tenant_id == tenant_id, FileBlob.checksum_sha256 == checksum, FileBlob.size_bytes == size, FileBlob.protection_discriminator == protection_discriminator)
+        .with_for_update()
         .one_or_none()
     )
     if blob:

src/govoplan_files/backend/storage/lifecycle.py

@@ -0,0 +1,796 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+import hashlib
+import json
+from typing import Callable, Iterable
+
+from sqlalchemy import func, inspect, or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryPlan,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+from govoplan_core.db.session import get_database
+from govoplan_files.backend.db.models import (
+    CampaignAttachmentUse,
+    FileAsset,
+    FileBlob,
+    FileConnectorSpace,
+    FileFolder,
+    FileFormEvidenceGrant,
+    FileShare,
+    FileVersion,
+)
+from govoplan_files.backend.storage.access import ensure_owner_access
+from govoplan_files.backend.storage.backends import (
+    StorageBackendError,
+    get_storage_backend,
+)
+from govoplan_files.backend.storage.common import FileStorageError, utcnow
+from govoplan_files.backend.storage.connector_spaces import connector_space_owner_id
+from govoplan_files.backend.storage.paths import normalize_folder
+from govoplan_files.backend.storage.share_state import effective_file_share_clause
+
+
+@dataclass(frozen=True, slots=True)
+class PurgePreviewItem:
+    file_id: str
+    filename: str
+    lifecycle_revision: int
+    deleted_at: datetime | None
+    retained_until: datetime | None
+    legal_hold: bool
+    blockers: tuple[str, ...]
+    blob_ids: tuple[str, ...]
+
+    def digest_payload(self) -> dict[str, object]:
+        return {
+            "file_id": self.file_id,
+            "filename": self.filename,
+            "lifecycle_revision": self.lifecycle_revision,
+            "deleted_at": _iso(self.deleted_at),
+            "retained_until": _iso(self.retained_until),
+            "legal_hold": self.legal_hold,
+            "blockers": list(self.blockers),
+            "blob_ids": list(self.blob_ids),
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class PurgePreview:
+    preview_sha256: str
+    items: tuple[PurgePreviewItem, ...]
+
+    @property
+    def eligible(self) -> bool:
+        return all(not item.blockers for item in self.items)
+
+
+@dataclass(frozen=True, slots=True)
+class PurgeResult:
+    recovery_operation_id: str
+    status: str
+    replayed: bool
+    purged_files: int
+    released_blobs: int
+
+
+@dataclass(frozen=True, slots=True)
+class BlobGcResult:
+    inspected_blobs: int
+    deleted_blobs: int
+    unresolved_operation_ids: tuple[str, ...]
+
+
+def set_asset_lifecycle(
+    session: Session,
+    asset: FileAsset,
+    *,
+    retained_until: datetime | None,
+    legal_hold: bool,
+    reason: str,
+    expected_revision: int,
+) -> FileAsset:
+    if asset.lifecycle_revision != expected_revision:
+        raise FileStorageError("File lifecycle settings changed; reload before saving")
+    asset.retained_until = retained_until
+    asset.legal_hold = bool(legal_hold)
+    asset.lifecycle_reason = reason.strip()
+    asset.lifecycle_revision += 1
+    session.add(asset)
+    session.flush()
+    return asset
+
+
+def get_asset_for_lifecycle(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    asset_id: str,
+    is_admin: bool,
+) -> FileAsset:
+    asset = session.get(FileAsset, asset_id)
+    if asset is None or asset.tenant_id != tenant_id:
+        raise FileStorageError("File not found")
+    ensure_owner_access(
+        session,
+        tenant_id=tenant_id,
+        owner_type=asset.owner_type,
+        owner_id=_asset_owner_id(asset),
+        user_id=user_id,
+        is_admin=is_admin,
+    )
+    return asset
+
+
+def restore_asset(session: Session, asset: FileAsset) -> bool:
+    if asset.deleted_at is None:
+        return False
+    collision = _asset_owner_query(session, asset).filter(
+        FileAsset.id != asset.id,
+        FileAsset.display_path == asset.display_path,
+        FileAsset.deleted_at.is_(None),
+    ).first()
+    if collision is not None:
+        raise FileStorageError(
+            "The file cannot be restored because its path is already in use"
+        )
+    asset.deleted_at = None
+    asset.lifecycle_revision += 1
+    session.add(asset)
+    session.flush()
+    return True
+
+
+def restore_folder(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    user_id: str,
+    path: str,
+    recursive: bool,
+    is_admin: bool,
+) -> tuple[int, int]:
+    ensure_owner_access(
+        session,
+        tenant_id=tenant_id,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        user_id=user_id,
+        is_admin=is_admin,
+    )
+    normalized = normalize_folder(path)
+    if not normalized:
+        raise FileStorageError("Folder path is required")
+    prefix = f"{normalized}/"
+    folders = _folder_owner_query(
+        session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id
+    ).filter(FileFolder.deleted_at.is_not(None))
+    assets = _asset_owner_query_by_values(
+        session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id
+    ).filter(FileAsset.deleted_at.is_not(None))
+    if recursive:
+        folders = folders.filter(
+            (FileFolder.path == normalized) | FileFolder.path.like(f"{prefix}%")
+        )
+        assets = assets.filter(FileAsset.display_path.like(f"{prefix}%"))
+    else:
+        folders = folders.filter(FileFolder.path == normalized)
+        assets = assets.filter(False)
+    folder_rows = folders.order_by(FileFolder.path.asc()).all()
+    asset_rows = assets.order_by(FileAsset.display_path.asc()).all()
+    if not folder_rows:
+        raise FileStorageError("Deleted folder not found")
+    for folder in folder_rows:
+        collision = _folder_owner_query(
+            session,
+            tenant_id=tenant_id,
+            owner_type=owner_type,
+            owner_id=owner_id,
+        ).filter(
+            FileFolder.id != folder.id,
+            FileFolder.path == folder.path,
+            FileFolder.deleted_at.is_(None),
+        ).first()
+        if collision is not None:
+            raise FileStorageError(
+                f"The folder cannot be restored because its path is in use: {folder.path}"
+            )
+    for asset in asset_rows:
+        collision = _asset_owner_query(session, asset).filter(
+            FileAsset.id != asset.id,
+            FileAsset.display_path == asset.display_path,
+            FileAsset.deleted_at.is_(None),
+        ).first()
+        if collision is not None:
+            raise FileStorageError(
+                f"A file cannot be restored because its path is in use: {asset.display_path}"
+            )
+    for folder in folder_rows:
+        folder.deleted_at = None
+        session.add(folder)
+    for asset in asset_rows:
+        asset.deleted_at = None
+        asset.lifecycle_revision += 1
+        session.add(asset)
+    session.flush()
+    return len(folder_rows), len(asset_rows)
+
+
+def restore_connector_space(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    space_id: str,
+    is_admin: bool,
+) -> FileConnectorSpace:
+    space = session.get(FileConnectorSpace, space_id)
+    if space is None or space.tenant_id != tenant_id or space.deleted_at is None:
+        raise FileStorageError("Deleted connector space not found")
+    ensure_owner_access(
+        session,
+        tenant_id=tenant_id,
+        owner_type=space.owner_type,
+        owner_id=connector_space_owner_id(space),
+        user_id=user_id,
+        is_admin=is_admin,
+    )
+    collision = _connector_space_owner_query(session, space).filter(
+        FileConnectorSpace.id != space.id,
+        FileConnectorSpace.label == space.label,
+        FileConnectorSpace.deleted_at.is_(None),
+    ).first()
+    if collision is not None:
+        raise FileStorageError(
+            "The connector space cannot be restored because its label is in use"
+        )
+    space.deleted_at = None
+    space.is_active = True
+    session.add(space)
+    session.flush()
+    return space
+
+
+def preview_asset_purge(
+    session: Session, *, tenant_id: str, file_ids: Iterable[str]
+) -> PurgePreview:
+    normalized_ids = sorted(set(str(value).strip() for value in file_ids if str(value).strip()))
+    if not normalized_ids or len(normalized_ids) > 100:
+        raise FileStorageError("Purge preview must contain between 1 and 100 files")
+    assets = (
+        session.query(FileAsset)
+        .filter(FileAsset.tenant_id == tenant_id, FileAsset.id.in_(normalized_ids))
+        .order_by(FileAsset.id.asc())
+        .all()
+    )
+    by_id = {asset.id: asset for asset in assets}
+    items: list[PurgePreviewItem] = []
+    for file_id in normalized_ids:
+        asset = by_id.get(file_id)
+        if asset is None:
+            items.append(
+                PurgePreviewItem(
+                    file_id=file_id,
+                    filename="",
+                    lifecycle_revision=0,
+                    deleted_at=None,
+                    retained_until=None,
+                    legal_hold=False,
+                    blockers=("not_found",),
+                    blob_ids=(),
+                )
+            )
+            continue
+        blockers = _asset_purge_blockers(session, asset)
+        blob_ids = tuple(
+            sorted(
+                row[0]
+                for row in session.query(FileVersion.blob_id)
+                .filter(FileVersion.file_asset_id == asset.id)
+                .distinct()
+                .all()
+            )
+        )
+        items.append(
+            PurgePreviewItem(
+                file_id=asset.id,
+                filename=asset.filename,
+                lifecycle_revision=asset.lifecycle_revision,
+                deleted_at=asset.deleted_at,
+                retained_until=asset.retained_until,
+                legal_hold=asset.legal_hold,
+                blockers=tuple(blockers),
+                blob_ids=blob_ids,
+            )
+        )
+    payload = [item.digest_payload() for item in items]
+    digest = hashlib.sha256(
+        json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    return PurgePreview(preview_sha256=digest, items=tuple(items))
+
+
+def execute_asset_purge(
+    session: Session,
+    *,
+    tenant_id: str,
+    file_ids: Iterable[str],
+    preview_sha256: str,
+    idempotency_key: str,
+    approval_reference: str,
+    before_commit: Callable[[list[FileAsset]], None] | None = None,
+) -> PurgeResult:
+    normalized_ids = sorted(set(str(value).strip() for value in file_ids if str(value).strip()))
+    if not normalized_ids or len(normalized_ids) > 100:
+        raise FileStorageError("Purge must contain between 1 and 100 files")
+    request = {
+        "tenant_id": tenant_id,
+        "file_ids": normalized_ids,
+        "preview_sha256": preview_sha256,
+        "approval_reference": approval_reference,
+    }
+    try:
+        started = begin_durable_recovery_operation(
+            get_database().SessionLocal,
+            identity=process_runtime_identity(),
+            module_id="files",
+            operation_type="asset-hard-purge",
+            idempotency_key=f"files-asset-purge:{idempotency_key}",
+            request=request,
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.IRREVERSIBLE,
+                preconditions=(
+                    "the actor holds the dedicated Files purge permission",
+                    "the signed-off preview still matches current lifecycle state",
+                    "every target is soft-deleted and free of retention or legal-hold blockers",
+                ),
+                verification_steps=(
+                    "verify every target FileAsset and FileVersion row is absent",
+                    "recalculate retained FileBlob reference counts before later garbage collection",
+                ),
+                approval_reference=approval_reference,
+            ),
+            precondition_evidence={
+                "preview_sha256": preview_sha256,
+                "target_count": len(normalized_ids),
+            },
+            lease_resource_key=f"files:purge:{tenant_id}",
+            lease_ttl_seconds=15 * 60,
+            resource_type="file_asset_batch",
+            resource_id=preview_sha256,
+            metadata={"resources": ["postgresql"], "bounded_target_count": len(normalized_ids)},
+            block_unresolved_resource=True,
+        )
+    except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc:
+        raise FileStorageError(
+            "Another purge or unresolved retry owns the tenant purge fence"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError, ValueError) as exc:
+        raise FileStorageError("The Files recovery ledger is unavailable; nothing was purged") from exc
+    if started.replayed or started.operation is None:
+        return PurgeResult(
+            recovery_operation_id=started.operation_id,
+            status=started.status,
+            replayed=True,
+            purged_files=0,
+            released_blobs=0,
+        )
+    operation = started.operation
+    try:
+        preview = preview_asset_purge(session, tenant_id=tenant_id, file_ids=normalized_ids)
+        if preview.preview_sha256 != preview_sha256:
+            operation.reject(
+                summary="The purge preview became stale before execution",
+                evidence=_verified_evidence(
+                    {"preview_matches": False},
+                    expected_preview_sha256=preview_sha256,
+                    observed_preview_sha256=preview.preview_sha256,
+                ),
+            )
+            raise FileStorageError("The purge preview is stale; preview again")
+        if not preview.eligible:
+            operation.reject(
+                summary="The purge was blocked by current lifecycle policy",
+                evidence=_verified_evidence(
+                    {"lifecycle_policy_allows": False},
+                    blocked_files=[
+                        {"file_id": item.file_id, "blockers": list(item.blockers)}
+                        for item in preview.items
+                        if item.blockers
+                    ]
+                ),
+            )
+            raise FileStorageError("One or more files are not eligible for purge")
+        assets = (
+            session.query(FileAsset)
+            .filter(FileAsset.tenant_id == tenant_id, FileAsset.id.in_(normalized_ids))
+            .with_for_update()
+            .all()
+        )
+        blob_ids = sorted({blob_id for item in preview.items for blob_id in item.blob_ids})
+        if before_commit is not None:
+            before_commit(assets)
+        session.query(FileShare).filter(
+            FileShare.file_asset_id.in_(normalized_ids)
+        ).delete(synchronize_session=False)
+        session.query(FileVersion).filter(
+            FileVersion.file_asset_id.in_(normalized_ids)
+        ).delete(synchronize_session=False)
+        for asset in assets:
+            session.delete(asset)
+        session.flush()
+        released_blobs = 0
+        for blob_id in blob_ids:
+            references = int(
+                session.query(func.count(FileVersion.id))
+                .filter(FileVersion.blob_id == blob_id)
+                .scalar()
+                or 0
+            )
+            blob = session.get(FileBlob, blob_id)
+            if blob is not None:
+                blob.ref_count = references
+                session.add(blob)
+                if references == 0:
+                    released_blobs += 1
+        operation.commit_verified_success(
+            session,
+            evidence=_verified_evidence(
+                {
+                    "assets_absent": True,
+                    "versions_absent": True,
+                    "blob_reference_counts_recalculated": True,
+                },
+                purged_file_ids=normalized_ids,
+                purged_files=len(assets),
+                released_blobs=released_blobs,
+                preview_sha256=preview_sha256,
+            ),
+        )
+        return PurgeResult(
+            recovery_operation_id=started.operation_id,
+            status=RecoveryStatus.SUCCEEDED.value,
+            replayed=False,
+            purged_files=len(assets),
+            released_blobs=released_blobs,
+        )
+    except FileStorageError:
+        session.rollback()
+        if not operation.closed:
+            operation.release_unresolved()
+        raise
+    except Exception as exc:
+        session.rollback()
+        if not operation.closed:
+            operation.reject(
+                summary="The purge failed before its database transaction committed",
+                evidence=_verified_evidence(
+                    {"database_transaction_committed": False},
+                    exception_type=type(exc).__name__,
+                ),
+            )
+        raise
+
+
+def garbage_collect_unreferenced_blobs(
+    session: Session,
+    *,
+    tenant_id: str,
+    limit: int,
+    approval_reference: str,
+) -> BlobGcResult:
+    candidate_rows = (
+        session.query(FileBlob)
+        .filter(FileBlob.tenant_id == tenant_id)
+        .filter(
+            or_(FileBlob.retained_until.is_(None), FileBlob.retained_until <= utcnow())
+        )
+        .filter(
+            ~session.query(FileVersion.id)
+            .filter(FileVersion.blob_id == FileBlob.id)
+            .exists()
+        )
+        .order_by(FileBlob.created_at.asc(), FileBlob.id.asc())
+        .limit(limit)
+        .all()
+    )
+    candidates = tuple(
+        (row.id, row.storage_key) for row in candidate_rows
+    )
+    # Close the read snapshot before the independent recovery transaction takes
+    # its durable SQLite/PostgreSQL write lock.
+    session.commit()
+    deleted = 0
+    unresolved: list[str] = []
+    backend = get_storage_backend()
+    for candidate_id, storage_key in candidates:
+        key_digest = hashlib.sha256(storage_key.encode("utf-8")).hexdigest()
+        request = {
+            "tenant_id": tenant_id,
+            "blob_id": candidate_id,
+            "storage_key_sha256": key_digest,
+            "approval_reference": approval_reference,
+        }
+        try:
+            started = begin_durable_recovery_operation(
+                get_database().SessionLocal,
+                identity=process_runtime_identity(),
+                module_id="files",
+                operation_type="blob-garbage-collection",
+                idempotency_key=(
+                    f"files-blob-gc:{candidate_id}:{key_digest[:12]}:"
+                    f"{hashlib.sha256(approval_reference.encode('utf-8')).hexdigest()[:12]}"
+                ),
+                request=request,
+                recovery_plan=RecoveryPlan(
+                    mode=RecoveryMode.FORWARD_RECOVERY,
+                    preconditions=(
+                        "the caller holds dedicated Files purge authority",
+                        "a fresh database check shows no FileVersion reference",
+                    ),
+                    forward_recovery_steps=(
+                        "verify the object is absent",
+                        "delete the unreferenced FileBlob row after absence is proven",
+                    ),
+                    verification_steps=(
+                        "recheck FileVersion references while holding the blob fence",
+                        "probe the exact managed storage key after deletion",
+                    ),
+                    approval_reference=approval_reference,
+                ),
+                precondition_evidence={
+                    "blob_id": candidate_id,
+                    "storage_key_sha256": key_digest,
+                    "observed_reference_count": 0,
+                },
+                lease_resource_key=f"files:blob:{tenant_id}:{candidate_id}",
+                lease_ttl_seconds=15 * 60,
+                resource_type="file_blob",
+                resource_id=candidate_id,
+                metadata={"resources": ["postgresql", "object-storage"], "storage_backend": backend.name},
+                block_unresolved_resource=True,
+            )
+        except (RecoveryOperationBusy, RecoveryOperationStateConflict):
+            continue
+        except (RecoveryGuaranteeError, RuntimeError, ValueError) as exc:
+            raise FileStorageError(
+                "The Files recovery ledger is unavailable; no blob bytes were deleted"
+            ) from exc
+        if started.replayed or started.operation is None:
+            continue
+        operation = started.operation
+        try:
+            blob = (
+                session.query(FileBlob)
+                .filter(FileBlob.id == candidate_id, FileBlob.tenant_id == tenant_id)
+                .with_for_update()
+                .one_or_none()
+            )
+            references = int(
+                session.query(func.count(FileVersion.id))
+                .filter(FileVersion.blob_id == candidate_id)
+                .scalar()
+                or 0
+            )
+            blob_retained = bool(
+                blob is not None
+                and blob.retained_until is not None
+                and _as_utc(blob.retained_until) > utcnow()
+            )
+        except Exception as exc:
+            session.rollback()
+            if not operation.closed:
+                operation.reject(
+                    summary="Blob eligibility could not be rechecked before deletion",
+                    evidence=_verified_evidence(
+                        {"external_effect_started": False},
+                        exception_type=type(exc).__name__,
+                    ),
+                )
+            raise FileStorageError(
+                "Blob eligibility could not be rechecked; no bytes were deleted"
+            ) from exc
+        if blob is None or references or blob_retained:
+            operation.reject(
+                summary="Blob garbage collection was no longer eligible",
+                evidence=_verified_evidence(
+                    {
+                        "blob_present": blob is not None,
+                        "reference_count_zero": references == 0,
+                        "blob_retention_expired": not blob_retained,
+                    },
+                    reference_count=references,
+                ),
+            )
+            session.rollback()
+            continue
+        try:
+            backend.delete(blob.storage_key)
+            object_present = backend.exists(blob.storage_key)
+        except (StorageBackendError, OSError) as exc:
+            try:
+                object_present = backend.exists(blob.storage_key)
+            except (StorageBackendError, OSError):
+                object_present = None
+            session.rollback()
+            if object_present is True:
+                operation.fail(
+                    summary="The unreferenced blob object was retained",
+                    evidence={"object_present": True, "exception_type": type(exc).__name__},
+                )
+            else:
+                operation.unresolved(
+                    status=RecoveryStatus.OUTCOME_UNKNOWN,
+                    summary="The blob deletion outcome could not be verified",
+                    evidence={"object_present": object_present, "exception_type": type(exc).__name__},
+                    failure_summary="Reconcile the exact blob key before retrying garbage collection",
+                )
+                unresolved.append(started.operation_id)
+            continue
+        if object_present:
+            session.rollback()
+            operation.fail(
+                summary="The object store did not delete the unreferenced blob",
+                evidence={"object_present": True},
+            )
+            continue
+        try:
+            session.delete(blob)
+            session.flush()
+            operation.commit_verified_success(
+                session,
+                evidence=_verified_evidence(
+                    {
+                        "object_absent": True,
+                        "database_blob_absent": True,
+                        "reference_count_zero": True,
+                    },
+                    blob_id=candidate_id,
+                    storage_key_sha256=key_digest,
+                    object_present=False,
+                    database_blob_present=False,
+                    reference_count=0,
+                ),
+            )
+            deleted += 1
+        except Exception as exc:
+            session.rollback()
+            if not operation.closed:
+                operation.unresolved(
+                    status=RecoveryStatus.RECOVERY_REQUIRED,
+                    summary="The object is absent but blob metadata still requires forward recovery",
+                    evidence={"object_present": False, "exception_type": type(exc).__name__},
+                    failure_summary="Delete the FileBlob row only after rechecking all FileVersion references",
+                )
+            unresolved.append(started.operation_id)
+    return BlobGcResult(
+        inspected_blobs=len(candidates),
+        deleted_blobs=deleted,
+        unresolved_operation_ids=tuple(unresolved),
+    )
+
+
+def _asset_purge_blockers(session: Session, asset: FileAsset) -> list[str]:
+    blockers: list[str] = []
+    if asset.deleted_at is None:
+        blockers.append("not_soft_deleted")
+    if asset.legal_hold:
+        blockers.append("legal_hold")
+    if asset.retained_until is not None and _as_utc(asset.retained_until) > utcnow():
+        blockers.append("retention_active")
+    if _table_exists(session, CampaignAttachmentUse.__tablename__) and session.query(
+        CampaignAttachmentUse.id
+    ).filter(CampaignAttachmentUse.file_asset_id == asset.id).first() is not None:
+        blockers.append("campaign_evidence")
+    if _table_exists(session, FileFormEvidenceGrant.__tablename__) and session.query(
+        FileFormEvidenceGrant.id
+    ).filter(FileFormEvidenceGrant.file_asset_id == asset.id).first() is not None:
+        blockers.append("form_evidence")
+    if session.query(FileShare.id).filter(
+        FileShare.file_asset_id == asset.id,
+        effective_file_share_clause(),
+    ).first() is not None:
+        blockers.append("active_share")
+    return blockers
+
+
+def _asset_owner_id(asset: FileAsset) -> str:
+    owner_id = asset.owner_user_id if asset.owner_type == "user" else asset.owner_group_id
+    if not owner_id:
+        raise FileStorageError("File has no valid owner")
+    return owner_id
+
+
+def _asset_owner_query(session: Session, asset: FileAsset):
+    return _asset_owner_query_by_values(
+        session,
+        tenant_id=asset.tenant_id,
+        owner_type=asset.owner_type,
+        owner_id=_asset_owner_id(asset),
+    )
+
+
+def _asset_owner_query_by_values(
+    session: Session, *, tenant_id: str, owner_type: str, owner_id: str
+):
+    query = session.query(FileAsset).filter(
+        FileAsset.tenant_id == tenant_id, FileAsset.owner_type == owner_type
+    )
+    if owner_type == "user":
+        return query.filter(FileAsset.owner_user_id == owner_id)
+    if owner_type == "group":
+        return query.filter(FileAsset.owner_group_id == owner_id)
+    raise FileStorageError("Files must be owned by a user or group")
+
+
+def _folder_owner_query(
+    session: Session, *, tenant_id: str, owner_type: str, owner_id: str
+):
+    query = session.query(FileFolder).filter(
+        FileFolder.tenant_id == tenant_id, FileFolder.owner_type == owner_type
+    )
+    if owner_type == "user":
+        return query.filter(FileFolder.owner_user_id == owner_id)
+    if owner_type == "group":
+        return query.filter(FileFolder.owner_group_id == owner_id)
+    raise FileStorageError("Folders must be owned by a user or group")
+
+
+def _connector_space_owner_query(session: Session, space: FileConnectorSpace):
+    query = session.query(FileConnectorSpace).filter(
+        FileConnectorSpace.tenant_id == space.tenant_id,
+        FileConnectorSpace.owner_type == space.owner_type,
+    )
+    owner_id = connector_space_owner_id(space)
+    if space.owner_type == "user":
+        return query.filter(FileConnectorSpace.owner_user_id == owner_id)
+    return query.filter(FileConnectorSpace.owner_group_id == owner_id)
+
+
+def _as_utc(value: datetime) -> datetime:
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+
+
+def _iso(value: datetime | None) -> str | None:
+    return _as_utc(value).isoformat() if value is not None else None
+
+
+def _table_exists(session: Session, table_name: str) -> bool:
+    return bool(session.bind is not None and inspect(session.bind).has_table(table_name))
+
+
+def _verified_evidence(
+    checks: dict[str, object], **details: object
+) -> dict[str, object]:
+    return {"verified": True, "checks": checks, **details}
+
+
+__all__ = [
+    "BlobGcResult",
+    "PurgePreview",
+    "PurgePreviewItem",
+    "PurgeResult",
+    "execute_asset_purge",
+    "garbage_collect_unreferenced_blobs",
+    "get_asset_for_lifecycle",
+    "preview_asset_purge",
+    "restore_asset",
+    "restore_connector_space",
+    "restore_folder",
+    "set_asset_lifecycle",
+]