Workflow progress indicator + UI improvements

src/workflows/mortgage_embedded_agent/demo.py

@@ -112,8 +112,14 @@ async def main() -> None:
         TEMPORAL_ADDRESS,
         data_converter=pydantic_data_converter,
     )
-    for case_id in case_ids:
-        await _run_case(client, case_id, str(image_dir))
+    # Run all cases concurrently so a case waiting on human review doesn't block the others.
+    results = await asyncio.gather(
+        *(_run_case(client, case_id, str(image_dir)) for case_id in case_ids),
+        return_exceptions=True,
+    )
+    for case_id, outcome in zip(case_ids, results, strict=True):
+        if isinstance(outcome, Exception):
+            print(f"Case {case_id} failed: {outcome}")
 
 
 if __name__ == "__main__":

src/workflows/mortgage_embedded_agent/mortgage_models.py

@@ -8,6 +8,16 @@
 
 Decision = Literal["APPROVED", "CONDITIONAL", "REJECTED", "HUMAN_REVIEW"]
 HumanDecision = Literal["APPROVED", "REJECTED"]
+WorkflowStage = Literal[
+    "STARTED",
+    "EXTRACTING_APPLICATION",
+    "RETRIEVING_POLICY",
+    "RUNNING_ANALYSES",
+    "CRITIC_REVIEW",
+    "DRAFTING_DECISION",
+    "AWAITING_HUMAN_REVIEW",
+    "COMPLETED",
+]
 
 
 class CollectionItem(BaseModel):
@@ -146,6 +156,14 @@ class UnderwritingInput(BaseModel):
     image_dir: str
 
 
+class WorkflowProgress(BaseModel):
+    """Current stage of the underwriting workflow, exposed via query."""
+
+    stage: WorkflowStage
+    detail: str = ""
+    updated_at: str
+
+
 class UnderwritingMetrics(BaseModel):
     """Computed underwriting metrics."""
 

src/workflows/mortgage_embedded_agent/mortgage_workflow.py

@@ -28,6 +28,8 @@
     UnderwritingAnalyses,
     UnderwritingInput,
     UnderwritingOutput,
