feat(#202): introduce Document lifecycle state machine

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

@@ -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),
     )
 
 

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):

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

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

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)

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

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: ...
 
 

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

document-parser/persistence/database.py

@@ -45,22 +45,59 @@ CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at);
 """
 
 
-_MIGRATIONS = [
+# Column migrations: (table, column_name, ddl). Idempotent — applied only
-    ("document_json", "ALTER TABLE analysis_jobs ADD COLUMN document_json TEXT"),
+# when the column is missing from the live schema. Order matters when a
-    ("chunks_json", "ALTER TABLE analysis_jobs ADD COLUMN chunks_json TEXT"),
+# later migration depends on an earlier one (none today).
-    ("progress_current", "ALTER TABLE analysis_jobs ADD COLUMN progress_current INTEGER"),
+_COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [
-    ("progress_total", "ALTER TABLE analysis_jobs ADD COLUMN progress_total INTEGER"),
+    ("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."""
+    """Apply additive column migrations, then any post-migration DDL.
-    cursor = await db.execute("PRAGMA table_info(analysis_jobs)")
+
-    existing = {row[1] for row in await cursor.fetchall()}
+    Existing columns are detected via PRAGMA `table_info`, so re-running
-    for col_name, ddl in _MIGRATIONS:
+    the migration on an already-up-to-date DB is a no-op.
-        if col_name not in existing:
+    """
+    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()
 
 

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)
+                """INSERT INTO documents (
-                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
+                       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:

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)

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",

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

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):

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 {