feat: live stage progression during workflow execution

Each workflow node now writes its current stage to the database via
report_progress(). The frontend polls GET /workflows/{id} every 3s
and the API reads the PROGRESS checkpoint to return the live current_node.

Result: the pipeline visualization shows stages lighting up one by one
during processing (Extract ✓ → Chunk ✓ → Embed pulsing → ...) instead
of jumping from PENDING to all-green COMPLETED.

Stages reporting progress: extract, chunk, embed, classify, knowledge,
reconcile, validate, decision. Each writes to a PROGRESS checkpoint
with its own DB session (committed immediately, visible to API polls).

backend/app/api/workflows.py

@@ -37,8 +37,16 @@ def _workflow_response(workflow: WorkflowRun) -> dict:
     elif workflow.status == WorkflowStatus.PENDING:
         current_node = "PENDING"
     else:
-        # RUNNING — use latest checkpoint
-        current_node = completed_stages[-1] if completed_stages else "EXTRACTION"
+        # RUNNING — read live progress from PROGRESS checkpoint
+        progress_cp = None
+        for cp in reversed(checkpoints):
+            if cp.agent_name == "PROGRESS":
+                progress_cp = cp
+                break
+        if progress_cp and progress_cp.state:
+            current_node = progress_cp.state.get("current_node", "EXTRACTION")
+        else:
+            current_node = "EXTRACTION"
 
     # Get validation results from checkpoint state if available
     validation_results = None

backend/app/workflow/nodes/chunk.py

