diff --git a/src/app/dashboard/estimate/[id]/intake/page.tsx b/src/app/dashboard/estimate/[id]/intake/page.tsx index 8f31089..0752b4c 100644 --- a/src/app/dashboard/estimate/[id]/intake/page.tsx +++ b/src/app/dashboard/estimate/[id]/intake/page.tsx @@ -25,6 +25,14 @@ export default async function EstimateIntakePage({ const identity = intakeIdentitySchema.safeParse(row.intakeExtraction); + let pipelineSubStage = null; + if (row.status === "processing") { + const { getEstimatePipelineSubStage } = await import( + "@/features/estimate-extraction-pipeline/progress" + ); + pipelineSubStage = await getEstimatePipelineSubStage(row.workflowRunId); + } + return (
); diff --git a/src/features/estimate-extraction-pipeline/progress.ts b/src/features/estimate-extraction-pipeline/progress.ts new file mode 100644 index 0000000..0869677 --- /dev/null +++ b/src/features/estimate-extraction-pipeline/progress.ts @@ -0,0 +1,68 @@ +import { mastra } from './index'; + +/** + * Read-only progress signal, decoupled from the app's DB: nothing here + * writes anything, anywhere. It reads Mastra's own workflow-run storage + * and derives which of the four non-suspending pipeline stages is + * furthest along. `triggerSummarizeEstimate()`/`resumeSummarizeEstimate()` + * are untouched; this is purely an additional read path a caller can + * consult, and any failure here (storage unreachable, run not found, + * unexpected shape) resolves to `null` rather than throwing, so a fault + * here can't break whatever is rendering the estimate's status. + * + * Each of the four stages is composed in `pipeline.ts` as a + * workflow-as-step (`.then(extractionFanoutWorkflow)` etc.), and Mastra + * persists a nested workflow-as-step's run as its OWN storage row — + * same `runId` as the parent `Generate Estimate` run, but keyed under + * its own `workflow_name` (its own `id`, e.g. `'Extraction'`) — rather + * than folding it into the parent run's own `steps` record. Confirmed + * directly against `mastra_workflow_snapshot`: a mid-flight run had + * `Extraction`/`Classify Findings` at `workflow_name='Extraction'` / + * `'Classify Findings'` with `status: 'success'`, `enrichment-fanout` at + * `'running'`, and no row yet for `'Presentation'`, while the PARENT + * run's own snapshot never gained entries for any of the four. So this + * reads each stage's snapshot directly via `workflowName`, not through + * `workflow.getWorkflowRunById()`'s parent-run `steps` record. + */ + +export type PipelineSubStageId = 'extraction' | 'classification' | 'enrichment' | 'presentation'; + +export type PipelineSubStage = { + stageId: PipelineSubStageId; + status: 'running' | 'success'; +}; + +/** + * Workflow ids as declared by each stage's own `createWorkflow({ id: ... })` + * in `pipeline.ts`'s composition root (extraction/steps.ts, + * classification/workflow.ts, enrichment/workflow.ts, + * presentation/workflow.ts). + */ +const SUB_STAGE_WORKFLOW_NAMES: { stageId: PipelineSubStageId; workflowName: string }[] = [ + { stageId: 'extraction', workflowName: 'Extraction' }, + { stageId: 'classification', workflowName: 'Classify Findings' }, + { stageId: 'enrichment', workflowName: 'enrichment-fanout' }, + { stageId: 'presentation', workflowName: 'Presentation' }, +]; + +export async function getEstimatePipelineSubStage( + workflowRunId: string | null | undefined, +): Promise { + if (!workflowRunId) return null; + + try { + const storage = mastra.getStorage(); + const workflowStore = await storage?.getStore('workflows'); + if (!workflowStore) return null; + + let current: PipelineSubStage | null = null; + for (const { stageId, workflowName } of SUB_STAGE_WORKFLOW_NAMES) { + const snapshot = await workflowStore.loadWorkflowSnapshot({ runId: workflowRunId, workflowName }); + if (!snapshot) continue; + current = { stageId, status: snapshot.status === 'success' ? 'success' : 'running' }; + } + return current; + } catch { + return null; + } +} diff --git a/src/features/estimate/components/estimate-intake-view.tsx b/src/features/estimate/components/estimate-intake-view.tsx index 79440ef..93e438a 100644 --- a/src/features/estimate/components/estimate-intake-view.tsx +++ b/src/features/estimate/components/estimate-intake-view.tsx @@ -5,6 +5,7 @@ import { useRouter } from "next/navigation"; import { FileSearch } from "lucide-react"; import type { IntakeIdentity } from "@/features/estimate-extraction-pipeline/intake"; +import type { PipelineSubStage } from "@/features/estimate-extraction-pipeline/progress"; import type { EstimateStatus } from "../db/schema"; import { EstimateConfirmationView } from "./estimate-confirmation-view"; import { EstimateRetryButton } from "./estimate-retry-button"; @@ -19,6 +20,7 @@ interface EstimateIntakeViewProps { identity: IntakeIdentity | null; phase: "identity" | "timeframe"; errorMessage: string | null; + pipelineSubStage?: PipelineSubStage | null; } export function EstimateIntakeView({ @@ -27,6 +29,7 @@ export function EstimateIntakeView({ identity, phase, errorMessage, + pipelineSubStage = null, }: EstimateIntakeViewProps) { const router = useRouter(); const isProcessing = status === "uploaded" || status === "processing"; @@ -45,6 +48,7 @@ export function EstimateIntakeView({ identityConfirmed={Boolean(identity && phase === "timeframe")} timeframeSelected={Boolean(identity && phase === "timeframe" && status === "processing")} errorMessage={errorMessage} + pipelineSubStage={pipelineSubStage} /> ); diff --git a/src/features/estimate/components/estimate-status-bar.tsx b/src/features/estimate/components/estimate-status-bar.tsx index 803dfce..988acff 100644 --- a/src/features/estimate/components/estimate-status-bar.tsx +++ b/src/features/estimate/components/estimate-status-bar.tsx @@ -1,6 +1,7 @@ "use client"; import type { EstimateStatus } from "@/features/estimate/db/schema"; +import type { PipelineSubStage, PipelineSubStageId } from "@/features/estimate-extraction-pipeline/progress"; import { Tooltip, TooltipContent, @@ -8,69 +9,100 @@ import { } from "@/design-systems/shadcn/components/tooltip"; import { cn } from "@/lib/utils"; -export type StageId = "uploaded" | "processing" | "analyzed" | "priced" | "delivered"; +export type StageId = "uploaded" | "extraction" | "classification" | "enrichment" | "delivered"; const STAGES: { id: StageId; label: string; description: string }[] = [ { id: "uploaded", label: "Uploaded", description: "Report received and queued." }, - { - id: "processing", - label: "Processing", - description: "AI is reading the inspection report and extracting billable items.", - }, - { id: "analyzed", label: "Analyzed", description: "Billable items extracted from the report." }, - { id: "priced", label: "Priced", description: "Local market pricing has been applied to each item." }, + { id: "extraction", label: "Extracted", description: "Reading every page of the report for billable items." }, + { id: "classification", label: "Analyzing", description: "Sorting extracted findings into trades and cost types." }, + { id: "enrichment", label: "Pricing", description: "Applying local market pricing to each line item." }, { id: "delivered", label: "Delivered", description: "Final estimate is ready to view and send." }, ]; -type EstimatePhase = - | "uploaded" - | "processing" - | "identity" - | "timeframe" - | "completed" - | "failed"; +/** Maps each real pipeline sub-stage directly onto its bar slot. */ +const SUB_STAGE_INDEX: Record = { + extraction: 1, + classification: 2, + enrichment: 3, + presentation: 4, +}; -function getPhase({ +interface Progress { + /** Last fully-completed index, -1 if none. */ + litUpTo: number; + /** Currently in-progress index, -1 if none. */ + activeIndex: number; + /** -1 if not failed. */ + failedIndex: number; +} + +function computeProgress({ + status, + pipelineSubStage, +}: { + status: EstimateStatus; + pipelineSubStage: PipelineSubStage | null; +}): Progress { + if (status === "completed") { + return { litUpTo: STAGES.length - 1, activeIndex: -1, failedIndex: -1 }; + } + + if (status === "failed") { + const knownIndex = pipelineSubStage ? SUB_STAGE_INDEX[pipelineSubStage.stageId] : 0; + const failedIndex = + pipelineSubStage?.status === "success" ? Math.min(knownIndex + 1, STAGES.length - 1) : knownIndex; + return { litUpTo: failedIndex - 1, activeIndex: -1, failedIndex }; + } + + if (pipelineSubStage) { + const index = SUB_STAGE_INDEX[pipelineSubStage.stageId]; + if (pipelineSubStage.status === "success") { + return { litUpTo: index, activeIndex: -1, failedIndex: -1 }; + } + return { litUpTo: index - 1, activeIndex: index, failedIndex: -1 }; + } + + // No sub-stage data yet: still uploaded, awaiting HITL confirmation, or + // in the initial parse/identity-extraction stretch before extraction + // starts. Nothing beyond "uploaded" is knowable yet. + return { litUpTo: 0, activeIndex: status === "processing" ? 0 : -1, failedIndex: -1 }; +} + +function captionLabel({ status, identityConfirmed, timeframeSelected, + pipelineSubStage, }: { status: EstimateStatus; identityConfirmed: boolean; timeframeSelected: boolean; -}): EstimatePhase { - if (status === "failed") return "failed"; - if (status === "completed") return "completed"; + pipelineSubStage: PipelineSubStage | null; +}): string { + if (status === "failed") return "Failed"; + if (status === "completed") return "Delivered"; if (status === "awaiting_confirmation") { - if (!identityConfirmed) return "identity"; - if (!timeframeSelected) return "timeframe"; + if (!identityConfirmed) return "Confirm identity"; + if (!timeframeSelected) return "Select timeframe"; } - if (status === "processing" || timeframeSelected) return "processing"; - return "uploaded"; -} - -function phaseLabel(phase: EstimatePhase): string { - if (phase === "identity") return "Confirm identity"; - if (phase === "timeframe") return "Select timeframe"; - if (phase === "completed") return "Completed"; - if (phase === "failed") return "Failed"; - if (phase === "processing") return "Processing"; + if (pipelineSubStage) return STAGES[SUB_STAGE_INDEX[pipelineSubStage.stageId]].label; + if (status === "processing") return "Processing"; return "Uploaded"; } -function litUpTo(phase: EstimatePhase): number { - if (phase === "processing") return 1; - if (phase === "identity") return 2; - if (phase === "timeframe") return 3; - if (phase === "completed") return STAGES.length - 1; - return 0; -} - interface EstimateStatusBarProps { status: EstimateStatus; identityConfirmed?: boolean; timeframeSelected?: boolean; errorMessage?: string | null; + /** + * Optional, best-effort sub-stage read from the pipeline's own run + * state (see `estimate-extraction-pipeline/progress.ts`). Absent or + * `null` (feature unavailable, read failed, or run hasn't reached a + * sub-stage yet) just leaves the bar at "Uploaded" until it resolves + * — never a required prop for correct rendering. + */ + pipelineSubStage?: PipelineSubStage | null; className?: string; } @@ -79,12 +111,12 @@ export function EstimateStatusBar({ identityConfirmed = false, timeframeSelected = false, errorMessage, + pipelineSubStage = null, className, }: EstimateStatusBarProps) { - const phase = getPhase({ status, identityConfirmed, timeframeSelected }); - const litIndex = litUpTo(phase); - const isFailed = phase === "failed"; - const isActive = phase === "processing" || phase === "identity" || phase === "timeframe"; + const { litUpTo, activeIndex, failedIndex } = computeProgress({ status, pipelineSubStage }); + const isFailed = failedIndex !== -1; + const label = captionLabel({ status, identityConfirmed, timeframeSelected, pipelineSubStage }); return (
{STAGES.map((stage, i) => { - const lit = !isFailed && i <= litIndex; - const active = isActive && i === litIndex; - const failed = isFailed && i === 1; - const tooltipText = failed - ? errorMessage ?? "Processing failed." - : stage.description; + const lit = !isFailed && i <= litUpTo; + const active = !isFailed && i === activeIndex; + const failed = isFailed && i === failedIndex; return ( @@ -125,7 +154,7 @@ export function EstimateStatusBar({ {lit && !active && !failed && " ✓"}
- {tooltipText} + {failed ? errorMessage ?? "Processing failed." : stage.description}
@@ -133,7 +162,7 @@ export function EstimateStatusBar({ })}
- {phaseLabel(phase)} + {label} ); diff --git a/src/features/estimate/components/estimates-list-view.tsx b/src/features/estimate/components/estimates-list-view.tsx index 5a9b6a1..0fa086e 100644 --- a/src/features/estimate/components/estimates-list-view.tsx +++ b/src/features/estimate/components/estimates-list-view.tsx @@ -2,6 +2,7 @@ import { db } from "@/db"; import { estimateRequestTable } from "../db/schema"; import { eq, desc } from "drizzle-orm"; import { authServerProvider } from "@/auth/server-provider"; +import type { PipelineSubStage } from "@/features/estimate-extraction-pipeline/progress"; import { headers } from "next/headers"; import Link from "next/link"; import { FileText } from "lucide-react"; @@ -22,6 +23,20 @@ export async function EstimatesListView() { .where(eq(estimateRequestTable.userId, session.user.id)) .orderBy(desc(estimateRequestTable.createdAt)); + let pipelineSubStages: (PipelineSubStage | null)[] = estimates.map(() => null); + if (estimates.some((upload) => upload.status === "processing")) { + const { getEstimatePipelineSubStage } = await import( + "@/features/estimate-extraction-pipeline/progress" + ); + pipelineSubStages = await Promise.all( + estimates.map((upload) => + upload.status === "processing" + ? getEstimatePipelineSubStage(upload.workflowRunId) + : Promise.resolve(null), + ), + ); + } + return (
@@ -51,7 +66,7 @@ export async function EstimatesListView() {
- {estimates.map((upload) => { + {estimates.map((upload, index) => { const leftBlock = (
@@ -93,6 +108,7 @@ export async function EstimatesListView() { identityConfirmed={Boolean(upload.intakeConfirmedAt)} timeframeSelected={Boolean(upload.timeframe)} errorMessage={upload.errorMessage} + pipelineSubStage={pipelineSubStages[index]} /> {upload.status === "failed" && ( diff --git a/src/features/estimate/components/recent-estimates-widget.tsx b/src/features/estimate/components/recent-estimates-widget.tsx index 6ac9a2d..0f760ba 100644 --- a/src/features/estimate/components/recent-estimates-widget.tsx +++ b/src/features/estimate/components/recent-estimates-widget.tsx @@ -2,6 +2,7 @@ import { db } from "@/db"; import { estimateRequestTable } from "../db/schema"; import { eq, desc } from "drizzle-orm"; import { authServerProvider } from "@/auth/server-provider"; +import type { PipelineSubStage } from "@/features/estimate-extraction-pipeline/progress"; import { headers } from "next/headers"; import { FileText, ArrowRight } from "lucide-react"; import { Button } from "@/design-systems/shadcn/components/button"; @@ -25,6 +26,20 @@ export async function RecentEstimatesWidget() { .orderBy(desc(estimateRequestTable.createdAt)) .limit(5); + let pipelineSubStages: (PipelineSubStage | null)[] = recentUploads.map(() => null); + if (recentUploads.some((upload) => upload.status === "processing")) { + const { getEstimatePipelineSubStage } = await import( + "@/features/estimate-extraction-pipeline/progress" + ); + pipelineSubStages = await Promise.all( + recentUploads.map((upload) => + upload.status === "processing" + ? getEstimatePipelineSubStage(upload.workflowRunId) + : Promise.resolve(null), + ), + ); + } + if (recentUploads.length === 0) { return (
@@ -50,7 +65,7 @@ export async function RecentEstimatesWidget() { return (
- {recentUploads.map((upload) => { + {recentUploads.map((upload, index) => { const content = ( <>
@@ -74,6 +89,7 @@ export async function RecentEstimatesWidget() { identityConfirmed={Boolean(upload.intakeConfirmedAt)} timeframeSelected={Boolean(upload.timeframe)} errorMessage={upload.errorMessage} + pipelineSubStage={pipelineSubStages[index]} /> {upload.status === "failed" && (