From ff0b795b6da856d815dec42209bd908411309891 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Wed, 29 Apr 2026 15:34:25 +0200 Subject: [PATCH] feat(#202): introduce Document lifecycle state machine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a first-class lifecycle state to every document, distinct from AnalysisJob.status. The lifecycle describes the document as a whole and is the foundation for the doc-centric pivot in 0.6.0. Domain - DocumentLifecycleState enum (Uploaded/Parsed/Chunked/Ingested/Stale/Failed) - Document.lifecycle_state and lifecycle_state_at fields - Document.transition_to() validates against a transition table (domain/lifecycle.py) and returns a DocumentLifecycleChanged event - InvalidLifecycleTransitionError on disallowed transitions Persistence - ALTER TABLE documents to add the two columns (default 'Uploaded') - New index idx_documents_lifecycle_state for filter perf - _COLUMN_MIGRATIONS refactored to support multiple tables - _POST_MIGRATION_DDL list for indexes on freshly-added columns - SqliteDocumentRepository.update_lifecycle() Services - AnalysisService drives transitions on parse / chunk / re-chunk / fail via _transition_document(); idempotent and resilient (logs WARN and continues if a stale state is somehow encountered) API - DocumentResponse exposes lifecycleState + lifecycleStateAt (additive — existing 'status' field kept for backwards compat) Frontend - Document type extended with lifecycleState and lifecycleStateAt - DocumentLifecycleState union literal mirroring the backend enum Tests - 24 new tests in test_lifecycle.py covering transitions, idempotency, invariant preservation, and event emission - test_repos.py: round-trip + every-enum-value check + update_lifecycle - test_chunking.py: rechunk path now mocks document_repo correctly Refs #202 --- document-parser/api/documents.py | 2 + document-parser/api/schemas.py | 5 + document-parser/domain/events.py | 35 +++++ document-parser/domain/exceptions.py | 37 ++++++ document-parser/domain/lifecycle.py | 83 ++++++++++++ document-parser/domain/models.py | 34 +++++ document-parser/domain/ports.py | 10 ++ document-parser/domain/value_objects.py | 25 ++++ document-parser/persistence/database.py | 59 +++++++-- document-parser/persistence/document_repo.py | 48 ++++++- document-parser/services/analysis_service.py | 51 ++++++++ document-parser/tests/test_chunking.py | 9 +- document-parser/tests/test_lifecycle.py | 129 +++++++++++++++++++ document-parser/tests/test_repos.py | 43 ++++++- frontend/src/shared/types.ts | 23 ++++ 15 files changed, 578 insertions(+), 15 deletions(-) create mode 100644 document-parser/domain/events.py create mode 100644 document-parser/domain/exceptions.py create mode 100644 document-parser/domain/lifecycle.py create mode 100644 document-parser/tests/test_lifecycle.py diff --git a/document-parser/api/documents.py b/document-parser/api/documents.py index 1289166..55f2c8e 100644 --- a/document-parser/api/documents.py +++ b/document-parser/api/documents.py @@ -34,6 +34,8 @@ def _to_response(doc) -> DocumentResponse: file_size=doc.file_size, page_count=doc.page_count, created_at=str(doc.created_at), + lifecycle_state=doc.lifecycle_state.value, + lifecycle_state_at=(str(doc.lifecycle_state_at) if doc.lifecycle_state_at else None), ) diff --git a/document-parser/api/schemas.py b/document-parser/api/schemas.py index 7b7d82e..707e146 100644 --- a/document-parser/api/schemas.py +++ b/document-parser/api/schemas.py @@ -56,6 +56,11 @@ class DocumentResponse(_CamelModel): file_size: int | None = None page_count: int | None = None created_at: str | datetime + # 0.6.0 — Document lifecycle state machine (#202). The lifecycle + # describes the document as a whole; `status` above is kept for + # backwards compat and currently still maps to `DOCUMENT_STATUS_UPLOADED`. + lifecycle_state: str = "Uploaded" + lifecycle_state_at: str | datetime | None = None class AnalysisResponse(_CamelModel): diff --git a/document-parser/domain/events.py b/document-parser/domain/events.py new file mode 100644 index 0000000..0b8c21d --- /dev/null +++ b/document-parser/domain/events.py @@ -0,0 +1,35 @@ +"""Domain events — frozen records that document state transitions. + +Events are produced by domain operations (typically returned from a +mutation method on an aggregate). They are pure data — no event bus is +wired in 0.6.0; services can choose to log, persist, or publish them +later. Keeping them here keeps the domain layer free of any event-bus +infrastructure. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from datetime import datetime + + from domain.value_objects import DocumentLifecycleState + + +@dataclass(frozen=True) +class DocumentLifecycleChanged: + """A Document lifecycle transition occurred. + + Attributes: + document_id: id of the document that transitioned. + previous: state the document was in before the transition. + current: state the document is in after the transition. + at: timestamp of the transition (UTC). + """ + + document_id: str + previous: DocumentLifecycleState + current: DocumentLifecycleState + at: datetime diff --git a/document-parser/domain/exceptions.py b/document-parser/domain/exceptions.py new file mode 100644 index 0000000..ce6d61b --- /dev/null +++ b/document-parser/domain/exceptions.py @@ -0,0 +1,37 @@ +"""Domain-level exceptions. + +Exceptions defined in this module are raised by domain operations and value +objects when an invariant is violated. They have no infrastructure +dependencies and are safe to import from any layer. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from domain.value_objects import DocumentLifecycleState + + +class DomainError(Exception): + """Base class for domain-level errors. Catch this when wiring API + layers if you want a single hook for any invariant violation.""" + + +class InvalidLifecycleTransitionError(DomainError): + """Raised when a Document.transition_to() call asks for a (source, + target) pair that is not in the allowed transition table. + + Carries `source` and `target` so callers can produce a useful error + message without re-discovering them. + """ + + def __init__( + self, + *, + source: DocumentLifecycleState, + target: DocumentLifecycleState, + ) -> None: + super().__init__(f"Invalid document lifecycle transition: {source.value} -> {target.value}") + self.source = source + self.target = target diff --git a/document-parser/domain/lifecycle.py b/document-parser/domain/lifecycle.py new file mode 100644 index 0000000..e2a0eb3 --- /dev/null +++ b/document-parser/domain/lifecycle.py @@ -0,0 +1,83 @@ +"""Document lifecycle state machine — pure domain logic. + +Defines the canonical state model for a Document in Docling Studio: + Uploaded → Parsed → Chunked → Ingested → (Stale|Chunked) → Ingested + +Stale is set by the auto-detect logic (#204) — never reached by a manual +action. Failed is reachable from any state and represents a pipeline error. + +This module contains: + - The transition table (which (from → to) pairs are allowed). + - `is_allowed_transition`: the pure check. + - `assert_transition`: raises `InvalidLifecycleTransitionError` when not allowed. + +The dataclass that carries the state lives in `domain.models.Document`. The +domain event emitted on transition lives in `domain.events`. +""" + +from __future__ import annotations + +from domain.exceptions import InvalidLifecycleTransitionError +from domain.value_objects import DocumentLifecycleState + +# Allowed transitions: from → set of allowed targets. +# Failed is always reachable as a terminal-ish state (see notes below). +# Self-loops are allowed only where they make pipeline sense (e.g. re-chunk). +_TRANSITIONS: dict[DocumentLifecycleState, frozenset[DocumentLifecycleState]] = { + DocumentLifecycleState.UPLOADED: frozenset( + { + DocumentLifecycleState.PARSED, + DocumentLifecycleState.CHUNKED, # parse + chunk in one pipeline call + DocumentLifecycleState.FAILED, + } + ), + DocumentLifecycleState.PARSED: frozenset( + { + DocumentLifecycleState.PARSED, # idempotent re-parse + DocumentLifecycleState.CHUNKED, + DocumentLifecycleState.FAILED, + } + ), + DocumentLifecycleState.CHUNKED: frozenset( + { + DocumentLifecycleState.CHUNKED, # re-chunk + DocumentLifecycleState.INGESTED, + DocumentLifecycleState.FAILED, + } + ), + DocumentLifecycleState.INGESTED: frozenset( + { + DocumentLifecycleState.STALE, + DocumentLifecycleState.CHUNKED, # re-chunk after ingest + DocumentLifecycleState.INGESTED, # re-push (idempotent) + DocumentLifecycleState.FAILED, + } + ), + DocumentLifecycleState.STALE: frozenset( + { + DocumentLifecycleState.INGESTED, + DocumentLifecycleState.CHUNKED, + DocumentLifecycleState.FAILED, + } + ), + # Failed is a recoverable state — the operator (or a retry) can move + # back to a non-terminal state by re-running the relevant pipeline step. + DocumentLifecycleState.FAILED: frozenset( + { + DocumentLifecycleState.UPLOADED, + DocumentLifecycleState.PARSED, + DocumentLifecycleState.CHUNKED, + } + ), +} + + +def is_allowed_transition(source: DocumentLifecycleState, target: DocumentLifecycleState) -> bool: + """Return True iff transitioning from `source` to `target` is allowed.""" + return target in _TRANSITIONS.get(source, frozenset()) + + +def assert_transition(source: DocumentLifecycleState, target: DocumentLifecycleState) -> None: + """Raise `InvalidLifecycleTransitionError` if the transition is not allowed.""" + if not is_allowed_transition(source, target): + raise InvalidLifecycleTransitionError(source=source, target=target) diff --git a/document-parser/domain/models.py b/document-parser/domain/models.py index e89883d..0a371ee 100644 --- a/document-parser/domain/models.py +++ b/document-parser/domain/models.py @@ -7,6 +7,10 @@ import uuid from dataclasses import dataclass, field from datetime import UTC, datetime +from domain.events import DocumentLifecycleChanged +from domain.lifecycle import assert_transition +from domain.value_objects import DocumentLifecycleState + class AnalysisStatus(enum.StrEnum): PENDING = "PENDING" @@ -32,6 +36,36 @@ class Document: page_count: int | None = None storage_path: str = "" created_at: datetime = field(default_factory=_utcnow) + lifecycle_state: DocumentLifecycleState = DocumentLifecycleState.UPLOADED + lifecycle_state_at: datetime | None = None + + def transition_to( + self, + target: DocumentLifecycleState, + *, + now: datetime | None = None, + ) -> DocumentLifecycleChanged: + """Move the document to `target`, validating the transition. + + Returns the corresponding `DocumentLifecycleChanged` event so the + caller (typically a service) can log / persist / publish it. The + event is pure data — no event bus is wired in 0.6.0. + + Raises: + InvalidLifecycleTransitionError: if (current → target) is not in + the allowed transition table. + """ + previous = self.lifecycle_state + assert_transition(previous, target) + at = now or _utcnow() + self.lifecycle_state = target + self.lifecycle_state_at = at + return DocumentLifecycleChanged( + document_id=self.id, + previous=previous, + current=target, + at=at, + ) @dataclass diff --git a/document-parser/domain/ports.py b/document-parser/domain/ports.py index 9431b21..967871d 100644 --- a/document-parser/domain/ports.py +++ b/document-parser/domain/ports.py @@ -9,12 +9,15 @@ from __future__ import annotations from typing import TYPE_CHECKING, Protocol, runtime_checkable if TYPE_CHECKING: + from datetime import datetime + from domain.models import AnalysisJob, Document from domain.value_objects import ( ChunkingOptions, ChunkResult, ConversionOptions, ConversionResult, + DocumentLifecycleState, LLMProviderType, ReasoningResult, ) @@ -85,6 +88,13 @@ class DocumentRepository(Protocol): async def update_page_count(self, doc_id: str, page_count: int) -> None: ... + async def update_lifecycle( + self, + doc_id: str, + state: DocumentLifecycleState, + at: datetime, + ) -> None: ... + async def delete(self, doc_id: str) -> bool: ... diff --git a/document-parser/domain/value_objects.py b/document-parser/domain/value_objects.py index 72132f5..784dd2c 100644 --- a/document-parser/domain/value_objects.py +++ b/document-parser/domain/value_objects.py @@ -14,6 +14,31 @@ DEFAULT_PAGE_WIDTH: float = 612.0 DEFAULT_PAGE_HEIGHT: float = 792.0 +class DocumentLifecycleState(StrEnum): + """Canonical lifecycle of a Document in Docling Studio. + + Distinct from `AnalysisStatus` (which describes a single conversion + attempt). The lifecycle describes the document as a whole: + + Uploaded raw file persisted, no parse yet + Parsed conversion produced a document tree + Chunked chunker produced a draft chunkset (pre-store) + Ingested chunkset has been embedded into at least one store + Stale a chunkset was edited after a successful push and the + corresponding store no longer matches (#204) + Failed a pipeline step failed; recoverable by retry + + Allowed transitions live in `domain.lifecycle._TRANSITIONS`. + """ + + UPLOADED = "Uploaded" + PARSED = "Parsed" + CHUNKED = "Chunked" + INGESTED = "Ingested" + STALE = "Stale" + FAILED = "Failed" + + @dataclass(frozen=True) class PageElement: type: str diff --git a/document-parser/persistence/database.py b/document-parser/persistence/database.py index c56620e..b79f5db 100644 --- a/document-parser/persistence/database.py +++ b/document-parser/persistence/database.py @@ -45,22 +45,59 @@ CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at); """ -_MIGRATIONS = [ - ("document_json", "ALTER TABLE analysis_jobs ADD COLUMN document_json TEXT"), - ("chunks_json", "ALTER TABLE analysis_jobs ADD COLUMN chunks_json TEXT"), - ("progress_current", "ALTER TABLE analysis_jobs ADD COLUMN progress_current INTEGER"), - ("progress_total", "ALTER TABLE analysis_jobs ADD COLUMN progress_total INTEGER"), +# Column migrations: (table, column_name, ddl). Idempotent — applied only +# when the column is missing from the live schema. Order matters when a +# later migration depends on an earlier one (none today). +_COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ + ("analysis_jobs", "document_json", "ALTER TABLE analysis_jobs ADD COLUMN document_json TEXT"), + ("analysis_jobs", "chunks_json", "ALTER TABLE analysis_jobs ADD COLUMN chunks_json TEXT"), + ( + "analysis_jobs", + "progress_current", + "ALTER TABLE analysis_jobs ADD COLUMN progress_current INTEGER", + ), + ( + "analysis_jobs", + "progress_total", + "ALTER TABLE analysis_jobs ADD COLUMN progress_total INTEGER", + ), + # 0.6.0 — Document lifecycle state machine (#202). + ( + "documents", + "lifecycle_state", + "ALTER TABLE documents ADD COLUMN lifecycle_state TEXT NOT NULL DEFAULT 'Uploaded'", + ), + ( + "documents", + "lifecycle_state_at", + "ALTER TABLE documents ADD COLUMN lifecycle_state_at TEXT", + ), +] + +# DDL statements run after column migrations — typically CREATE INDEX +# IF NOT EXISTS for indexes on freshly-added columns. +_POST_MIGRATION_DDL: list[str] = [ + "CREATE INDEX IF NOT EXISTS idx_documents_lifecycle_state ON documents(lifecycle_state)", ] async def _run_migrations(db: aiosqlite.Connection) -> None: - """Add columns that may be missing in older databases.""" - cursor = await db.execute("PRAGMA table_info(analysis_jobs)") - existing = {row[1] for row in await cursor.fetchall()} - for col_name, ddl in _MIGRATIONS: - if col_name not in existing: + """Apply additive column migrations, then any post-migration DDL. + + Existing columns are detected via PRAGMA `table_info`, so re-running + the migration on an already-up-to-date DB is a no-op. + """ + columns_by_table: dict[str, set[str]] = {} + for table, col_name, ddl in _COLUMN_MIGRATIONS: + if table not in columns_by_table: + cursor = await db.execute(f"PRAGMA table_info({table})") + columns_by_table[table] = {row[1] for row in await cursor.fetchall()} + if col_name not in columns_by_table[table]: await db.execute(ddl) - logger.info("Migration: added column %s to analysis_jobs", col_name) + columns_by_table[table].add(col_name) + logger.info("Migration: added column %s to %s", col_name, table) + for ddl in _POST_MIGRATION_DDL: + await db.execute(ddl) await db.commit() diff --git a/document-parser/persistence/document_repo.py b/document-parser/persistence/document_repo.py index beb81c5..22f64b9 100644 --- a/document-parser/persistence/document_repo.py +++ b/document-parser/persistence/document_repo.py @@ -5,15 +5,34 @@ from __future__ import annotations from datetime import UTC, datetime from domain.models import Document +from domain.value_objects import DocumentLifecycleState from persistence.database import get_connection +def _parse_iso(value: str | None) -> datetime | None: + if value is None or value == "": + return None + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + return parsed + + def _row_to_document(row) -> Document: created = row["created_at"] if isinstance(created, str): created = datetime.fromisoformat(created) if created.tzinfo is None: created = created.replace(tzinfo=UTC) + # `_run_migrations` guarantees both lifecycle columns exist by the time + # any read happens, so direct access is safe. + lifecycle_value = row["lifecycle_state"] + lifecycle_state = ( + DocumentLifecycleState(lifecycle_value) + if lifecycle_value + else DocumentLifecycleState.UPLOADED + ) + lifecycle_state_at = _parse_iso(row["lifecycle_state_at"]) return Document( id=row["id"], filename=row["filename"], @@ -22,6 +41,8 @@ def _row_to_document(row) -> Document: page_count=row["page_count"], storage_path=row["storage_path"], created_at=created, + lifecycle_state=lifecycle_state, + lifecycle_state_at=lifecycle_state_at, ) @@ -32,8 +53,10 @@ class SqliteDocumentRepository: """Persist a new document record.""" async with get_connection() as db: await db.execute( - """INSERT INTO documents (id, filename, content_type, file_size, page_count, storage_path, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?)""", + """INSERT INTO documents ( + id, filename, content_type, file_size, page_count, + storage_path, created_at, lifecycle_state, lifecycle_state_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( doc.id, doc.filename, @@ -42,6 +65,8 @@ class SqliteDocumentRepository: doc.page_count, doc.storage_path, str(doc.created_at), + doc.lifecycle_state.value, + str(doc.lifecycle_state_at) if doc.lifecycle_state_at else None, ), ) await db.commit() @@ -72,6 +97,25 @@ class SqliteDocumentRepository: ) await db.commit() + async def update_lifecycle( + self, + doc_id: str, + state: DocumentLifecycleState, + at: datetime, + ) -> None: + """Persist a lifecycle state transition for a document. + + Callers must have already validated the transition via + `Document.transition_to()` — this method does not check the + transition graph; it just writes. + """ + async with get_connection() as db: + await db.execute( + "UPDATE documents SET lifecycle_state = ?, lifecycle_state_at = ? WHERE id = ?", + (state.value, str(at), doc_id), + ) + await db.commit() + async def delete(self, doc_id: str) -> bool: """Delete a document by ID. Returns True if a row was removed.""" async with get_connection() as db: diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 1c6030f..04ad7dd 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -16,6 +16,7 @@ from typing import TYPE_CHECKING import pypdfium2 as pdfium +from domain.exceptions import InvalidLifecycleTransitionError from domain.models import AnalysisJob, AnalysisStatus from domain.services import classify_error, merge_results from domain.value_objects import ( @@ -23,6 +24,7 @@ from domain.value_objects import ( ChunkResult, ConversionOptions, ConversionResult, + DocumentLifecycleState, ) if TYPE_CHECKING: @@ -162,6 +164,10 @@ class AnalysisService: chunks_json = json.dumps([_chunk_to_dict(c) for c in chunks]) await self._analysis_repo.update_chunks(job_id, chunks_json) + # Re-chunk drives the document into Chunked (idempotent if already + # Chunked; #204 will mark per-store links Stale separately). + await self._transition_document(job.document_id, DocumentLifecycleState.CHUNKED) + return chunks async def update_chunk_text(self, job_id: str, chunk_index: int, text: str) -> list[dict]: @@ -292,11 +298,47 @@ class AnalysisService: if job: job.mark_failed(error) await self._analysis_repo.update_status(job) + await self._transition_document(job.document_id, DocumentLifecycleState.FAILED) except OSError: logger.exception("Database I/O error marking job %s as failed", job_id) except Exception: logger.exception("Unexpected error marking job %s as failed", job_id) + async def _transition_document( + self, + document_id: str, + target: DocumentLifecycleState, + ) -> None: + """Drive a Document lifecycle transition (#202). + + Idempotent on the target — if the document is already in the + requested state, no write happens. Invalid transitions are + logged at WARNING and swallowed so a lifecycle hiccup does not + crash an otherwise-successful pipeline run. + """ + doc = await self._document_repo.find_by_id(document_id) + if doc is None: + return + if doc.lifecycle_state == target: + return + try: + event = doc.transition_to(target) + except InvalidLifecycleTransitionError: + logger.warning( + "Skipped invalid lifecycle transition for doc %s: %s -> %s", + document_id, + doc.lifecycle_state.value, + target.value, + ) + return + await self._document_repo.update_lifecycle(document_id, event.current, event.at) + logger.info( + "lifecycle_changed doc_id=%s from=%s to=%s", + document_id, + event.previous.value, + event.current.value, + ) + async def _run_analysis( self, job_id: str, @@ -379,6 +421,15 @@ class AnalysisService: if result.page_count: await self._document_repo.update_page_count(job.document_id, result.page_count) + # Drive the document lifecycle (#202): chunks present → Chunked, + # otherwise → Parsed. + target_state = ( + DocumentLifecycleState.CHUNKED + if chunks_json is not None + else DocumentLifecycleState.PARSED + ) + await self._transition_document(job.document_id, target_state) + await self._write_tree_to_neo4j(job, result.document_json) logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count) diff --git a/document-parser/tests/test_chunking.py b/document-parser/tests/test_chunking.py index 5b37b7e..5f6bd2d 100644 --- a/document-parser/tests/test_chunking.py +++ b/document-parser/tests/test_chunking.py @@ -10,7 +10,7 @@ import pytest from fastapi.testclient import TestClient from api.schemas import ChunkBboxResponse, ChunkingOptionsRequest, ChunkResponse, RechunkRequest -from domain.models import AnalysisJob, AnalysisStatus +from domain.models import AnalysisJob, AnalysisStatus, Document from domain.value_objects import ChunkBbox, ChunkingOptions, ChunkResult from main import app @@ -529,6 +529,13 @@ class TestRemoteChunkingPath: ) analysis_repo.find_by_id = AsyncMock(return_value=job) analysis_repo.update_chunks = AsyncMock(return_value=True) + # rechunk() now drives a Document lifecycle transition (#202), so + # the document_repo must return a real Document and accept the + # update_lifecycle write. + document_repo.find_by_id = AsyncMock( + return_value=Document(id="d1", filename="test.pdf", storage_path="/tmp/test.pdf") + ) + document_repo.update_lifecycle = AsyncMock() chunks = await service.rechunk( "j-remote", diff --git a/document-parser/tests/test_lifecycle.py b/document-parser/tests/test_lifecycle.py new file mode 100644 index 0000000..f63ac52 --- /dev/null +++ b/document-parser/tests/test_lifecycle.py @@ -0,0 +1,129 @@ +"""Tests for the Document lifecycle state machine (#202).""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from domain.events import DocumentLifecycleChanged +from domain.exceptions import InvalidLifecycleTransitionError +from domain.lifecycle import assert_transition, is_allowed_transition +from domain.models import Document +from domain.value_objects import DocumentLifecycleState + +ALLOWED_TRANSITIONS = [ + # Uploaded + (DocumentLifecycleState.UPLOADED, DocumentLifecycleState.PARSED), + (DocumentLifecycleState.UPLOADED, DocumentLifecycleState.CHUNKED), + (DocumentLifecycleState.UPLOADED, DocumentLifecycleState.FAILED), + # Parsed + (DocumentLifecycleState.PARSED, DocumentLifecycleState.PARSED), + (DocumentLifecycleState.PARSED, DocumentLifecycleState.CHUNKED), + (DocumentLifecycleState.PARSED, DocumentLifecycleState.FAILED), + # Chunked + (DocumentLifecycleState.CHUNKED, DocumentLifecycleState.CHUNKED), + (DocumentLifecycleState.CHUNKED, DocumentLifecycleState.INGESTED), + (DocumentLifecycleState.CHUNKED, DocumentLifecycleState.FAILED), + # Ingested + (DocumentLifecycleState.INGESTED, DocumentLifecycleState.STALE), + (DocumentLifecycleState.INGESTED, DocumentLifecycleState.CHUNKED), + (DocumentLifecycleState.INGESTED, DocumentLifecycleState.INGESTED), + (DocumentLifecycleState.INGESTED, DocumentLifecycleState.FAILED), + # Stale + (DocumentLifecycleState.STALE, DocumentLifecycleState.INGESTED), + (DocumentLifecycleState.STALE, DocumentLifecycleState.CHUNKED), + (DocumentLifecycleState.STALE, DocumentLifecycleState.FAILED), + # Failed (recoverable) + (DocumentLifecycleState.FAILED, DocumentLifecycleState.UPLOADED), + (DocumentLifecycleState.FAILED, DocumentLifecycleState.PARSED), + (DocumentLifecycleState.FAILED, DocumentLifecycleState.CHUNKED), +] + + +@pytest.mark.parametrize(("source", "target"), ALLOWED_TRANSITIONS) +def test_allowed_transitions_pass( + source: DocumentLifecycleState, target: DocumentLifecycleState +) -> None: + assert is_allowed_transition(source, target) + # assert_transition does not raise. + assert_transition(source, target) + + +def test_disallowed_transitions_are_rejected() -> None: + """Spot-check the most surprising disallowed pairs. + + Exhaustive enumeration of every disallowed pair would just mirror the + transition table; this list captures the cases a reader is most + likely to argue about. + """ + forbidden = [ + # Uploaded cannot skip directly to Ingested or Stale. + (DocumentLifecycleState.UPLOADED, DocumentLifecycleState.INGESTED), + (DocumentLifecycleState.UPLOADED, DocumentLifecycleState.STALE), + # Parsed cannot become Ingested without going through Chunked. + (DocumentLifecycleState.PARSED, DocumentLifecycleState.INGESTED), + # Stale is not directly reachable from Parsed (no per-store yet). + (DocumentLifecycleState.PARSED, DocumentLifecycleState.STALE), + # Failed cannot be reached as a "self-loop" — it always reflects a + # new failure, never an idempotent re-mark. + (DocumentLifecycleState.FAILED, DocumentLifecycleState.FAILED), + # Failed cannot jump directly to Ingested or Stale — must re-do + # the relevant pipeline step first. + (DocumentLifecycleState.FAILED, DocumentLifecycleState.INGESTED), + (DocumentLifecycleState.FAILED, DocumentLifecycleState.STALE), + ] + for source, target in forbidden: + assert not is_allowed_transition(source, target) + with pytest.raises(InvalidLifecycleTransitionError) as excinfo: + assert_transition(source, target) + assert excinfo.value.source == source + assert excinfo.value.target == target + + +def test_document_default_state_is_uploaded() -> None: + doc = Document(filename="test.pdf", storage_path="/tmp/test.pdf") + assert doc.lifecycle_state == DocumentLifecycleState.UPLOADED + assert doc.lifecycle_state_at is None + + +def test_transition_returns_event_and_mutates_state() -> None: + doc = Document(id="doc-1", filename="t.pdf", storage_path="/tmp/t.pdf") + before = datetime(2026, 4, 29, 10, tzinfo=UTC) + + event = doc.transition_to(DocumentLifecycleState.PARSED, now=before) + + assert isinstance(event, DocumentLifecycleChanged) + assert event.document_id == "doc-1" + assert event.previous == DocumentLifecycleState.UPLOADED + assert event.current == DocumentLifecycleState.PARSED + assert event.at == before + # State mutated on the dataclass. + assert doc.lifecycle_state == DocumentLifecycleState.PARSED + assert doc.lifecycle_state_at == before + + +def test_invalid_transition_does_not_mutate_state() -> None: + doc = Document(id="doc-1", filename="t.pdf", storage_path="/tmp/t.pdf") + assert doc.lifecycle_state == DocumentLifecycleState.UPLOADED + + with pytest.raises(InvalidLifecycleTransitionError): + doc.transition_to(DocumentLifecycleState.STALE) + + # State is untouched. + assert doc.lifecycle_state == DocumentLifecycleState.UPLOADED + assert doc.lifecycle_state_at is None + + +def test_idempotent_self_transitions_emit_event() -> None: + """Self-loops (re-parse, re-chunk, re-push) are explicitly allowed + so the pipeline can drive transitions without checking the source + state first. They still emit an event for observability.""" + doc = Document(id="doc-1", filename="t.pdf", storage_path="/tmp/t.pdf") + doc.transition_to(DocumentLifecycleState.CHUNKED) + assert doc.lifecycle_state == DocumentLifecycleState.CHUNKED + + event = doc.transition_to(DocumentLifecycleState.CHUNKED) + + assert event.previous == DocumentLifecycleState.CHUNKED + assert event.current == DocumentLifecycleState.CHUNKED diff --git a/document-parser/tests/test_repos.py b/document-parser/tests/test_repos.py index dc29d38..4c3179b 100644 --- a/document-parser/tests/test_repos.py +++ b/document-parser/tests/test_repos.py @@ -1,10 +1,11 @@ """Tests for persistence repositories using a temporary SQLite database.""" -from datetime import datetime +from datetime import UTC, datetime import pytest from domain.models import AnalysisJob, AnalysisStatus, Document +from domain.value_objects import DocumentLifecycleState from persistence.analysis_repo import SqliteAnalysisRepository from persistence.database import init_db from persistence.document_repo import SqliteDocumentRepository @@ -81,6 +82,46 @@ class TestDocumentRepo: deleted = await document_repo.delete("nonexistent") assert deleted is False + async def test_default_lifecycle_state_is_uploaded(self, document_repo): + """Fresh document round-trip preserves the default Uploaded state.""" + doc = Document(id="doc-1", filename="t.pdf", storage_path="/tmp/t.pdf") + await document_repo.insert(doc) + + found = await document_repo.find_by_id("doc-1") + assert found is not None + assert found.lifecycle_state == DocumentLifecycleState.UPLOADED + assert found.lifecycle_state_at is None + + async def test_update_lifecycle_persists_state_and_timestamp(self, document_repo): + doc = Document(id="doc-1", filename="t.pdf", storage_path="/tmp/t.pdf") + await document_repo.insert(doc) + + when = datetime(2026, 4, 29, 12, 0, tzinfo=UTC) + await document_repo.update_lifecycle("doc-1", DocumentLifecycleState.PARSED, when) + + found = await document_repo.find_by_id("doc-1") + assert found is not None + assert found.lifecycle_state == DocumentLifecycleState.PARSED + assert found.lifecycle_state_at is not None + assert found.lifecycle_state_at == when + + async def test_lifecycle_state_round_trips_for_each_value(self, document_repo): + """Every enum value must serialize cleanly into and out of SQLite.""" + when = datetime(2026, 4, 29, 12, 0, tzinfo=UTC) + for value in DocumentLifecycleState: + doc = Document( + id=f"doc-{value.value}", + filename="t.pdf", + storage_path="/tmp/t.pdf", + lifecycle_state=value, + lifecycle_state_at=when, + ) + await document_repo.insert(doc) + + found = await document_repo.find_by_id(f"doc-{value.value}") + assert found is not None + assert found.lifecycle_state == value + class TestAnalysisRepo: async def _insert_doc(self, document_repo): diff --git a/frontend/src/shared/types.ts b/frontend/src/shared/types.ts index c6d0a8f..67d6618 100644 --- a/frontend/src/shared/types.ts +++ b/frontend/src/shared/types.ts @@ -1,3 +1,22 @@ +/** + * Canonical document lifecycle state — mirrors the backend enum + * `DocumentLifecycleState` (see `domain/value_objects.py`). + * + * - `Uploaded` raw file persisted, no parse yet + * - `Parsed` conversion produced a document tree + * - `Chunked` chunker produced a draft chunkset + * - `Ingested` chunkset has been embedded into at least one store + * - `Stale` chunkset edited after a successful push (per-store concept) + * - `Failed` a pipeline step failed; recoverable by retry + */ +export type DocumentLifecycleState = + | 'Uploaded' + | 'Parsed' + | 'Chunked' + | 'Ingested' + | 'Stale' + | 'Failed' + export interface Document { id: string filename: string @@ -5,6 +24,10 @@ export interface Document { fileSize: number | null pageCount: number | null createdAt: string + /** Canonical lifecycle state. Drives the status badge in `/docs`. */ + lifecycleState: DocumentLifecycleState + /** ISO timestamp of the last lifecycle transition (UTC). */ + lifecycleStateAt: string | null } export interface PipelineOptions {