From 9926a4b3fd00b9afd1c890d7b4bebdd3c8563e49 Mon Sep 17 00:00:00 2001 From: Vadim Malanov Date: Sun, 14 Jun 2026 19:22:26 +0300 Subject: [PATCH] feat: add asset manifest knowledge ingest --- README.md | 73 +++- app/api/routes_ingestion.py | 48 ++- app/api/schemas.py | 120 +++++- .../versions/0002_asset_ingest_jobs.py | 72 ++++ app/db/models.py | 39 ++ app/ingestion/knowledge_ingest.py | 408 ++++++++++++++++++ app/ingestion/pipeline.py | 40 +- app/storage/minio_client.py | 4 +- tests/test_alembic.py | 25 +- tests/test_knowledge_ingest.py | 262 +++++++++++ 10 files changed, 1055 insertions(+), 36 deletions(-) create mode 100644 app/db/migrations/versions/0002_asset_ingest_jobs.py create mode 100644 app/ingestion/knowledge_ingest.py create mode 100644 tests/test_knowledge_ingest.py diff --git a/README.md b/README.md index 89345b3..dc649d5 100644 --- a/README.md +++ b/README.md @@ -53,10 +53,79 @@ curl http://localhost:8000/api/v1/health | jq . Open the interactive Swagger docs at . -## Ingest documents +## Ingest TeamHUB assets + +TeamHUB modules should call `POST /api/v1/knowledge-ingest` with an +`AssetManifest`. The original file stays in the owner module's MinIO/S3 bucket; +LegacyHUB reads it by `bucket` + `object_key`, verifies `size_bytes` and +`sha256`, records an idempotent ingest job by +`asset_id + sha256 + manifest_version`, then builds OCR/search projections. + +```bash +curl -X POST http://localhost:8000/api/v1/knowledge-ingest \ + -H "Content-Type: application/json" \ + -H "X-API-Key: ${LEGACYHUB_API_KEY}" \ + -H "Idempotency-Key: evt_01J..." \ + -d '{ + "manifest": { + "manifest_version": "1.0", + "asset": { + "asset_id": "asset_01J...", + "owner_module": "mailhub", + "owner_record_type": "document", + "owner_record_id": "doc_01J...", + "asset_kind": "document", + "title": "incoming-letter.pdf", + "original_filename": "incoming-letter.pdf", + "content_type": "application/pdf", + "size_bytes": 184220, + "sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "created_at": "2026-06-14T12:00:00Z", + "created_by": "staff:123" + }, + "storage": { + "provider": "s3", + "bucket": "teamhub-mailhub-originals", + "object_key": "mailhub/2026/06/14/asset_01J/original/incoming-letter.pdf", + "version_id": null + }, + "security": { + "gate_status": "approved", + "classification": "internal", + "av_status": "clean", + "content_scan_status": "clean", + "pii_status": "reviewed", + "approved_at": "2026-06-14T12:03:00Z", + "approved_by": "staff:123", + "quarantine_reason": null + }, + "retention": { + "policy_id": "mailhub-default", + "retain_until": null, + "legal_hold": false, + "legal_hold_reason": null + }, + "derivatives": [], + "links": [] + }, + "ingest_options": { + "generate_derivatives": true, + "index_full_text": true, + "index_vectors": true, + "emit_events": true + } + }' +``` + +Manifests with `security.gate_status != "approved"` are rejected without reading +object storage. Container-local paths such as `/data/input/...`, `C:\...`, or +`file://...` are invalid in the manifest contract. + +## Ingest local folders (deprecated) Mount a folder into the container at `/data/input` (the compose file already -mounts `./data/input` for you), drop PDFs into it, and call: +mounts `./data/input` for you), drop PDFs into it, and call the compatibility +endpoint: ```bash curl -X POST http://localhost:8000/api/v1/ingest/folder \ diff --git a/app/api/routes_ingestion.py b/app/api/routes_ingestion.py index eaf8e08..bd1e38f 100644 --- a/app/api/routes_ingestion.py +++ b/app/api/routes_ingestion.py @@ -5,30 +5,64 @@ from __future__ import annotations import uuid from pathlib import Path -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Header, HTTPException, Response, status -from app.api.schemas import IngestFolderRequest, IngestFolderResponse +from app.api.schemas import ( + AssetManifestEnvelope, + IngestFolderRequest, + IngestFolderResponse, + KnowledgeIngestResponse, +) +from app.ingestion.knowledge_ingest import KnowledgeIngestError, accept_knowledge_ingest from app.logging_config import get_logger logger = get_logger(__name__) -router = APIRouter(prefix="/ingest", tags=["ingestion"]) +router = APIRouter(tags=["ingestion"]) -@router.post("/folder", response_model=IngestFolderResponse) -def ingest_folder(req: IngestFolderRequest) -> IngestFolderResponse: +@router.post( + "/knowledge-ingest", + response_model=KnowledgeIngestResponse, +) +def knowledge_ingest( + req: AssetManifestEnvelope, + response: Response, + idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"), +) -> KnowledgeIngestResponse: + """Accept a TeamHUB AssetManifest and queue LegacyHUB projection ingest.""" + try: + result = accept_knowledge_ingest(req, idempotency_key=idempotency_key) + except KnowledgeIngestError as exc: + raise HTTPException( + status_code=exc.status_code, + detail={"reason_code": exc.reason_code, "message": exc.message}, + ) from exc + response.status_code = ( + status.HTTP_202_ACCEPTED if result.status == "accepted" else status.HTTP_200_OK + ) + return result + + +@router.post("/ingest/folder", response_model=IngestFolderResponse, deprecated=True) +def ingest_folder(req: IngestFolderRequest, response: Response) -> IngestFolderResponse: """Discover all PDFs under ``path`` and queue them for processing. The request returns immediately after the discovery pass. Per-document OCR / extraction / indexing happens asynchronously in Celery workers. + + Deprecated for TeamHUB integrations: use ``POST /api/v1/knowledge-ingest`` + with an AssetManifest and object-storage reference instead. """ + response.headers["Deprecation"] = "true" + response.headers["Link"] = '; rel="successor-version"' folder = Path(req.path) if not folder.exists() or not folder.is_dir(): raise HTTPException(status_code=400, detail=f"Folder not found: {req.path}") # Lazy import - keeps module load light. - from app.ingestion.scanner import discover_documents - from app.workers.tasks import process_document + from app.ingestion.scanner import discover_documents # noqa: PLC0415 + from app.workers.tasks import process_document # noqa: PLC0415 run_id = uuid.uuid4() discovered, queued, dups, invalid = 0, 0, 0, 0 diff --git a/app/api/schemas.py b/app/api/schemas.py index e06191e..4b8c270 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -6,8 +6,7 @@ import uuid from datetime import datetime from typing import Any, Literal -from pydantic import BaseModel, Field - +from pydantic import BaseModel, Field, model_validator # ---------------- Health ---------------- @@ -39,6 +38,123 @@ class IngestFolderResponse(BaseModel): invalid_files: int +AssetKind = Literal["document", "image", "audio", "video", "archive", "dataset", "other"] +KnowledgeIngestStatus = Literal["accepted", "duplicate", "rejected"] + + +def _looks_like_local_path(value: str) -> bool: + stripped = value.strip() + if not stripped: + return False + lower = stripped.lower() + if lower.startswith("file://"): + return True + if stripped.startswith(("/", "\\")): + return True + if len(stripped) >= 3 and stripped[1] == ":" and stripped[2] in {"\\", "/"}: + return stripped[0].isalpha() + if "\\" in stripped: + return True + return False + + +def _iter_string_values(value: Any, prefix: str = ""): + if isinstance(value, str): + yield prefix, value + return + if isinstance(value, dict): + for key, child in value.items(): + child_prefix = f"{prefix}.{key}" if prefix else str(key) + yield from _iter_string_values(child, child_prefix) + return + if isinstance(value, list): + for idx, child in enumerate(value): + yield from _iter_string_values(child, f"{prefix}[{idx}]") + + +class AssetIdentity(BaseModel): + asset_id: str = Field(..., min_length=1) + owner_module: str = Field(..., min_length=1) + owner_record_type: str = Field(..., min_length=1) + owner_record_id: str = Field(..., min_length=1) + asset_kind: AssetKind + title: str | None = None + original_filename: str | None = None + content_type: str = Field(..., min_length=1) + size_bytes: int = Field(..., ge=0) + sha256: str = Field(..., min_length=64, max_length=64, pattern=r"^[0-9a-fA-F]{64}$") + created_at: datetime | None = None + created_by: str | None = None + + +class AssetStorageRef(BaseModel): + provider: str = Field(..., min_length=1) + bucket: str = Field(..., min_length=1) + object_key: str = Field(..., min_length=1) + version_id: str | None = None + region: str | None = None + kms_key_id: str | None = None + + +class AssetSecurityState(BaseModel): + gate_status: str = Field(..., min_length=1) + classification: str | None = None + av_status: str | None = None + content_scan_status: str | None = None + pii_status: str | None = None + approved_at: datetime | None = None + approved_by: str | None = None + quarantine_reason: str | None = None + + +class AssetRetentionState(BaseModel): + policy_id: str = Field(..., min_length=1) + retain_until: datetime | None = None + legal_hold: bool + legal_hold_reason: str | None = None + + +class AssetManifest(BaseModel): + manifest_version: str = Field(..., min_length=1) + asset: AssetIdentity + storage: AssetStorageRef + security: AssetSecurityState + retention: AssetRetentionState + derivatives: list[dict[str, Any]] = Field(default_factory=list) + links: list[dict[str, Any]] = Field(default_factory=list) + source: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def reject_local_filesystem_paths(self) -> AssetManifest: + dumped = self.model_dump(mode="json") + for path, value in _iter_string_values(dumped): + if _looks_like_local_path(value): + raise ValueError( + f"AssetManifest MUST NOT contain local filesystem paths: {path}" + ) + return self + + +class KnowledgeIngestOptions(BaseModel): + generate_derivatives: bool = True + index_full_text: bool = True + index_vectors: bool = True + emit_events: bool = True + + +class AssetManifestEnvelope(BaseModel): + manifest: AssetManifest + ingest_options: KnowledgeIngestOptions = Field(default_factory=KnowledgeIngestOptions) + + +class KnowledgeIngestResponse(BaseModel): + status: KnowledgeIngestStatus + ingest_job_id: uuid.UUID | None = None + asset_id: str + idempotency_key: str | None = None + reason_code: str | None = None + + class DocumentSummary(BaseModel): id: uuid.UUID original_file_name: str diff --git a/app/db/migrations/versions/0002_asset_ingest_jobs.py b/app/db/migrations/versions/0002_asset_ingest_jobs.py new file mode 100644 index 0000000..6222443 --- /dev/null +++ b/app/db/migrations/versions/0002_asset_ingest_jobs.py @@ -0,0 +1,72 @@ +"""asset manifest ingest jobs + +Revision ID: 0002_asset_ingest_jobs +Revises: 0001_initial +Create Date: 2026-06-14 + +""" +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0002_asset_ingest_jobs" +down_revision: str | None = "0001_initial" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "asset_ingest_jobs", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column( + "document_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("documents.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("asset_id", sa.Text, nullable=False), + sa.Column("manifest_version", sa.String(32), nullable=False), + sa.Column("sha256", sa.String(64), nullable=False), + sa.Column("owner_module", sa.Text, nullable=False), + sa.Column("owner_record_type", sa.Text, nullable=False), + sa.Column("owner_record_id", sa.Text, nullable=False), + sa.Column("status", sa.String(32), nullable=False, server_default="QUEUED"), + sa.Column("idempotency_key", sa.Text, nullable=True), + sa.Column("storage_bucket", sa.Text, nullable=False), + sa.Column("storage_key", sa.Text, nullable=False), + sa.Column("storage_version_id", sa.Text, nullable=True), + sa.Column("manifest", postgresql.JSONB, nullable=False), + sa.Column( + "ingest_options", + postgresql.JSONB, + nullable=False, + server_default=sa.text("'{}'::jsonb"), + ), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.UniqueConstraint( + "asset_id", + "sha256", + "manifest_version", + name="uq_asset_ingest_identity", + ), + ) + op.create_index("ix_asset_ingest_document", "asset_ingest_jobs", ["document_id"]) + op.create_index("ix_asset_ingest_asset", "asset_ingest_jobs", ["asset_id"]) + op.create_index( + "ix_asset_ingest_owner", + "asset_ingest_jobs", + ["owner_module", "owner_record_type", "owner_record_id"], + ) + + +def downgrade() -> None: + op.drop_index("ix_asset_ingest_owner", table_name="asset_ingest_jobs") + op.drop_index("ix_asset_ingest_asset", table_name="asset_ingest_jobs") + op.drop_index("ix_asset_ingest_document", table_name="asset_ingest_jobs") + op.drop_table("asset_ingest_jobs") diff --git a/app/db/models.py b/app/db/models.py index 05a0425..354e0e7 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -246,6 +246,45 @@ class IngestionRun(Base): ) +class AssetIngestJob(Base): + __tablename__ = "asset_ingest_jobs" + __table_args__ = ( + UniqueConstraint( + "asset_id", + "sha256", + "manifest_version", + name="uq_asset_ingest_identity", + ), + Index("ix_asset_ingest_document", "document_id"), + Index("ix_asset_ingest_asset", "asset_id"), + Index("ix_asset_ingest_owner", "owner_module", "owner_record_type", "owner_record_id"), + ) + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + document_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), ForeignKey("documents.id", ondelete="CASCADE"), nullable=False + ) + asset_id: Mapped[str] = mapped_column(Text, nullable=False) + manifest_version: Mapped[str] = mapped_column(String(32), nullable=False) + sha256: Mapped[str] = mapped_column(String(64), nullable=False) + owner_module: Mapped[str] = mapped_column(Text, nullable=False) + owner_record_type: Mapped[str] = mapped_column(Text, nullable=False) + owner_record_id: Mapped[str] = mapped_column(Text, nullable=False) + status: Mapped[str] = mapped_column(String(32), nullable=False, default="QUEUED") + idempotency_key: Mapped[str | None] = mapped_column(Text, nullable=True) + storage_bucket: Mapped[str] = mapped_column(Text, nullable=False) + storage_key: Mapped[str] = mapped_column(Text, nullable=False) + storage_version_id: Mapped[str | None] = mapped_column(Text, nullable=True) + manifest_json: Mapped[dict[str, Any]] = mapped_column("manifest", JSONB, nullable=False) + ingest_options: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False + ) + + class ProcessingEvent(Base): __tablename__ = "processing_events" __table_args__ = ( diff --git a/app/ingestion/knowledge_ingest.py b/app/ingestion/knowledge_ingest.py new file mode 100644 index 0000000..a763db6 --- /dev/null +++ b/app/ingestion/knowledge_ingest.py @@ -0,0 +1,408 @@ +"""TeamHUB AssetManifest ingest orchestration. + +This API boundary accepts module-owned object storage references, verifies the +original bytes, records an idempotent ingest job, and queues the existing +per-document OCR/extraction/indexing pipeline. +""" + +from __future__ import annotations + +import shutil +import tempfile +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError + +from app.api.schemas import ( + AssetManifest, + AssetManifestEnvelope, + KnowledgeIngestOptions, + KnowledgeIngestResponse, +) +from app.config import settings +from app.db.models import ( + ArtifactType, + AssetIngestJob, + Document, + DocumentStatus, + ProcessingEvent, +) +from app.db.session import session_scope +from app.logging_config import get_logger +from app.storage.artifacts import ensure_artifact +from app.storage.local_paths import work_dir_for +from app.storage.minio_client import MinioStorage, get_storage +from app.utils.hashing import sha256_file + +logger = get_logger(__name__) + + +@dataclass(frozen=True) +class VerifiedAssetObject: + local_path: Path + size_bytes: int + sha256: str + + +@dataclass(frozen=True) +class ExistingKnowledgeIngestJob: + job_id: uuid.UUID + + +@dataclass(frozen=True) +class KnowledgeIngestRegistration: + job_id: uuid.UUID + document_id: uuid.UUID | None + duplicate: bool + + +class KnowledgeIngestError(Exception): + def __init__(self, status_code: int, reason_code: str, message: str) -> None: + super().__init__(message) + self.status_code = status_code + self.reason_code = reason_code + self.message = message + + +class KnowledgeIngestRepository(Protocol): + def find_existing_job(self, manifest: AssetManifest) -> ExistingKnowledgeIngestJob | None: + ... + + def register_verified_manifest( + self, + manifest: AssetManifest, + ingest_options: KnowledgeIngestOptions, + verified: VerifiedAssetObject, + idempotency_key: str | None, + ) -> KnowledgeIngestRegistration: + ... + + +class SqlAlchemyKnowledgeIngestRepository: + def find_existing_job(self, manifest: AssetManifest) -> ExistingKnowledgeIngestJob | None: + with session_scope() as db: + job = db.execute(_identity_query(manifest)).scalar_one_or_none() + if job is None: + return None + return ExistingKnowledgeIngestJob(job_id=job.id) + + def register_verified_manifest( + self, + manifest: AssetManifest, + ingest_options: KnowledgeIngestOptions, + verified: VerifiedAssetObject, + idempotency_key: str | None, + ) -> KnowledgeIngestRegistration: + try: + with session_scope() as db: + existing_job = db.execute(_identity_query(manifest)).scalar_one_or_none() + if existing_job is not None: + return KnowledgeIngestRegistration( + job_id=existing_job.id, + document_id=existing_job.document_id, + duplicate=True, + ) + + existing_doc = db.execute( + select(Document).where(Document.sha256 == manifest.asset.sha256.lower()) + ).scalar_one_or_none() + + if existing_doc is None: + document_id = uuid.uuid4() + doc = Document( + id=document_id, + source_path=_storage_uri(manifest), + original_file_name=_original_filename(manifest), + sha256=manifest.asset.sha256.lower(), + file_size_bytes=verified.size_bytes, + mime_type=manifest.asset.content_type, + status=DocumentStatus.STORED_ORIGINAL, + ) + db.add(doc) + db.add( + ProcessingEvent( + document_id=document_id, + stage=DocumentStatus.DISCOVERED, + level="INFO", + message="Asset manifest accepted", + data=_event_data(manifest, idempotency_key), + ) + ) + else: + document_id = existing_doc.id + if existing_doc.status == DocumentStatus.DISCOVERED: + existing_doc.status = DocumentStatus.STORED_ORIGINAL + + ensure_artifact( + db, + document_id=document_id, + artifact_type=ArtifactType.ORIGINAL_PDF, + bucket=manifest.storage.bucket, + key=manifest.storage.object_key, + checksum=manifest.asset.sha256.lower(), + ) + + job_id = uuid.uuid4() + db.add( + AssetIngestJob( + id=job_id, + document_id=document_id, + asset_id=manifest.asset.asset_id, + manifest_version=manifest.manifest_version, + sha256=manifest.asset.sha256.lower(), + owner_module=manifest.asset.owner_module, + owner_record_type=manifest.asset.owner_record_type, + owner_record_id=manifest.asset.owner_record_id, + status="QUEUED", + idempotency_key=idempotency_key, + storage_bucket=manifest.storage.bucket, + storage_key=manifest.storage.object_key, + storage_version_id=manifest.storage.version_id, + manifest_json=manifest.model_dump(mode="json"), + ingest_options=ingest_options.model_dump(mode="json"), + ) + ) + db.add( + ProcessingEvent( + run_id=job_id, + document_id=document_id, + stage=DocumentStatus.STORED_ORIGINAL, + level="INFO", + message="Original referenced from owner object storage", + data=_event_data(manifest, idempotency_key), + ) + ) + + return KnowledgeIngestRegistration( + job_id=job_id, + document_id=document_id, + duplicate=False, + ) + except IntegrityError: + existing = self.find_existing_job(manifest) + if existing is not None: + return KnowledgeIngestRegistration( + job_id=existing.job_id, + document_id=None, + duplicate=True, + ) + raise + + +def accept_knowledge_ingest( + req: AssetManifestEnvelope, + *, + idempotency_key: str | None = None, + storage: MinioStorage | None = None, + repository: KnowledgeIngestRepository | None = None, + enqueue_document=None, + cache_original=None, + temp_dir: Path | None = None, +) -> KnowledgeIngestResponse: + manifest = req.manifest + if manifest.security.gate_status.lower() != "approved": + return KnowledgeIngestResponse( + status="rejected", + asset_id=manifest.asset.asset_id, + idempotency_key=idempotency_key, + reason_code="security_gate_not_approved", + ) + + repo = repository or SqlAlchemyKnowledgeIngestRepository() + existing = repo.find_existing_job(manifest) + if existing is not None: + return KnowledgeIngestResponse( + status="duplicate", + ingest_job_id=existing.job_id, + asset_id=manifest.asset.asset_id, + idempotency_key=idempotency_key, + ) + + object_storage = storage or get_storage() + enqueue = enqueue_document or enqueue_process_document + cache = cache_original or cache_verified_original + + with _DownloadTarget(temp_dir) as local_path: + try: + object_storage.get_to_path( + manifest.storage.bucket, + manifest.storage.object_key, + local_path, + version_id=manifest.storage.version_id, + ) + except Exception as exc: + raise KnowledgeIngestError( + 400, + "object_read_failed", + f"Could not read object from storage: {manifest.storage.bucket}/{manifest.storage.object_key}", + ) from exc + + verified = _verify_downloaded_object(local_path, manifest) + registration = repo.register_verified_manifest( + manifest, + req.ingest_options, + verified, + idempotency_key, + ) + + if registration.duplicate: + return KnowledgeIngestResponse( + status="duplicate", + ingest_job_id=registration.job_id, + asset_id=manifest.asset.asset_id, + idempotency_key=idempotency_key, + ) + if registration.document_id is None: + raise KnowledgeIngestError(500, "document_registration_failed", "Document was not registered") + + cache(registration.document_id, verified.sha256, verified.local_path) + enqueue(registration.document_id, registration.job_id) + + logger.info( + "knowledge_ingest.accepted", + asset_id=manifest.asset.asset_id, + job_id=str(registration.job_id), + document_id=str(registration.document_id), + ) + return KnowledgeIngestResponse( + status="accepted", + ingest_job_id=registration.job_id, + asset_id=manifest.asset.asset_id, + idempotency_key=idempotency_key, + ) + + +def enqueue_process_document(document_id: uuid.UUID, run_id: uuid.UUID) -> None: + from app.workers.tasks import process_document # noqa: PLC0415 + + process_document.delay(str(document_id), str(run_id)) + + +def cache_verified_original(document_id: uuid.UUID, sha256: str, source_path: Path) -> None: + dest = work_dir_for(document_id) / f"{sha256}.pdf" + if dest.exists() and sha256_file(dest) == sha256: + return + shutil.copyfile(source_path, dest) + + +def load_asset_metadata_for_document(document_id: uuid.UUID) -> dict[str, object]: + with session_scope() as db: + job = ( + db.execute( + select(AssetIngestJob) + .where(AssetIngestJob.document_id == document_id) + .order_by(AssetIngestJob.created_at.desc()) + ) + .scalars() + .first() + ) + if job is None: + return {} + return _metadata_from_manifest(job.manifest_json) + + +def _identity_query(manifest: AssetManifest): + return select(AssetIngestJob).where( + AssetIngestJob.asset_id == manifest.asset.asset_id, + AssetIngestJob.sha256 == manifest.asset.sha256.lower(), + AssetIngestJob.manifest_version == manifest.manifest_version, + ) + + +def _verify_downloaded_object(path: Path, manifest: AssetManifest) -> VerifiedAssetObject: + size = path.stat().st_size + if size != manifest.asset.size_bytes: + raise KnowledgeIngestError( + 400, + "size_mismatch", + f"Object size {size} does not match manifest size {manifest.asset.size_bytes}", + ) + sha = sha256_file(path) + if sha != manifest.asset.sha256.lower(): + raise KnowledgeIngestError( + 400, + "sha256_mismatch", + "Object SHA256 does not match manifest SHA256", + ) + return VerifiedAssetObject(local_path=path, size_bytes=size, sha256=sha) + + +def _storage_uri(manifest: AssetManifest) -> str: + uri = f"s3://{manifest.storage.bucket}/{manifest.storage.object_key}" + if manifest.storage.version_id: + return f"{uri}?versionId={manifest.storage.version_id}" + return uri + + +def _original_filename(manifest: AssetManifest) -> str: + if manifest.asset.original_filename: + return manifest.asset.original_filename + if manifest.asset.title: + return manifest.asset.title + return manifest.storage.object_key.rsplit("/", 1)[-1] or manifest.asset.asset_id + + +def _event_data(manifest: AssetManifest, idempotency_key: str | None) -> dict[str, object]: + return { + "asset_id": manifest.asset.asset_id, + "manifest_version": manifest.manifest_version, + "owner_module": manifest.asset.owner_module, + "owner_record_type": manifest.asset.owner_record_type, + "owner_record_id": manifest.asset.owner_record_id, + "idempotency_key": idempotency_key, + "storage": manifest.storage.model_dump(mode="json"), + "security": manifest.security.model_dump(mode="json"), + "retention": manifest.retention.model_dump(mode="json"), + } + + +def _metadata_from_manifest(manifest: dict[str, object]) -> dict[str, object]: + asset = manifest.get("asset") if isinstance(manifest, dict) else {} + storage = manifest.get("storage") if isinstance(manifest, dict) else {} + retention = manifest.get("retention") if isinstance(manifest, dict) else {} + security = manifest.get("security") if isinstance(manifest, dict) else {} + if not isinstance(asset, dict): + asset = {} + if not isinstance(storage, dict): + storage = {} + if not isinstance(retention, dict): + retention = {} + if not isinstance(security, dict): + security = {} + return { + "asset_id": asset.get("asset_id"), + "owner_module": asset.get("owner_module"), + "owner_record_type": asset.get("owner_record_type"), + "owner_record_id": asset.get("owner_record_id"), + "manifest_version": manifest.get("manifest_version"), + "retention": retention, + "legal_hold": retention.get("legal_hold"), + "security_gate_status": security.get("gate_status"), + "source_storage": storage, + } + + +class _DownloadTarget: + def __init__(self, temp_dir: Path | None) -> None: + self.temp_dir = temp_dir + self._managed_dir: tempfile.TemporaryDirectory[str] | None = None + self.path: Path | None = None + + def __enter__(self) -> Path: + if self.temp_dir is None: + parent = Path(settings.app_work_dir) / "_knowledge_ingest" + parent.mkdir(parents=True, exist_ok=True) + self._managed_dir = tempfile.TemporaryDirectory(dir=parent) + self.path = Path(self._managed_dir.name) / "original" + else: + self.temp_dir.mkdir(parents=True, exist_ok=True) + self.path = self.temp_dir / f"{uuid.uuid4()}.original" + return self.path + + def __exit__(self, exc_type, exc, tb) -> None: + if self._managed_dir is not None: + self._managed_dir.cleanup() diff --git a/app/ingestion/pipeline.py b/app/ingestion/pipeline.py index b11abbe..138f7d5 100644 --- a/app/ingestion/pipeline.py +++ b/app/ingestion/pipeline.py @@ -9,7 +9,7 @@ from __future__ import annotations import json import uuid -from datetime import datetime, timezone +from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -25,16 +25,17 @@ from app.db.models import ( Page, ProcessingEvent, ) -from app.storage.artifacts import ensure_artifact from app.db.session import session_scope from app.indexing import opensearch_client, qdrant_client from app.indexing.embeddings import get_embedder from app.ingestion.chunker import ChunkRecord, chunk_extraction from app.ingestion.docling_extractor import ExtractionResult, extract from app.ingestion.figure_processor import persist_figures +from app.ingestion.knowledge_ingest import load_asset_metadata_for_document from app.ingestion.ocr import run_ocr from app.ingestion.table_processor import persist_tables from app.logging_config import get_logger +from app.storage.artifacts import ensure_artifact from app.storage.local_paths import ( key_docling_json, key_markdown, @@ -47,7 +48,9 @@ from app.utils.language import detect_language logger = get_logger(__name__) -def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) -> dict[str, Any]: +def process_document_id( # noqa: PLR0911, PLR0912, PLR0915 + document_id: uuid.UUID, run_id: uuid.UUID | None = None +) -> dict[str, Any]: """Top-level entry called by the Celery task. Wraps the pipeline in error handling so the task always either succeeds or marks the document FAILED. """ @@ -84,7 +87,7 @@ def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) try: _emit_event(document_id, run_id, DocumentStatus.OCR_STARTED, "OCR started") ocr_result = run_ocr(local_pdf, ocr_pdf, languages=settings.ocr_languages) - except Exception as exc: # noqa: BLE001 + except Exception as exc: logger.exception("pipeline.ocr_failed", document_id=str(document_id)) return _fail(document_id, run_id, DocumentStatus.OCR_FAILED, f"OCR failed: {exc}") @@ -116,7 +119,7 @@ def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) try: _emit_event(document_id, run_id, DocumentStatus.EXTRACTION_STARTED, "Docling extraction started") extraction = extract(ocr_result.output_path) - except Exception as exc: # noqa: BLE001 + except Exception as exc: logger.exception("pipeline.docling_failed", document_id=str(document_id)) return _fail(document_id, run_id, DocumentStatus.EXTRACTION_FAILED, f"Docling failed: {exc}") @@ -140,6 +143,7 @@ def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) chunk_records = chunk_extraction(extraction) sample_text = "\n".join(p.text for p in extraction.pages[:3] if p.text) lang = detect_language(sample_text) + asset_metadata = load_asset_metadata_for_document(document_id) with session_scope() as db: _ensure_artifact(db, document_id, ArtifactType.MARKDOWN, storage.derived_bucket, md_key) @@ -159,7 +163,7 @@ def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) # Replace chunks idempotently: drop all and re-insert. db.execute(delete(Chunk).where(Chunk.document_id == document_id)) for cr in chunk_records: - db.add(_to_chunk_row(document_id, page_id_by_number, cr)) + db.add(_to_chunk_row(document_id, page_id_by_number, cr, asset_metadata)) doc.status = DocumentStatus.CHUNKING_COMPLETED db.add( @@ -192,7 +196,7 @@ def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) for (chunk_id, _text, payload), vec in zip(qdrant_points, vectors, strict=True) ] qdrant_client.upsert_chunks(triples) - except Exception as exc: # noqa: BLE001 + except Exception as exc: logger.exception("pipeline.indexing_failed", document_id=str(document_id)) return _fail(document_id, run_id, DocumentStatus.FAILED, f"Indexing failed: {exc}") @@ -218,8 +222,24 @@ def process_document_id(document_id: uuid.UUID, run_id: uuid.UUID | None = None) # ---------------- helpers ---------------- def _to_chunk_row( - document_id: uuid.UUID, page_id_by_number: dict[int, uuid.UUID], cr: ChunkRecord + document_id: uuid.UUID, + page_id_by_number: dict[int, uuid.UUID], + cr: ChunkRecord, + asset_metadata: dict[str, object] | None = None, ) -> Chunk: + metadata = dict(cr.metadata) + if asset_metadata: + metadata["asset_manifest"] = asset_metadata + for key in ( + "asset_id", + "owner_module", + "owner_record_type", + "owner_record_id", + "manifest_version", + "legal_hold", + ): + if key in asset_metadata: + metadata[key] = asset_metadata[key] return Chunk( document_id=document_id, page_id=page_id_by_number.get(cr.page_number), @@ -232,7 +252,7 @@ def _to_chunk_row( token_count=cr.token_count, ocr_confidence=None, quality_flags=cr.quality_flags, - chunk_metadata=cr.metadata, + chunk_metadata=metadata, ) @@ -305,7 +325,7 @@ def _build_index_payloads( "language_hint": language_hint, "metadata": row.chunk_metadata or {}, "quality_flags": row.quality_flags or {}, - "created_at": (row.created_at or datetime.now(tz=timezone.utc)).isoformat(), + "created_at": (row.created_at or datetime.now(tz=UTC)).isoformat(), } ) text_preview = text[:512] diff --git a/app/storage/minio_client.py b/app/storage/minio_client.py index a7e430f..93e819b 100644 --- a/app/storage/minio_client.py +++ b/app/storage/minio_client.py @@ -83,9 +83,9 @@ class MinioStorage: metadata=metadata or {}, ) - def get_to_path(self, bucket: str, key: str, dest: Path) -> Path: + def get_to_path(self, bucket: str, key: str, dest: Path, version_id: str | None = None) -> Path: dest.parent.mkdir(parents=True, exist_ok=True) - self.client.fget_object(bucket, key, str(dest)) + self.client.fget_object(bucket, key, str(dest), version_id=version_id) return dest def exists(self, bucket: str, key: str) -> bool: diff --git a/tests/test_alembic.py b/tests/test_alembic.py index 1036bff..2a680c6 100644 --- a/tests/test_alembic.py +++ b/tests/test_alembic.py @@ -13,11 +13,13 @@ full backing-service compose stack to be online. from __future__ import annotations -import os import sys from pathlib import Path import pytest +from alembic import command +from alembic.config import Config +from alembic.script import ScriptDirectory ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) @@ -29,9 +31,6 @@ def alembic_cfg(tmp_path, monkeypatch): db_file = tmp_path / "legacyhub.db" monkeypatch.setenv("POSTGRES_HOST", "127.0.0.1") monkeypatch.setenv("POSTGRES_PORT", "5432") - # Force a fresh Settings + Alembic env that ignores the configured PG. - from alembic.config import Config - cfg = Config(str(ROOT / "alembic.ini")) cfg.set_main_option("script_location", str(ROOT / "app" / "db" / "migrations")) cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_file}") @@ -40,11 +39,10 @@ def alembic_cfg(tmp_path, monkeypatch): def test_migration_offline_emits_sql(alembic_cfg, tmp_path): """Offline mode generates SQL for every table; verify ``documents`` appears - and at least one JSONB-equivalent column is rendered. SQLite has no JSONB - but Alembic's offline mode happily emits the raw DDL for inspection. + and the TeamHUB AssetManifest ingest job table is part of the linear head. + SQLite has no JSONB but Alembic's offline mode happily emits the raw DDL for + inspection. """ - from alembic import command - out_file = tmp_path / "upgrade.sql" # ``--sql`` mode bypasses dialect-specific runtime, perfect for a fast check. with out_file.open("w", encoding="utf-8") as f: @@ -59,17 +57,18 @@ def test_migration_offline_emits_sql(alembic_cfg, tmp_path): assert "CREATE TABLE documents" in sql assert "CREATE TABLE chunks" in sql assert "CREATE TABLE processing_events" in sql + assert "CREATE TABLE asset_ingest_jobs" in sql # Constraint sanity assert "uq_chunks_doc_idx" in sql assert "uq_pages_doc_page" in sql + assert "uq_asset_ingest_identity" in sql def test_revision_history_is_linear(alembic_cfg): - """The current project has a single linear history at 0001_initial.""" - from alembic.script import ScriptDirectory - + """The project keeps a single linear migration head.""" script = ScriptDirectory.from_config(alembic_cfg) heads = script.get_heads() assert len(heads) == 1, f"expected one head, got: {heads}" - initial = next(iter(script.walk_revisions())) - assert initial.revision == "0001_initial" + assert heads == ["0002_asset_ingest_jobs"] + revisions = [rev.revision for rev in script.walk_revisions()] + assert revisions == ["0002_asset_ingest_jobs", "0001_initial"] diff --git a/tests/test_knowledge_ingest.py b/tests/test_knowledge_ingest.py new file mode 100644 index 0000000..77b031c --- /dev/null +++ b/tests/test_knowledge_ingest.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +import hashlib +import uuid +from pathlib import Path +from typing import Any + +from fastapi.testclient import TestClient + +from app.api import routes_ingestion +from app.api.schemas import AssetManifestEnvelope, KnowledgeIngestResponse +from app.config import settings +from app.ingestion.knowledge_ingest import ( + ExistingKnowledgeIngestJob, + KnowledgeIngestRegistration, + accept_knowledge_ingest, +) +from app.main import app + + +def _manifest_payload(*, payload: bytes | None = None, gate_status: str = "approved") -> dict[str, Any]: + body = payload or b"%PDF-1.4\nTeamHUB asset manifest\n" + sha = hashlib.sha256(body).hexdigest() + return { + "manifest": { + "manifest_version": "1.0", + "asset": { + "asset_id": "asset_01JZ8W0MAILHUBDOC0001", + "owner_module": "mailhub", + "owner_record_type": "document", + "owner_record_id": "doc_01JZ8W0MAILHUBDOC0001", + "asset_kind": "document", + "title": "incoming-letter.pdf", + "original_filename": "incoming-letter.pdf", + "content_type": "application/pdf", + "size_bytes": len(body), + "sha256": sha, + "created_at": "2026-06-14T12:00:00Z", + "created_by": "staff:123", + }, + "storage": { + "provider": "s3", + "bucket": "teamhub-mailhub-originals", + "object_key": "mailhub/2026/06/14/asset_01JZ8W0/original/incoming-letter.pdf", + "version_id": None, + }, + "security": { + "gate_status": gate_status, + "classification": "internal", + "av_status": "clean", + "content_scan_status": "clean", + "pii_status": "reviewed", + "approved_at": "2026-06-14T12:03:00Z", + "approved_by": "staff:123", + "quarantine_reason": None, + }, + "retention": { + "policy_id": "mailhub-default", + "retain_until": None, + "legal_hold": False, + "legal_hold_reason": None, + }, + "derivatives": [], + "links": [], + "source": { + "correlation_id": "corr_01JZ8W0", + "source_system": "mailhub", + "source_event_id": "evt_01JZ8W0", + }, + }, + "ingest_options": { + "generate_derivatives": True, + "index_full_text": True, + "index_vectors": True, + "emit_events": True, + }, + } + + +def test_knowledge_ingest_route_accepts_asset_manifest(monkeypatch): + job_id = uuid.uuid4() + seen: dict[str, Any] = {} + + def fake_accept(req: AssetManifestEnvelope, *, idempotency_key: str | None = None): + seen["asset_id"] = req.manifest.asset.asset_id + seen["idempotency_key"] = idempotency_key + return KnowledgeIngestResponse( + status="accepted", + ingest_job_id=job_id, + asset_id=req.manifest.asset.asset_id, + idempotency_key=idempotency_key, + ) + + monkeypatch.setattr(routes_ingestion, "accept_knowledge_ingest", fake_accept) + + client = TestClient(app) + res = client.post( + f"{settings.app_api_prefix}/knowledge-ingest", + json=_manifest_payload(), + headers={"Idempotency-Key": "evt_01JZ8W0"}, + ) + + assert res.status_code == 202, res.text + assert res.json() == { + "status": "accepted", + "ingest_job_id": str(job_id), + "asset_id": "asset_01JZ8W0MAILHUBDOC0001", + "idempotency_key": "evt_01JZ8W0", + "reason_code": None, + } + assert seen == { + "asset_id": "asset_01JZ8W0MAILHUBDOC0001", + "idempotency_key": "evt_01JZ8W0", + } + + +def test_knowledge_ingest_rejects_local_filesystem_paths(monkeypatch): + def must_not_run(*_args, **_kwargs): + raise AssertionError("service must not run for invalid local paths") + + monkeypatch.setattr(routes_ingestion, "accept_knowledge_ingest", must_not_run) + body = _manifest_payload() + body["manifest"]["storage"]["object_key"] = "/data/input/incoming-letter.pdf" + + client = TestClient(app) + res = client.post(f"{settings.app_api_prefix}/knowledge-ingest", json=body) + + assert res.status_code == 422 + assert "local filesystem paths" in res.text + + +def test_knowledge_ingest_route_returns_rejected_for_unapproved_gate(): + client = TestClient(app) + res = client.post( + f"{settings.app_api_prefix}/knowledge-ingest", + json=_manifest_payload(gate_status="quarantined"), + ) + + assert res.status_code == 200, res.text + assert res.json()["status"] == "rejected" + assert res.json()["reason_code"] == "security_gate_not_approved" + + +def test_knowledge_ingest_rejects_unapproved_security_gate(): + class MustNotTouchStorage: + def get_to_path(self, *_args, **_kwargs): + raise AssertionError("storage must not be read before security approval") + + class MustNotTouchRepository: + def find_existing_job(self, *_args, **_kwargs): + raise AssertionError("repository must not be read before security approval") + + req = AssetManifestEnvelope.model_validate(_manifest_payload(gate_status="quarantined")) + + res = accept_knowledge_ingest( + req, + storage=MustNotTouchStorage(), + repository=MustNotTouchRepository(), + enqueue_document=lambda _document_id, _job_id: None, + cache_original=lambda _document_id, _sha, _path: None, + ) + + assert res.status == "rejected" + assert res.reason_code == "security_gate_not_approved" + assert res.ingest_job_id is None + + +def test_knowledge_ingest_verifies_object_and_enqueues_once(tmp_path: Path): + payload = b"%PDF-1.4\nverified asset bytes\n" + req = AssetManifestEnvelope.model_validate(_manifest_payload(payload=payload)) + document_id = uuid.uuid4() + job_id = uuid.uuid4() + + class FakeStorage: + def __init__(self) -> None: + self.calls: list[tuple[str, str, str | None]] = [] + + def get_to_path(self, bucket: str, key: str, dest: Path, version_id: str | None = None): + self.calls.append((bucket, key, version_id)) + dest.write_bytes(payload) + return dest + + class FakeRepository: + def __init__(self) -> None: + self.verified_size: int | None = None + self.verified_sha256: str | None = None + + def find_existing_job(self, *_args, **_kwargs): + return None + + def register_verified_manifest(self, manifest, ingest_options, verified, idempotency_key): + self.verified_size = verified.size_bytes + self.verified_sha256 = verified.sha256 + return KnowledgeIngestRegistration( + job_id=job_id, + document_id=document_id, + duplicate=False, + ) + + storage = FakeStorage() + repository = FakeRepository() + enqueued: list[tuple[uuid.UUID, uuid.UUID]] = [] + cached: list[tuple[uuid.UUID, str, Path]] = [] + + res = accept_knowledge_ingest( + req, + idempotency_key="evt_01JZ8W0", + storage=storage, + repository=repository, + enqueue_document=lambda doc_id, run_id: enqueued.append((doc_id, run_id)), + cache_original=lambda doc_id, sha, path: cached.append((doc_id, sha, path)), + temp_dir=tmp_path, + ) + + assert res.status == "accepted" + assert res.ingest_job_id == job_id + assert res.idempotency_key == "evt_01JZ8W0" + assert storage.calls == [ + ( + "teamhub-mailhub-originals", + "mailhub/2026/06/14/asset_01JZ8W0/original/incoming-letter.pdf", + None, + ) + ] + assert repository.verified_size == len(payload) + assert repository.verified_sha256 == hashlib.sha256(payload).hexdigest() + assert enqueued == [(document_id, job_id)] + assert len(cached) == 1 + assert cached[0][0] == document_id + assert cached[0][1] == hashlib.sha256(payload).hexdigest() + assert cached[0][2].read_bytes() == payload + + +def test_knowledge_ingest_duplicate_does_not_read_or_enqueue(): + existing_job_id = uuid.uuid4() + req = AssetManifestEnvelope.model_validate(_manifest_payload()) + + class MustNotTouchStorage: + def get_to_path(self, *_args, **_kwargs): + raise AssertionError("duplicate manifest must not be downloaded again") + + class FakeRepository: + def find_existing_job(self, manifest): + assert manifest.asset.asset_id == "asset_01JZ8W0MAILHUBDOC0001" + return ExistingKnowledgeIngestJob(job_id=existing_job_id) + + def register_verified_manifest(self, *_args, **_kwargs): + raise AssertionError("duplicate manifest must not be registered again") + + res = accept_knowledge_ingest( + req, + storage=MustNotTouchStorage(), + repository=FakeRepository(), + enqueue_document=lambda _document_id, _job_id: (_ for _ in ()).throw( + AssertionError("duplicate manifest must not enqueue a job") + ), + cache_original=lambda _document_id, _sha, _path: None, + ) + + assert res.status == "duplicate" + assert res.ingest_job_id == existing_job_id + assert res.asset_id == "asset_01JZ8W0MAILHUBDOC0001"