Audit and repair intent workflow runtime

Author: tannerlinsley

src/routes/admin/intent.tsx

@@ -12,15 +12,18 @@ import {
   AlertTriangle,
   CheckCircle2,
   Clock,
+  Wrench,
 } from 'lucide-react'
 import { Button } from '~/ui'
 import { Card } from '~/components/Card'
 import { formatDistanceToNow } from '~/utils/dates'
 import {
   getIntentAdminStats,
+  getIntentWorkflowHealth,
   listIntentPackages,
   listFailedVersions,
   listIntentWorkflowRuns,
+  repairIntentWorkflowStore,
   triggerIntentDiscover,
   triggerIntentProcess,
   retryIntentVersion,
@@ -42,6 +45,7 @@ const QK = {
   packages: ['admin', 'intent', 'packages'] as const,
   failed: ['admin', 'intent', 'failed'] as const,
   workflows: ['admin', 'intent', 'workflows'] as const,
+  health: ['admin', 'intent', 'workflow-health'] as const,
 }
 
 // ---------------------------------------------------------------------------
@@ -77,13 +81,19 @@ function IntentAdminPage() {
     refetchInterval: 10_000,
   })
 
+  const healthQuery = useQuery({
+    queryKey: QK.health,
+    queryFn: () => getIntentWorkflowHealth(),
+    refetchInterval: 10_000,
+  })
+
   const discoverMutation = useMutation({
     mutationFn: () => triggerIntentDiscover(),
     onSuccess: invalidateAll,
   })
 
   const processMutation = useMutation({
-    mutationFn: (limit: number) => triggerIntentProcess({ data: { limit } }),
+    mutationFn: () => triggerIntentProcess(),
     onSuccess: invalidateAll,
   })
 
@@ -92,6 +102,11 @@ function IntentAdminPage() {
     onSuccess: invalidateAll,
   })
 