@@ -17,6 +17,9 @@ def chunk(
     tracker = get_or_create_tracker(str(state.workflow_run_id))
     tracker.start_stage("chunking")
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "CHUNKING")
+
     result = chunking_service.chunk_document(
         document_version_id=state.document_version_id,
         sections=state.extracted_sections,

backend/app/workflow/nodes/classify.py

@@ -10,6 +10,9 @@ def classify(
 
     state.current_node = "CLASSIFICATION"
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "CLASSIFICATION")
+
     state.document_type = "GENERAL"
 
     return state

backend/app/workflow/nodes/decision.py

@@ -11,6 +11,9 @@ def decide(state: WorkflowState) -> WorkflowState:
 
     state.current_node = "DECISION"
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "DECISION")
+
     agent = DecisionAgent()
 
     state.decision = agent.decide(

backend/app/workflow/nodes/embed.py

@@ -22,6 +22,9 @@ def embed(
     tracker = get_or_create_tracker(str(state.workflow_run_id))
     tracker.start_stage("embedding")
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "EMBEDDING")
+
     chunks = (
         db.query(DocumentChunk)
         .filter(

backend/app/workflow/nodes/extract.py

@@ -17,6 +17,9 @@ def extract(state: WorkflowState) -> WorkflowState:
     tracker = get_or_create_tracker(str(state.workflow_run_id))
     tracker.start_stage("extraction")
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "EXTRACTION")
+
     try:
         sections = extract_text_sections(
             state.document_path

backend/app/workflow/nodes/knowledge.py

@@ -93,6 +93,9 @@ def knowledge(
         state.metadata["knowledge_items_created"] = 0
         return state
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "KNOWLEDGE_EXTRACTION")
+
     sections = []
 
     for index, section in enumerate(state.extracted_sections):
@@ -144,9 +147,8 @@ def knowledge(
     except Exception as first_err:
         # On failure, retry with fewer sections to reduce output size
         import logging
-        logging.getLogger(__name__).warning(
-            "Knowledge extraction failed, retrying with reduced input: %s", first_err
-        )
+        logger = logging.getLogger(__name__)
+        logger.warning("Knowledge extraction failed: %s", str(first_err)[:100])
         # Take only first half of sections to reduce output length
         half = max(1, len(sections) // 2)
         reduced_prompt = f"{KNOWLEDGE_PROMPT}\n" + "\n".join(sections[:half])
@@ -154,8 +156,9 @@ def knowledge(
             raw_result = structured_model.invoke(reduced_prompt)
             response = raw_result["parsed"]
             raw_response = raw_result.get("raw")
-        except Exception:
+        except Exception as second_err:
             # Final fallback: return empty extraction rather than crash
+            logger.error("Knowledge extraction retry also failed: %s", str(second_err)[:100])
             state.metadata["knowledge_items_created"] = 0
             state.metadata["extraction_error"] = str(first_err)[:200]
             tracker.end_stage("knowledge_extraction")
@@ -168,13 +171,29 @@ def knowledge(
             usage = meta.get("token_usage") or meta.get("usage") or {}
             input_tokens = usage.get("prompt_tokens", 0) or usage.get("input_tokens", 0)
             output_tokens = usage.get("completion_tokens", 0) or usage.get("output_tokens", 0)
-            tracker.record_llm_usage(
-                "knowledge_extraction",
-                input_tokens=input_tokens,
-                output_tokens=output_tokens,
-            )
-    except Exception:
-        pass  # Non-fatal
+            if input_tokens or output_tokens:
+                tracker.record_llm_usage(
+                    "knowledge_extraction",
+                    input_tokens=input_tokens,
+                    output_tokens=output_tokens,
+                )
+            else:
+                import logging
+                logging.getLogger(__name__).warning("Token usage was 0. Meta: %s", meta)
+        elif raw_response and hasattr(raw_response, "usage_metadata"):
+            um = raw_response.usage_metadata
+            if um:
+                tracker.record_llm_usage(
+                    "knowledge_extraction",
+                    input_tokens=getattr(um, "input_tokens", 0) or 0,
+                    output_tokens=getattr(um, "output_tokens", 0) or 0,
+                )
+        else:
+            import logging
+            logging.getLogger(__name__).warning("No raw_response or no response_metadata. raw_response type: %s", type(raw_response))
+    except Exception as e:
+        import logging
+        logging.getLogger(__name__).warning("Token capture exception: %s", e)
 
     tracker.end_stage("knowledge_extraction")
 

backend/app/workflow/nodes/reconcile.py

@@ -21,6 +21,9 @@ def reconcile(
     tracker = get_or_create_tracker(str(state.workflow_run_id))
     tracker.start_stage("reconciliation")
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "RECONCILIATION")
+
     service = ReconciliationService(db)
 
     results = service.reconcile_document(

backend/app/workflow/nodes/validate.py

@@ -22,6 +22,9 @@ def validate(
     tracker = get_or_create_tracker(str(state.workflow_run_id))
     tracker.start_stage("validation")
 
+    from app.workflow.progress import report_progress
+    report_progress(state.workflow_run_id, "VALIDATION")
+
     knowledge_repository = KnowledgeRepository(db)
     rule_repository = RuleRepository(db)
 

backend/app/workflow/progress.py

@@ -0,0 +1,37 @@
+"""
+Workflow progress tracking.
+
+Persists the current node to the workflow_checkpoints table so the
+frontend can poll and show live stage progression during execution.
+"""
+from __future__ import annotations
+
+from app.database.database import SessionLocal
+from app.models.workflow_checkpoint import WorkflowCheckpoint
+
+
+def report_progress(workflow_run_id, stage_name: str):
+    """
+    Write/update the current stage to the database.
+    Uses its own session so it commits immediately and is visible to API polls.
+    """
+    db = SessionLocal()
+    try:
+        # Upsert: replace the PROGRESS checkpoint
+        db.query(WorkflowCheckpoint).filter(
+            WorkflowCheckpoint.workflow_run_id == workflow_run_id,
+            WorkflowCheckpoint.agent_name == "PROGRESS",
+        ).delete(synchronize_session=False)
+
+        checkpoint = WorkflowCheckpoint(
+            workflow_run_id=workflow_run_id,
+            agent_name="PROGRESS",
+            state={"current_node": stage_name},
+            message=stage_name,
+        )
+        db.add(checkpoint)
+        db.commit()
+    except Exception:
+        db.rollback()
+    finally:
+        db.close()