feat: add verification queue for ambiguous AI-Radar signals

migrations/0007_verification_queue.sql

@@ -0,0 +1,48 @@
+PRAGMA foreign_keys = ON;
+
+CREATE TABLE IF NOT EXISTS verification_queue (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  source_id INTEGER NOT NULL,
+  external_id TEXT NOT NULL,
+  candidate_json TEXT NOT NULL,
+  reason TEXT NOT NULL,
+  signal_score INTEGER NOT NULL DEFAULT 0,
+  status TEXT NOT NULL DEFAULT 'pending' CHECK(status IN ('pending','verified','discarded')),
+  attempts INTEGER NOT NULL DEFAULT 0,
+  first_seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  next_check_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  expires_at TEXT NOT NULL,
+  last_checked_at TEXT,
+  resolved_item_id INTEGER,
+  last_error TEXT,
+  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  UNIQUE(source_id, external_id),
+  FOREIGN KEY(source_id) REFERENCES sources(id) ON DELETE CASCADE,
+  FOREIGN KEY(resolved_item_id) REFERENCES items(id) ON DELETE SET NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_verification_queue_due
+  ON verification_queue(status, next_check_at, signal_score DESC);
+
+CREATE TABLE IF NOT EXISTS source_link_baselines (
+  source_id INTEGER PRIMARY KEY,
+  initialized_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  FOREIGN KEY(source_id) REFERENCES sources(id) ON DELETE CASCADE
+);
+
+-- Model dates are publication/observation metadata, never offer expiration dates.
+UPDATE items
+SET expires_at = NULL
+WHERE kind IN ('new_model','model_api_available','model_open_source','model_benchmark','discovered_model')
+  AND expires_at IS NOT NULL;
+
+-- Force one full generic-web fetch after deployment so existing links become a baseline
+-- instead of being mistaken for newly discovered links. content_hash is deliberately kept.
+UPDATE sources
+SET etag = NULL,
+    last_modified = NULL,
+    next_fetch_at = CURRENT_TIMESTAMP,
+    updated_at = CURRENT_TIMESTAMP
+WHERE type = 'web'
+  AND (config_json IS NULL OR config_json NOT LIKE '%discoveryProvider%');

src/collectors.ts

@@ -2,7 +2,7 @@ import type { Candidate, CollectResult, SourceConfig, SourceRow } from './types'
 import { parseJson, sha256, stripHtml, textExcerpt } from './utils';
 
 function conditionalHeaders(source: SourceRow): Headers {
-  const headers = new Headers({ 'user-agent': 'AI-Radar/0.2 (+https://github.com/Felix8686/TokenRadar)', accept: '*/*' });
+  const headers = new Headers({ 'user-agent': 'AI-Radar/0.3 (+https://github.com/Felix8686/TokenRadar)', accept: '*/*' });
   if (source.etag) headers.set('if-none-match', source.etag);
   if (source.last_modified) headers.set('if-modified-since', source.last_modified);
   const config = parseJson<SourceConfig>(source.config_json, {});
@@ -236,6 +236,45 @@ async function collectArtificialAnalysisModels(source: SourceRow): Promise<Colle
   };
 }
 
