feat: persist DiskSage file lineage ontology

backend/alembic/versions/0018_disksage_file_lineage.py

@@ -0,0 +1,86 @@
+"""persist scoped, encrypted DiskSage file lineage envelopes
+
+Revision ID: 0018_disksage_file_lineage
+Revises: 0017_merge_newsdom_carddav_heads
+Create Date: 2026-08-13 00:00:00.000000
+
+The Rust DiskSage verifier remains authoritative for copy/provider proof. Naruon
+stores the validated envelope for workspace-scoped provenance and exposes only
+the redacted graph projection in list responses.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0018_disksage_file_lineage"
+down_revision = "0017_merge_newsdom_carddav_heads"
+branch_labels = None
+depends_on = None
+
+_TABLE = "disksage_file_lineage_records"
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    inspector = sa.inspect(connection)
+    if not inspector.has_table(_TABLE):
+        op.create_table(
+            _TABLE,
+            sa.Column("lineage_record_uid", sa.String(length=96), nullable=False),
+            sa.Column("user_id", sa.String(), nullable=False),
+            sa.Column("organization_id", sa.String(), nullable=True),
+            sa.Column("workspace_id", sa.String(), nullable=False),
+            sa.Column("lineage_fingerprint", sa.String(length=64), nullable=False),
+            sa.Column("envelope_sha256", sa.String(length=64), nullable=False),
+            sa.Column("schema_version", sa.Integer(), nullable=False),
+            sa.Column("schema_kind", sa.String(length=96), nullable=False),
+            sa.Column("source_kind", sa.String(length=64), nullable=False),
+            sa.Column("archive_kind", sa.String(length=64), nullable=False),
+            sa.Column("raw_content_sha256", sa.String(length=64), nullable=False),
+            sa.Column("raw_content_blake3", sa.String(length=64), nullable=False),
+            sa.Column("content_bytes", sa.BigInteger(), nullable=False),
+            sa.Column("ontology_class", sa.String(length=256), nullable=False),
+            sa.Column("ontology_relation_count", sa.Integer(), nullable=False),
+            sa.Column("ontology_predicates", sa.JSON(), nullable=False),
+            sa.Column("provider_name", sa.String(length=32), nullable=False),
+            sa.Column("provider_sync_confirmed", sa.Boolean(), nullable=False),
+            sa.Column(
+                "provider_sync_state",
+                sa.String(length=32),
+                nullable=False,
+                server_default="unknown",
+            ),
+            sa.Column("envelope_json_encrypted", sa.String(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("lineage_record_uid"),
+            sa.UniqueConstraint(
+                "user_id",
+                "workspace_id",
+                "lineage_fingerprint",
+                name="uq_disksage_lineage_workspace_fingerprint",
+            ),
+        )
+
+    op.create_index(
+        "ix_disksage_lineage_scope_time",
+        _TABLE,
+        ["user_id", "organization_id", "workspace_id", "created_at"],
+        if_not_exists=True,
+    )
+    op.create_index(
+        "ix_disksage_lineage_ontology_class",
+        _TABLE,
+        ["workspace_id", "ontology_class"],
+        if_not_exists=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_disksage_lineage_ontology_class", table_name=_TABLE, if_exists=True
+    )
+    op.drop_index("ix_disksage_lineage_scope_time", table_name=_TABLE, if_exists=True)
+    connection = op.get_bind()
+    if sa.inspect(connection).has_table(_TABLE):
+        op.drop_table(_TABLE)

backend/alembic/versions/0019_disksage_provider_sync_state.py

@@ -0,0 +1,43 @@
+"""persist provider-native sync state for DiskSage lineage envelopes
+
+Revision ID: 0019_disksage_provider_sync_state
+Revises: 0018_disksage_file_lineage
+Create Date: 2026-08-13 00:00:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0019_disksage_provider_sync_state"
+down_revision = "0018_disksage_file_lineage"
+branch_labels = None
+depends_on = None
+
+_TABLE = "disksage_file_lineage_records"
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table(_TABLE) and not any(
+        column["name"] == "provider_sync_state"
+        for column in inspector.get_columns(_TABLE)
+    ):
+        op.add_column(
+            _TABLE,
+            sa.Column(
+                "provider_sync_state",
+                sa.String(length=32),
+                nullable=False,
+                server_default="unknown",
+            ),
+        )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table(_TABLE) and any(
+        column["name"] == "provider_sync_state"
+        for column in inspector.get_columns(_TABLE)
+    ):
+        op.drop_column(_TABLE, "provider_sync_state")

backend/alembic/versions/0020_disksage_lineage_column_names.py

