feat: add human review approval gate and workflow resume

backend/alembic/versions/add_waiting_for_review.py

@@ -0,0 +1,27 @@
+"""add waiting for review workflow status
+
+Revision ID: add_waiting_for_review
+Revises: 585aa57edb27
+"""
+
+from alembic import op
+
+
+revision = "add_waiting_for_review"
+down_revision = "585aa57edb27"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "ALTER TYPE workflow_status "
+        "ADD VALUE IF NOT EXISTS 'WAITING_FOR_REVIEW'"
+    )
+
+
+def downgrade() -> None:
+    # PostgreSQL does not support removing a value directly
+    # from an enum type. A downgrade would require recreating
+    # the enum and migrating the column.
+    pass

backend/app/api/workflows.py

@@ -0,0 +1,60 @@
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.core.dependencies import get_current_user
+from app.database.session import get_db
+from app.models.user import User
+from app.models.workflow_run import WorkflowRun
+from app.models.workspace import Workspace
+
+router = APIRouter(
+    prefix="/workflows",
+    tags=["Workflows"],
+)
+
+
+@router.get("/{workflow_id}")
+def get_workflow(
+    workflow_id: UUID,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    workflow = (
+        db.query(WorkflowRun)
+        .filter(WorkflowRun.id == workflow_id)
+        .first()
+    )
+
+    if workflow is None:
+        raise HTTPException(
+            status_code=404,
+            detail="Workflow not found.",
+        )
+
+    workspace = (
+        db.query(Workspace)
+        .filter(
+            Workspace.id == workflow.workspace_id,
+            Workspace.created_by == current_user.id,
+        )
+        .first()
+    )
+
+    if workspace is None:
+        raise HTTPException(
+            status_code=403,
+            detail="You do not have access to this workflow.",
+        )
+
+    return {
+        "id": str(workflow.id),
+        "workspace_id": str(workflow.workspace_id),
+        "document_version_id": str(
+            workflow.document_version_id
+        ),
+        "status": workflow.status.value,
+        "started_at": workflow.started_at,
+        "completed_at": workflow.completed_at,
+    }

backend/app/main.py

@@ -1,14 +1,17 @@
-from fastapi import FastAPI
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+
+from app.api import auth, documents, proposals, workflows
 from app.api.workspaces import router as workspace_router
-from app.api import auth, documents
-from app.api import proposals
+
+
 app = FastAPI(
     title="DocWeave",
     description="Agentic Document Intelligence Platform",
     version="0.1.0",
 )
 
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -17,10 +20,26 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-app.include_router(auth.router, prefix="/auth", tags=["auth"])
-app.include_router(documents.router, prefix="/documents", tags=["documents"])
+
+app.include_router(
+    auth.router,
+    prefix="/auth",
+    tags=["auth"],
+)
+
+app.include_router(
+    documents.router,
+    prefix="/documents",
+    tags=["documents"],
+)
+
 app.include_router(workspace_router)
+
 app.include_router(proposals.router)
+
+app.include_router(workflows.router)
+
+
 @app.get("/")
 def root():
     return {

backend/app/models/workflow_run.py

@@ -19,6 +19,7 @@ from app.database.database import Base
 class WorkflowStatus(str, enum.Enum):
     PENDING = "PENDING"
     RUNNING = "RUNNING"
+    WAITING_FOR_REVIEW = "WAITING_FOR_REVIEW"
     COMPLETED = "COMPLETED"
     FAILED = "FAILED"
     CANCELLED = "CANCELLED"

backend/app/repositories/workflow_repository.py

@@ -18,6 +18,23 @@ class WorkflowRepository:
 
         return workflow
 
+    def get(self, workflow_id):
+        return (
+            self.db.query(WorkflowRun)
+            .filter(WorkflowRun.id == workflow_id)
+            .first()
+        )
+
+    def get_by_document_version(self, document_version_id):
+        return (
+            self.db.query(WorkflowRun)
+            .filter(
+                WorkflowRun.document_version_id == document_version_id,
+            )
+            .order_by(WorkflowRun.started_at.desc())
+            .first()
+        )
+
     def update_status(
         self,
         workflow: WorkflowRun,

backend/app/services/document_service.py

@@ -12,6 +12,7 @@ from app.storage.local_storage import LocalStorage
 from app.services.workflow_service import WorkflowService
 from app.workflow import WorkflowExecutor, WorkflowState
 
+
 class DocumentService:
     def __init__(self, db: Session):
         self.repository = DocumentRepository(db)
@@ -51,9 +52,14 @@ class DocumentService:
 
         self.repository.refresh(document)
         self.repository.refresh(version)
-#todo:
-# Move workflow orchestration out of DocumentService.
-# The upload service should only persist documents.
+
+        # TODO:
+        # Move workflow orchestration out of DocumentService.
+        #
+        # The upload service should eventually only persist documents.
+        # Workflow execution should be handled by a dedicated workflow
+        # application/service layer.
+
         workflow_service = WorkflowService(self.repository.db)
 
         workflow = workflow_service.start_workflow(
@@ -68,7 +74,12 @@ class DocumentService:
             document_path=version.storage_path,
         )
 
-        executor = WorkflowExecutor(self.repository.db)
+        # The workflow pauses before the human_review node when the
+        # decision agent determines that human review is required.
+        executor = WorkflowExecutor(
+            self.repository.db,
+            interrupt_before=["human_review"],
+        )
 
         try:
             executor.execute(
@@ -77,4 +88,5 @@ class DocumentService:
             )
         finally:
             executor.close()
+
         return document, version

backend/app/services/proposal_review_service.py

@@ -10,9 +10,8 @@ from app.models.commit import Commit
 from app.models.knowledge_item import KnowledgeItem, KnowledgeStatus
 from app.models.proposal import Proposal, ProposalStatus, ProposalType
 from app.models.review import Review, ReviewDecision
-from app.models.user import User
 from app.models.workspace import Workspace
-
+from app.models.workflow_run import WorkflowStatus
 
 class ProposalReviewService:
     """
@@ -50,6 +49,169 @@ class ProposalReviewService:
             .all()
         )
 
+    def get_pending_for_document_version(
+        self,
+        workspace_id: uuid.UUID,
+        document_version_id: uuid.UUID,
+    ) -> list[Proposal]:
+        """
+        Return pending proposals associated with a specific document version.
+
+        CREATE proposals normally point directly at the newly extracted
+        KnowledgeItem.
+
+        UPDATE proposals point at the existing canonical KnowledgeItem,
+        so their source_knowledge_item_id inside proposed_changes is also
+        checked.
+        """
+
+        knowledge_items = (
+            self.db.query(KnowledgeItem.id)
+            .filter(
+                KnowledgeItem.workspace_id == workspace_id,
+                KnowledgeItem.document_version_id == document_version_id,
+            )
+            .all()
+        )
+
+        knowledge_item_ids = {
+            str(item_id)
+            for (item_id,) in knowledge_items
+        }
+
+        if not knowledge_item_ids:
+            return []
+
+        proposals = (
+            self.db.query(Proposal)
+            .filter(
+                Proposal.workspace_id == workspace_id,
+                Proposal.status == ProposalStatus.PENDING,
+            )
+            .order_by(Proposal.created_at.asc())
+            .all()
+        )
+
+        relevant = []
+
+        for proposal in proposals:
+            # CREATE / DELETE / directly-associated proposals
+            if (
+                proposal.knowledge_item_id is not None
+                and str(proposal.knowledge_item_id)
+                in knowledge_item_ids
+            ):
+                relevant.append(proposal)
+                continue
+
+            # UPDATE proposals reference the newly extracted source item
+            # through proposed_changes.
+            changes = proposal.proposed_changes or {}
+
+            source_id = changes.get("source_knowledge_item_id")
+
+            if source_id and str(source_id) in knowledge_item_ids:
+                relevant.append(proposal)
+
+        return relevant
+
+    def _get_proposal_document_version_id(
+        self,
+        proposal: Proposal,
+    ) -> uuid.UUID | None:
+        """
+        Resolve the document version that generated this proposal.
+
+        CREATE/DELETE proposals normally point directly at the extracted
+        KnowledgeItem.
+
+        UPDATE proposals point at the existing canonical item, so the
+        newly extracted source KnowledgeItem is stored in
+        proposed_changes["source_knowledge_item_id"].
+        """
+
+        source_item_id = None
+
+        changes = proposal.proposed_changes or {}
+
+        if proposal.proposal_type == ProposalType.UPDATE:
+            source_item_id = changes.get(
+                "source_knowledge_item_id"
+            )
+
+        if source_item_id:
+            source_item = (
+                self.db.query(KnowledgeItem)
+                .filter(
+                    KnowledgeItem.id == uuid.UUID(
+                        str(source_item_id)
+                    )
+                )
+                .first()
+            )
+
+            if source_item is not None:
+                return source_item.document_version_id
+
+        if proposal.knowledge_item_id is not None:
+            item = (
+                self.db.query(KnowledgeItem)
+                .filter(
+                    KnowledgeItem.id == proposal.knowledge_item_id
+                )
+                .first()
+            )
+
+            if item is not None:
+                return item.document_version_id
+
+        return None
+    
+    def _resume_workflow_if_ready(
+        self,
+        proposal: Proposal,
+    ) -> None:
+        document_version_id = (
+            self._get_proposal_document_version_id(
+                proposal
+            )
+        )
+
+        if document_version_id is None:
+            return
+
+        pending = self.get_pending_for_document_version(
+            proposal.workspace_id,
+            document_version_id,
+        )
+
+        if pending:
+            return
+
+        from app.repositories.workflow_repository import (
+            WorkflowRepository,
+        )
+        from app.workflow import WorkflowExecutor
+
+        workflow_repository = WorkflowRepository(self.db)
+
+        workflow = workflow_repository.get_by_document_version(
+            document_version_id
+        )
+
+        if workflow is None:
+            return
+
+        if workflow.status != WorkflowStatus.WAITING_FOR_REVIEW:
+            return
+
+        executor = WorkflowExecutor(self.db)
+
+        try:
+            executor.resume(workflow)
+        finally:
+            executor.close()
+    
     def approve(
         self,
         proposal_id: uuid.UUID,
@@ -90,6 +252,8 @@ class ProposalReviewService:
             self.db.commit()
             self.db.refresh(commit)
 
+            self._resume_workflow_if_ready(proposal)
+
             return commit
 
         except Exception:
@@ -125,8 +289,9 @@ class ProposalReviewService:
             self.db.commit()
             self.db.refresh(review)
 
-            return review
+            self._resume_workflow_if_ready(proposal)
 
+            return review
         except Exception:
             self.db.rollback()
             raise
@@ -226,15 +391,16 @@ class ProposalReviewService:
 
         existing_item.status = KnowledgeStatus.ACTIVE
 
-        # The newly extracted item was the evidence that caused
-        # this proposal. It should not become a second active
-        # canonical item.
         source_id = changes.get("source_knowledge_item_id")
 
         if source_id:
-            source_item = self.db.query(KnowledgeItem).filter(
-                KnowledgeItem.id == uuid.UUID(source_id)
-            ).first()
+            source_item = (
+                self.db.query(KnowledgeItem)
+                .filter(
+                    KnowledgeItem.id == uuid.UUID(source_id)
+                )
+                .first()
+            )
 
             if source_item is not None:
                 source_item.status = KnowledgeStatus.SUPERSEDED
@@ -308,9 +474,7 @@ class ProposalReviewService:
         if proposal.status != ProposalStatus.PENDING:
             raise HTTPException(
                 status_code=409,
-                detail=(
-                    "Proposal has already been reviewed."
-                ),
+                detail="Proposal has already been reviewed.",
             )
 
     def _verify_workspace_access(
@@ -330,7 +494,5 @@ class ProposalReviewService:
         if workspace is None:
             raise HTTPException(
                 status_code=403,
-                detail=(
-                    "You do not have access to this workspace."
-                ),
+                detail="You do not have access to this workspace.",
             )

backend/app/services/workflow_service.py

@@ -25,6 +25,9 @@ class WorkflowService:
 
         return workflow
 
+    def get_workflow(self, workflow_id):
+        return self.repository.get(workflow_id)
+
     def complete_workflow(self, workflow):
         workflow = self.repository.update_status(
             workflow,
@@ -34,4 +37,15 @@ class WorkflowService:
         self.repository.commit()
         self.repository.refresh(workflow)
 
+        return workflow
+
+    def wait_for_review(self, workflow):
+        workflow = self.repository.update_status(
+            workflow,
+            WorkflowStatus.WAITING_FOR_REVIEW,
+        )
+
+        self.repository.commit()
+        self.repository.refresh(workflow)
+
         return workflow

backend/app/workflow/executor.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import psycopg
-from sqlalchemy.orm import Session
 from langgraph.checkpoint.postgres import PostgresSaver
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
+from langgraph.types import Command
+from sqlalchemy.orm import Session
 
 from app.core.config import DATABASE_URL
 from app.services.workflow_service import WorkflowService
@@ -48,6 +49,27 @@ class WorkflowExecutor:
             interrupt_before=interrupt_before,
         )
 
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _is_completed(state) -> bool:
+        """
+        LangGraph may return either a WorkflowState instance or
+        a dictionary depending on the configured state schema and
+        execution path.
+        """
+
+        if isinstance(state, dict):
+            return bool(state.get("completed", False))
+
+        return bool(getattr(state, "completed", False))
+
+    # ------------------------------------------------------------------
+    # Initial execution
+    # ------------------------------------------------------------------
+
     def execute(
         self,
         state: WorkflowState | None,
@@ -65,9 +87,43 @@ class WorkflowExecutor:
             config=config,
         )
 
-        self.workflow_service.complete_workflow(workflow)
+        if self._is_completed(final_state):
+            self.workflow_service.complete_workflow(workflow)
+        else:
+            self.workflow_service.wait_for_review(workflow)
+
+        return final_state
+
+    # ------------------------------------------------------------------
+    # Resume after human review
+    # ------------------------------------------------------------------
+
+    def resume(
+        self,
+        workflow,
+    ) -> WorkflowState:
+
+        config = {
+            "configurable": {
+                "thread_id": str(workflow.id),
+            }
+        }
+
+        final_state = self.graph.invoke(
+            Command(resume=True),
+            config=config,
+        )
+
+        if self._is_completed(final_state):
+            self.workflow_service.complete_workflow(workflow)
+        else:
+            self.workflow_service.wait_for_review(workflow)
 
         return final_state
 
+    # ------------------------------------------------------------------
+    # Cleanup
+    # ------------------------------------------------------------------
+
     def close(self):
         self.checkpoint_connection.close()

backend/app/workflow/graph.py

@@ -2,6 +2,7 @@
 
 from sqlalchemy.orm import Session
 from langgraph.graph import END, START, StateGraph
+from langgraph.types import interrupt
 
 from app.llm.client import LLMClient
 from app.repositories.knowledge_repository import KnowledgeRepository
@@ -31,6 +32,10 @@ def build_workflow(
 ):
     """
     Build the DocWeave document intelligence workflow.
+
+    The human review branch uses LangGraph's durable interrupt()
+    mechanism so that the workflow state is checkpointed and can
+    later be resumed with Command(resume=...).
     """
 
     llm_client = LLMClient()
@@ -110,10 +115,39 @@ def build_workflow(
         ),
     )
 
-    # Temporary branch target.
-    # This will become the real durable human gate later.
-    def human_review(state: WorkflowState) -> WorkflowState:
+    # ------------------------------------------------------------------
+    # Durable human review gate
+    # ------------------------------------------------------------------
+
+    def human_review(
+        state: WorkflowState,
+    ) -> WorkflowState:
+        """
+        Durable human approval gate.
+
+        LangGraph checkpoints the current state when interrupt()
+        is reached. The workflow can later continue from this
+        checkpoint using Command(resume=...).
+        """
+
         state.current_node = "HUMAN_REVIEW"
+
+        interrupt(
+            {
+                "type": "human_review",
+                "workflow_run_id": str(
+                    state.workflow_run_id
+                ),
+                "document_version_id": str(
+                    state.document_version_id
+                ),
+                "message": (
+                    "Human review is required before "
+                    "the workflow can continue."
+                ),
+            }
+        )
+
         return state
 
     graph.add_node(
@@ -170,7 +204,10 @@ def build_workflow(
         "decision",
     )
 
-    # Decision controls the workflow path.
+    # ------------------------------------------------------------------
+    # Decision routing
+    # ------------------------------------------------------------------
+
     graph.add_conditional_edges(
         "decision",
         route_after_decision,
@@ -180,13 +217,16 @@ def build_workflow(
         },
     )
 
+    # ------------------------------------------------------------------
+    # Normal completion path
+    # ------------------------------------------------------------------
+
     graph.add_edge(
         "link",
         "complete",
     )
 
-    # Temporary review branch returns to completion.
-    # We will replace this with the real human gate later.
+    # After human approval/resume, continue to completion.
     graph.add_edge(
         "human_review",
         "complete",
@@ -197,7 +237,11 @@ def build_workflow(
         END,
     )
 
+    # ------------------------------------------------------------------
+    # Compile
+    # ------------------------------------------------------------------
+
     return graph.compile(
         checkpointer=checkpointer,
         interrupt_before=interrupt_before,
-    )
+    )