+  const repairWorkflowMutation = useMutation({
+    mutationFn: () => repairIntentWorkflowStore(),
+    onSuccess: invalidateAll,
+  })
+
   const seedMutation = useMutation({
     mutationFn: (name: string) => seedIntentPackage({ data: { name } }),
     onSuccess: () => {
@@ -160,19 +175,19 @@ function IntentAdminPage() {
           <Button
             size="sm"
             color="green"
-            onClick={() => processMutation.mutate(10)}
+            onClick={() => processMutation.mutate()}
             disabled={
               processMutation.isPending ||
               (stats?.pendingVersions ?? 0) + (stats?.failedVersions ?? 0) === 0
             }
-            title="Download tarballs and extract skills for up to 10 pending versions"
+            title="Download tarballs and extract skills until the workflow nears its time budget"
           >
             <Play
               className={
                 processMutation.isPending ? 'animate-pulse w-4 h-4' : 'w-4 h-4'
               }
             />
-            {processMutation.isPending ? 'Processing...' : 'Process 10 Pending'}
+            {processMutation.isPending ? 'Processing...' : 'Process Queue'}
           </Button>
           {(stats?.failedVersions ?? 0) > 0 && (
             <Button
@@ -235,10 +250,17 @@ function IntentAdminPage() {
       {processMutation.data && (
         <ResultBanner
           title={`Processed ${processMutation.data.processed} version(s)`}
-          items={processMutation.data.results.map(
-            (r) =>
-              `${r.packageName}@${r.version}: ${r.status === 'synced' ? `${r.skillCount} skills` : `FAILED — ${r.error}`}`,
-          )}
+          items={[
+            ...(processMutation.data.deferred > 0
+              ? [
+                  `${processMutation.data.deferred} version(s) deferred by the time budget`,
+                ]
+              : []),
+            ...processMutation.data.results.map(
+              (r) =>
+                `${r.packageName}@${r.version}: ${r.status === 'synced' ? `${r.skillCount} skills` : `FAILED — ${r.error}`}`,
+            ),
+          ]}
           errors={processMutation.data.results
             .filter((r) => r.status === 'failed')
             .map((r) => `${r.packageName}@${r.version}: ${r.error}`)}
@@ -253,6 +275,17 @@ function IntentAdminPage() {
           onDismiss={() => resetFailedMutation.reset()}
         />
       )}
+      {repairWorkflowMutation.data && (
+        <ResultBanner
+          title="Workflow store repaired"
+          items={[
+            `${repairWorkflowMutation.data.staleRunsMarkedErrored} stale run(s) marked errored`,
+            `${repairWorkflowMutation.data.unregisteredSchedulesDeleted} obsolete schedule(s) deleted`,
+          ]}
+          errors={[]}
+          onDismiss={() => repairWorkflowMutation.reset()}
+        />
+      )}
       {githubDiscoverMutation.data && (
         <ResultBanner
           title="GitHub discovery complete"
@@ -332,6 +365,13 @@ function IntentAdminPage() {
         />
       </div>
 
+      <WorkflowHealthSection
+        health={healthQuery.data}
+        loading={healthQuery.isLoading}
+        repairing={repairWorkflowMutation.isPending}
+        onRepair={() => repairWorkflowMutation.mutate()}
+      />
+
       <WorkflowRunsSection
         runs={workflowsQuery.data ?? []}
         loading={workflowsQuery.isLoading}
@@ -393,6 +433,154 @@ function StatCard({
   )
 }
 
+function WorkflowHealthSection({
+  health,
+  loading,
+  repairing,
+  onRepair,
+}: {
+  readonly health:
+    | {
+        checkedAt: Date
+        staleRunMs: number
+        staleRuns: Array<{
+          runId: string
+          workflowId: string
+          status: string
+          updatedAt: Date
+          leaseExpiresAt: Date | null
+        }>
+        unregisteredSchedules: Array<{
+          scheduleId: string
+          workflowId: string
+          enabled: boolean
+          nextFireAt: Date | null
+          updatedAt: Date
+        }>
+        schedules: Array<{
+          scheduleId: string
+          workflowId: string
+          enabled: boolean
+          nextFireAt: Date | null
+          updatedAt: Date
+        }>
+        latestRuns: Array<{
+          runId: string
+          workflowId: string
+          status: string
+          updatedAt: Date
+        }>
+        statusCounts: Array<{ status: string; count: number }>
+      }
+    | undefined
+  readonly loading: boolean
+  readonly repairing: boolean
+  readonly onRepair: () => void
+}) {
+  const staleRunCount = health?.staleRuns.length ?? 0
+  const obsoleteScheduleCount = health?.unregisteredSchedules.length ?? 0
+  const needsRepair = staleRunCount > 0 || obsoleteScheduleCount > 0
+
+  return (
+    <div className="mb-6">
+      <div className="mb-2 flex items-center justify-between gap-3">
+        <h2 className="text-sm font-semibold text-gray-700 dark:text-gray-300 flex items-center gap-1.5">
+          <CheckCircle2 className="w-4 h-4" />
+          Workflow Health
+        </h2>
+        <Button
+          size="xs"
+          variant="secondary"
+          onClick={onRepair}
+          disabled={repairing || loading || !needsRepair}
+          title="Mark stale runs as errored and delete schedules for workflows that are no longer registered"
+        >
+          <Wrench
+            className={repairing ? 'w-3.5 h-3.5 animate-pulse' : 'w-3.5 h-3.5'}
+          />
+          {repairing ? 'Repairing...' : 'Repair Store'}
+        </Button>
+      </div>
+      {loading ? (
+        <div className="h-28 rounded-xl bg-gray-100 dark:bg-gray-800 animate-pulse" />
+      ) : (
+        <div className="rounded-xl border border-gray-200 dark:border-gray-800 overflow-hidden">
+          <div className="grid grid-cols-2 md:grid-cols-4 divide-x divide-y md:divide-y-0 divide-gray-100 dark:divide-gray-800">
+            <HealthMetric
+              label="Stale runs"
+              value={staleRunCount}
+              tone={staleRunCount > 0 ? 'red' : 'green'}
+            />
+            <HealthMetric
+              label="Obsolete schedules"
+              value={obsoleteScheduleCount}
+              tone={obsoleteScheduleCount > 0 ? 'amber' : 'green'}
+            />
+            <HealthMetric
+              label="Registered schedules"
+              value={health?.schedules.length ?? 0}
+              tone="default"
+            />
+            <HealthMetric
+              label="Tracked runs"
+              value={
+                health?.statusCounts.reduce(
+                  (total, row) => total + row.count,
+                  0,
+                ) ?? 0
+              }
+              tone="default"
+            />
+          </div>
+          {health && needsRepair && (
+            <div className="border-t border-gray-100 dark:border-gray-800 bg-amber-50/60 dark:bg-amber-950/20 px-3 py-2 text-xs text-amber-800 dark:text-amber-200">
+              {[...health.staleRuns, ...health.unregisteredSchedules]
+                .slice(0, 4)
+                .map((item) =>
+                  'runId' in item
+                    ? `${item.workflowId}: ${item.status} since ${formatDistanceToNow(
+                        item.updatedAt,
+                        { addSuffix: true },
+                      )}`
+                    : `${item.workflowId}: obsolete schedule ${item.scheduleId}`,
+                )
+                .join(' | ')}
+            </div>
+          )}
+        </div>
+      )}
+    </div>
+  )
+}
+
+function HealthMetric({
+  label,
+  value,
+  tone,
+}: {
+  readonly label: string
+  readonly value: number
+  readonly tone: 'default' | 'green' | 'amber' | 'red'
+}) {
+  const valueClass = {
+    default: 'text-gray-900 dark:text-white',
+    green: 'text-emerald-600 dark:text-emerald-400',
+    amber: 'text-amber-600 dark:text-amber-400',
+    red: 'text-red-600 dark:text-red-400',
+  }[tone]
+
+  return (
+    <div className="bg-white dark:bg-gray-900 px-3 py-2">
+      <div className="text-xs text-gray-500 dark:text-gray-400 mb-1">
+        {label}
+      </div>
+      <div className={`text-xl font-semibold tabular-nums ${valueClass}`}>
+        {value.toLocaleString()}
+      </div>
+    </div>
+  )
+}
+
 function WorkflowRunsSection({
   runs,
   loading,

src/server/scheduled.server.ts

@@ -2,26 +2,28 @@ import { materializeWorkflowSchedules } from '@tanstack/workflow-runtime'
 import { pruneStaleCacheRows } from '~/utils/github-content-cache.server'
 import { refreshHomepageNpmStatsSummary } from '~/utils/homepage-npm-stats.server'
 import { refreshGitHubOrgStats } from '~/utils/stats.functions'
-import { workflowRuntime } from '~/utils/workflow-runtime.server'
+import {
+  reconcileWorkflowRuntimeStore,
+  workflowRuntime,
+} from '~/utils/workflow-runtime.server'
 
 const CONTENT_CACHE_PRUNE_CRON = '0 9 * * *'
-const STATS_AND_INTENT_DISCOVER_CRON = '0 */6 * * *'
-const INTENT_PROCESS_CRON = '*/15 * * * *'
+const STATS_REFRESH_CRON = '0 */6 * * *'
+const WORKFLOW_SWEEP_CRON = '* * * * *'
 const WORKFLOW_SWEEP_MAX_DURATION_MS = 25_000
 
 export async function runScheduledTasks(cron: string, scheduledTime: number) {
   switch (cron) {
     case CONTENT_CACHE_PRUNE_CRON:
       await runContentCachePrune(scheduledTime)
       return
-    case STATS_AND_INTENT_DISCOVER_CRON:
+    case STATS_REFRESH_CRON:
       await Promise.all([
         runGitHubStatsRefresh(scheduledTime),
         runHomepageNpmStatsSummaryRefresh(scheduledTime),
-        runWorkflowSweep(cron, scheduledTime),
       ])
       return
-    case INTENT_PROCESS_CRON:
+    case WORKFLOW_SWEEP_CRON:
       await runWorkflowSweep(cron, scheduledTime)
       return
     default:
@@ -34,6 +36,7 @@ async function runWorkflowSweep(cron: string, scheduledTime: number) {
   console.log('[workflow-sweep] Starting workflow sweep...')
 
   try {
+    const reconciliation = await reconcileWorkflowRuntimeStore()
     const materialized = await materializeWorkflowSchedules(workflowRuntime, {
       now: scheduledTime,
     })
@@ -48,7 +51,7 @@ async function runWorkflowSweep(cron: string, scheduledTime: number) {
     const duration = Date.now() - startTime
 
     console.log(
-      `[workflow-sweep] Completed in ${duration}ms - materialized: ${materialized.length}, scheduled: ${JSON.stringify(sweep.summary.scheduled)}, timers: ${JSON.stringify(sweep.summary.timers)}, remaining: ${sweep.remainingMayExist}`,
+      `[workflow-sweep] Completed in ${duration}ms - staleRuns: ${reconciliation.staleRunsMarkedErrored}, prunedSchedules: ${reconciliation.unregisteredSchedulesDeleted}, materialized: ${materialized.length}, scheduled: ${JSON.stringify(sweep.summary.scheduled)}, timers: ${JSON.stringify(sweep.summary.timers)}, remaining: ${sweep.remainingMayExist}`,
     )
     console.log(
       '[workflow-sweep] Scheduled time:',

src/utils/intent-admin.functions.ts

@@ -4,16 +4,17 @@ import {
   deleteIntentPackage as deleteIntentPackageServer,
   discoverViaGitHub as discoverViaGitHubServer,
   getIntentAdminStats as getIntentAdminStatsServer,
+  getIntentWorkflowHealth as getIntentWorkflowHealthServer,
   listFailedVersions as listFailedVersionsServer,
   listIntentPackages as listIntentPackagesServer,
   listIntentWorkflowRuns as listIntentWorkflowRunsServer,
+  repairIntentWorkflowStore as repairIntentWorkflowStoreServer,
   resetFailedVersions as resetFailedVersionsServer,
   retryIntentVersion as retryIntentVersionServer,
   seedIntentPackage as seedIntentPackageServer,
   triggerIntentDiscover as triggerIntentDiscoverServer,
   triggerIntentProcess as triggerIntentProcessServer,
 } from '~/utils/intent-admin.server'
-import { pageSizeSchema } from './schemas'
 
 export const getIntentAdminStats = createServerFn({ method: 'GET' }).handler(
   async () => getIntentAdminStatsServer(),
@@ -31,17 +32,21 @@ export const listIntentWorkflowRuns = createServerFn({ method: 'GET' }).handler(
   async () => listIntentWorkflowRunsServer(),
 )
 
+export const getIntentWorkflowHealth = createServerFn({
+  method: 'GET',
+}).handler(async () => getIntentWorkflowHealthServer())
+
+export const repairIntentWorkflowStore = createServerFn({
+  method: 'POST',
+}).handler(async () => repairIntentWorkflowStoreServer())
+
 export const triggerIntentDiscover = createServerFn({ method: 'POST' }).handler(
   async () => triggerIntentDiscoverServer(),
 )
 
-export const triggerIntentProcess = createServerFn({ method: 'POST' })
-  .validator(
-    v.object({
-      limit: v.optional(pageSizeSchema, 10),
-    }),
-  )
-  .handler(async ({ data }) => triggerIntentProcessServer({ data }))
+export const triggerIntentProcess = createServerFn({ method: 'POST' }).handler(
+  async () => triggerIntentProcessServer(),
+)
 
 export const retryIntentVersion = createServerFn({ method: 'POST' })
   .validator(