diff --git a/src/workflows/mortgage_embedded_agent/demo.py b/src/workflows/mortgage_embedded_agent/demo.py index e2ffe30..84c932c 100644 --- a/src/workflows/mortgage_embedded_agent/demo.py +++ b/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__": diff --git a/src/workflows/mortgage_embedded_agent/mortgage_models.py b/src/workflows/mortgage_embedded_agent/mortgage_models.py index 83b433a..4b23101 100644 --- a/src/workflows/mortgage_embedded_agent/mortgage_models.py +++ b/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.""" diff --git a/src/workflows/mortgage_embedded_agent/mortgage_workflow.py b/src/workflows/mortgage_embedded_agent/mortgage_workflow.py index af6e514..b3079d5 100644 --- a/src/workflows/mortgage_embedded_agent/mortgage_workflow.py +++ b/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, diff --git a/src/workflows/mortgage_embedded_agent/mortgage_workflow_tests.py b/src/workflows/mortgage_embedded_agent/mortgage_workflow_tests.py index 89169a2..fb42f3a 100644 --- a/src/workflows/mortgage_embedded_agent/mortgage_workflow_tests.py +++ b/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" diff --git a/src/workflows/mortgage_embedded_agent/review_app.py b/src/workflows/mortgage_embedded_agent/review_app.py index d51be13..fc5c16e 100644 --- a/src/workflows/mortgage_embedded_agent/review_app.py +++ b/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 'Running' + 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'{_escape(label)}' + + +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"" f"{_escape(case_id)}" f"{_escape(entry.get('created_at', ''))}" - f"{_escape(status)}" + f"{_status_badge(status, progress, decision)}" f"" ) @@ -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"

Case {case_id}

Unable to load workflow: {exc}

", + f'

← All cases

' + f"

Case {_escape(case_id)}

Unable to load workflow: {_escape(exc)}

", + ) + progress = await _fetch_progress(handle) if status_label == "RUNNING" else None + progress_block = "" + if progress: + progress_block = ( + f'

{_escape(progress.detail)} (updated {_escape(progress.updated_at)})

' ) 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"""

Final Outcome

-

{_escape(result.final_decision)}

+

{_status_badge(status_label, None, final_decision)}

Risk Score: {_escape(result.risk_score)}

Human Review Required: {_escape(result.human_review_required)}

Decision Memo (Preview):

@@ -509,12 +587,14 @@ async def case_view(case_id: str, message: str | None = None, error: str | None """ body = f""" +

← All cases

Case {_escape(case_id)}{f" — {_escape(packet.display_name)}" if packet else ""}

Workflow ID: {_escape(_workflow_id(case_id))}

{alert_block}

Status

-

{_escape(status_label)}

+

{_status_badge(status_label, progress, final_decision)}

+ {progress_block}
{final_decision_block} @@ -543,7 +623,8 @@ async def case_view(case_id: str, message: str | None = None, error: str | None
""" 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) diff --git a/src/workflows/mortgage_fixed_flow/mortgage_models.py b/src/workflows/mortgage_fixed_flow/mortgage_models.py index 05044d4..d37397e 100644 --- a/src/workflows/mortgage_fixed_flow/mortgage_models.py +++ b/src/workflows/mortgage_fixed_flow/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.""" diff --git a/src/workflows/mortgage_fixed_flow/mortgage_workflow.py b/src/workflows/mortgage_fixed_flow/mortgage_workflow.py index c4257cb..69ccb4b 100644 --- a/src/workflows/mortgage_fixed_flow/mortgage_workflow.py +++ b/src/workflows/mortgage_fixed_flow/mortgage_workflow.py @@ -26,6 +26,8 @@ UnderwritingAnalyses, UnderwritingInput, UnderwritingOutput, + WorkflowProgress, + WorkflowStage, ) from .mortgage_utils import ( compute_metrics, @@ -44,6 +46,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: @@ -63,11 +75,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), @@ -89,6 +108,7 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput: "decision": "overall underwriting decision thresholds", } + self._set_stage("RETRIEVING_POLICY", "Looking up underwriting policy") policy_futures = { key: workflow.start_activity( retrieve_policy_context, @@ -101,7 +121,9 @@ async def run(self, input_data: UnderwritingInput) -> UnderwritingOutput: # Run specialist analyses in a fixed, deterministic order. analyses = UnderwritingAnalyses(credit="", income="", assets="", collateral="") - for agent in ["credit", "income", "assets", "collateral"]: + agents = ["credit", "income", "assets", "collateral"] + for index, agent in enumerate(agents, start=1): + self._set_stage("RUNNING_ANALYSES", f"{agent.title()} analysis ({index} of {len(agents)})") result = await workflow.execute_activity( run_agent_analysis, AgentTask( @@ -116,6 +138,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, @@ -130,6 +153,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( @@ -202,10 +226,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, diff --git a/src/workflows/mortgage_fixed_flow/review_app.py b/src/workflows/mortgage_fixed_flow/review_app.py index 597ba3e..928f7bd 100644 --- a/src/workflows/mortgage_fixed_flow/review_app.py +++ b/src/workflows/mortgage_fixed_flow/review_app.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import html import json import os @@ -13,11 +14,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") @@ -140,6 +141,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; @@ -274,15 +279,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 'Running' + 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'{_escape(label)}' + + +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: @@ -304,37 +369,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"" f"{_escape(case_id)}" f"{_escape(entry.get('created_at', ''))}" - f"{_escape(status)}" + f"{_status_badge(status, progress, decision)}" f"" ) @@ -390,16 +459,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"

Case {case_id}

Unable to load workflow: {exc}

", + f'

← All cases

' + f"

Case {_escape(case_id)}

Unable to load workflow: {_escape(exc)}

", + ) + progress = await _fetch_progress(handle) if status_label == "RUNNING" else None + progress_block = "" + if progress: + progress_block = ( + f'

{_escape(progress.detail)} (updated {_escape(progress.updated_at)})

' ) 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"""

Final Outcome

-

{_escape(result.final_decision)}

+

{_status_badge(status_label, None, final_decision)}

Risk Score: {_escape(result.risk_score)}

Human Review Required: {_escape(result.human_review_required)}

Decision Memo (Preview):

@@ -520,12 +598,14 @@ async def case_view(case_id: str, message: str | None = None, error: str | None """ body = f""" +

← All cases

Case {_escape(case_id)}{f" — {_escape(packet.display_name)}" if packet else ""}

Workflow ID: {_escape(_workflow_id(case_id))}

{alert_block}

Status

-

{_escape(status_label)}

+

{_status_badge(status_label, progress, final_decision)}

+ {progress_block}
{final_decision_block} @@ -554,7 +634,8 @@ async def case_view(case_id: str, message: str | None = None, error: str | None
""" 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)