@@ -0,0 +1,39 @@
+"""rename ambiguous DiskSage lineage column names
+
+Revision ID: 0020_disksage_lineage_column_names
+Revises: 0019_disksage_provider_sync_state
+Create Date: 2026-08-13 00:00:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0020_disksage_lineage_column_names"
+down_revision = "0019_disksage_provider_sync_state"
+branch_labels = None
+depends_on = None
+
+_TABLE = "disksage_file_lineage_records"
+
+
+def _rename_if_present(inspector: sa.Inspector, old: str, new: str) -> None:
+    columns = {column["name"] for column in inspector.get_columns(_TABLE)}
+    if old in columns and new not in columns:
+        op.alter_column(_TABLE, old, new_column_name=new)
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if inspector.has_table(_TABLE):
+        _rename_if_present(inspector, "bytes", "content_bytes")
+        inspector = sa.inspect(op.get_bind())
+        _rename_if_present(inspector, "provider", "provider_name")
+
+
+def downgrade() -> None:
+    # This revision repairs legacy installations that still have the old
+    # single-word names.  Revision 0019 and the canonical 0018 schema both
+    # use content_bytes/provider_name, so a downgrade must not reintroduce
+    # the legacy shape or leave the model and migration chain inconsistent.
+    pass

backend/alembic/versions/0021_disksage_lineage_scope_index.py

@@ -0,0 +1,39 @@
+"""align the DiskSage lineage scope index with the list query"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0021_disksage_lineage_scope_index"
+down_revision = "0020_disksage_lineage_column_names"
+branch_labels = None
+depends_on = None
+
+_TABLE = "disksage_file_lineage_records"
+_INDEX = "ix_disksage_lineage_scope_time"
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if not inspector.has_table(_TABLE):
+        return
+    op.drop_index(_INDEX, table_name=_TABLE, if_exists=True)
+    op.create_index(
+        _INDEX,
+        _TABLE,
+        ["user_id", "workspace_id", "created_at"],
+        if_not_exists=True,
+    )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if not inspector.has_table(_TABLE):
+        return
+    op.drop_index(_INDEX, table_name=_TABLE, if_exists=True)
+    op.create_index(
+        _INDEX,
+        _TABLE,
+        ["user_id", "organization_id", "workspace_id", "created_at"],
+        if_not_exists=True,
+    )

backend/api/disksage.py

