feat: build document intelligence workflow

backend/alembic/versions/5b7572e0875a_add_document_chunks.py

@@ -0,0 +1,41 @@
+"""add document chunks
+
+Revision ID: 5b7572e0875a
+Revises: 0fd360eb89e4
+Create Date: 2026-08-10 15:17:52.645173
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = '5b7572e0875a'
+down_revision: Union[str, None] = '0fd360eb89e4'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('document_chunks',
+    sa.Column('id', sa.UUID(), nullable=False),
+    sa.Column('document_version_id', sa.UUID(), nullable=False),
+    sa.Column('chunk_index', sa.Integer(), nullable=False),
+    sa.Column('text', sa.Text(), nullable=False),
+    sa.Column('page_number', sa.Integer(), nullable=True),
+    sa.Column('section', sa.Text(), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['document_version_id'], ['document_versions.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_document_chunks_document_version_id'), 'document_chunks', ['document_version_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_document_chunks_document_version_id'), table_name='document_chunks')
+    op.drop_table('document_chunks')
+    # ### end Alembic commands ###

backend/app/models/__init__.py

@@ -7,6 +7,7 @@ from .user import User
 from .workspace import Workspace
 from .document import Document
 from .document_version import DocumentVersion
+from .document_chunk import DocumentChunk
 
 from .knowledge_item import KnowledgeItem
 from .knowledge_evidence import KnowledgeEvidence

backend/app/models/document_chunk.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, Text, func
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.database.database import Base
+
+
+class DocumentChunk(Base):
+    """
+    A retrievable chunk of a specific document version.
+
+    Chunks preserve lightweight provenance so retrieval results
+    can be traced back to their original document location.
+    """
+
+    __tablename__ = "document_chunks"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        primary_key=True,
+        default=uuid.uuid4,
+    )
+
+    document_version_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey(
+            "document_versions.id",
+            ondelete="CASCADE",
+        ),
+        nullable=False,
+        index=True,
+    )
+
+    chunk_index: Mapped[int] = mapped_column(
+        Integer,
+        nullable=False,
+    )
+
+    text: Mapped[str] = mapped_column(
+        Text,
+        nullable=False,
+    )
+
+    page_number: Mapped[int | None] = mapped_column(
+        Integer,
+        nullable=True,
+    )
+
+    section: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+
+    document_version: Mapped["DocumentVersion"] = relationship(
+        back_populates="chunks",
+    )
+
+    def __repr__(self) -> str:
+        return (
+            f"<DocumentChunk("
+            f"document_version={self.document_version_id}, "
+            f"index={self.chunk_index})>"
+        )

backend/app/models/document_version.py

@@ -18,6 +18,10 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database.database import Base
 
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from app.models.document_chunk import DocumentChunk
 
 class DocumentVersionStatus(str, enum.Enum):
     UPLOADED = "UPLOADED"
@@ -129,6 +133,11 @@ class DocumentVersion(Base):
         cascade="all, delete-orphan",
     )
 
+    chunks: Mapped[list["DocumentChunk"]] = relationship(
+        back_populates="document_version",
+        cascade="all, delete-orphan",
+    )
+
     workflow_runs: Mapped[list["WorkflowRun"]] = relationship(
         back_populates="document_version",
         cascade="all, delete-orphan",

backend/app/repositories/knowledge_repository.py

@@ -79,4 +79,21 @@ class KnowledgeRepository(BaseRepository[KnowledgeItem]):
                 == document_version_id
             )
             .all()
+        )
+    
+    def find_candidates(
+        self,
+        *,
+        workspace_id,
+        knowledge_type,
+        title,
+    ):
+        return (
+            self.db.query(KnowledgeItem)
+            .filter(
+                KnowledgeItem.workspace_id == workspace_id,
+                KnowledgeItem.type == knowledge_type,
+                KnowledgeItem.title.ilike(f"%{title}%"),
+            )
+            .all()
         )

backend/app/services/chunking_service.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from app.models.document_chunk import DocumentChunk
+
+
+class ChunkingService:
+    """
+    Converts extracted document sections into persistent,
+    retrieval-ready document chunks.
+    """
+
+    CHUNK_SIZE = 500
+    OVERLAP = 50
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    def chunk_document(
+        self,
+        *,
+        document_version_id,
+        sections: list[dict],
+    ) -> dict:
+
+        # Make chunk generation idempotent.
+        # Re-running the workflow won't create duplicate chunks.
+        self.db.query(DocumentChunk).filter(
+            DocumentChunk.document_version_id
+            == document_version_id
+        ).delete(
+            synchronize_session=False
+        )
+
+        chunk_index = 0
+
+        for section in sections:
+            text = (section.get("text") or "").strip()
+
+            if not text:
+                continue
+
+            metadata = section.get("metadata") or {}
+
+            chunks = self._chunk_text(text)
+
+            for chunk_text in chunks:
+                self.db.add(
+                    DocumentChunk(
+                        document_version_id=document_version_id,
+                        chunk_index=chunk_index,
+                        text=chunk_text,
+                        page_number=metadata.get("page"),
+                        section=(
+                            metadata.get("section_title")
+                            or metadata.get("section_type")
+                        ),
+                    )
+                )
+
+                chunk_index += 1
+
+        self.db.commit()
+
+        return {
+            "chunks_created": chunk_index,
+        }
+
+    @classmethod
+    def _chunk_text(
+        cls,
+        text: str,
+    ) -> list[str]:
+
+        if len(text) <= cls.CHUNK_SIZE:
+            return [text]
+
+        chunks = []
+
+        start = 0
+        text_length = len(text)
+
+        while start < text_length:
+
+            end = min(
+                start + cls.CHUNK_SIZE,
+                text_length,
+            )
+
+            chunk = text[start:end].strip()
+
+            if chunk:
+                chunks.append(chunk)
+
+            if end >= text_length:
+                break
+
+            start = end - cls.OVERLAP
+
+        return chunks

backend/app/services/document_processor.py

@@ -107,6 +107,65 @@ def clean_text(text: str) -> str:
     text = re.sub(r"\n{3,}", "\n\n", text)
     return text.strip()
 
+def extraction_quality(sections: list[TextSection]) -> float:
+    """
+    Estimate whether extracted PDF text is substantial enough
+    to be considered usable.
+
+    Returns a score from 0.0 to 1.0.
+    """
+
+    if not sections:
+        return 0.0
+
+    total_chars = sum(len(section.text) for section in sections)
+
+    if total_chars == 0:
+        return 0.0
+
+    text = "\n".join(section.text for section in sections)
+
+    words = re.findall(r"\b\w+\b", text)
+    if not words:
+        return 0.0
+
+    alphanumeric_chars = sum(
+        char.isalnum()
+        for char in text
+    )
+
+    alphanumeric_ratio = alphanumeric_chars / max(len(text), 1)
+
+    word_count = len(words)
+
+    score = 0.0
+
+    # Amount of actual text
+    if total_chars >= 2000:
+        score += 0.4
+    elif total_chars >= 1000:
+        score += 0.25
+    elif total_chars >= 500:
+        score += 0.1
+
+    # Number of words
+    if word_count >= 300:
+        score += 0.3
+    elif word_count >= 150:
+        score += 0.2
+    elif word_count >= 75:
+        score += 0.1
+
+    # Mostly actual text rather than symbols/noise
+    if alphanumeric_ratio >= 0.75:
+        score += 0.3
+    elif alphanumeric_ratio >= 0.60:
+        score += 0.2
+    elif alphanumeric_ratio >= 0.45:
+        score += 0.1
+
+    return min(score, 1.0)
+
 
 def detect_type(file_path: str, mime_type: str = None):
     extension = os.path.splitext(file_path)[1].lower()
@@ -285,7 +344,10 @@ def extract_pdf_docling(file_path: str) -> list[TextSection]:
 
 def extract_pdf_ocr(file_path: str) -> list[TextSection]:
     try:
-        images = convert_from_path(file_path)
+        images = convert_from_path(
+            file_path,
+            poppler_path=os.getenv("POPPLER_PATH"),
+        )
         sections = []
 
         for page_number, image in enumerate(images, start=1):
@@ -305,21 +367,73 @@ def extract_pdf_ocr(file_path: str) -> list[TextSection]:
 
 
 def extract_pdf_sections(file_path: str) -> list[TextSection]:
-    """PyMuPDF → Docling → OCR cascade."""
-    sections = extract_pdf_text_pymupdf(file_path) if fitz else extract_pdf_text_pypdf(file_path)
+    """
+    Extract PDF text using a quality-aware cascade:
+
+        PyMuPDF → Docling → OCR
+    """
+
+    # ------------------------------------------------------------
+    # 1. PyMuPDF
+    # ------------------------------------------------------------
 
-    if sum(len(s.text) for s in sections) > 500:
+    sections = (
+        extract_pdf_text_pymupdf(file_path)
+        if fitz
+        else extract_pdf_text_pypdf(file_path)
+    )
+
+    score = extraction_quality(sections)
+
+    logger.info(
+        "PyMuPDF extraction quality: %.2f (%s chars)",
+        score,
+        sum(len(s.text) for s in sections),
+    )
+
+    if score >= 0.6:
         return sections
 
-    logger.info("PyMuPDF extraction weak, trying Docling")
+    # ------------------------------------------------------------
+    # 2. Docling
+    # ------------------------------------------------------------
+
+    logger.info(
+        "PyMuPDF extraction quality insufficient, trying Docling"
+    )
+
     sections = extract_pdf_docling(file_path)
 
-    if sum(len(s.text) for s in sections) > 500:
+    score = extraction_quality(sections)
+
+    logger.info(
+        "Docling extraction quality: %.2f (%s chars)",
+        score,
+        sum(len(s.text) for s in sections),
+    )
+
+    if score >= 0.6:
         return sections
 
-    logger.info("Docling failed, OCR triggered")
-    return extract_pdf_ocr(file_path)
+    # ------------------------------------------------------------
+    # 3. OCR
+    # ------------------------------------------------------------
+
+    logger.info(
+        "Docling extraction quality insufficient, triggering OCR"
+    )
 
+    sections = extract_pdf_ocr(file_path)
+
+    score = extraction_quality(sections)
+
+    logger.info(
+        "OCR extraction quality: %.2f (%s chars)",
+        score,
+        sum(len(s.text) for s in sections),
+    )
+
+    return sections
 
 # ---------------------------------------------------------------------------
 # Other format extractors

backend/app/services/knowledge_link_service.py

@@ -0,0 +1,172 @@
+from __future__ import annotations
+
+from difflib import SequenceMatcher
+
+from sqlalchemy.orm import Session
+
+from app.models.knowledge_item import KnowledgeItem
+from app.models.knowledge_link import (
+    KnowledgeLink,
+    RelationshipType,
+)
+
+
+class KnowledgeLinkService:
+    """
+    Builds semantic relationships between KnowledgeItems.
+
+    Link creation is intentionally conservative:
+    only strong, deterministic relationships are created.
+    """
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    def link_document(
+        self,
+        *,
+        workspace_id,
+        document_version_id,
+    ) -> dict:
+        current_items = (
+            self.db.query(KnowledgeItem)
+            .filter(
+                KnowledgeItem.workspace_id == workspace_id,
+                KnowledgeItem.document_version_id == document_version_id,
+            )
+            .all()
+        )
+
+        existing_items = (
+            self.db.query(KnowledgeItem)
+            .filter(
+                KnowledgeItem.workspace_id == workspace_id,
+                KnowledgeItem.document_version_id != document_version_id,
+            )
+            .all()
+        )
+
+        created = 0
+
+        for source in current_items:
+            for target in existing_items:
+                if source.id == target.id:
+                    continue
+
+                relationship = self._infer_relationship(
+                    source,
+                    target,
+                )
+
+                if relationship is None:
+                    continue
+
+                relationship_type, confidence = relationship
+
+                if self._link_exists(
+                    source.id,
+                    target.id,
+                    relationship_type,
+                ):
+                    continue
+
+                self.db.add(
+                    KnowledgeLink(
+                        source_item_id=source.id,
+                        target_item_id=target.id,
+                        relationship_type=relationship_type,
+                        confidence=confidence,
+                    )
+                )
+
+                created += 1
+
+        self.db.commit()
+
+        return {
+            "links_created": created,
+        }
+
+    def _infer_relationship(
+        self,
+        source: KnowledgeItem,
+        target: KnowledgeItem,
+    ):
+        title_similarity = self._similarity(
+            source.title,
+            target.title,
+        )
+
+        value_similarity = self._similarity(
+            source.value,
+            target.value,
+        )
+
+        # Same concept with highly similar values.
+        if (
+            source.type == target.type
+            and title_similarity >= 0.85
+            and value_similarity >= 0.80
+        ):
+            return (
+                RelationshipType.RELATED_TO,
+                round(
+                    (title_similarity + value_similarity) / 2,
+                    3,
+                ),
+            )
+
+        # A method/treatment can be related to an entity/claim.
+        if (
+            source.type.name == "METHOD"
+            and target.type.name in {"ENTITY", "CLAIM", "OBSERVATION"}
+            and title_similarity >= 0.65
+        ):
+            return (
+                RelationshipType.USES,
+                round(title_similarity, 3),
+            )
+
+        # Metrics commonly describe claims/observations.
+        if (
+            source.type.name == "METRIC"
+            and target.type.name in {"CLAIM", "OBSERVATION"}
+            and title_similarity >= 0.65
+        ):
+            return (
+                RelationshipType.REFERENCES,
+                round(title_similarity, 3),
+            )
+
+        return None
+
+    def _link_exists(
+        self,
+        source_item_id,
+        target_item_id,
+        relationship_type: RelationshipType,
+    ) -> bool:
+        return (
+            self.db.query(KnowledgeLink)
+            .filter(
+                KnowledgeLink.source_item_id == source_item_id,
+                KnowledgeLink.target_item_id == target_item_id,
+                KnowledgeLink.relationship_type == relationship_type,
+            )
+            .first()
+            is not None
+        )
+
+    @staticmethod
+    def _similarity(
+        left: str | None,
+        right: str | None,
+    ) -> float:
+        if not left or not right:
+            return 0.0
+
+        return SequenceMatcher(
+            None,
+            left.lower().strip(),
+            right.lower().strip(),
+        ).ratio()

backend/app/services/reconciliation_service.py

@@ -0,0 +1,223 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from app.models.knowledge_item import KnowledgeItem
+from app.models.proposal import Proposal, ProposalStatus, ProposalType
+from app.repositories.knowledge_repository import KnowledgeRepository
+from difflib import SequenceMatcher
+
+class ReconciliationService:
+    """
+    Compares newly extracted knowledge against the existing
+    Knowledge Register and creates proposals for changes.
+
+    Reconciliation never directly mutates existing knowledge.
+    All changes go through PENDING proposals.
+    """
+
+    def __init__(self, db: Session):
+        self.db = db
+        self.repository = KnowledgeRepository(db)
+
+    def reconcile_document(
+        self,
+        *,
+        workspace_id,
+        document_version_id,
+    ) -> dict:
+        new_items = self.repository.list_by_document_version(
+            document_version_id
+        )
+
+        existing_items = (
+            self.db.query(KnowledgeItem)
+            .filter(
+                KnowledgeItem.workspace_id == workspace_id,
+                KnowledgeItem.document_version_id != document_version_id,
+            )
+            .all()
+        )
+
+        new_count = 0
+        duplicate_count = 0
+        conflict_count = 0
+
+        for item in new_items:
+            match = self._find_match(item, existing_items)
+
+            if match is None:
+                self._create_create_proposal(
+                    workspace_id=workspace_id,
+                    item=item,
+                )
+                new_count += 1
+                continue
+
+            if self._is_duplicate(item, match):
+                duplicate_count += 1
+                continue
+
+            self._create_update_proposal(
+                workspace_id=workspace_id,
+                existing_item=match,
+                new_item=item,
+            )
+            conflict_count += 1
+
+        self.db.commit()
+
+        return {
+            "new": new_count,
+            "duplicates": duplicate_count,
+            "conflicts": conflict_count,
+        }
+
+    def _find_match(
+        self,
+        item: KnowledgeItem,
+        existing_items: list[KnowledgeItem],
+    ) -> KnowledgeItem | None:
+
+        best_match = None
+        best_score = 0.0
+
+        new_title = self._normalize(item.title)
+
+        for existing in existing_items:
+            if existing.type != item.type:
+                continue
+
+            existing_title = self._normalize(existing.title)
+
+            # Exact match
+            if new_title == existing_title:
+                return existing
+
+            # Similarity match
+            score = SequenceMatcher(
+                None,
+                new_title,
+                existing_title,
+            ).ratio()
+
+            if score > best_score:
+                best_score = score
+                best_match = existing
+
+        if best_score >= 0.80:
+            return best_match
+
+        return None
+
+    def _is_duplicate(
+        self,
+        new_item: KnowledgeItem,
+        existing_item: KnowledgeItem,
+    ) -> bool:
+
+        return (
+            self._normalize(new_item.value)
+            == self._normalize(existing_item.value)
+        )
+
+    def _pending_proposal_exists(
+        self,
+        knowledge_item_id,
+        proposal_type: ProposalType,
+    ) -> bool:
+
+        return (
+            self.db.query(Proposal)
+            .filter(
+                Proposal.knowledge_item_id == knowledge_item_id,
+                Proposal.proposal_type == proposal_type,
+                Proposal.status == ProposalStatus.PENDING,
+            )
+            .first()
+            is not None
+        )
+
+    def _create_create_proposal(
+        self,
+        *,
+        workspace_id,
+        item: KnowledgeItem,
+    ):
+        if self._pending_proposal_exists(
+            item.id,
+            ProposalType.CREATE,
+        ):
+            return
+
+        proposal = Proposal(
+            workspace_id=workspace_id,
+            knowledge_item_id=item.id,
+            proposal_type=ProposalType.CREATE,
+            status=ProposalStatus.PENDING,
+            summary=f"Create knowledge item: {item.title}",
+            rationale=(
+                "No matching knowledge item was found in the "
+                "existing Knowledge Register."
+            ),
+            proposed_changes={
+                "type": item.type.value,
+                "title": item.title,
+                "value": item.value,
+                "summary": item.summary,
+                "attributes": item.attributes,
+                "confidence": item.confidence,
+            },
+        )
+
+        self.db.add(proposal)
+
+    def _create_update_proposal(
+        self,
+        *,
+        workspace_id,
+        existing_item: KnowledgeItem,
+        new_item: KnowledgeItem,
+    ):
+        if self._pending_proposal_exists(
+            existing_item.id,
+            ProposalType.UPDATE,
+        ):
+            return
+
+        proposal = Proposal(
+            workspace_id=workspace_id,
+            knowledge_item_id=existing_item.id,
+            proposal_type=ProposalType.UPDATE,
+            status=ProposalStatus.PENDING,
+            summary=f"Update knowledge item: {existing_item.title}",
+            rationale=(
+                "A matching knowledge item exists, but the newly "
+                "extracted value differs from the registered value."
+            ),
+            proposed_changes={
+                "existing": {
+                    "value": existing_item.value,
+                    "summary": existing_item.summary,
+                    "attributes": existing_item.attributes,
+                    "confidence": existing_item.confidence,
+                },
+                "proposed": {
+                    "value": new_item.value,
+                    "summary": new_item.summary,
+                    "attributes": new_item.attributes,
+                    "confidence": new_item.confidence,
+                },
+                "source_knowledge_item_id": str(new_item.id),
+            },
+        )
+
+        self.db.add(proposal)
+    @staticmethod
+    def _normalize(value: str | None) -> str:
+        if not value:
+            return ""
+
+        words = value.lower().strip().split()
+
+        return " ".join(sorted(words))

backend/app/workflow/executor.py

@@ -2,25 +2,19 @@ from __future__ import annotations
 
 from sqlalchemy.orm import Session
 
-from app.llm.client import LLMClient
-from app.repositories.knowledge_repository import KnowledgeRepository
 from app.services.workflow_service import WorkflowService
-from app.workflow.nodes.classify import classify
-from app.workflow.nodes.complete import complete
-from app.workflow.nodes.extract import extract
-from app.workflow.nodes.knowledge import knowledge
+from app.workflow.graph import build_workflow
 from app.workflow.state import WorkflowState
 
 
 class WorkflowExecutor:
     """