+const LINK_SIGNAL = /\b(?:free|credits?|pricing|price|discount|coupon|promo|api|models?|release|launch|introducing|announcement|open[-\s]?source|benchmark|coding|agent)\b|免费|额度|价格|优惠|折扣|模型|发布|推出|开源|评测|编程|智能体/i;
+const CONTENT_PATH = /\/(?:resources?|blog|news|releases?|models?|api|pricing|docs?|changelog)(?:\/|$)/i;
+
+export function extractLinkedPageCandidates(source: SourceRow, html: string): Candidate[] {
+  let base: URL;
+  try { base = new URL(source.url); } catch { return []; }
+  const ranked = new Map<string, { candidate: Candidate; score: number }>();
+  const regex = /<a\b[^>]*href=["']([^"']+)["'][^>]*>([\s\S]*?)<\/a>/gi;
+  for (const match of html.matchAll(regex)) {
+    const href = match[1].trim();
+    if (!href || href.startsWith('#') || /^(?:mailto|javascript|tel):/i.test(href)) continue;
+    let target: URL;
+    try { target = new URL(href, base); } catch { continue; }
+    if (!['http:', 'https:'].includes(target.protocol) || target.origin !== base.origin) continue;
+    target.hash = '';
+    if (target.pathname === base.pathname && target.search === base.search) continue;
+    if (/\.(?:png|jpe?g|gif|svg|webp|css|js|ico|pdf|zip)(?:$|\?)/i.test(target.pathname)) continue;
+    if (/\/(?:login|signin|signup|privacy|terms|contact|about)(?:\/|$)/i.test(target.pathname)) continue;
+    const label = stripHtml(match[2]).replace(/\s+/g, ' ').trim().slice(0, 220);
+    let score = 0;
+    if (LINK_SIGNAL.test(label)) score += 45;
+    if (LINK_SIGNAL.test(`${target.pathname} ${target.search}`)) score += 35;
+    if (CONTENT_PATH.test(target.pathname)) score += 20;
+    if (score < 20) continue;
+    const url = target.toString();
+    const candidate: Candidate = {
+      externalId: `linked:${url}`,
+      title: label || target.pathname.split('/').filter(Boolean).pop() || url,
+      summary: `Linked page observed from ${source.name}`,
+      rawExcerpt: label || undefined,
+      url,
+      observationKind: 'linked_page',
+    };
+    const previous = ranked.get(url);
+    if (!previous || score > previous.score) ranked.set(url, { candidate, score });
+  }
+  return [...ranked.values()].sort((a, b) => b.score - a.score).slice(0, 60).map((entry) => entry.candidate);
+}
+
 async function collectWeb(source: SourceRow): Promise<CollectResult> {
   const config = parseJson<SourceConfig>(source.config_json, {});
   if (config.discoveryProvider === 'openrouter_models') return collectOpenRouterModels(source);
@@ -248,13 +287,21 @@ async function collectWeb(source: SourceRow): Promise<CollectResult> {
   const normalized = sanitizeExcerpt(body) || stripHtml(body).replace(/\s+/g, ' ').trim();
   const contentHash = await sha256(normalized);
   const excerpt = normalized.length <= 1200 ? normalized : `${normalized.slice(0, 1200)}…`;
+  const pageChange: Candidate = {
+    externalId: contentHash,
+    title: `${source.name} changed`,
+    summary: excerpt,
+    rawExcerpt: excerpt,
+    url: source.url,
+    observationKind: 'page_change',
+  };
   return {
     statusCode: response.status,
     notModified: false,
     etag: response.headers.get('etag') || undefined,
     lastModified: response.headers.get('last-modified') || undefined,
     contentHash,
-    candidates: [{ externalId: contentHash, title: `${source.name} changed`, summary: excerpt, rawExcerpt: excerpt, url: source.url }],
+    candidates: [pageChange, ...extractLinkedPageCandidates(source, body)],
   };
 }
 

src/db.ts

@@ -1,4 +1,4 @@
-import type { Candidate, Classification, ItemRow, SourceRow, SourceTier } from './types';
+import type { Candidate, Classification, ItemRow, SourceRow, SourceTier, VerificationQueueRow } from './types';
 import { addMinutesIso, isoNow, sha256 } from './utils';
 
 const DISCOVERY_KINDS = new Set<Classification['kind']>(['new_model','model_api_available','model_open_source','model_benchmark']);
@@ -13,6 +13,10 @@ export async function getDueSources(db: D1Database, limit: number): Promise<Sour
   return result.results || [];
 }
 