@@ -0,0 +1,145 @@
+import datetime
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.exc import IntegrityError
+
+from api.auth import AuthContext, get_auth_context
+from db.models import DiskSageFileLineageRecord
+from db.session import get_db
+from services.disksage_file_lineage import (
+    FileLineageEnvelope,
+    FileLineageSummary,
+    canonical_envelope_json,
+    canonical_envelope_sha256,
+    ontology_predicates,
+)
+from core.runtime_secrets import EncryptionKeyMissingError
+
+router = APIRouter(prefix="/api/disksage", tags=["disksage"])
+
+
+def _summary(record: DiskSageFileLineageRecord) -> FileLineageSummary:
+    return FileLineageSummary(
+        lineage_record_uid=record.lineage_record_uid,
+        lineage_fingerprint=record.lineage_fingerprint,
+        schema_version=record.schema_version,
+        source_kind=record.source_kind,
+        archive_kind=record.archive_kind,
+        raw_content_sha256=record.raw_content_sha256,
+        raw_content_blake3=record.raw_content_blake3,
+        content_bytes=record.content_bytes,
+        ontology_class=record.ontology_class,
+        ontology_relation_count=record.ontology_relation_count,
+        ontology_predicates=list(record.ontology_predicates or []),
+        provider_name=record.provider_name,
+        provider_sync_confirmed=record.provider_sync_confirmed,
+        provider_sync_state=record.provider_sync_state,
+        created_at=record.created_at.isoformat(),
+    )
+
+
+def _encrypted_envelope_json(envelope: FileLineageEnvelope) -> str:
+    return canonical_envelope_json(envelope)
+
+
+@router.post("/file-lineage", response_model=FileLineageSummary, status_code=201)
+async def ingest_file_lineage(
+    envelope: FileLineageEnvelope,
+    auth_context: AuthContext = Depends(get_auth_context),
+    db: AsyncSession = Depends(get_db),
+) -> FileLineageSummary:
+    """Persist one validated DiskSage envelope without authorizing source eviction."""
+
+    envelope_sha256 = canonical_envelope_sha256(envelope)
+    lineage_fingerprint = envelope.cloud_copy.lineage_fingerprint
+    existing_result = await db.execute(
+        select(DiskSageFileLineageRecord).where(
+            DiskSageFileLineageRecord.user_id == auth_context.user_id,
+            DiskSageFileLineageRecord.workspace_id == auth_context.workspace_id,
+            DiskSageFileLineageRecord.lineage_fingerprint == lineage_fingerprint,
+        )
+    )
+    existing = existing_result.scalar_one_or_none()
+    if existing is not None:
+        if existing.envelope_sha256 != envelope_sha256:
+            raise HTTPException(
+                status_code=409,
+                detail="lineage fingerprint is already bound to different evidence",
+            )
+        return _summary(existing)
+
+    try:
+        record = DiskSageFileLineageRecord(
+            lineage_record_uid=f"disksage_lineage_{uuid.uuid4().hex}",
+            user_id=auth_context.user_id,
+            organization_id=auth_context.organization_id,
+            workspace_id=auth_context.workspace_id,
+            lineage_fingerprint=lineage_fingerprint,
+            envelope_sha256=envelope_sha256,
+            schema_version=envelope.schema_version,
+            schema_kind=envelope.schema_kind,
+            source_kind=envelope.source_kind,
+            archive_kind=envelope.archive_kind,
+            raw_content_sha256=envelope.raw_content_sha256,
+            raw_content_blake3=envelope.raw_content_blake3,
+            content_bytes=envelope.bytes,
+            ontology_class=envelope.ontology_class,
+            ontology_relation_count=len(envelope.ontology_relations),
+            ontology_predicates=ontology_predicates(envelope),
+            provider_name=envelope.cloud_copy.provider,
+            provider_sync_confirmed=envelope.cloud_copy.provider_sync_confirmed,
+            provider_sync_state=envelope.cloud_copy.provider_sync_state or "unknown",
+            envelope_json_encrypted=_encrypted_envelope_json(envelope),
+            created_at=datetime.datetime.now(datetime.timezone.utc),
+        )
+        db.add(record)
+        await db.commit()
+        await db.refresh(record)
+    except IntegrityError:
+        await db.rollback()
+        replayed_result = await db.execute(
+            select(DiskSageFileLineageRecord).where(
+                DiskSageFileLineageRecord.user_id == auth_context.user_id,
+                DiskSageFileLineageRecord.workspace_id == auth_context.workspace_id,
+                DiskSageFileLineageRecord.lineage_fingerprint == lineage_fingerprint,
+            )
+        )
+        replayed = replayed_result.scalar_one_or_none()
+        if replayed is None:
+            raise
+        if replayed.envelope_sha256 != envelope_sha256:
+            raise HTTPException(
+                status_code=409,
+                detail="lineage fingerprint is already bound to different evidence",
+            ) from None
+        return _summary(replayed)
+    except EncryptionKeyMissingError as error:
+        await db.rollback()
+        raise HTTPException(
+            status_code=503,
+            detail="Server encryption key is not configured. Contact your workspace administrator.",
+        ) from error
+    return _summary(record)
+
+
+@router.get("/file-lineage", response_model=list[FileLineageSummary])
+async def list_file_lineage(
+    limit: int = Query(default=50, ge=1, le=100),
+    auth_context: AuthContext = Depends(get_auth_context),
+    db: AsyncSession = Depends(get_db),
+) -> list[FileLineageSummary]:
+    """List redacted lineage graph projections; encrypted envelope values stay server-side."""
+
+    result = await db.execute(
+        select(DiskSageFileLineageRecord)
+        .where(
+            DiskSageFileLineageRecord.user_id == auth_context.user_id,
+            DiskSageFileLineageRecord.workspace_id == auth_context.workspace_id,
+        )
+        .order_by(DiskSageFileLineageRecord.created_at.desc())
+        .limit(limit)
+    )
+    return [_summary(record) for record in result.scalars().all()]

backend/core/runtime_secrets.py

@@ -18,6 +18,12 @@
 ENCRYPTION_KEY_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$")
 
 
+class EncryptionKeyMissingError(RuntimeError):
+    """Raised when encryption is requested without the configured active key."""
+
+    error_code = "encryption-key-missing"
+
+
 @dataclass(frozen=True)
 class RuntimeEncryptionKey:
     key_id: str
@@ -147,7 +153,7 @@ def build_encryption_keyring(
     previous_keys_value: str | None = None,
 ) -> EncryptionKeyRing:
     if active_key_value is None or not active_key_value.strip():
-        raise RuntimeError(
+        raise EncryptionKeyMissingError(
             "ENCRYPTION_KEY is required. Refusing to encrypt without a configured key."
         )