diff --git a/document-parser/domain/lifecycle_aggregation.py b/document-parser/domain/lifecycle_aggregation.py new file mode 100644 index 0000000..a94a724 --- /dev/null +++ b/document-parser/domain/lifecycle_aggregation.py @@ -0,0 +1,58 @@ +"""Aggregate the per-(document, store) link states into a single +document-level lifecycle state. + +The doc lifecycle column is the materialized result of this rule. It is +recomputed any time a link write happens. Read paths use the stored +column directly (cheap) — they do not call this rule on every GET. + +The rule prefers "more concerning" states first: + + any link FAILED -> Document FAILED + any link STALE -> Document STALE + any link INGESTED -> Document INGESTED + no links -> keep the document's current pre-link state + (Uploaded / Parsed / Chunked) — the caller is + responsible for not overwriting that. + +If you find yourself wanting a fourth case, you probably want a new +link state, not a new aggregation branch. + +This module is pure: no I/O, no datetime — just data in / data out. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from domain.value_objects import DocumentLifecycleState, DocumentStoreLinkState + +if TYPE_CHECKING: + from collections.abc import Iterable + + from domain.models import DocumentStoreLink + + +def aggregate_lifecycle( + links: Iterable[DocumentStoreLink], + *, + fallback: DocumentLifecycleState, +) -> DocumentLifecycleState: + """Compute the aggregate document lifecycle state. + + Args: + links: every `DocumentStoreLink` for the document. May be empty. + fallback: the lifecycle state to return when there are no links — + typically the document's current pre-link state (`Uploaded`, + `Parsed`, or `Chunked`). + + Returns: + The aggregate `DocumentLifecycleState`. + """ + states = {link.state for link in links} + if DocumentStoreLinkState.FAILED in states: + return DocumentLifecycleState.FAILED + if DocumentStoreLinkState.STALE in states: + return DocumentLifecycleState.STALE + if DocumentStoreLinkState.INGESTED in states: + return DocumentLifecycleState.INGESTED + return fallback diff --git a/document-parser/domain/models.py b/document-parser/domain/models.py index 0a371ee..7c05040 100644 --- a/document-parser/domain/models.py +++ b/document-parser/domain/models.py @@ -9,7 +9,11 @@ from datetime import UTC, datetime from domain.events import DocumentLifecycleChanged from domain.lifecycle import assert_transition -from domain.value_objects import DocumentLifecycleState +from domain.value_objects import ( + DocumentLifecycleState, + DocumentStoreLinkState, + StoreKind, +) class AnalysisStatus(enum.StrEnum): @@ -130,3 +134,70 @@ class AnalysisJob: self.status = AnalysisStatus.FAILED self.error_message = error self.completed_at = _utcnow() + + +@dataclass +class Store: + """A logical destination for ingested chunks. + + A `Store` represents a named, configurable target for ingestion (e.g. + `rh-corpus-v3`, `legal-v1`). It is decoupled from the underlying + `VectorStore` adapter (which represents the *technology*, like + OpenSearch). One adapter can serve many stores by namespacing the + physical index name on `slug`. + + See design doc `docs/design/203-per-store-ingestion-state.md`. + """ + + id: str = field(default_factory=_new_id) + name: str = "" + slug: str = "" + kind: StoreKind = StoreKind.OPENSEARCH + embedder: str = "" + config: dict = field(default_factory=dict) + is_default: bool = False + created_at: datetime = field(default_factory=_utcnow) + + +@dataclass +class DocumentStoreLink: + """A live record of a document's presence in a single store. + + The link carries the per-pair state (Ingested / Stale / Failed) plus + metadata used by the auto-stale detection (#204) and the chunks + editor (#205). One row per (document, store) pair — enforced by a + UNIQUE constraint at the persistence layer. + """ + + id: str = field(default_factory=_new_id) + document_id: str = "" + store_id: str = "" + state: DocumentStoreLinkState = DocumentStoreLinkState.INGESTED + chunkset_hash: str | None = None + last_push_at: datetime | None = None + last_run_id: str | None = None + error_message: str | None = None + + def mark_ingested( + self, + *, + hash_: str, + at: datetime, + run_id: str | None = None, + ) -> None: + """Record a successful push: chunkset hash + timestamp + run id.""" + self.state = DocumentStoreLinkState.INGESTED + self.chunkset_hash = hash_ + self.last_push_at = at + self.last_run_id = run_id + self.error_message = None + + def mark_stale(self) -> None: + """Mark the link as stale (source chunkset drifted from pushed).""" + self.state = DocumentStoreLinkState.STALE + self.error_message = None + + def mark_failed(self, *, error: str) -> None: + """Record a failed push attempt.""" + self.state = DocumentStoreLinkState.FAILED + self.error_message = error diff --git a/document-parser/domain/ports.py b/document-parser/domain/ports.py index 967871d..d6929b4 100644 --- a/document-parser/domain/ports.py +++ b/document-parser/domain/ports.py @@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, Protocol, runtime_checkable if TYPE_CHECKING: from datetime import datetime - from domain.models import AnalysisJob, Document + from domain.models import AnalysisJob, Document, DocumentStoreLink, Store from domain.value_objects import ( ChunkingOptions, ChunkResult, @@ -98,6 +98,36 @@ class DocumentRepository(Protocol): async def delete(self, doc_id: str) -> bool: ... +class StoreRepository(Protocol): + """Port for `Store` persistence (introduced by #203).""" + + async def insert(self, store: Store) -> None: ... + + async def find_all(self) -> list[Store]: ... + + async def find_by_slug(self, slug: str) -> Store | None: ... + + async def find_by_id(self, store_id: str) -> Store | None: ... + + async def get_default(self) -> Store | None: ... + + +class DocumentStoreLinkRepository(Protocol): + """Port for `DocumentStoreLink` persistence (introduced by #203).""" + + async def upsert(self, link: DocumentStoreLink) -> None: + """Insert or update by (document_id, store_id).""" + ... + + async def find_for_document(self, document_id: str) -> list[DocumentStoreLink]: ... + + async def find_for_store(self, store_id: str) -> list[DocumentStoreLink]: ... + + async def find_one(self, document_id: str, store_id: str) -> DocumentStoreLink | None: ... + + async def delete(self, document_id: str, store_id: str) -> bool: ... + + class AnalysisRepository(Protocol): """Port for analysis job persistence.""" diff --git a/document-parser/domain/value_objects.py b/document-parser/domain/value_objects.py index 784dd2c..d3f1baf 100644 --- a/document-parser/domain/value_objects.py +++ b/document-parser/domain/value_objects.py @@ -39,6 +39,28 @@ class DocumentLifecycleState(StrEnum): FAILED = "Failed" +class StoreKind(StrEnum): + """Backing technology of a Store. Today only OpenSearch is implemented; + the enum is here so future backends (Pinecone, Qdrant, pgvector) can be + added without touching the persistence schema.""" + + OPENSEARCH = "opensearch" + + +class DocumentStoreLinkState(StrEnum): + """State of a (document, store) ingestion link. + + Distinct from `DocumentLifecycleState` — the document lifecycle is the + aggregate over all per-store links. A link is `Ingested` when its + chunkset hash matches the source; `Stale` when the source has drifted + after the last push; `Failed` when the last push attempt errored. + """ + + 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 b79f5db..7398f4c 100644 --- a/document-parser/persistence/database.py +++ b/document-parser/persistence/database.py @@ -42,6 +42,34 @@ CREATE TABLE IF NOT EXISTS analysis_jobs ( CREATE INDEX IF NOT EXISTS idx_analysis_jobs_status ON analysis_jobs(status); CREATE INDEX IF NOT EXISTS idx_analysis_jobs_created_at ON analysis_jobs(created_at); CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at); + +-- 0.6.0 — Per (document, store) ingestion state (#203). +CREATE TABLE IF NOT EXISTS stores ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + slug TEXT NOT NULL UNIQUE, + kind TEXT NOT NULL, + embedder TEXT NOT NULL, + config TEXT NOT NULL DEFAULT '{}', + is_default INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS document_store_links ( + id TEXT PRIMARY KEY, + document_id TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE, + store_id TEXT NOT NULL REFERENCES stores(id) ON DELETE CASCADE, + state TEXT NOT NULL, + chunkset_hash TEXT, + last_push_at TEXT, + last_run_id TEXT, + error_message TEXT, + UNIQUE (document_id, store_id) +); + +CREATE INDEX IF NOT EXISTS idx_dsl_doc ON document_store_links(document_id); +CREATE INDEX IF NOT EXISTS idx_dsl_store ON document_store_links(store_id); +CREATE INDEX IF NOT EXISTS idx_dsl_state ON document_store_links(state); """ @@ -107,10 +135,27 @@ async def init_db() -> None: async with aiosqlite.connect(DB_PATH) as db: await db.executescript(_SCHEMA) await _run_migrations(db) + await _seed_default_store(db) await db.commit() logger.info("Database initialized at %s", DB_PATH) +async def _seed_default_store(db: aiosqlite.Connection) -> None: + """Insert the canonical `default` store on first boot. + + Idempotent — uses INSERT OR IGNORE keyed on the unique slug. The + embedder is read from the DEFAULT_EMBEDDER env var with a sensible + fallback so existing single-index deployments keep working. + """ + embedder = os.environ.get("DEFAULT_EMBEDDER", "bge-m3") + await db.execute( + """INSERT OR IGNORE INTO stores + (id, name, slug, kind, embedder, config, is_default, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now'))""", + ("default", "default", "default", "opensearch", embedder, "{}", 1), + ) + + async def get_db() -> aiosqlite.Connection: """Open a new database connection with row factory and FK enforcement.""" db = await aiosqlite.connect(DB_PATH) diff --git a/document-parser/persistence/document_store_link_repo.py b/document-parser/persistence/document_store_link_repo.py new file mode 100644 index 0000000..fa38d83 --- /dev/null +++ b/document-parser/persistence/document_store_link_repo.py @@ -0,0 +1,98 @@ +"""Document-Store link repository — SQLite CRUD on `document_store_links`.""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from domain.models import DocumentStoreLink +from domain.value_objects import DocumentStoreLinkState +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_link(row) -> DocumentStoreLink: + return DocumentStoreLink( + id=row["id"], + document_id=row["document_id"], + store_id=row["store_id"], + state=DocumentStoreLinkState(row["state"]), + chunkset_hash=row["chunkset_hash"], + last_push_at=_parse_iso(row["last_push_at"]), + last_run_id=row["last_run_id"], + error_message=row["error_message"], + ) + + +class SqliteDocumentStoreLinkRepository: + """SQLite implementation of the DocumentStoreLinkRepository port.""" + + async def upsert(self, link: DocumentStoreLink) -> None: + """Insert a new link or update the existing (document_id, store_id) row.""" + async with get_connection() as db: + await db.execute( + """INSERT INTO document_store_links + (id, document_id, store_id, state, chunkset_hash, + last_push_at, last_run_id, error_message) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(document_id, store_id) DO UPDATE SET + state = excluded.state, + chunkset_hash = excluded.chunkset_hash, + last_push_at = excluded.last_push_at, + last_run_id = excluded.last_run_id, + error_message = excluded.error_message""", + ( + link.id, + link.document_id, + link.store_id, + link.state.value, + link.chunkset_hash, + str(link.last_push_at) if link.last_push_at else None, + link.last_run_id, + link.error_message, + ), + ) + await db.commit() + + async def find_for_document(self, document_id: str) -> list[DocumentStoreLink]: + async with get_connection() as db: + cursor = await db.execute( + "SELECT * FROM document_store_links WHERE document_id = ?", + (document_id,), + ) + rows = await cursor.fetchall() + return [_row_to_link(r) for r in rows] + + async def find_for_store(self, store_id: str) -> list[DocumentStoreLink]: + async with get_connection() as db: + cursor = await db.execute( + "SELECT * FROM document_store_links WHERE store_id = ?", + (store_id,), + ) + rows = await cursor.fetchall() + return [_row_to_link(r) for r in rows] + + async def find_one(self, document_id: str, store_id: str) -> DocumentStoreLink | None: + async with get_connection() as db: + cursor = await db.execute( + "SELECT * FROM document_store_links WHERE document_id = ? AND store_id = ?", + (document_id, store_id), + ) + row = await cursor.fetchone() + return _row_to_link(row) if row else None + + async def delete(self, document_id: str, store_id: str) -> bool: + async with get_connection() as db: + cursor = await db.execute( + "DELETE FROM document_store_links WHERE document_id = ? AND store_id = ?", + (document_id, store_id), + ) + await db.commit() + return cursor.rowcount > 0 diff --git a/document-parser/persistence/store_repo.py b/document-parser/persistence/store_repo.py new file mode 100644 index 0000000..fd67913 --- /dev/null +++ b/document-parser/persistence/store_repo.py @@ -0,0 +1,83 @@ +"""Store repository — SQLite CRUD for the `stores` table.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime + +from domain.models import Store +from domain.value_objects import StoreKind +from persistence.database import get_connection + + +def _row_to_store(row) -> Store: + created = row["created_at"] + if isinstance(created, str): + created = datetime.fromisoformat(created) + if created.tzinfo is None: + created = created.replace(tzinfo=UTC) + config_raw = row["config"] or "{}" + try: + config = json.loads(config_raw) + except (TypeError, ValueError): + config = {} + return Store( + id=row["id"], + name=row["name"], + slug=row["slug"], + kind=StoreKind(row["kind"]), + embedder=row["embedder"], + config=config, + is_default=bool(row["is_default"]), + created_at=created, + ) + + +class SqliteStoreRepository: + """SQLite implementation of the StoreRepository port.""" + + async def insert(self, store: Store) -> None: + async with get_connection() as db: + await db.execute( + """INSERT INTO stores + (id, name, slug, kind, embedder, config, is_default, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + store.id, + store.name, + store.slug, + store.kind.value, + store.embedder, + json.dumps(store.config), + 1 if store.is_default else 0, + str(store.created_at), + ), + ) + await db.commit() + + async def find_all(self) -> list[Store]: + async with get_connection() as db: + cursor = await db.execute("SELECT * FROM stores ORDER BY is_default DESC, name ASC") + rows = await cursor.fetchall() + return [_row_to_store(r) for r in rows] + + async def find_by_slug(self, slug: str) -> Store | None: + async with get_connection() as db: + cursor = await db.execute("SELECT * FROM stores WHERE slug = ?", (slug,)) + row = await cursor.fetchone() + return _row_to_store(row) if row else None + + async def find_by_id(self, store_id: str) -> Store | None: + async with get_connection() as db: + cursor = await db.execute("SELECT * FROM stores WHERE id = ?", (store_id,)) + row = await cursor.fetchone() + return _row_to_store(row) if row else None + + async def get_default(self) -> Store | None: + """Return the seeded `default` store, if any.""" + async with get_connection() as db: + cursor = await db.execute( + "SELECT * FROM stores WHERE is_default = 1 ORDER BY created_at ASC LIMIT 1" + ) + row = await cursor.fetchone() + return _row_to_store(row) if row else None diff --git a/document-parser/tests/test_lifecycle_aggregation.py b/document-parser/tests/test_lifecycle_aggregation.py new file mode 100644 index 0000000..ef0dbe3 --- /dev/null +++ b/document-parser/tests/test_lifecycle_aggregation.py @@ -0,0 +1,72 @@ +"""Tests for the document lifecycle aggregation rule (#203).""" + +from __future__ import annotations + +from domain.lifecycle_aggregation import aggregate_lifecycle +from domain.models import DocumentStoreLink +from domain.value_objects import DocumentLifecycleState, DocumentStoreLinkState + + +def _link(state: DocumentStoreLinkState) -> DocumentStoreLink: + return DocumentStoreLink( + id=f"link-{state.value}", + document_id="doc-1", + store_id=f"store-{state.value}", + state=state, + ) + + +def test_no_links_returns_fallback() -> None: + assert ( + aggregate_lifecycle([], fallback=DocumentLifecycleState.CHUNKED) + == DocumentLifecycleState.CHUNKED + ) + + +def test_single_ingested_link_makes_doc_ingested() -> None: + assert ( + aggregate_lifecycle( + [_link(DocumentStoreLinkState.INGESTED)], + fallback=DocumentLifecycleState.CHUNKED, + ) + == DocumentLifecycleState.INGESTED + ) + + +def test_any_stale_link_makes_doc_stale() -> None: + """Stale outranks Ingested — a doc with one stale store is considered stale.""" + assert ( + aggregate_lifecycle( + [ + _link(DocumentStoreLinkState.INGESTED), + _link(DocumentStoreLinkState.STALE), + ], + fallback=DocumentLifecycleState.CHUNKED, + ) + == DocumentLifecycleState.STALE + ) + + +def test_any_failed_link_makes_doc_failed() -> None: + """Failed outranks every other link state.""" + assert ( + aggregate_lifecycle( + [ + _link(DocumentStoreLinkState.INGESTED), + _link(DocumentStoreLinkState.STALE), + _link(DocumentStoreLinkState.FAILED), + ], + fallback=DocumentLifecycleState.CHUNKED, + ) + == DocumentLifecycleState.FAILED + ) + + +def test_only_failed_link_makes_doc_failed() -> None: + assert ( + aggregate_lifecycle( + [_link(DocumentStoreLinkState.FAILED)], + fallback=DocumentLifecycleState.CHUNKED, + ) + == DocumentLifecycleState.FAILED + ) diff --git a/document-parser/tests/test_store_repo.py b/document-parser/tests/test_store_repo.py new file mode 100644 index 0000000..7126d38 --- /dev/null +++ b/document-parser/tests/test_store_repo.py @@ -0,0 +1,190 @@ +"""Tests for the Store and DocumentStoreLink repositories (#203).""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from domain.models import Document, DocumentStoreLink, Store +from domain.value_objects import DocumentStoreLinkState, StoreKind +from persistence.database import init_db +from persistence.document_repo import SqliteDocumentRepository +from persistence.document_store_link_repo import SqliteDocumentStoreLinkRepository +from persistence.store_repo import SqliteStoreRepository + + +@pytest.fixture(autouse=True) +async def setup_db(monkeypatch, tmp_path): + db_path = str(tmp_path / "test.db") + monkeypatch.setattr("persistence.database.DB_PATH", db_path) + await init_db() + yield + + +@pytest.fixture +def store_repo(): + return SqliteStoreRepository() + + +@pytest.fixture +def link_repo(): + return SqliteDocumentStoreLinkRepository() + + +@pytest.fixture +def document_repo(): + return SqliteDocumentRepository() + + +class TestStoreRepo: + async def test_default_store_seeded_on_init(self, store_repo): + """init_db() must seed exactly one `default` store on a fresh DB.""" + default = await store_repo.get_default() + assert default is not None + assert default.slug == "default" + assert default.is_default is True + assert default.kind == StoreKind.OPENSEARCH + + async def test_seed_is_idempotent(self, store_repo): + """Re-running init_db() must not create a second default store.""" + await init_db() + await init_db() + all_stores = await store_repo.find_all() + slugs = [s.slug for s in all_stores] + assert slugs.count("default") == 1 + + async def test_insert_and_find_by_slug(self, store_repo): + store = Store( + id="s-rh", + name="rh-corpus-v3", + slug="rh-corpus-v3", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + config={"index_name": "rh-corpus-v3"}, + ) + await store_repo.insert(store) + + found = await store_repo.find_by_slug("rh-corpus-v3") + assert found is not None + assert found.embedder == "bge-m3" + assert found.config == {"index_name": "rh-corpus-v3"} + + async def test_find_all_orders_default_first(self, store_repo): + await store_repo.insert( + Store( + id="s-rh", + name="rh", + slug="rh", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + ) + ) + await store_repo.insert( + Store( + id="s-legal", + name="legal", + slug="legal", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + ) + ) + all_stores = await store_repo.find_all() + assert all_stores[0].slug == "default" # seeded default comes first + + +class TestDocumentStoreLinkRepo: + async def _seed_doc(self, document_repo, doc_id: str = "doc-1") -> Document: + doc = Document(id=doc_id, filename="t.pdf", storage_path="/tmp/t.pdf") + await document_repo.insert(doc) + return doc + + async def test_upsert_creates_then_updates(self, link_repo, document_repo): + await self._seed_doc(document_repo) + link = DocumentStoreLink( + id="l-1", + document_id="doc-1", + store_id="default", + state=DocumentStoreLinkState.INGESTED, + ) + await link_repo.upsert(link) + + # Update same (doc, store) — no second row. + link.mark_ingested( + hash_="abc123", + at=datetime(2026, 4, 29, 12, tzinfo=UTC), + run_id="run-1", + ) + await link_repo.upsert(link) + + all_for_doc = await link_repo.find_for_document("doc-1") + assert len(all_for_doc) == 1 + assert all_for_doc[0].chunkset_hash == "abc123" + assert all_for_doc[0].last_run_id == "run-1" + + async def test_unique_constraint_on_doc_store_pair(self, link_repo, document_repo): + """Two links with the same (doc, store) collapse to one via upsert.""" + await self._seed_doc(document_repo) + await link_repo.upsert( + DocumentStoreLink( + id="l-1", + document_id="doc-1", + store_id="default", + state=DocumentStoreLinkState.INGESTED, + ) + ) + await link_repo.upsert( + DocumentStoreLink( + id="l-2", + document_id="doc-1", + store_id="default", + state=DocumentStoreLinkState.STALE, + ) + ) + rows = await link_repo.find_for_document("doc-1") + assert len(rows) == 1 + assert rows[0].state == DocumentStoreLinkState.STALE + + async def test_find_one_returns_none_when_absent(self, link_repo, document_repo): + await self._seed_doc(document_repo) + assert await link_repo.find_one("doc-1", "default") is None + + async def test_cascade_delete_when_doc_removed(self, link_repo, document_repo): + await self._seed_doc(document_repo) + await link_repo.upsert( + DocumentStoreLink( + id="l-1", + document_id="doc-1", + store_id="default", + state=DocumentStoreLinkState.INGESTED, + ) + ) + await document_repo.delete("doc-1") + rows = await link_repo.find_for_document("doc-1") + assert rows == [] + + async def test_state_round_trips(self, link_repo, document_repo): + await self._seed_doc(document_repo) + for state in DocumentStoreLinkState: + link = DocumentStoreLink( + id=f"l-{state.value}", + document_id="doc-1", + store_id=f"store-{state.value}", + state=state, + ) + # Need a store row for FK; seed minimal stores. + from persistence.store_repo import SqliteStoreRepository + + await SqliteStoreRepository().insert( + Store( + id=f"store-{state.value}", + name=f"s-{state.value}", + slug=f"s-{state.value}", + kind=StoreKind.OPENSEARCH, + embedder="bge-m3", + ) + ) + await link_repo.upsert(link) + found = await link_repo.find_one("doc-1", f"store-{state.value}") + assert found is not None + assert found.state == state