+export async function getSourceById(db:D1Database,id:number):Promise<SourceRow|null>{
+  return await db.prepare('SELECT * FROM sources WHERE id=?1 LIMIT 1').bind(id).first<SourceRow>();
+}
+
 export async function saveFetchSuccess(db: D1Database, source: SourceRow, state: { etag?: string; lastModified?: string; contentHash?: string; statusCode: number; changed: boolean; durationMs: number }): Promise<void> {
   const now = new Date();
   await db.batch([
@@ -57,6 +61,50 @@ export async function filterNewCandidates(db:D1Database,source:SourceRow,candida
   return unseen.filter((_,index)=>!duplicates.has(fingerprints[index]));
 }
 
+export async function hasLinkBaseline(db:D1Database,sourceId:number):Promise<boolean>{
+  const row=await db.prepare('SELECT source_id FROM source_link_baselines WHERE source_id=?1 LIMIT 1').bind(sourceId).first<{source_id:number}>();
+  return Boolean(row);
+}
+
+export async function markLinkBaseline(db:D1Database,sourceId:number):Promise<void>{
+  await db.prepare('INSERT OR IGNORE INTO source_link_baselines(source_id,initialized_at) VALUES(?1,CURRENT_TIMESTAMP)').bind(sourceId).run();
+}
+
+export async function enqueueVerification(db:D1Database,source:SourceRow,candidate:Candidate,reason:string,signalScore:number):Promise<void>{
+  const externalId=await sourceEntryId(candidate);
+  const now=new Date();
+  const expiresAt=new Date(now.getTime()+72*60*60*1000).toISOString();
+  await db.prepare(`INSERT INTO verification_queue(source_id,external_id,candidate_json,reason,signal_score,status,attempts,first_seen_at,next_check_at,expires_at,updated_at)
+    VALUES(?1,?2,?3,?4,?5,'pending',0,?6,?6,?7,?6)
+    ON CONFLICT(source_id,external_id) DO UPDATE SET candidate_json=excluded.candidate_json,reason=excluded.reason,signal_score=MAX(verification_queue.signal_score,excluded.signal_score),updated_at=excluded.updated_at
+    WHERE verification_queue.status='pending'`)
+    .bind(source.id,externalId,JSON.stringify(candidate),reason,Math.max(0,Math.min(100,signalScore)),now.toISOString(),expiresAt).run();
+}
+
+export async function expireVerificationQueue(db:D1Database):Promise<void>{
+  const now=isoNow();
+  await db.prepare(`UPDATE verification_queue SET status='discarded',updated_at=?1 WHERE status='pending' AND (expires_at<=?1 OR attempts>=3)`).bind(now).run();
+}
+
+export async function getDueVerification(db:D1Database,limit=5):Promise<VerificationQueueRow[]>{
+  await expireVerificationQueue(db);
+  const result=await db.prepare(`SELECT * FROM verification_queue WHERE status='pending' AND next_check_at<=?1 AND expires_at>?1 AND attempts<3 ORDER BY signal_score DESC,first_seen_at ASC LIMIT ?2`).bind(isoNow(),Math.max(1,Math.min(20,limit))).all<VerificationQueueRow>();
+  return result.results||[];
+}
+
+export async function markVerificationRetry(db:D1Database,row:VerificationQueueRow,error?:unknown):Promise<void>{
+  const attempts=row.attempts+1;
+  const expired=attempts>=3||new Date(row.expires_at).getTime()<=Date.now();
+  const nextCheck=new Date(Date.now()+12*60*60*1000).toISOString();
+  const message=error instanceof Error?error.message:error?String(error):null;
+  await db.prepare(`UPDATE verification_queue SET attempts=?1,status=?2,next_check_at=?3,last_checked_at=?4,last_error=?5,updated_at=?4 WHERE id=?6`)
+    .bind(attempts,expired?'discarded':'pending',nextCheck,isoNow(),message?.slice(0,800)||null,row.id).run();
+}
+
+export async function markVerificationResolved(db:D1Database,id:number,itemId:number|null):Promise<void>{
+  await db.prepare(`UPDATE verification_queue SET status='verified',resolved_item_id=?1,last_checked_at=?2,updated_at=?2 WHERE id=?3`).bind(itemId,isoNow(),id).run();
+}
+
 export async function registerDiscoveredSource(db:D1Database,source:SourceRow,candidate:Candidate,c:Classification):Promise<boolean>{
   if(source.source_tier!=='discovery'||!DISCOVERY_KINDS.has(c.kind)||c.priority==='P3'||!candidate.url)return false;
   let url:string;

src/index.ts

@@ -1,31 +1,61 @@
-import type { Env, ItemRow, SourceRow, SourceTier } from './types';
+import type { Candidate, Env, ItemRow, SourceRow, SourceTier } from './types';
 import { collectSource } from './collectors';
 import { classifyDeterministically, maybeEnrichWithAi } from './rules';
-import { expireTemporarySources, filterNewCandidates, getDueSources, getRecentItems, insertItem, markPushed, noteSourceValue, registerDiscoveredSource, rememberSourceEntries, rememberSourceEntry, saveFetchFailure, saveFetchSuccess } from './db';
+import { enqueueVerification, expireTemporarySources, filterNewCandidates, getDueSources, getRecentItems, hasLinkBaseline, insertItem, markLinkBaseline, markPushed, noteSourceValue, registerDiscoveredSource, rememberSourceEntries, rememberSourceEntry, saveFetchFailure, saveFetchSuccess } from './db';
 import { generateDailyReport, getDailyReport, getLatestReport } from './daily';
 import { pushP1 } from './telegram';
+import { directP1Allowed, needsVerification, processVerificationQueue, verificationSignalScore } from './verification';
 import { jsonResponse } from './utils';
 
+async function establishLinkBaseline(env:Env,source:SourceRow,candidates:Candidate[]):Promise<Candidate[]>{
+  const links=candidates.filter(candidate=>candidate.observationKind==='linked_page');
+  if(!links.length)return candidates;
+  if(await hasLinkBaseline(env.DB,source.id))return candidates;
+  await rememberSourceEntries(env.DB,source,links);
+  await markLinkBaseline(env.DB,source.id);
+  return candidates.filter(candidate=>candidate.observationKind!=='linked_page');
+}
+
 async function processSource(env:Env,source:SourceRow):Promise<void>{
   const started=Date.now();
   try{
     const result=await collectSource(source);
+    const candidatesForRun=result.notModified?result.candidates:await establishLinkBaseline(env,source,result.candidates);
     const changed=!result.notModified&&Boolean(result.contentHash&&result.contentHash!==source.content_hash);
     const isBootstrap=!source.content_hash;
     if(!result.notModified&&changed){
       if(isBootstrap){
-        await rememberSourceEntries(env.DB,source,result.candidates);
+        await rememberSourceEntries(env.DB,source,candidatesForRun);
       }else{
-        const candidates=await filterNewCandidates(env.DB,source,result.candidates);
+        const candidates=await filterNewCandidates(env.DB,source,candidatesForRun);
+        const hasNewLinkedPage=candidates.some(candidate=>candidate.observationKind==='linked_page');
         for(const candidate of candidates){
           const deterministic=classifyDeterministically(source,candidate);
-          const c=await maybeEnrichWithAi(env,source,candidate,deterministic);
+
+          // If a listing page changed because it contains a newly observed content link,
+          // verify that exact linked page instead of pushing the generic "xxx changed" observation.
+          if(candidate.observationKind==='page_change'&&hasNewLinkedPage){
+            await rememberSourceEntry(env.DB,source,candidate);
+            continue;
+          }
+
+          if(needsVerification(source,candidate,deterministic)){
+            await enqueueVerification(env.DB,source,candidate,candidate.observationKind||deterministic.kind,verificationSignalScore(source,candidate));
+            await rememberSourceEntry(env.DB,source,candidate);
+            continue;
+          }
+
+          // A plain catalog observation is internal low-confidence data. It can be retained as P3,
+          // but it is never allowed to become an immediate high-value alert without verification.
+          const c=deterministic.kind==='discovered_model'
+            ? {...deterministic,score:Math.min(39,deterministic.score),priority:'P3' as const}
+            : await maybeEnrichWithAi(env,source,candidate,deterministic);
           const saved=await insertItem(env.DB,source,candidate,c);
           await rememberSourceEntry(env.DB,source,candidate);
           if(!saved.inserted||!saved.id)continue;
           await registerDiscoveredSource(env.DB,source,candidate,c);
           await noteSourceValue(env.DB,source,c);
-          if(c.priority==='P1'){
+          if(directP1Allowed(candidate,c)){
             const item:ItemRow&{source_name:string}={id:saved.id,source_id:source.id,title:candidate.title,summary:candidate.summary||null,url:candidate.url||null,kind:c.kind,priority:c.priority,score:c.score,source_confidence:c.sourceConfidence,verification_status:c.verificationStatus,vendor:c.vendor||null,product:c.product||null,previous_price:c.previousPrice??null,current_price:c.currentPrice??null,currency:c.currency||null,expires_at:c.expiresAt||null,discovered_at:new Date().toISOString(),published_at:candidate.publishedAt||null,pushed_at:null,source_name:source.name};
             if(await pushP1(env,item,c.summaryZh))await markPushed(env.DB,saved.id);
           }
@@ -37,7 +67,16 @@ async function processSource(env:Env,source:SourceRow):Promise<void>{
     await saveFetchFailure(env.DB,source,error,Date.now()-started);
   }
 }
-async function harvest(env:Env):Promise<{processed:number}>{await expireTemporarySources(env.DB);const limit=Math.max(1,Math.min(50,Number(env.SOURCE_BATCH_SIZE||10)));const sources=await getDueSources(env.DB,limit);for(const source of sources)await processSource(env,source);return{processed:sources.length};}
+
+async function harvest(env:Env):Promise<{processed:number;verificationProcessed:number;verificationVerified:number}>{
+  await expireTemporarySources(env.DB);
+  const limit=Math.max(1,Math.min(50,Number(env.SOURCE_BATCH_SIZE||10)));
+  const sources=await getDueSources(env.DB,limit);
+  for(const source of sources)await processSource(env,source);
+  const verification=await processVerificationQueue(env,5);
+  return{processed:sources.length,verificationProcessed:verification.processed,verificationVerified:verification.verified};
+}
+
 function isAdmin(request:Request,env:Env):boolean{if(!env.ADMIN_TOKEN)return false;return request.headers.get('authorization')===`Bearer ${env.ADMIN_TOKEN}`;}
 export function toPublicItem(row:ItemRow&{source_name:string;last_verified_at:string|null}){return{id:row.id,title:row.title,summary:row.summary,url:row.url,kind:row.kind,vendor:row.vendor,product:row.product,previous_price:row.previous_price,current_price:row.current_price,currency:row.currency,expires_at:row.expires_at,discovered_at:row.discovered_at,published_at:row.published_at,source_name:row.source_name,verification_status:row.verification_status,last_verified_at:row.last_verified_at};}
 async function listSources(env:Env):Promise<Response>{const rows=await env.DB.prepare(`SELECT id,name,url,type,trust_level,enabled,interval_minutes,source_tier,discovered_from_source_id,expires_at,hit_count,next_fetch_at,last_fetch_at,last_success_at,failure_count,status FROM sources ORDER BY id DESC`).all();return jsonResponse(rows.results||[]);}

src/types.ts

@@ -19,6 +19,8 @@ export type SourceTier = 'core' | 'discovery' | 'temporary' | 'candidate';
 export type DiscoveryProvider = 'openrouter_models' | 'huggingface_models' | 'artificial_analysis_models';
 export type TrustLevel = 'A' | 'B' | 'C' | 'D';
 export type Priority = 'P1' | 'P2' | 'P3';
+export type ObservationKind = 'page_change' | 'linked_page';
+export type VerificationQueueStatus = 'pending' | 'verified' | 'discarded';
 export type ItemKind =
   | 'free_credit'
   | 'limited_offer'
@@ -57,6 +59,7 @@ export interface Candidate {
   externalId?: string; title: string; summary?: string; url?: string; publishedAt?: string; rawExcerpt?: string;
   signalKind?: Extract<ItemKind, 'new_model' | 'model_api_available' | 'model_open_source' | 'model_benchmark' | 'discovered_model'>;
   vendorHint?: string; productHint?: string;
+  observationKind?: ObservationKind;
 }
 
 export interface CollectResult {
@@ -71,6 +74,23 @@ export interface Classification {
   aiEnriched: boolean;
 }
 
+export interface VerificationQueueRow {
+  id: number;
+  source_id: number;
+  external_id: string;
+  candidate_json: string;
+  reason: string;
+  signal_score: number;
+  status: VerificationQueueStatus;
+  attempts: number;
+  first_seen_at: string;
+  next_check_at: string;
+  expires_at: string;
+  last_checked_at: string | null;
+  resolved_item_id: number | null;
+  last_error: string | null;
+}
+
 export interface ItemRow {
   id: number; source_id: number; title: string; summary: string | null; url: string | null;
   kind: ItemKind; priority: Priority; score: number; source_confidence: string; verification_status: string;