"""Governed asynchronous document-upload and review HTTP API.""" from __future__ import annotations import asyncio import re import uuid from datetime import datetime from pathlib import PurePosixPath from typing import Annotated, Literal, cast from fastapi import APIRouter, Depends, Header, Query, Request, status from pydantic import ( BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator, ) from starlette.requests import ClientDisconnect from app.adapters.local_storage import ( LocalStorageError, LocalUploadStorage, StorageErrorCode, ) from app.core.config import Settings, get_settings from app.core.demo_identity import ( ACCESS_SCOPE_ID, BAILIAN_ACCESS_SCOPE_ID, BAILIAN_KNOWLEDGE_BASE_ID, KNOWLEDGE_BASE_ID, ) from app.core.problems import ApiProblem from app.persistence.document_review import ( DocumentReviewConflictError, DocumentReviewError, DocumentReviewNotFoundError, DocumentReviewResult, DocumentReviewStateError, PostgresDocumentReviewRepository, ) from app.persistence.documents import ( CompletedUpload, DocumentActor, DocumentDetail, DocumentListPage, DocumentPersistenceError, DocumentsRepository, DocumentSummary, DocumentUpload, IdempotencyConflictError, PostgresDocumentsRepository, ReviewBlock, ReviewBundle, ReviewChunk, ReviewPage, ReviewVersion, SafeJob, UploadStateConflictError, idempotency_key_hash, upload_request_fingerprint, ) _DOCX_MIME = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" _SECRET = re.compile(r"(?i)(?:sk-[A-Za-z0-9_-]{16,}|Bearer\s+[A-Za-z0-9._~+/-]{16,})") _SYNTHETIC_ACTOR = DocumentActor( subject="synthetic-demo-maintainer", knowledge_base_id=KNOWLEDGE_BASE_ID, access_scope_id=ACCESS_SCOPE_ID, ) _BAILIAN_SYNTHETIC_ACTOR = DocumentActor( subject="synthetic-bailian-maintainer", knowledge_base_id=BAILIAN_KNOWLEDGE_BASE_ID, access_scope_id=BAILIAN_ACCESS_SCOPE_ID, ) class StrictModel(BaseModel): model_config = ConfigDict(extra="forbid") class CreateDocumentUploadRequest(StrictModel): filename: str = Field(min_length=1, max_length=240) declared_mime_type: Literal[ "text/plain", "text/markdown", "application/pdf", "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ] expected_size: int = Field(ge=1, le=2_147_483_648) expected_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") @field_validator("filename") @classmethod def validate_filename(cls, value: str) -> str: if ( value != value.strip() or "\x00" in value or "/" in value or "\\" in value or _SECRET.search(value) ): raise ValueError("filename is not safe") return value @field_validator("declared_mime_type") @classmethod def validate_extension_matches_mime(cls, mime: str, info: ValidationInfo) -> str: filename = cast_filename(info.data.get("filename")) suffix = PurePosixPath(filename).suffix.lower() accepted = { ".txt": {"text/plain"}, ".md": {"text/plain", "text/markdown"}, ".markdown": {"text/plain", "text/markdown"}, ".pdf": {"application/pdf"}, ".docx": {_DOCX_MIME}, } if mime not in accepted.get(suffix, set()): raise ValueError("filename extension does not match media type") return mime class UploadResponse(StrictModel): id: uuid.UUID filename: str declared_mime_type: str expected_size: int expected_sha256: str actual_size: int | None actual_sha256: str | None status: Literal["CREATED", "STORED", "COMPLETED"] document_id: uuid.UUID | None parse_job_id: uuid.UUID | None created_at: datetime updated_at: datetime completed_at: datetime | None replayed: bool = False class SafeJobResponse(StrictModel): id: uuid.UUID job_type: str stage: str status: Literal["QUEUED", "RUNNING", "SUCCEEDED", "FAILED", "CANCELLED"] progress: int = Field(ge=0, le=100) attempt: int = Field(ge=0) max_attempts: int = Field(ge=1) last_error_code: str | None created_at: datetime updated_at: datetime finished_at: datetime | None class DocumentSummaryResponse(StrictModel): id: uuid.UUID filename: str mime_type: str raw_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") status: str active_version_id: uuid.UUID | None created_at: datetime updated_at: datetime class DocumentDetailResponse(StrictModel): document: DocumentSummaryResponse version_count: int = Field(ge=0) page_count: int = Field(ge=0) block_count: int = Field(ge=0) chunk_count: int = Field(ge=0) class DocumentListResponse(StrictModel): items: list[DocumentSummaryResponse] next_cursor: uuid.UUID | None class CompleteUploadResponse(StrictModel): upload: UploadResponse document: DocumentSummaryResponse job: SafeJobResponse class ReviewVersionResponse(StrictModel): id: uuid.UUID review_state: str review_revision: int = Field(ge=0) status: str parser_profile_hash: str chunk_profile_hash: str cloud_policy_id: str outbound_manifest_sha256: str | None expected_chunk_count: int | None error_code: str | None created_at: datetime completed_at: datetime | None class ReviewPageResponse(StrictModel): id: uuid.UUID ordinal: int page_number: int | None text: str text_sha256: str line_start: int line_end: int class ReviewBlockResponse(StrictModel): id: uuid.UUID ordinal: int kind: str text: str text_sha256: str section_path: list[str] anchor_id: str char_start: int char_end: int line_start: int line_end: int page_start: int | None page_end: int | None class ReviewChunkResponse(StrictModel): ordinal: int display_text: str cloud_text: str cloud_text_sha256: str embedding_text_sha256: str token_count: int page_start: int | None page_end: int | None section_path: list[str] approval_status: str index_status: str class ReviewBundleResponse(StrictModel): document: DocumentSummaryResponse version: ReviewVersionResponse | None pages: list[ReviewPageResponse] blocks: list[ReviewBlockResponse] chunks: list[ReviewChunkResponse] next_ordinal: int | None class DocumentReviewDecisionRequest(StrictModel): decision: Literal["APPROVE", "REJECT"] reason_code: Literal[ "SYNTHETIC_REVIEW_APPROVED", "RIGHTS_NOT_VERIFIED", "CONTENT_QUALITY_REJECTED", "CLOUD_PROCESSING_REJECTED", ] expected_revision: int = Field(ge=0) outbound_manifest_sha256: str | None = Field( default=None, pattern=r"^[0-9a-f]{64}$", ) @model_validator(mode="after") def validate_decision_contract(self) -> DocumentReviewDecisionRequest: if self.decision == "APPROVE": if ( self.reason_code != "SYNTHETIC_REVIEW_APPROVED" or self.outbound_manifest_sha256 is None ): raise ValueError("approval requires the reviewed manifest") elif ( self.reason_code == "SYNTHETIC_REVIEW_APPROVED" or self.outbound_manifest_sha256 is not None ): raise ValueError("rejection requires a rejection reason and no manifest") return self class DocumentReviewDecisionResponse(StrictModel): document_id: uuid.UUID document_version_id: uuid.UUID decision: Literal["APPROVE", "REJECT"] review_state: Literal["CLOUD_APPROVED", "REJECTED"] review_revision: int = Field(ge=1) outbound_manifest_sha256: str | None embedding_profile_hash: str | None job: SafeJobResponse | None def cast_filename(value: object) -> str: return value if isinstance(value, str) else "" def get_document_actor( settings: Annotated[Settings, Depends(get_settings)], ) -> DocumentActor: """Return a server-owned synthetic grant; requests cannot select a scope.""" if settings.document_namespace_mode == "bailian": return _BAILIAN_SYNTHETIC_ACTOR return _SYNTHETIC_ACTOR def get_documents_repository( settings: Annotated[Settings, Depends(get_settings)], ) -> DocumentsRepository: return PostgresDocumentsRepository(settings) def get_upload_storage( settings: Annotated[Settings, Depends(get_settings)], ) -> LocalUploadStorage: return LocalUploadStorage( settings.upload_root, max_bytes=settings.max_upload_mb * 1024 * 1024, ) def get_document_review_repository( settings: Annotated[Settings, Depends(get_settings)], ) -> PostgresDocumentReviewRepository: return PostgresDocumentReviewRepository(settings) def parse_idempotency_key( value: Annotated[str, Header(alias="Idempotency-Key")], ) -> uuid.UUID: try: return uuid.UUID(value) except (ValueError, AttributeError): raise ApiProblem( status=422, code="IDEMPOTENCY_KEY_INVALID", title="Idempotency key is invalid", detail="Idempotency-Key must be a UUID.", ) from None router = APIRouter( prefix="/api/v1", tags=["documents"], ) @router.post( "/document-uploads", response_model=UploadResponse, status_code=status.HTTP_201_CREATED, operation_id="createDocumentUpload", ) def create_document_upload( body: CreateDocumentUploadRequest, request: Request, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], settings: Annotated[Settings, Depends(get_settings)], key: Annotated[uuid.UUID, Depends(parse_idempotency_key)], ) -> UploadResponse: if body.expected_size > settings.max_upload_mb * 1024 * 1024: raise ApiProblem( status=413, code="UPLOAD_TOO_LARGE", title="Upload is too large", detail="The declared upload size exceeds the configured limit.", ) fingerprint = upload_request_fingerprint( filename=body.filename, declared_mime_type=body.declared_mime_type, expected_size=body.expected_size, expected_sha256=body.expected_sha256, ) try: upload, created = repository.create_upload( actor=actor, idempotency_key_hash=idempotency_key_hash(actor, key), request_fingerprint=fingerprint, filename=body.filename, declared_mime_type=body.declared_mime_type, expected_size=body.expected_size, expected_sha256=body.expected_sha256, storage_key=uuid.uuid4(), trace_id=_trace_id(request), ) except IdempotencyConflictError: raise ApiProblem( status=409, code="IDEMPOTENCY_CONFLICT", title="Idempotency conflict", detail="The key was already used for a different upload declaration.", ) from None except DocumentPersistenceError: raise _persistence_problem() from None return _upload_response(upload, replayed=not created) @router.put( "/document-uploads/{upload_id}/content", response_model=UploadResponse, operation_id="storeDocumentUploadContent", openapi_extra={ "requestBody": { "required": True, "content": { "application/octet-stream": {"schema": {"type": "string", "format": "binary"}} }, } }, ) async def store_document_upload_content( upload_id: uuid.UUID, request: Request, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], storage: Annotated[LocalUploadStorage, Depends(get_upload_storage)], ) -> UploadResponse: if request.headers.get("content-type", "").split(";", 1)[0].strip().lower() != ( "application/octet-stream" ): raise ApiProblem( status=415, code="UPLOAD_CONTENT_TYPE_INVALID", title="Upload content type is invalid", detail="Upload bytes require application/octet-stream.", ) try: upload = await asyncio.to_thread(repository.get_upload, actor, upload_id) except DocumentPersistenceError: raise _persistence_problem() from None if upload is None: raise _not_found_problem() if upload.status == "COMPLETED": return _upload_response(upload) declared_length = request.headers.get("content-length") if declared_length is not None: try: length = int(declared_length) except ValueError: raise ApiProblem( status=400, code="CONTENT_LENGTH_INVALID", title="Content length is invalid", detail="Content-Length must be a decimal byte count.", ) from None if length != upload.expected_size: raise ApiProblem( status=422, code="UPLOAD_SIZE_MISMATCH", title="Upload size mismatch", detail="The streamed byte count must match the upload declaration.", ) try: stored = await storage.store( storage_key=upload.storage_key, chunks=request.stream(), expected_size=upload.expected_size, expected_sha256=upload.expected_sha256, ) updated = await asyncio.to_thread( repository.mark_upload_stored, actor=actor, upload_id=upload_id, actual_size=stored.byte_size, actual_sha256=stored.sha256, trace_id=_trace_id(request), ) except ClientDisconnect: raise ApiProblem( status=400, code="UPLOAD_INTERRUPTED", title="Upload was interrupted", detail="The upload stream ended before it was complete.", ) from None except LocalStorageError as exc: raise _storage_problem(exc.code) from None except UploadStateConflictError: raise _state_problem() from None except DocumentPersistenceError: raise _persistence_problem() from None return _upload_response(updated) @router.post( "/document-uploads/{upload_id}/complete", response_model=CompleteUploadResponse, status_code=status.HTTP_202_ACCEPTED, operation_id="completeDocumentUpload", ) def complete_document_upload( upload_id: uuid.UUID, request: Request, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], ) -> CompleteUploadResponse: try: completed = repository.complete_upload( actor=actor, upload_id=upload_id, trace_id=_trace_id(request), ) except UploadStateConflictError: raise _state_problem() from None except DocumentPersistenceError: raise _persistence_problem() from None return _completed_response(completed) @router.get( "/document-uploads/{upload_id}", response_model=UploadResponse, operation_id="getDocumentUpload", ) def get_document_upload( upload_id: uuid.UUID, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], ) -> UploadResponse: try: upload = repository.get_upload(actor, upload_id) except DocumentPersistenceError: raise _persistence_problem() from None if upload is None: raise _not_found_problem() return _upload_response(upload) @router.get( "/document-jobs/{job_id}", response_model=SafeJobResponse, operation_id="getDocumentJob", ) def get_document_job( job_id: uuid.UUID, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], ) -> SafeJobResponse: try: job = repository.get_job(actor, job_id) except DocumentPersistenceError: raise _persistence_problem() from None if job is None: raise _not_found_problem() return _job_response(job) @router.get( "/documents", response_model=DocumentListResponse, operation_id="listDocuments", ) def list_documents( actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], cursor: Annotated[uuid.UUID | None, Query()] = None, limit: Annotated[int, Query(ge=1, le=100)] = 20, ) -> DocumentListResponse: try: page = repository.list_documents(actor, cursor=cursor, limit=limit) except DocumentPersistenceError: raise _persistence_problem() from None return _document_list_response(page) @router.get( "/documents/{document_id}", response_model=DocumentDetailResponse, operation_id="getDocument", ) def get_document( document_id: uuid.UUID, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], ) -> DocumentDetailResponse: try: detail = repository.get_document(actor, document_id) except DocumentPersistenceError: raise _persistence_problem() from None if detail is None: raise _not_found_problem() return _document_detail_response(detail) @router.get( "/documents/{document_id}/review-bundle", response_model=ReviewBundleResponse, operation_id="getDocumentReviewBundle", ) def get_document_review_bundle( document_id: uuid.UUID, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[DocumentsRepository, Depends(get_documents_repository)], after_ordinal: Annotated[int, Query(ge=-1)] = -1, limit: Annotated[int, Query(ge=1, le=100)] = 50, ) -> ReviewBundleResponse: try: bundle = repository.get_review_bundle( actor, document_id, after_ordinal=after_ordinal, limit=limit, ) except DocumentPersistenceError: raise _persistence_problem() from None if bundle is None: raise _not_found_problem() return _review_bundle_response(bundle) @router.post( "/documents/{document_id}/review-decisions", response_model=DocumentReviewDecisionResponse, status_code=status.HTTP_202_ACCEPTED, operation_id="createDocumentReviewDecision", ) def create_document_review_decision( document_id: uuid.UUID, body: DocumentReviewDecisionRequest, request: Request, actor: Annotated[DocumentActor, Depends(get_document_actor)], repository: Annotated[ PostgresDocumentReviewRepository, Depends(get_document_review_repository), ], ) -> DocumentReviewDecisionResponse: try: result = repository.apply_decision( actor=actor, document_id=document_id, decision=body.decision, reason_code=body.reason_code, expected_revision=body.expected_revision, outbound_manifest_sha256=body.outbound_manifest_sha256, trace_id=_trace_id(request), ) except DocumentReviewNotFoundError: raise _not_found_problem() from None except DocumentReviewConflictError: raise ApiProblem( status=412, code="REVIEW_REVISION_CONFLICT", title="Review revision conflict", detail="The review bundle changed; reload it before deciding.", ) from None except DocumentReviewStateError: raise ApiProblem( status=409, code="REVIEW_STATE_CONFLICT", title="Document is not reviewable", detail="The latest document version is not eligible for this decision.", ) from None except DocumentReviewError: raise _persistence_problem() from None return _review_decision_response(result) def _trace_id(request: Request) -> uuid.UUID: try: return uuid.UUID(str(request.state.trace_id)) except (ValueError, AttributeError): return uuid.uuid4() def _upload_response(upload: DocumentUpload, *, replayed: bool = False) -> UploadResponse: return UploadResponse( id=upload.id, filename=upload.filename, declared_mime_type=upload.declared_mime_type, expected_size=upload.expected_size, expected_sha256=upload.expected_sha256, actual_size=upload.actual_size, actual_sha256=upload.actual_sha256, status=cast_upload_status(upload.status), document_id=upload.document_id, parse_job_id=upload.parse_job_id, created_at=upload.created_at, updated_at=upload.updated_at, completed_at=upload.completed_at, replayed=replayed, ) def cast_upload_status(value: str) -> Literal["CREATED", "STORED", "COMPLETED"]: if value not in {"CREATED", "STORED", "COMPLETED"}: raise DocumentPersistenceError return cast(Literal["CREATED", "STORED", "COMPLETED"], value) def _job_response(job: SafeJob) -> SafeJobResponse: return SafeJobResponse( id=job.id, job_type=job.job_type, stage=job.stage, status=cast_job_status(job.status), progress=job.progress, attempt=job.attempt, max_attempts=job.max_attempts, last_error_code=job.last_error_code, created_at=job.created_at, updated_at=job.updated_at, finished_at=job.finished_at, ) def cast_job_status( value: str, ) -> Literal["QUEUED", "RUNNING", "SUCCEEDED", "FAILED", "CANCELLED"]: if value not in {"QUEUED", "RUNNING", "SUCCEEDED", "FAILED", "CANCELLED"}: raise DocumentPersistenceError return cast(Literal["QUEUED", "RUNNING", "SUCCEEDED", "FAILED", "CANCELLED"], value) def _document_response(document: DocumentSummary) -> DocumentSummaryResponse: return DocumentSummaryResponse( id=document.id, filename=document.filename, mime_type=document.mime_type, raw_sha256=document.raw_sha256, status=document.status, active_version_id=document.active_version_id, created_at=document.created_at, updated_at=document.updated_at, ) def _document_detail_response(detail: DocumentDetail) -> DocumentDetailResponse: return DocumentDetailResponse( document=_document_response(detail.document), version_count=detail.version_count, page_count=detail.page_count, block_count=detail.block_count, chunk_count=detail.chunk_count, ) def _document_list_response(page: DocumentListPage) -> DocumentListResponse: return DocumentListResponse( items=[_document_response(item) for item in page.items], next_cursor=page.next_cursor, ) def _completed_response(value: CompletedUpload) -> CompleteUploadResponse: return CompleteUploadResponse( upload=_upload_response(value.upload), document=_document_response(value.document), job=_job_response(value.job), ) def _review_version_response(value: ReviewVersion) -> ReviewVersionResponse: return ReviewVersionResponse( id=value.id, review_state=value.review_state, review_revision=value.review_revision, status=value.status, parser_profile_hash=value.parser_profile_hash, chunk_profile_hash=value.chunk_profile_hash, cloud_policy_id=value.cloud_policy_id, outbound_manifest_sha256=value.outbound_manifest_sha256, expected_chunk_count=value.expected_chunk_count, error_code=value.error_code, created_at=value.created_at, completed_at=value.completed_at, ) def _review_page_response(value: ReviewPage) -> ReviewPageResponse: return ReviewPageResponse( id=value.id, ordinal=value.ordinal, page_number=value.page_number, text=value.text, text_sha256=value.text_sha256, line_start=value.line_start, line_end=value.line_end, ) def _review_block_response(value: ReviewBlock) -> ReviewBlockResponse: return ReviewBlockResponse( id=value.id, ordinal=value.ordinal, kind=value.kind, text=value.text, text_sha256=value.text_sha256, section_path=list(value.section_path), anchor_id=value.anchor_id, char_start=value.char_start, char_end=value.char_end, line_start=value.line_start, line_end=value.line_end, page_start=value.page_start, page_end=value.page_end, ) def _review_chunk_response(value: ReviewChunk) -> ReviewChunkResponse: return ReviewChunkResponse( ordinal=value.ordinal, display_text=value.display_text, cloud_text=value.cloud_text, cloud_text_sha256=value.cloud_text_sha256, embedding_text_sha256=value.embedding_text_sha256, token_count=value.token_count, page_start=value.page_start, page_end=value.page_end, section_path=list(value.section_path), approval_status=value.approval_status, index_status=value.index_status, ) def _review_bundle_response(value: ReviewBundle) -> ReviewBundleResponse: return ReviewBundleResponse( document=_document_response(value.document), version=(_review_version_response(value.version) if value.version is not None else None), pages=[_review_page_response(item) for item in value.pages], blocks=[_review_block_response(item) for item in value.blocks], chunks=[_review_chunk_response(item) for item in value.chunks], next_ordinal=value.next_ordinal, ) def _review_decision_response( value: DocumentReviewResult, ) -> DocumentReviewDecisionResponse: return DocumentReviewDecisionResponse( document_id=value.document_id, document_version_id=value.document_version_id, decision=value.decision, review_state=value.review_state, review_revision=value.review_revision, outbound_manifest_sha256=value.outbound_manifest_sha256, embedding_profile_hash=value.embedding_profile_hash, job=_job_response(value.job) if value.job is not None else None, ) def _not_found_problem() -> ApiProblem: return ApiProblem( status=404, code="DOCUMENT_RESOURCE_NOT_FOUND", title="Document resource not found", detail="The requested resource is unavailable to the current identity.", ) def _state_problem() -> ApiProblem: return ApiProblem( status=409, code="UPLOAD_STATE_CONFLICT", title="Upload state conflict", detail="The upload is not ready for this operation.", ) def _persistence_problem() -> ApiProblem: return ApiProblem( status=503, code="DOCUMENT_PERSISTENCE_UNAVAILABLE", title="Document service unavailable", detail="Document metadata is temporarily unavailable.", ) def _storage_problem(code: StorageErrorCode) -> ApiProblem: mapping = { StorageErrorCode.TOO_LARGE: (413, "UPLOAD_TOO_LARGE", "Upload is too large"), StorageErrorCode.SIZE_MISMATCH: ( 422, "UPLOAD_SIZE_MISMATCH", "Upload size mismatch", ), StorageErrorCode.HASH_MISMATCH: ( 422, "UPLOAD_HASH_MISMATCH", "Upload digest mismatch", ), StorageErrorCode.OBJECT_CONFLICT: ( 409, "UPLOAD_OBJECT_CONFLICT", "Upload object conflict", ), StorageErrorCode.INVALID_CONTRACT: ( 422, "UPLOAD_CONTRACT_INVALID", "Upload contract is invalid", ), StorageErrorCode.ROOT_UNSAFE: ( 503, "UPLOAD_STORAGE_UNSAFE", "Upload storage unavailable", ), StorageErrorCode.IO_UNAVAILABLE: ( 503, "UPLOAD_STORAGE_UNAVAILABLE", "Upload storage unavailable", ), } status_code, public_code, title = mapping[code] return ApiProblem( status=status_code, code=public_code, title=title, detail="The upload content could not be stored under the declared contract.", )