feat: add external funnel monitoring and conversion alerts

apps/web/src/app/api/mcp/[tool]/route.ts

@@ -14,6 +14,7 @@ import { logAuditEvent } from "@/lib/audit";
 import { redis } from "@/lib/redis";
 import { authHold, capturePayment } from "@/lib/payment";
 import { sendTrafficAlert } from "@/lib/traffic-alert";
+import { evaluateExternalFunnelHealth } from "@/lib/traffic-conversion-monitor";
 import crypto from "crypto";
 import { z } from "zod";
 
@@ -182,6 +183,13 @@ export async function POST(request: NextRequest, props: { params: Promise<{ tool
           },
         });
 
+        if (isPublicIp) {
+          void evaluateExternalFunnelHealth({
+            surface: "mcp/list_open_tasks",
+            periodMinutes: 10,
+          });
+        }
+
         void prisma.auditEvent.count({
           where: {
             createdAt: {
@@ -288,6 +296,11 @@ export async function POST(request: NextRequest, props: { params: Promise<{ tool
           },
         });
 
+        void evaluateExternalFunnelHealth({
+          surface: "mcp/claim_task",
+          periodMinutes: 10,
+        });
+
         // Set Redis TTL key (3600 seconds)
         await redis.set(`vw:task:${claim.task_id}:executing`, claim.claim_token, "EX", 3600);
 
@@ -364,6 +377,11 @@ export async function POST(request: NextRequest, props: { params: Promise<{ tool
           },
         });
 
+        void evaluateExternalFunnelHealth({
+          surface: "mcp/submit_solution",
+          periodMinutes: 10,
+        });
+
         // Async trigger E2B Sandbox evaluation
         const taskObj = await prisma.task.findUnique({ where: { id: submission.task_id }});
         if (taskObj && typeof taskObj.acceptance_criteria === "object" && taskObj.acceptance_criteria !== null) {

apps/web/src/app/api/open-tasks/route.ts

@@ -3,6 +3,7 @@ import { prisma } from "@/lib/prisma";
 import { TaskStatus } from "@agent-bounty/contracts";
 import { sendTrafficAlert } from "@/lib/traffic-alert";
 import { isIP } from "node:net";
+import { evaluateExternalFunnelHealth } from "@/lib/traffic-conversion-monitor";
 
 export const dynamic = "force-dynamic";
 
@@ -235,6 +236,13 @@ export async function GET(request: Request) {
     }
   }).catch(() => {});
 
+  if (isPublicIp) {
+    void evaluateExternalFunnelHealth({
+      surface: "public-open-tasks",
+      periodMinutes: 10,
+    });
+  }
+
   return NextResponse.json({
     platform: "VibeWork",
     version: "v1",

apps/web/src/app/api/traffic/route.ts

@@ -162,6 +162,14 @@ export async function GET(request: NextRequest) {
     return false;
   };
 
+  const isInternalActor = (params: {
+    actorType: string | null | undefined;
+    actorId: string | null | undefined;
+  }) => {
+    if (params.actorType === "AGENT") return false;
+    return isInternalActorId(params.actorId);
+  };
+
   const externalActorSummary = externalActorRows
     .map((row) => ({
       actorId: row.actorId || "unknown",
@@ -246,7 +254,11 @@ export async function GET(request: NextRequest) {
   });
 
   const recentExternalEvents = recentEvents.filter((event) =>
-    event.action.startsWith("EXTERNAL_") && !isInternalActorId(event.actorId)
+    event.action.startsWith("EXTERNAL_") &&
+    !isInternalActor({
+      actorType: event.actorType,
+      actorId: event.actorId,
+    })
   );
 
   const recentInternalEvents = recentEvents.filter(

apps/web/src/app/traffic/page.tsx

@@ -17,6 +17,10 @@ const EVENT_LABELS: Record<string, string> = {
   EXTERNAL_SUBMIT_SOLUTION_ERROR: "外部提交失敗",
   EXTERNAL_LIST_OPEN_TASKS_ERROR: "外部公開流量端點錯誤",
   EXTERNAL_LIST_OPEN_TASKS_MCP_ERROR: "外部 MCP 流量端點錯誤",
+  EXTERNAL_FUNNEL_CLAIM_STALL: "外部曝光後未接案",
+  EXTERNAL_FUNNEL_SUBMIT_STALL: "外部接案後未提交",
+  EXTERNAL_FUNNEL_PASS_STALL: "外部提交後未 PASS",
+  EXTERNAL_FUNNEL_PAYOUT_STALL: "PASS 後未出金",
   JUDGE_COMPLETE: "AI 交件判定完成",
 };
 
@@ -61,6 +65,11 @@ function isInternalActorId(value: string | null | undefined) {
   return false;
 }
 
+function isInternalActor(input: { actorType: string | null | undefined; actorId: string | null | undefined }) {
+  if (input.actorType === "AGENT") return false;
+  return isInternalActorId(input.actorId);
+}
+
 function isAuthorizedToken(token: string | undefined, tokenHeader: string | undefined) {
   if (!token) return true;
   return tokenHeader === token;
@@ -245,7 +254,13 @@ async function getTrafficSummary(minutes: number) {
     externalActorSummary,
     externalEventTypes,
     internalEventTypes,
-    recentExternalEvents: recentEvents.filter((event) => event.action.startsWith("EXTERNAL_") && !isInternalActorId(event.actorId)),
+    recentExternalEvents: recentEvents.filter((event) =>
+      event.action.startsWith("EXTERNAL_") &&
+      !isInternalActor({
+        actorType: event.actorType,
+        actorId: event.actorId,
+      })
+    ),
     recentInternalEvents: recentEvents.filter((event) => !event.action.startsWith("EXTERNAL_")),
     conversionSummary,
     conversionRates,

apps/web/src/lib/traffic-conversion-monitor.ts

@@ -0,0 +1,216 @@
+import { prisma } from "./prisma";
+import { redis } from "./redis";
+import { sendTrafficAlert } from "./traffic-alert";
+
+type FunnelSummary = {
+  discoveryEvents: number;
+  claimEvents: number;
+  submitEvents: number;
+  judgePassEvents: number;
+  judgeFailEvents: number;
+  payoutCaptured: number;
+  payoutReleased: number;
+  periodMinutes: number;
+};
+
+type MonitorInput = {
+  surface: string;
+  periodMinutes?: number;
+};
+
+const DEFAULT_PERIOD_MINUTES = 10;
+const ALERT_TTL_SECONDS = 600;
+
+function asRecordJson(value: unknown): Record<string, unknown> | undefined {
+  if (typeof value === "object" && value !== null && !Array.isArray(value)) {
+    return value as Record<string, unknown>;
+  }
+  return undefined;
+}
+
+function isMissingLedgerTableError(error: unknown) {
+  return (
+    typeof error === "object" &&
+    error !== null &&
+    "code" in error &&
+    (error as { code: string }).code === "P2021"
+  );
+}
+
+async function fetchFunnelSummary(minutes: number): Promise<FunnelSummary> {
+  const since = new Date(Date.now() - minutes * 60 * 1000);
+
+  const summaryRows = await prisma.auditEvent.groupBy({
+    by: ["action"],
+    where: {
+      createdAt: { gte: since },
+      action: {
+        startsWith: "EXTERNAL_",
+      },
+    },
+    _count: { _all: true },
+  });
+
+  const judgeRows = await prisma.auditEvent.findMany({
+    where: {
+      createdAt: { gte: since },
+      action: "JUDGE_COMPLETE",
+    },
+    select: { metadata: true },
+  });
+
+  let payoutCaptured = 0;
+  let payoutReleased = 0;
+  try {
+    [payoutCaptured, payoutReleased] = await Promise.all([
+      prisma.ledgerEntry.count({
+        where: {
+          created_at: { gte: since },
+          phase: "CAPTURE",
+          response_status: "SUCCESS",
+        },
+      }),
+      prisma.ledgerEntry.count({
+        where: {
+          created_at: { gte: since },
+          phase: "RELEASE",
+          response_status: "SUCCESS",
+        },
+      }),
+    ]);
+  } catch (error) {
+    if (!isMissingLedgerTableError(error)) {
+      throw error;
+    }
+  }
+
+  const actionSummary = Object.fromEntries(
+    summaryRows.map((row) => [row.action, row._count._all])
+  );
+
+  const discoveryEvents =
+    (actionSummary["EXTERNAL_LIST_OPEN_TASKS"] || 0) +
+    (actionSummary["EXTERNAL_LIST_OPEN_TASKS_MCP"] || 0);
+  const claimEvents = actionSummary["EXTERNAL_CLAIM_TASK_SUCCESS"] || 0;
+  const submitEvents = actionSummary["EXTERNAL_SUBMIT_SOLUTION_SUCCESS"] || 0;
+
+  const judgePassEvents = judgeRows.filter((row) => {
+    const metadata = asRecordJson(row.metadata);
+    return metadata?.overall_result === "PASS";
+  }).length;
+
+  const judgeFailEvents = judgeRows.filter((row) => {
+    const metadata = asRecordJson(row.metadata);
+    return metadata?.overall_result === "FAIL";
+  }).length;
+
+  return {
+    discoveryEvents,
+    claimEvents,
+    submitEvents,
+    judgePassEvents,
+    judgeFailEvents,
+    payoutCaptured,
+    payoutReleased,
+    periodMinutes: minutes,
+  };
+}
+
+function buildAlertMessage(rule: string, summary: FunnelSummary) {
+  const { periodMinutes, discoveryEvents, claimEvents, submitEvents, judgePassEvents, payoutCaptured } = summary;
+
+  switch (rule) {
+    case "EXTERNAL_FUNNEL_CLAIM_STALL":
+      return `外部曝光已達 ${discoveryEvents}（最近 ${periodMinutes} 分鐘），但尚無接案（EXTERNAL_CLAIM_TASK_SUCCESS = ${claimEvents}）。請檢查任務是否包含可直接執行的 npx 指令與明確交付條件。`;
+    case "EXTERNAL_FUNNEL_SUBMIT_STALL":
+      return `外部已有 ${claimEvents} 筆接案，但近期 ${periodMinutes} 分鐘無任何提交（EXTERNAL_SUBMIT_SOLUTION_SUCCESS = ${submitEvents}）。請先加速回傳格式與驗收測試規格。`;
+    case "EXTERNAL_FUNNEL_PASS_STALL":
+      return `外部已提交 ${submitEvents} 次但尚無 PASS（JUDGE_RESULT PASS = ${judgePassEvents}）。請先檢查 task acceptance_criteria 與測試欄位是否可自動驗證。`;
+    case "EXTERNAL_FUNNEL_PAYOUT_STALL":
+      return `有 PASS 但未收款（payout CAPTURE 成功 = ${payoutCaptured}）。請確認支付授權、Stripe key 與 capture 任務是否正常。`;
+    default:
+      return "外部 AI 流量轉化斷崖異常。";
+  }
+}
+
+function alertRules(summary: FunnelSummary): Array<{ action: string; message: string }> {
+  const alerts: Array<{ action: string; message: string }> = [];
+
+  if (summary.discoveryEvents > 0 && summary.claimEvents === 0) {
+    alerts.push({
+      action: "EXTERNAL_FUNNEL_CLAIM_STALL",
+      message: buildAlertMessage("EXTERNAL_FUNNEL_CLAIM_STALL", summary),
+    });
+  }
+
+  if (summary.claimEvents > 0 && summary.submitEvents === 0) {
+    alerts.push({
+      action: "EXTERNAL_FUNNEL_SUBMIT_STALL",
+      message: buildAlertMessage("EXTERNAL_FUNNEL_SUBMIT_STALL", summary),
+    });
+  }
+
+  if (summary.submitEvents > 0 && summary.judgePassEvents === 0) {
+    alerts.push({
+      action: "EXTERNAL_FUNNEL_PASS_STALL",
+      message: buildAlertMessage("EXTERNAL_FUNNEL_PASS_STALL", summary),
+    });
+  }
+
+  if (summary.judgePassEvents > 0 && summary.payoutCaptured === 0) {
+    alerts.push({
+      action: "EXTERNAL_FUNNEL_PAYOUT_STALL",
+      message: buildAlertMessage("EXTERNAL_FUNNEL_PAYOUT_STALL", summary),
+    });
+  }
+
+  return alerts;
+}
+
+async function shouldEmitAlert(key: string): Promise<boolean> {
+  try {
+    const result = await redis.set(key, String(Date.now()), "NX", "EX", ALERT_TTL_SECONDS);
+    return result === "OK";
+  } catch (error) {
+    console.warn("[traffic-monitor] redis dedupe failed", error);
+    return true;
+  }
+}
+
+export async function evaluateExternalFunnelHealth(input: MonitorInput): Promise<void> {
+  const periodMinutes = Math.max(input.periodMinutes ?? DEFAULT_PERIOD_MINUTES, 5);
+
+  try {
+    const summary = await fetchFunnelSummary(periodMinutes);
+    const alerts = alertRules(summary);
+
+    await Promise.all(
+      alerts.map(async (rule) => {
+        const dedupeKey = `vw:traffic-funnel:${rule.action}:${input.surface}`;
+        const shouldNotify = await shouldEmitAlert(dedupeKey);
+        if (!shouldNotify) return;
+
+        await sendTrafficAlert({
+          level: "warning",
+          action: rule.action,
+          surface: input.surface,
+          actorType: "SYSTEM",
+          actorId: "traffic-monitor",
+          message: rule.message,
+          metadata: {
+            discovery_events: summary.discoveryEvents,
+            claim_events: summary.claimEvents,
+            submit_events: summary.submitEvents,
+            judge_pass_events: summary.judgePassEvents,
+            judge_fail_events: summary.judgeFailEvents,
+            payout_captured: summary.payoutCaptured,
+            payout_released: summary.payoutReleased,
+            period_minutes: summary.periodMinutes,
+          },
+        });
+      })
+    );
+  } catch (error) {
+    console.error("[traffic-monitor] evaluateExternalFunnelHealth failed", error);
+  }
+}