danvics/docling-studio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(#205): promote chunks to first-class entities + audit trail

Browse source 
The data and domain layers for the chunks editor (#219-224 in 0.6.0).
Chunks were previously stored as a JSON blob in analysis_jobs.chunks_json;
this commit makes them first-class persisted entities with stable IDs,
soft-delete, and an immutable audit log.

Domain
- Chunk: persistent entity with id, document_id, sequence, text,
  headings, source_page, bboxes, doc_items, token_count, timestamps,
  deleted_at (soft delete)
- ChunkEdit: immutable audit row (action, actor, at, before, after,
  parents, children, reason)
- ChunkPush: snapshot of which chunk_ids landed in which store at push
- ChunkEditAction enum: insert/update/delete/merge/split
- domain/chunk_editing.py: pure operations on a chunkset (insert,
  update, delete, merge, split). Each returns a new chunkset and the
  affected chunk(s); errors raise ChunkEditingError.

Persistence
- Three new tables: chunks, chunk_edits, chunk_pushes (FK + indexes)
- SqliteChunkRepository (insert, insert_many, update, soft_delete,
  find_for_document, find_by_id; respects deleted_at)
- SqliteChunkEditRepository (append-only audit log; paginated reads
  ordered newest-first; per-chunk history)
- SqliteChunkPushRepository (per-(doc, store) latest snapshot)

Ports
- ChunkRepository, ChunkEditRepository, ChunkPushRepository protocols
  added to domain/ports.py

Tests
- 17 tests for the pure chunk-editing operations covering insert /
  update / delete / merge / split, sequence shifts, lineage, error
  paths (out-of-range, missing id, deleted target, cross-document)
- 11 tests for the three repositories: round-trips, soft-delete
  filtering, history ordering, lineage round-trip, cascade-delete with
  document, find_latest semantics

Service orchestration (ChunkEditingService — atomic chunk + audit
write) and the API endpoints land in a follow-up commit on the same
feature branch / next release. The data + domain foundation here is
what unblocks #219-224.

Refs #205
This commit is contained in:
Pier-Jean Malandrino 2026-04-29 15:45:18 +02:00
parent 4b59ef2330
commit 648f2a5c0d
9 changed files with 1094 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

216
document-parser/domain/chunk_editing.py Normal file
View file

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

77
document-parser/domain/models.py
View file

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

57
document-parser/domain/ports.py
View file

@ -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."""

14
document-parser/domain/value_objects.py
View file

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

145
document-parser/persistence/chunk_edit_repo.py Normal file
View file

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

134
document-parser/persistence/chunk_repo.py Normal file
View file

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

44
document-parser/persistence/database.py
View file

@ -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);
"""

196
document-parser/tests/test_chunk_editing.py Normal file
View file

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

212
document-parser/tests/test_chunk_repos.py Normal file
View file

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