diff --git a/document-parser/domain/chunk_editing.py b/document-parser/domain/chunk_editing.py new file mode 100644 index 0000000..bd143b9 --- /dev/null +++ b/document-parser/domain/chunk_editing.py @@ -0,0 +1,216 @@ +"""Pure-domain operations on a chunkset. + +These functions take a chunkset as input and return a new chunkset +(plus, where appropriate, the chunks that were created). They do not +touch the database, do not record audit rows, and do not raise +infrastructure errors. The `ChunkEditingService` (in `services/`) +wraps each call with audit-record generation and atomic persistence. + +All operations preserve `sequence` ordering: insertions and splits +shift subsequent sequences upward; deletions / merges leave gaps. +Sequences are 0-based and only required to be strictly increasing +within a document; gaps are explicitly allowed so we never rewrite +many rows for a single edit. +""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime + +from domain.models import Chunk + + +class ChunkEditingError(Exception): + """Raised by chunk-editing operations on invalid input (missing id, + out-of-range offset, etc.). Subclasses `Exception` rather than + `DomainError` so the API layer can map all of them to 4xx without + a wider catch.""" + + +def _utcnow() -> datetime: + return datetime.now(UTC) + + +def _new_id() -> str: + return uuid.uuid4().hex + + +def _index_of(chunks: list[Chunk], chunk_id: str) -> int: + for idx, c in enumerate(chunks): + if c.id == chunk_id: + return idx + raise ChunkEditingError(f"chunk not found: {chunk_id}") + + +def insert( + chunks: list[Chunk], + *, + at_position: int, + text: str, + document_id: str, + headings: list[str] | None = None, + source_page: int | None = None, +) -> tuple[list[Chunk], Chunk]: + """Insert a fresh chunk at position `at_position`. + + Returns the updated chunkset and the new chunk. Subsequent chunks' + sequences are shifted by +1. + """ + if at_position < 0 or at_position > len(chunks): + raise ChunkEditingError(f"insert position out of range: {at_position}") + now = _utcnow() + new_chunk = Chunk( + id=_new_id(), + document_id=document_id, + sequence=at_position, + text=text, + headings=list(headings or []), + source_page=source_page, + created_at=now, + updated_at=now, + ) + out = list(chunks) + for c in out[at_position:]: + c.sequence += 1 + out.insert(at_position, new_chunk) + return out, new_chunk + + +def update( + chunks: list[Chunk], + chunk_id: str, + *, + text: str | None = None, + headings: list[str] | None = None, +) -> tuple[list[Chunk], Chunk]: + """Update text and/or headings of a chunk in-place. + + Returns the updated chunkset and the modified chunk. The chunk's + id and sequence are preserved. + """ + idx = _index_of(chunks, chunk_id) + target = chunks[idx] + if target.deleted_at is not None: + raise ChunkEditingError(f"chunk is deleted: {chunk_id}") + if text is not None: + target.text = text + if headings is not None: + target.headings = list(headings) + target.updated_at = _utcnow() + return list(chunks), target + + +def delete(chunks: list[Chunk], chunk_id: str) -> tuple[list[Chunk], Chunk]: + """Soft-delete a chunk. Returns the updated chunkset and the + deleted chunk (still present in the list, with `deleted_at` set).""" + idx = _index_of(chunks, chunk_id) + target = chunks[idx] + if target.deleted_at is not None: + return list(chunks), target # idempotent + target.deleted_at = _utcnow() + target.updated_at = target.deleted_at + return list(chunks), target + + +def merge( + chunks: list[Chunk], + chunk_ids: list[str], + *, + separator: str = "\n", +) -> tuple[list[Chunk], Chunk]: + """Merge `chunk_ids` (in order) into a single new chunk. + + The new chunk takes the headings of the first source and the + smallest source page. The `id`s of the sources are returned in + the new chunk's lineage via the audit row written by the service + layer. + + Returns the updated chunkset and the new merged chunk. The sources + are removed from the chunkset (hard-removed from the list — the + service is responsible for soft-deleting their persisted rows so + history queries still resolve them). + """ + if len(chunk_ids) < 2: + raise ChunkEditingError("merge requires at least two chunks") + indices = [_index_of(chunks, cid) for cid in chunk_ids] + sources = [chunks[i] for i in indices] + if any(c.deleted_at for c in sources): + raise ChunkEditingError("cannot merge a deleted chunk") + + document_id = sources[0].document_id + if any(c.document_id != document_id for c in sources): + raise ChunkEditingError("merge across documents is not allowed") + + merged_text = separator.join(c.text for c in sources) + merged_headings = list(sources[0].headings) + merged_page = min((c.source_page for c in sources if c.source_page is not None), default=None) + merged_sequence = min(c.sequence for c in sources) + + now = _utcnow() + new_chunk = Chunk( + id=_new_id(), + document_id=document_id, + sequence=merged_sequence, + text=merged_text, + headings=merged_headings, + source_page=merged_page, + created_at=now, + updated_at=now, + ) + + source_ids = {c.id for c in sources} + out = [c for c in chunks if c.id not in source_ids] + out.append(new_chunk) + out.sort(key=lambda c: c.sequence) + return out, new_chunk + + +def split( + chunks: list[Chunk], chunk_id: str, *, at_offset: int +) -> tuple[list[Chunk], Chunk, Chunk]: + """Split a chunk's text at `at_offset` into two new chunks. + + The source chunk is removed from the chunkset (the service-layer + persistence path soft-deletes its row so history queries can + resolve it). The two new chunks inherit headings and source_page + from the source. + + Returns the updated chunkset and the two new chunks `(left, right)`. + """ + idx = _index_of(chunks, chunk_id) + target = chunks[idx] + if target.deleted_at is not None: + raise ChunkEditingError(f"chunk is deleted: {chunk_id}") + if at_offset <= 0 or at_offset >= len(target.text): + raise ChunkEditingError(f"split offset out of range for chunk of length {len(target.text)}") + + now = _utcnow() + left = Chunk( + id=_new_id(), + document_id=target.document_id, + sequence=target.sequence, + text=target.text[:at_offset], + headings=list(target.headings), + source_page=target.source_page, + created_at=now, + updated_at=now, + ) + right = Chunk( + id=_new_id(), + document_id=target.document_id, + sequence=target.sequence + 1, + text=target.text[at_offset:], + headings=list(target.headings), + source_page=target.source_page, + created_at=now, + updated_at=now, + ) + + out = [c for c in chunks if c.id != target.id] + for c in out: + if c.sequence > target.sequence: + c.sequence += 1 + out.extend((left, right)) + out.sort(key=lambda c: c.sequence) + return out, left, right diff --git a/document-parser/domain/models.py b/document-parser/domain/models.py index 7c05040..826178c 100644 --- a/document-parser/domain/models.py +++ b/document-parser/domain/models.py @@ -10,6 +10,9 @@ from datetime import UTC, datetime from domain.events import DocumentLifecycleChanged from domain.lifecycle import assert_transition from domain.value_objects import ( + ChunkBbox, + ChunkDocItem, + ChunkEditAction, DocumentLifecycleState, DocumentStoreLinkState, StoreKind, @@ -201,3 +204,77 @@ class DocumentStoreLink: """Record a failed push attempt.""" self.state = DocumentStoreLinkState.FAILED self.error_message = error + + +@dataclass +class Chunk: + """A persisted chunk — first-class entity introduced by #205. + + Replaces the legacy `analysis_jobs.chunks_json` blob. The `id` is + stable across edits except for split/merge which produce new chunks + with new ids; the lineage is recorded in `chunk_edits` rows. + + `sequence` controls ordering within a document. Gaps are allowed — + splits push subsequent chunks' sequences without rewriting them. + + `deleted_at` is non-null when the chunk has been soft-deleted; the + row stays in the table so the audit trail keeps its before/after + pointers valid. + """ + + id: str = field(default_factory=_new_id) + document_id: str = "" + sequence: int = 0 + text: str = "" + headings: list[str] = field(default_factory=list) + source_page: int | None = None + bboxes: list[ChunkBbox] = field(default_factory=list) + doc_items: list[ChunkDocItem] = field(default_factory=list) + token_count: int | None = None + created_at: datetime = field(default_factory=_utcnow) + updated_at: datetime = field(default_factory=_utcnow) + deleted_at: datetime | None = None + + +@dataclass(frozen=True) +class ChunkEdit: + """An immutable audit record describing one mutating operation on + the chunkset of a document. Written atomically with the chunk + write it describes. + + `before` and `after` are JSON-serializable snapshots of the chunk + state. They are `None` for the start/end of the chunk's life: + - `before is None` for INSERT + - `after is None` for DELETE + - For MERGE: a single result row carries `parents = [chunk_ids…]` + - For SPLIT: two result rows each carry `parents = [source_id]` + and the source's edit row carries `children = [new_id, new_id]` + """ + + id: str + document_id: str + chunk_id: str | None + action: ChunkEditAction + actor: str + at: datetime + before: dict | None = None + after: dict | None = None + parents: list[str] = field(default_factory=list) + children: list[str] = field(default_factory=list) + reason: str | None = None + + +@dataclass(frozen=True) +class ChunkPush: + """Snapshot of which chunks were pushed to which store, when. + + Lets the API answer 'show me the chunkset that was in store X at + push N' without replaying the full audit log. + """ + + id: str + document_id: str + store_id: str + chunkset_hash: str + chunk_ids: list[str] + pushed_at: datetime diff --git a/document-parser/domain/ports.py b/document-parser/domain/ports.py index d6929b4..f46c8b7 100644 --- a/document-parser/domain/ports.py +++ b/document-parser/domain/ports.py @@ -11,7 +11,15 @@ from typing import TYPE_CHECKING, Protocol, runtime_checkable if TYPE_CHECKING: from datetime import datetime - from domain.models import AnalysisJob, Document, DocumentStoreLink, Store + from domain.models import ( + AnalysisJob, + Chunk, + ChunkEdit, + ChunkPush, + Document, + DocumentStoreLink, + Store, + ) from domain.value_objects import ( ChunkingOptions, ChunkResult, @@ -128,6 +136,53 @@ class DocumentStoreLinkRepository(Protocol): async def delete(self, document_id: str, store_id: str) -> bool: ... +class ChunkRepository(Protocol): + """Port for first-class chunk persistence (introduced by #205).""" + + async def insert(self, chunk: Chunk) -> None: ... + + async def insert_many(self, chunks: list[Chunk]) -> None: ... + + async def update(self, chunk: Chunk) -> None: ... + + async def soft_delete(self, chunk_id: str, *, at: datetime) -> bool: ... + + async def find_for_document( + self, + document_id: str, + *, + include_deleted: bool = False, + ) -> list[Chunk]: ... + + async def find_by_id(self, chunk_id: str) -> Chunk | None: ... + + +class ChunkEditRepository(Protocol): + """Port for the immutable chunk_edits audit log (introduced by #205).""" + + async def insert(self, edit: ChunkEdit) -> None: ... + + async def find_for_document( + self, + document_id: str, + *, + limit: int = 50, + offset: int = 0, + ) -> list[ChunkEdit]: ... + + async def find_for_chunk(self, chunk_id: str) -> list[ChunkEdit]: ... + + +class ChunkPushRepository(Protocol): + """Port for chunk_pushes snapshots (introduced by #205).""" + + async def insert(self, push: ChunkPush) -> None: ... + + async def find_by_id(self, push_id: str) -> ChunkPush | None: ... + + async def find_latest(self, document_id: str, store_id: str) -> ChunkPush | None: ... + + 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 d3f1baf..cf9b258 100644 --- a/document-parser/domain/value_objects.py +++ b/document-parser/domain/value_objects.py @@ -61,6 +61,20 @@ class DocumentStoreLinkState(StrEnum): FAILED = "Failed" +class ChunkEditAction(StrEnum): + """The five mutating operations the chunks editor supports. + + Recorded on every `ChunkEdit` row so the audit trail can answer "who + did what, when, and why" without resorting to JSON-path matching. + """ + + INSERT = "insert" + UPDATE = "update" + DELETE = "delete" + MERGE = "merge" + SPLIT = "split" + + @dataclass(frozen=True) class PageElement: type: str diff --git a/document-parser/persistence/chunk_edit_repo.py b/document-parser/persistence/chunk_edit_repo.py new file mode 100644 index 0000000..eadaf34 --- /dev/null +++ b/document-parser/persistence/chunk_edit_repo.py @@ -0,0 +1,145 @@ +"""Chunk edit / push repositories — immutable audit log + push snapshots (#205).""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime + +from domain.models import ChunkEdit, ChunkPush +from domain.value_objects import ChunkEditAction +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_edit(row) -> ChunkEdit: + return ChunkEdit( + id=row["id"], + document_id=row["document_id"], + chunk_id=row["chunk_id"], + action=ChunkEditAction(row["action"]), + actor=row["actor"], + at=_parse_iso(row["at"]) or datetime.now(UTC), + before=json.loads(row["before_json"]) if row["before_json"] else None, + after=json.loads(row["after_json"]) if row["after_json"] else None, + parents=json.loads(row["parents_json"]) if row["parents_json"] else [], + children=json.loads(row["children_json"]) if row["children_json"] else [], + reason=row["reason"], + ) + + +def _row_to_push(row) -> ChunkPush: + return ChunkPush( + id=row["id"], + document_id=row["document_id"], + store_id=row["store_id"], + chunkset_hash=row["chunkset_hash"], + chunk_ids=json.loads(row["chunk_ids"]) if row["chunk_ids"] else [], + pushed_at=_parse_iso(row["pushed_at"]) or datetime.now(UTC), + ) + + +class SqliteChunkEditRepository: + """SQLite implementation of the ChunkEditRepository port. + + The audit log is append-only: this repo offers `insert` and reads, + but no update or delete. Aborted edits should not produce an audit + row in the first place — that contract is enforced in the service + layer (audit + chunk write happen in the same SQL transaction). + """ + + async def insert(self, edit: ChunkEdit) -> None: + async with get_connection() as db: + await db.execute( + """INSERT INTO chunk_edits + (id, document_id, chunk_id, action, actor, at, + before_json, after_json, parents_json, children_json, reason) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + edit.id, + edit.document_id, + edit.chunk_id, + edit.action.value, + edit.actor, + str(edit.at), + json.dumps(edit.before) if edit.before is not None else None, + json.dumps(edit.after) if edit.after is not None else None, + json.dumps(edit.parents), + json.dumps(edit.children), + edit.reason, + ), + ) + await db.commit() + + async def find_for_document( + self, + document_id: str, + *, + limit: int = 50, + offset: int = 0, + ) -> list[ChunkEdit]: + async with get_connection() as db: + cursor = await db.execute( + """SELECT * FROM chunk_edits + WHERE document_id = ? + ORDER BY at DESC + LIMIT ? OFFSET ?""", + (document_id, limit, offset), + ) + rows = await cursor.fetchall() + return [_row_to_edit(r) for r in rows] + + async def find_for_chunk(self, chunk_id: str) -> list[ChunkEdit]: + async with get_connection() as db: + cursor = await db.execute( + "SELECT * FROM chunk_edits WHERE chunk_id = ? ORDER BY at ASC", + (chunk_id,), + ) + rows = await cursor.fetchall() + return [_row_to_edit(r) for r in rows] + + +class SqliteChunkPushRepository: + """SQLite implementation of the ChunkPushRepository port.""" + + async def insert(self, push: ChunkPush) -> None: + async with get_connection() as db: + await db.execute( + """INSERT INTO chunk_pushes + (id, document_id, store_id, chunkset_hash, chunk_ids, pushed_at) + VALUES (?, ?, ?, ?, ?, ?)""", + ( + push.id, + push.document_id, + push.store_id, + push.chunkset_hash, + json.dumps(push.chunk_ids), + str(push.pushed_at), + ), + ) + await db.commit() + + async def find_by_id(self, push_id: str) -> ChunkPush | None: + async with get_connection() as db: + cursor = await db.execute("SELECT * FROM chunk_pushes WHERE id = ?", (push_id,)) + row = await cursor.fetchone() + return _row_to_push(row) if row else None + + async def find_latest(self, document_id: str, store_id: str) -> ChunkPush | None: + async with get_connection() as db: + cursor = await db.execute( + """SELECT * FROM chunk_pushes + WHERE document_id = ? AND store_id = ? + ORDER BY pushed_at DESC + LIMIT 1""", + (document_id, store_id), + ) + row = await cursor.fetchone() + return _row_to_push(row) if row else None diff --git a/document-parser/persistence/chunk_repo.py b/document-parser/persistence/chunk_repo.py new file mode 100644 index 0000000..c0aa24c --- /dev/null +++ b/document-parser/persistence/chunk_repo.py @@ -0,0 +1,134 @@ +"""Chunk repository — SQLite CRUD for first-class chunks (#205).""" + +from __future__ import annotations + +import json +from dataclasses import asdict +from datetime import UTC, datetime + +from domain.models import Chunk +from domain.value_objects import ChunkBbox, ChunkDocItem +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_chunk(row) -> Chunk: + headings = json.loads(row["headings"]) if row["headings"] else [] + bboxes_raw = json.loads(row["bboxes"]) if row["bboxes"] else [] + doc_items_raw = json.loads(row["doc_items"]) if row["doc_items"] else [] + return Chunk( + id=row["id"], + document_id=row["document_id"], + sequence=row["sequence"], + text=row["text"], + headings=headings, + source_page=row["source_page"], + bboxes=[ChunkBbox(page=b["page"], bbox=b["bbox"]) for b in bboxes_raw], + doc_items=[ChunkDocItem(self_ref=d["self_ref"], label=d["label"]) for d in doc_items_raw], + token_count=row["token_count"], + created_at=_parse_iso(row["created_at"]) or datetime.now(UTC), + updated_at=_parse_iso(row["updated_at"]) or datetime.now(UTC), + deleted_at=_parse_iso(row["deleted_at"]), + ) + + +def _chunk_to_params(c: Chunk) -> tuple: + return ( + c.id, + c.document_id, + c.sequence, + c.text, + json.dumps(c.headings), + c.source_page, + json.dumps([asdict(b) for b in c.bboxes]), + json.dumps([asdict(d) for d in c.doc_items]), + c.token_count, + str(c.created_at), + str(c.updated_at), + str(c.deleted_at) if c.deleted_at else None, + ) + + +_INSERT_SQL = """INSERT INTO chunks + (id, document_id, sequence, text, headings, source_page, + bboxes, doc_items, token_count, created_at, updated_at, deleted_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""" + + +class SqliteChunkRepository: + """SQLite implementation of the ChunkRepository port.""" + + async def insert(self, chunk: Chunk) -> None: + async with get_connection() as db: + await db.execute(_INSERT_SQL, _chunk_to_params(chunk)) + await db.commit() + + async def insert_many(self, chunks: list[Chunk]) -> None: + if not chunks: + return + async with get_connection() as db: + await db.executemany(_INSERT_SQL, [_chunk_to_params(c) for c in chunks]) + await db.commit() + + async def update(self, chunk: Chunk) -> None: + async with get_connection() as db: + await db.execute( + """UPDATE chunks SET + sequence = ?, text = ?, headings = ?, source_page = ?, + bboxes = ?, doc_items = ?, token_count = ?, + updated_at = ?, deleted_at = ? + WHERE id = ?""", + ( + chunk.sequence, + chunk.text, + json.dumps(chunk.headings), + chunk.source_page, + json.dumps([asdict(b) for b in chunk.bboxes]), + json.dumps([asdict(d) for d in chunk.doc_items]), + chunk.token_count, + str(chunk.updated_at), + str(chunk.deleted_at) if chunk.deleted_at else None, + chunk.id, + ), + ) + await db.commit() + + async def soft_delete(self, chunk_id: str, *, at: datetime) -> bool: + async with get_connection() as db: + cursor = await db.execute( + "UPDATE chunks SET deleted_at = ?, updated_at = ? WHERE id = ?", + (str(at), str(at), chunk_id), + ) + await db.commit() + return cursor.rowcount > 0 + + async def find_for_document( + self, + document_id: str, + *, + include_deleted: bool = False, + ) -> list[Chunk]: + clause = "" if include_deleted else " AND deleted_at IS NULL" + async with get_connection() as db: + cursor = await db.execute( + f"""SELECT * FROM chunks + WHERE document_id = ?{clause} + ORDER BY sequence ASC""", + (document_id,), + ) + rows = await cursor.fetchall() + return [_row_to_chunk(r) for r in rows] + + async def find_by_id(self, chunk_id: str) -> Chunk | None: + async with get_connection() as db: + cursor = await db.execute("SELECT * FROM chunks WHERE id = ?", (chunk_id,)) + row = await cursor.fetchone() + return _row_to_chunk(row) if row else None diff --git a/document-parser/persistence/database.py b/document-parser/persistence/database.py index 7398f4c..526974b 100644 --- a/document-parser/persistence/database.py +++ b/document-parser/persistence/database.py @@ -70,6 +70,50 @@ CREATE TABLE IF NOT EXISTS document_store_links ( 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); + +-- 0.6.0 — Chunks promoted to first-class entities + audit trail (#205). +CREATE TABLE IF NOT EXISTS chunks ( + id TEXT PRIMARY KEY, + document_id TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE, + sequence INTEGER NOT NULL, + text TEXT NOT NULL, + headings TEXT NOT NULL DEFAULT '[]', + source_page INTEGER, + bboxes TEXT NOT NULL DEFAULT '[]', + doc_items TEXT NOT NULL DEFAULT '[]', + token_count INTEGER, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + deleted_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_chunks_doc ON chunks(document_id); +CREATE INDEX IF NOT EXISTS idx_chunks_doc_seq ON chunks(document_id, sequence); + +CREATE TABLE IF NOT EXISTS chunk_edits ( + id TEXT PRIMARY KEY, + document_id TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE, + chunk_id TEXT, + action TEXT NOT NULL, + actor TEXT NOT NULL DEFAULT 'system', + at TEXT NOT NULL, + before_json TEXT, + after_json TEXT, + parents_json TEXT NOT NULL DEFAULT '[]', + children_json TEXT NOT NULL DEFAULT '[]', + reason TEXT +); +CREATE INDEX IF NOT EXISTS idx_chunk_edits_doc_at ON chunk_edits(document_id, at); +CREATE INDEX IF NOT EXISTS idx_chunk_edits_chunk ON chunk_edits(chunk_id); + +CREATE TABLE IF NOT EXISTS chunk_pushes ( + 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, + chunkset_hash TEXT NOT NULL, + chunk_ids TEXT NOT NULL, + pushed_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_chunk_pushes_doc_store ON chunk_pushes(document_id, store_id); """ diff --git a/document-parser/tests/test_chunk_editing.py b/document-parser/tests/test_chunk_editing.py new file mode 100644 index 0000000..38b8bbc --- /dev/null +++ b/document-parser/tests/test_chunk_editing.py @@ -0,0 +1,196 @@ +"""Tests for the pure-domain chunk-editing operations (#205).""" + +from __future__ import annotations + +import pytest + +from domain.chunk_editing import ( + ChunkEditingError, + delete, + insert, + merge, + split, + update, +) +from domain.models import Chunk + + +def _make(document_id: str = "doc-1", count: int = 3) -> list[Chunk]: + out: list[Chunk] = [] + for i in range(count): + out.append( + Chunk( + id=f"c-{i}", + document_id=document_id, + sequence=i, + text=f"text {i}", + headings=[f"H{i}"], + source_page=i + 1, + ) + ) + return out + + +# --------------------------------------------------------------------------- +# insert +# --------------------------------------------------------------------------- + + +def test_insert_at_end_appends_chunk_with_correct_sequence() -> None: + chunks = _make(count=2) + out, new = insert(chunks, at_position=2, text="new", document_id="doc-1") + assert len(out) == 3 + assert out[2].id == new.id + assert new.sequence == 2 + # Existing chunks untouched. + assert out[0].sequence == 0 + assert out[1].sequence == 1 + + +def test_insert_in_middle_shifts_subsequent_sequences() -> None: + chunks = _make(count=3) + out, new = insert(chunks, at_position=1, text="middle", document_id="doc-1") + sequences = [c.sequence for c in out] + assert sequences == [0, 1, 2, 3] + assert out[1].id == new.id + # Original chunk-1 was shifted to sequence 2. + assert next(c for c in out if c.id == "c-1").sequence == 2 + + +def test_insert_out_of_range_raises() -> None: + chunks = _make(count=2) + with pytest.raises(ChunkEditingError): + insert(chunks, at_position=99, text="x", document_id="doc-1") + + +# --------------------------------------------------------------------------- +# update +# --------------------------------------------------------------------------- + + +def test_update_changes_text_in_place() -> None: + chunks = _make() + out, modified = update(chunks, "c-1", text="edited") + assert modified.id == "c-1" + assert modified.text == "edited" + # Identity preserved. + assert next(c for c in out if c.id == "c-1").text == "edited" + + +def test_update_can_change_headings_only() -> None: + chunks = _make() + _, modified = update(chunks, "c-0", headings=["X"]) + assert modified.headings == ["X"] + + +def test_update_unknown_chunk_raises() -> None: + chunks = _make() + with pytest.raises(ChunkEditingError): + update(chunks, "missing", text="x") + + +def test_update_deleted_chunk_raises() -> None: + chunks = _make() + delete(chunks, "c-1") + with pytest.raises(ChunkEditingError): + update(chunks, "c-1", text="x") + + +# --------------------------------------------------------------------------- +# delete +# --------------------------------------------------------------------------- + + +def test_delete_marks_deleted_at() -> None: + chunks = _make() + out, deleted = delete(chunks, "c-1") + target = next(c for c in out if c.id == "c-1") + assert target.deleted_at is not None + assert deleted is target + + +def test_delete_is_idempotent() -> None: + chunks = _make() + out_a, _ = delete(chunks, "c-1") + out_b, _ = delete(out_a, "c-1") + target_a = next(c for c in out_a if c.id == "c-1") + target_b = next(c for c in out_b if c.id == "c-1") + assert target_a.deleted_at == target_b.deleted_at + + +# --------------------------------------------------------------------------- +# merge +# --------------------------------------------------------------------------- + + +def test_merge_removes_sources_and_returns_new_chunk() -> None: + chunks = _make(count=3) + out, merged = merge(chunks, ["c-0", "c-1"]) + ids = {c.id for c in out} + assert "c-0" not in ids + assert "c-1" not in ids + assert merged.id in ids + assert merged.text == "text 0\ntext 1" + # Merged chunk inherits headings from first source. + assert merged.headings == ["H0"] + # Source page is the min. + assert merged.source_page == 1 + + +def test_merge_requires_at_least_two_chunks() -> None: + chunks = _make() + with pytest.raises(ChunkEditingError): + merge(chunks, ["c-0"]) + + +def test_merge_across_documents_raises() -> None: + chunks = _make(document_id="doc-1", count=2) + other = Chunk(id="x-0", document_id="doc-2", sequence=10, text="other") + chunks.append(other) + with pytest.raises(ChunkEditingError): + merge(chunks, ["c-0", "x-0"]) + + +# --------------------------------------------------------------------------- +# split +# --------------------------------------------------------------------------- + + +def test_split_creates_two_new_chunks() -> None: + chunks = _make(count=2) + out, left, right = split(chunks, "c-0", at_offset=4) + assert left.text == "text" + assert right.text == " 0" + ids = {c.id for c in out} + # Source replaced by two new chunks. + assert "c-0" not in ids + assert left.id in ids + assert right.id in ids + + +def test_split_preserves_headings_on_both_halves() -> None: + chunks = _make(count=1) + _, left, right = split(chunks, "c-0", at_offset=2) + assert left.headings == ["H0"] + assert right.headings == ["H0"] + + +def test_split_shifts_subsequent_sequences() -> None: + chunks = _make(count=3) + out, _, _ = split(chunks, "c-0", at_offset=2) + # After split, c-1 (was seq=1) and c-2 (was seq=2) shift up by 1. + seqs = sorted(c.sequence for c in out) + assert seqs == [0, 1, 2, 3] + + +def test_split_at_zero_offset_raises() -> None: + chunks = _make() + with pytest.raises(ChunkEditingError): + split(chunks, "c-0", at_offset=0) + + +def test_split_beyond_text_length_raises() -> None: + chunks = _make() + target = next(c for c in chunks if c.id == "c-0") + with pytest.raises(ChunkEditingError): + split(chunks, "c-0", at_offset=len(target.text)) diff --git a/document-parser/tests/test_chunk_repos.py b/document-parser/tests/test_chunk_repos.py new file mode 100644 index 0000000..93174c6 --- /dev/null +++ b/document-parser/tests/test_chunk_repos.py @@ -0,0 +1,212 @@ +"""Tests for the chunk / chunk_edit / chunk_push repositories (#205).""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from domain.models import Chunk, ChunkEdit, ChunkPush, Document +from domain.value_objects import ChunkBbox, ChunkEditAction +from persistence.chunk_edit_repo import ( + SqliteChunkEditRepository, + SqliteChunkPushRepository, +) +from persistence.chunk_repo import SqliteChunkRepository +from persistence.database import init_db +from persistence.document_repo import SqliteDocumentRepository + + +@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 +async def doc(): + repo = SqliteDocumentRepository() + document = Document(id="doc-1", filename="t.pdf", storage_path="/tmp/t.pdf") + await repo.insert(document) + return document + + +@pytest.fixture +def chunk_repo(): + return SqliteChunkRepository() + + +@pytest.fixture +def edit_repo(): + return SqliteChunkEditRepository() + + +@pytest.fixture +def push_repo(): + return SqliteChunkPushRepository() + + +class TestChunkRepo: + async def test_insert_and_find_for_document(self, doc, chunk_repo): + c = Chunk( + id="c-1", + document_id=doc.id, + sequence=0, + text="alpha", + headings=["A"], + source_page=1, + bboxes=[ChunkBbox(page=1, bbox=[0, 0, 10, 10])], + ) + await chunk_repo.insert(c) + + out = await chunk_repo.find_for_document(doc.id) + assert len(out) == 1 + assert out[0].id == "c-1" + assert out[0].text == "alpha" + assert out[0].headings == ["A"] + assert out[0].bboxes[0].bbox == [0, 0, 10, 10] + + async def test_insert_many_round_trips(self, doc, chunk_repo): + chunks = [ + Chunk(id=f"c-{i}", document_id=doc.id, sequence=i, text=f"t{i}") for i in range(5) + ] + await chunk_repo.insert_many(chunks) + + out = await chunk_repo.find_for_document(doc.id) + assert len(out) == 5 + # Returned in sequence order. + assert [c.id for c in out] == [f"c-{i}" for i in range(5)] + + async def test_soft_delete_excludes_chunk_unless_requested(self, doc, chunk_repo): + await chunk_repo.insert(Chunk(id="c-1", document_id=doc.id, sequence=0, text="a")) + await chunk_repo.soft_delete("c-1", at=datetime(2026, 4, 29, 12, tzinfo=UTC)) + + active = await chunk_repo.find_for_document(doc.id) + assert active == [] + + all_chunks = await chunk_repo.find_for_document(doc.id, include_deleted=True) + assert len(all_chunks) == 1 + assert all_chunks[0].deleted_at is not None + + async def test_update_persists_text_and_sequence(self, doc, chunk_repo): + c = Chunk(id="c-1", document_id=doc.id, sequence=0, text="old") + await chunk_repo.insert(c) + + c.text = "new" + c.sequence = 5 + c.updated_at = datetime(2026, 4, 29, 12, tzinfo=UTC) + await chunk_repo.update(c) + + out = await chunk_repo.find_by_id("c-1") + assert out is not None + assert out.text == "new" + assert out.sequence == 5 + + async def test_cascade_delete_with_document(self, doc, chunk_repo): + await chunk_repo.insert(Chunk(id="c-1", document_id=doc.id, sequence=0, text="a")) + await SqliteDocumentRepository().delete(doc.id) + out = await chunk_repo.find_for_document(doc.id, include_deleted=True) + assert out == [] + + +class TestChunkEditRepo: + async def test_insert_and_history(self, doc, edit_repo): + edit = ChunkEdit( + id="e-1", + document_id=doc.id, + chunk_id="c-1", + action=ChunkEditAction.UPDATE, + actor="user@example", + at=datetime(2026, 4, 29, 12, tzinfo=UTC), + before={"text": "old"}, + after={"text": "new"}, + reason="typo", + ) + await edit_repo.insert(edit) + + out = await edit_repo.find_for_document(doc.id) + assert len(out) == 1 + assert out[0].action == ChunkEditAction.UPDATE + assert out[0].before == {"text": "old"} + assert out[0].after == {"text": "new"} + assert out[0].reason == "typo" + + async def test_history_orders_newest_first(self, doc, edit_repo): + for i in range(3): + await edit_repo.insert( + ChunkEdit( + id=f"e-{i}", + document_id=doc.id, + chunk_id=f"c-{i}", + action=ChunkEditAction.INSERT, + actor="system", + at=datetime(2026, 4, 29, 12 + i, tzinfo=UTC), + ) + ) + out = await edit_repo.find_for_document(doc.id) + ids = [e.id for e in out] + assert ids == ["e-2", "e-1", "e-0"] + + async def test_find_for_chunk_filters(self, doc, edit_repo): + for i in range(3): + await edit_repo.insert( + ChunkEdit( + id=f"e-{i}", + document_id=doc.id, + chunk_id="c-target" if i == 1 else f"c-{i}", + action=ChunkEditAction.UPDATE, + actor="system", + at=datetime(2026, 4, 29, 12 + i, tzinfo=UTC), + ) + ) + out = await edit_repo.find_for_chunk("c-target") + assert [e.id for e in out] == ["e-1"] + + async def test_lineage_round_trips(self, doc, edit_repo): + edit = ChunkEdit( + id="e-merge", + document_id=doc.id, + chunk_id="c-merged", + action=ChunkEditAction.MERGE, + actor="system", + at=datetime(2026, 4, 29, 12, tzinfo=UTC), + parents=["c-1", "c-2"], + children=["c-merged"], + ) + await edit_repo.insert(edit) + out = await edit_repo.find_for_document(doc.id) + assert out[0].parents == ["c-1", "c-2"] + assert out[0].children == ["c-merged"] + + +class TestChunkPushRepo: + async def test_insert_and_find_latest(self, doc, push_repo): + first = ChunkPush( + id="p-1", + document_id=doc.id, + store_id="default", + chunkset_hash="hash-old", + chunk_ids=["c-1", "c-2"], + pushed_at=datetime(2026, 4, 29, 10, tzinfo=UTC), + ) + latest = ChunkPush( + id="p-2", + document_id=doc.id, + store_id="default", + chunkset_hash="hash-new", + chunk_ids=["c-1", "c-3"], + pushed_at=datetime(2026, 4, 29, 11, tzinfo=UTC), + ) + await push_repo.insert(first) + await push_repo.insert(latest) + + found = await push_repo.find_latest(doc.id, "default") + assert found is not None + assert found.id == "p-2" + assert found.chunkset_hash == "hash-new" + assert found.chunk_ids == ["c-1", "c-3"] + + async def test_find_latest_returns_none_when_absent(self, doc, push_repo): + assert await push_repo.find_latest(doc.id, "default") is None