-    Executes the document workflow.
+    Executes the document workflow through LangGraph.
     """
 
     def __init__(self, db: Session):
         self.workflow_service = WorkflowService(db)
-        self.llm_client = LLMClient()
-        self.knowledge_repository = KnowledgeRepository(db)
+        self.graph = build_workflow(db)
 
     def execute(
         self,
@@ -29,21 +23,12 @@ class WorkflowExecutor:
     ) -> WorkflowState:
 
         try:
-            state = extract(state)
-
-            state = classify(state)
-
-            state = knowledge(
-                state,
-                self.llm_client,
-                self.knowledge_repository,
-            )
-
-            state = complete(state)
+            final_state = self.graph.invoke(state)
 
             self.workflow_service.complete_workflow(workflow)
 
-            return state
+            return final_state
 
         except Exception:
+            self.workflow_service.fail_workflow(workflow)
             raise

backend/app/workflow/graph.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+from langgraph.graph import END, START, StateGraph
+
+from app.llm.client import LLMClient
+from app.repositories.knowledge_repository import KnowledgeRepository
+from app.services.chunking_service import ChunkingService
+from app.services.knowledge_link_service import KnowledgeLinkService
+
+from app.workflow.nodes.chunk import chunk
+from app.workflow.nodes.classify import classify
+from app.workflow.nodes.complete import complete
+from app.workflow.nodes.extract import extract
+from app.workflow.nodes.knowledge import knowledge
+from app.workflow.nodes.link import link
+from app.workflow.nodes.reconcile import reconcile
+from app.workflow.state import WorkflowState
+
+
+def build_workflow(db: Session):
+    """
+    Build the DocWeave document intelligence workflow.
+    """
+
+    llm_client = LLMClient()
+    knowledge_repository = KnowledgeRepository(db)
+    chunking_service = ChunkingService(db)
+    knowledge_link_service = KnowledgeLinkService(db)
+
+    graph = StateGraph(WorkflowState)
+
+    # ------------------------------------------------------------------
+    # Nodes
+    # ------------------------------------------------------------------
+
+    graph.add_node("extract", extract)
+
+    graph.add_node(
+        "chunk",
+        lambda state: chunk(
+            state,
+            chunking_service,
+        ),
+    )
+
+    graph.add_node("classify", classify)
+
+    graph.add_node(
+        "knowledge",
+        lambda state: knowledge(
+            state,
+            llm_client,
+            knowledge_repository,
+        ),
+    )
+
+    graph.add_node(
+        "reconciliation",
+        lambda state: reconcile(
+            state,
+            db,
+        ),
+    )
+
+    graph.add_node(
+        "link",
+        lambda state: link(
+            state,
+            knowledge_link_service,
+        ),
+    )
+
+    graph.add_node("complete", complete)
+
+    # ------------------------------------------------------------------
+    # Workflow edges
+    # ------------------------------------------------------------------
+
+    graph.add_edge(START, "extract")
+    graph.add_edge("extract", "chunk")
+    graph.add_edge("chunk", "classify")
+    graph.add_edge("classify", "knowledge")
+    graph.add_edge("knowledge", "reconciliation")
+    graph.add_edge("reconciliation", "link")
+    graph.add_edge("link", "complete")
+    graph.add_edge("complete", END)
+
+    return graph.compile()

backend/app/workflow/nodes/chunk.py

@@ -0,0 +1,23 @@
+from app.services.chunking_service import ChunkingService
+from app.workflow.state import WorkflowState
+
+
+def chunk(
+    state: WorkflowState,
+    chunking_service: ChunkingService,
+) -> WorkflowState:
+    """
+    Persist retrieval-ready document chunks while preserving
+    lightweight provenance.
+    """
+
+    state.current_node = "CHUNKING"
+
+    result = chunking_service.chunk_document(
+        document_version_id=state.document_version_id,
+        sections=state.extracted_sections,
+    )
+
+    state.metadata["chunks_created"] = result["chunks_created"]
+
+    return state

backend/app/workflow/nodes/link.py

@@ -0,0 +1,25 @@
+from app.services.knowledge_link_service import KnowledgeLinkService
+from app.workflow.state import WorkflowState
+
+
+def link(
+    state: WorkflowState,
+    knowledge_link_service: KnowledgeLinkService,
+) -> WorkflowState:
+    """
+    Create semantic links between the newly extracted knowledge
+    and the existing Knowledge Register.
+    """
+
+    state.current_node = "KNOWLEDGE_LINKING"
+
+    result = knowledge_link_service.link_document(
+        workspace_id=state.workspace_id,
+        document_version_id=state.document_version_id,
+    )
+
+    state.metadata["knowledge_links_created"] = (
+        result["links_created"]
+    )
+
+    return state

backend/app/workflow/nodes/reconcile.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from app.services.reconciliation_service import ReconciliationService
+from app.workflow.state import WorkflowState
+
+
+def reconcile(
+    state: WorkflowState,
+    db: Session,
+) -> WorkflowState:
+    """
+    Reconcile newly extracted knowledge against the
+    existing Knowledge Register.
+    """
+
+    state.current_node = "RECONCILIATION"
+
+    service = ReconciliationService(db)
+
+    results = service.reconcile_document(
+        workspace_id=state.workspace_id,
+        document_version_id=state.document_version_id,
+    )
+
+    state.metadata["reconciliation"] = results
+
+    return state