+    WorkflowProgress,
+    WorkflowStage,
 )
 from .mortgage_utils import (
     compute_metrics,
@@ -46,6 +48,16 @@ class MortgageUnderwritingWorkflow:
     def __init__(self) -> None:
         self._human_review: HumanReviewResult | None = None
         self._review_packet: HumanReviewPacket | None = None
+        self._progress = WorkflowProgress(stage="STARTED", updated_at=workflow.now().isoformat())
+
+    def _set_stage(self, stage: WorkflowStage, detail: str = "") -> None:
+        """Record the current stage so the UI can display progress."""
+
+        self._progress = WorkflowProgress(
+            stage=stage,
+            detail=detail,
+            updated_at=workflow.now().isoformat(),
+        )
 
     @workflow.signal
     async def submit_human_review(self, review: HumanReviewInput) -> None:
@@ -65,11 +77,18 @@ def get_review_packet(self) -> HumanReviewPacket | None:
 
         return self._review_packet
 
+    @workflow.query
+    def get_progress(self) -> WorkflowProgress:
+        """Expose the current workflow stage for the UI."""
+
+        return self._progress
+
     @workflow.run
     async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
         workflow.logger.info("Starting underwriting for case %s", input_data.case_id)
 
         # OCR intake: convert uploaded images into a validated MortgageApplication.
+        self._set_stage("EXTRACTING_APPLICATION", "Reading loan form images")
         applicant = await workflow.execute_activity(
             extract_application_from_images,
             ApplicationOcrTask(case_id=input_data.case_id, image_dir=input_data.image_dir),
@@ -92,6 +111,7 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
             "supervisor": "underwriting workflow routing and documentation sequencing",
         }
 
+        self._set_stage("RETRIEVING_POLICY", "Looking up underwriting policy")
         policy_futures = {
             key: workflow.start_activity(
                 retrieve_policy_context,
@@ -134,6 +154,10 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
             if next_agent not in remaining:
                 next_agent = remaining[0]
 
+            self._set_stage(
+                "RUNNING_ANALYSES",
+                f"{next_agent.title()} analysis ({len(completed) + 1} of 4)",
+            )
             agent_task = AgentTask(
                 agent_name=next_agent.title(),
                 applicant=sanitized_payload,
@@ -156,6 +180,10 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
             if agent not in completed
         ]
         for agent in remaining:
+            self._set_stage(
+                "RUNNING_ANALYSES",
+                f"{agent.title()} analysis ({len(completed) + 1} of 4)",
+            )
             result = await workflow.execute_activity(
                 run_agent_analysis,
                 AgentTask(
@@ -171,6 +199,7 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
 
         # Critic pass to check for missed risks or inconsistencies.
         critic_policy = policy_context["decision"]
+        self._set_stage("CRITIC_REVIEW", "Checking analyses for missed risks")
 
         critic_review = await workflow.execute_activity(
             run_critic_review,
@@ -185,6 +214,7 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
         )
 
         # Decision memo from LLM (structured JSON with decision + memo).
+        self._set_stage("DRAFTING_DECISION", "Drafting decision memo")
         decision_result = await workflow.execute_activity(
             run_decision_memo,
             DecisionTask(
@@ -256,10 +286,16 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput:
                 risk_score=risk_score,
             )
             # Pause until a human reviewer signals the workflow.
+            self._set_stage(
+                "AWAITING_HUMAN_REVIEW",
+                decision_recommendation.human_review_reason or "Conditional decision needs review",
+            )
             await workflow.wait_condition(lambda: self._human_review is not None)
             human_review = self._human_review
             final_decision = human_review.decision
 
+        self._set_stage("COMPLETED", f"Final decision: {final_decision}")
+
         # Emit the final underwriting output for the case.
         return UnderwritingOutput(
             case_id=input_data.case_id,

src/workflows/mortgage_embedded_agent/mortgage_workflow_tests.py

@@ -113,6 +113,10 @@ async def test_workflow_human_review_signal(client) -> None:
         assert packet.decision_recommendation.decision == "CONDITIONAL"
         assert packet.display_name
 
+        progress = await handle.query(MortgageUnderwritingWorkflow.get_progress)
+        assert progress.stage == "AWAITING_HUMAN_REVIEW"
+        assert progress.detail == "test review"
+
         await handle.signal(
             MortgageUnderwritingWorkflow.submit_human_review,
             HumanReviewInput(
@@ -130,3 +134,6 @@ async def test_workflow_human_review_signal(client) -> None:
         assert result.analyses.assets
         assert result.analyses.collateral
         assert "[APPLICANT_NAME]" not in result.decision_memo
+
+        progress = await handle.query(MortgageUnderwritingWorkflow.get_progress)
+        assert progress.stage == "COMPLETED"

src/workflows/mortgage_embedded_agent/review_app.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import html
 import json
 import os
@@ -12,11 +13,11 @@
 
 from fastapi import FastAPI, File, Form, UploadFile
 from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse, Response
-from temporalio.client import Client
+from temporalio.client import Client, WorkflowHandle
 from temporalio.contrib.pydantic import pydantic_data_converter
 from temporalio.exceptions import TemporalError
 
-from .mortgage_models import HumanReviewInput, UnderwritingInput, UnderwritingOutput
+from .mortgage_models import HumanReviewInput, UnderwritingInput, UnderwritingOutput, WorkflowProgress
 from .mortgage_workflow import MortgageUnderwritingWorkflow
 
 TEMPORAL_ADDRESS = os.environ.get("TEMPORAL_ADDRESS", "localhost:7233")
@@ -129,6 +130,10 @@ def _page(title: str, body: str, refresh_seconds: int | None = None) -> HTMLResp
       font-size: 12px;
       font-weight: 600;
     }}
+    .badge-waiting {{ background: #fef3c7; color: #92400e; }}
+    .badge-approved {{ background: #d1fae5; color: #065f46; }}
+    .badge-rejected {{ background: #fee2e2; color: #991b1b; }}
+    .badge-inactive {{ background: #e5e7eb; color: #374151; }}
     table {{
       width: 100%;
       border-collapse: collapse;
@@ -263,15 +268,75 @@ def _safe_extension(filename: str) -> str:
     return ".png"
 
 
-async def _fetch_status(client: Client, case_id: str) -> str:
-    handle = client.get_workflow_handle(_workflow_id(case_id))
+STAGE_LABELS = {
+    "STARTED": "Starting",
+    "EXTRACTING_APPLICATION": "Reading application",
+    "RETRIEVING_POLICY": "Retrieving policy",
+    "RUNNING_ANALYSES": "Analyzing application",
+    "CRITIC_REVIEW": "Reviewing analyses",
+    "DRAFTING_DECISION": "Drafting decision",
+    "AWAITING_HUMAN_REVIEW": "Awaiting human review",
+    "COMPLETED": "Finishing up",
+}
+
+
+def _humanize(value: str) -> str:
+    return value.replace("_", " ").capitalize()
+
+
+def _status_badge(status: str, progress: WorkflowProgress | None, decision: str | None) -> str:
+    """Colored, human-readable status badge.
+
+    Running cases show their current stage; completed cases show the final decision.
+    """
+    if status == "RUNNING":
+        if progress is None:
+            return '<span class="badge">Running</span>'
+        label = STAGE_LABELS.get(progress.stage, _humanize(progress.stage))
+        css = "badge badge-waiting" if progress.stage == "AWAITING_HUMAN_REVIEW" else "badge"
+    elif status == "COMPLETED":
+        label = f"Completed – {_humanize(decision)}" if decision else "Completed"
+        css = {
+            "APPROVED": "badge badge-approved",
+            "REJECTED": "badge badge-rejected",
+        }.get(decision or "", "badge badge-inactive")
+    else:
+        # Failed, timed out, terminated, canceled, not found, etc.
+        label = _humanize(status)
+        css = "badge badge-inactive"
+    return f'<span class="{css}">{_escape(label)}</span>'
+
+
+async def _fetch_decision(handle: WorkflowHandle) -> str | None:
+    try:
+        result = await handle.result()
+    except (TemporalError, ValueError):
+        return None
+    return result.final_decision
+
+
+async def _fetch_progress(handle: WorkflowHandle) -> WorkflowProgress | None:
+    try:
+        return await handle.query(MortgageUnderwritingWorkflow.get_progress)
+    except TemporalError:
+        return None
+
+
+async def _fetch_status(client: Client, case_id: str) -> tuple[str, WorkflowProgress | None, str | None]:
+    """Return the execution status, the current stage if running, and the decision if completed."""
+    handle = client.get_workflow_handle(_workflow_id(case_id), result_type=UnderwritingOutput)
     try:
         description = await handle.describe()
-        if description.status:
-            return description.status.name.replace("WORKFLOW_EXECUTION_STATUS_", "")
     except TemporalError:
-        return "NOT_FOUND"
-    return "UNKNOWN"
+        return "NOT_FOUND", None, None
+    if not description.status:
+        return "UNKNOWN", None, None
+    status = description.status.name.replace("WORKFLOW_EXECUTION_STATUS_", "")
+    # Only query running workflows; querying closed ones forces a full replay on the worker.
+    progress = await _fetch_progress(handle) if status == "RUNNING" else None
+    # Completed results come straight from the server's history, so no worker is needed.
+    decision = await _fetch_decision(handle) if status == "COMPLETED" else None
+    return status, progress, decision
 
 
 def _redirect_with_message(case_id: str, message: str | None = None, error: str | None = None) -> RedirectResponse:
@@ -293,37 +358,41 @@ async def index() -> HTMLResponse:
 
     entries_by_case = {entry.get("case_id", ""): entry for entry in entries if entry.get("case_id")}
 
-    # Pull in any running workflows that were started outside the UI (e.g., demo script).
+    # Pull in running and completed workflows, including ones started outside the UI (e.g., demo script).
     try:
         async for workflow in client.list_workflows(
-            query='WorkflowType = "MortgageUnderwritingWorkflow" and ExecutionStatus = "Running"',
-            limit=50,
+            query=(
+                'WorkflowType = "MortgageUnderwritingWorkflow" '
+                'and (ExecutionStatus = "Running" or ExecutionStatus = "Completed")'
+            ),
+            limit=200,
         ):
             workflow_id = workflow.id
             if not workflow_id.startswith("mortgage-"):
                 continue
             case_id = workflow_id.replace("mortgage-", "", 1)
             if case_id not in entries_by_case:
-                created_at = workflow.start_time.isoformat()
                 entries_by_case[case_id] = {
                     "case_id": case_id,
-                    "created_at": created_at,
+                    "created_at": workflow.start_time.isoformat(),
                     "image_dir": "",
                 }
     except TemporalError:
         pass
 
-    entries = sorted(entries_by_case.values(), key=lambda item: item.get("created_at", ""), reverse=True)
+    entries = sorted(entries_by_case.values(), key=lambda item: item.get("case_id", ""))
+
+    # Look up every case concurrently so the page (and its auto-refresh) stays fast.
+    statuses = await asyncio.gather(*(_fetch_status(client, entry["case_id"]) for entry in entries))
 
     rows = []
-    for entry in entries:
-        case_id = entry.get("case_id", "")
-        status = await _fetch_status(client, case_id) if case_id else "UNKNOWN"
+    for entry, (status, progress, decision) in zip(entries, statuses, strict=True):
+        case_id = entry["case_id"]
         rows.append(
             f"<tr>"
             f"<td><a href=\"/case/{_escape(case_id)}\">{_escape(case_id)}</a></td>"
             f"<td>{_escape(entry.get('created_at', ''))}</td>"
-            f"<td><span class=\"badge\">{_escape(status)}</span></td>"
+            f"<td>{_status_badge(status, progress, decision)}</td>"
             f"</tr>"
         )
 
@@ -379,16 +448,25 @@ async def case_view(case_id: str, message: str | None = None, error: str | None
     except TemporalError as exc:
         return _page(
             "Case Not Found",
-            f"<h1>Case {case_id}</h1><p>Unable to load workflow: {exc}</p>",
+            f'<p><a href="/">&larr; All cases</a></p>'
+            f"<h1>Case {_escape(case_id)}</h1><p>Unable to load workflow: {_escape(exc)}</p>",
+        )
+    progress = await _fetch_progress(handle) if status_label == "RUNNING" else None
+    progress_block = ""
+    if progress:
+        progress_block = (
+            f'<p class="muted">{_escape(progress.detail)} (updated {_escape(progress.updated_at)})</p>'
         )
     final_decision_block = ""
+    final_decision: str | None = None
     if status_label == "COMPLETED":
         try:
             result = await handle.result()
+            final_decision = result.final_decision
             final_decision_block = f"""
 <div class="card">
   <h2>Final Outcome</h2>
-  <p><span class="badge">{_escape(result.final_decision)}</span></p>
+  <p>{_status_badge(status_label, None, final_decision)}</p>
   <p><strong>Risk Score:</strong> {_escape(result.risk_score)}</p>
   <p><strong>Human Review Required:</strong> {_escape(result.human_review_required)}</p>
   <p><strong>Decision Memo (Preview):</strong></p>
@@ -509,12 +587,14 @@ async def case_view(case_id: str, message: str | None = None, error: str | None
 """
 
     body = f"""
+<p><a href="/">&larr; All cases</a></p>
 <h1>Case {_escape(case_id)}{f" — {_escape(packet.display_name)}" if packet else ""}</h1>
 <p class="muted">Workflow ID: {_escape(_workflow_id(case_id))}</p>
 {alert_block}
 <div class="card">
   <h2>Status</h2>
-  <p><span class="badge">{_escape(status_label)}</span></p>
+  <p>{_status_badge(status_label, progress, final_decision)}</p>
+  {progress_block}
 </div>
 {final_decision_block}
 
@@ -543,7 +623,8 @@ async def case_view(case_id: str, message: str | None = None, error: str | None
 </div>
 """ if allow_submit else ""}
 """
-    refresh_seconds = 10 if status_label == "RUNNING" else None
+    # Don't auto-refresh while the review form is shown, or it wipes the reviewer's input.
+    refresh_seconds = 10 if status_label == "RUNNING" and not allow_submit else None
     return _page(f"Case {case_id}", body, refresh_seconds=refresh_seconds)