TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { activityAnomaly, classifyChange, computeConfidence, computeImpact, computeImportance, computeSignalScore, describeChange, describeFieldChanges, diffIsEmpty, diffJson, diffList, diffText, evaluateChange, eventGroupOf, eventTypeSpec, newId, nextIntervalSeconds, NOISE_CLASSES, sha256, SILENT_ELIGIBLE_TYPES, slugify, sourceImportanceFromTier, summarizeDiff, type DiffResult, type HeuristicResult, type NormalizedContent, type Observation, type SemanticResult, type SensorEndpoint, type Tier, FEED_CHANNELS } from "@websensor/core";2import { getConnector, NormalizeError, PARSER_VERSION, scrapflyAvailable, scrapflyFetch } from "@websensor/connectors";3import { changes, db, eventEntities, events, interpretations, sensorRuns, sensors, snapshots, sql, textArray, type Sensor, type Source } from "@websensor/db";4import { getBlobStore } from "@websensor/store";5import { evaluateAlerts, type PublishedEvent } from "./alerts";6import { assessNovelty, clusterEvent, recentAnnouncementSimilarity } from "./cluster";7import { config, log } from "./config";8import { bumpEntityCounters, resolveEntities } from "./entities";9import { interpretChange, llmAvailable, renderDiff, type Interpretation } from "./interpret";10import { bumpDaily, bumpSourceDaily, m } from "./metrics";11import { publishChange, publishEvent } from "./redis";1213export type RunOutcome = "baseline" | "unchanged" | "not_modified" | "changed" | "event" | "error" | "missing" | "rate_limited" | "parse_error";1415interface CanonicalBlob {16 mode: NormalizedContent["mode"];17 text?: string;18 json?: unknown;19 items?: { key: string; [k: string]: unknown }[];20 compareFields?: string[];21 title?: string | null;22 headings?: string[];23 extractionConfidence: number;24}2526const ANNOUNCEMENT_SENSORS = new Set(["RSS", "ATOM", "STATUSPAGE", "GITHUB_RELEASE", "REST_API"]);2728export async function runSensor(sensor: Sensor, source: Source): Promise<RunOutcome> {29 const started = new Date();30 const runId = newId("run");31 const connector = getConnector(sensor.connector);32 const endpoint: SensorEndpoint = { id: sensor.id, sourceId: sensor.sourceId, name: sensor.name, url: sensor.url, type: sensor.type as SensorEndpoint["type"], tier: sensor.tier as Tier, connector: sensor.connector, config: sensor.config, etag: sensor.etag, lastModified: sensor.lastModified, state: sensor.state ?? null };33 const stop = m.fetchDuration.startTimer({ connector: sensor.connector });34 let obs: Observation;35 try {36 obs = await connector.fetch(endpoint);37 } catch (e) {38 obs = { sensorId: sensor.id, url: sensor.url, fetchedAt: new Date(), notModified: false, error: { code: "connector_threw", message: (e as Error).message }, meta: { status: 0, url: sensor.url, finalUrl: sensor.url, contentType: null, contentLength: 0, etag: null, lastModified: null, durationMs: Date.now() - started.getTime(), redirects: 0, method: "GET", headers: {} } };39 }40 stop();41 // Anti-bot response on a source that permits the Scrapfly fallback (acquisition step 14).42 if (!obs.error && [403, 429, 503].includes(obs.meta.status) && (source.fallback as { scrapfly?: boolean }).scrapfly && scrapflyAvailable() && ["http", "rss", "sitemap"].includes(sensor.connector)) {43 const via = await scrapflyFetch(sensor.id, sensor.url, { renderJs: Boolean((sensor.config as { renderJs?: boolean }).renderJs) });44 log.info({ sensor: sensor.id, direct: obs.meta.status, via: via.error ? via.error.code : via.meta.status }, "scrapfly fallback");45 if (!via.error && via.meta.status < 400 && via.body) {46 obs = via;47 await bumpDaily({ scrapfly_calls: 1 });48 }49 }50 m.bytes.inc({ connector: sensor.connector }, obs.meta.contentLength);51 if (obs.meta.status) m.httpStatus.inc({ status: String(obs.meta.status) });5253 const finish = async (outcome: RunOutcome, extra: { error?: string; snapshotId?: string; changed?: boolean; eventAt?: Date; rateLimitedUntil?: Date | null } = {}): Promise<RunOutcome> => {54 const isErr = outcome === "error" || outcome === "parse_error" || outcome === "rate_limited";55 const consecutive = isErr ? sensor.consecutiveErrors + 1 : 0;56 const health = outcome === "rate_limited" ? "RATE_LIMITED" : isErr ? (consecutive >= 5 ? "ERROR" : "DEGRADED") : outcome === "missing" ? "DEGRADED" : "UP";57 // Shadow sensors (Source Factory) keep their lifecycle status until evaluateShadows() decides.58 const status = !sensor.enabled ? "DISABLED" : sensor.status === "SHADOW" ? "SHADOW" : health === "ERROR" || health === "RATE_LIMITED" ? "DEGRADED" : "ACTIVE";59 const changes7d = await countChanges7d(sensor.id);60 const events7d = await countEvents7d(sensor.id);61 let interval = nextIntervalSeconds({ tier: sensor.tier as Tier, baseIntervalSeconds: sensor.baseIntervalSeconds, lastChangeAt: extra.changed ? new Date() : sensor.lastChangeAt, changes7d, events7d, consecutiveErrors: consecutive, lastWas304: outcome === "not_modified" });62 if (extra.rateLimitedUntil) interval = Math.max(interval, Math.ceil((extra.rateLimitedUntil.getTime() - Date.now()) / 1000));63 const avg = sensor.avgLatencyMs ? Math.round(sensor.avgLatencyMs * 0.8 + obs.meta.durationMs * 0.2) : obs.meta.durationMs;64 await db.insert(sensorRuns).values({ id: runId, sensorId: sensor.id, startedAt: started, finishedAt: new Date(), httpStatus: obs.meta.status || null, outcome, error: extra.error?.slice(0, 500) ?? obs.error?.message.slice(0, 500) ?? null, durationMs: obs.meta.durationMs, bytes: obs.meta.contentLength, fetchMethod: obs.meta.method, snapshotId: extra.snapshotId ?? null });65 await db66 .update(sensors)67 .set({68 lastCheckAt: new Date(),69 lastStatus: obs.meta.status || null,70 lastError: isErr || outcome === "missing" ? (extra.error ?? obs.error?.message ?? `HTTP ${obs.meta.status}`).slice(0, 500) : null,71 consecutiveErrors: consecutive,72 health,73 status,74 totalRuns: sensor.totalRuns + 1,75 totalNotModified: sensor.totalNotModified + (outcome === "not_modified" ? 1 : 0),76 avgLatencyMs: avg,77 nextCheckAt: new Date(Date.now() + interval * 1000),78 updatedAt: new Date(),79 ...(extra.changed ? { lastChangeAt: new Date() } : {}),80 ...(extra.eventAt ? { lastEventAt: extra.eventAt } : {}),81 ...(!sensor.validatedAt && !isErr && outcome !== "missing" ? { validatedAt: new Date() } : {}),82 ...(obs.meta.status >= 200 && obs.meta.status < 300 ? { etag: obs.meta.etag ?? sensor.etag, lastModified: obs.meta.lastModified ?? sensor.lastModified } : {}),83 })84 .where(sql`id = ${sensor.id}`);85 m.checks.inc({ connector: sensor.connector, outcome });86 await bumpDaily({ checks: 1, bytes: obs.meta.contentLength, not_modified: outcome === "not_modified" ? 1 : 0, errors: isErr ? 1 : 0 });87 await bumpSourceDaily(source.id, { checks: 1, not_modified: outcome === "not_modified" ? 1 : 0, errors: isErr ? 1 : 0 });88 return outcome;89 };9091 // ---- Transport-level outcomes -------------------------------------------------------92 if (obs.error) return finish("error", { error: `${obs.error.code}: ${obs.error.message}` });93 if (obs.notModified) return finish("not_modified");94 const st = obs.meta.status;95 if (st === 429) {96 const ra = Number(obs.meta.headers["retry-after"] ?? 0);97 return finish("rate_limited", { error: "HTTP 429", rateLimitedUntil: new Date(Date.now() + Math.min(6 * 3600e3, (ra > 0 ? ra : 900) * 1000)) });98 }99 if (st === 404 || st === 410) return handleMissing(sensor, source, obs, finish);100 if (st >= 400) return finish("error", { error: `HTTP ${st}` });101 if (!obs.body && obs.meta.method !== "HEAD") return finish("error", { error: "empty body" });102103 // ---- Normalize --------------------------------------------------------------------104 let norm: NormalizedContent;105 const tNorm = m.stageLatency.startTimer({ stage: "normalize" });106 try {107 norm = await connector.normalize(endpoint, obs);108 } catch (e) {109 tNorm();110 const msg = e instanceof NormalizeError ? `${e.code}: ${e.message}` : (e as Error).message;111 return finish("parse_error", { error: msg });112 }113 tNorm();114 if (norm.state) await db.update(sensors).set({ state: norm.state }).where(sql`id = ${sensor.id}`);115116 // Restore: page was flagged missing but is back.117 const missingState = (sensor.state as { missing?: { count: number; removedEventAt?: string } } | null)?.missing;118 if (missingState) await db.update(sensors).set({ state: { ...(norm.state ?? sensor.state ?? {}), missing: null } }).where(sql`id = ${sensor.id}`);119120 // ---- Compare with previous snapshot -------------------------------------------------121 const prev = sensor.lastSnapshotId ? (await db.select().from(snapshots).where(sql`id = ${sensor.lastSnapshotId}`))[0] : undefined;122 if (prev && prev.canonicalHash === norm.canonicalHash) {123 await touchUrl(sensor, source, false);124 return finish("unchanged");125 }126127 // Store snapshot (raw + canonical) — evidence is immutable.128 const store = getBlobStore();129 const raw = obs.body ? await store.put(obs.body) : null;130 const canonicalBlob: CanonicalBlob = { mode: norm.mode, text: norm.text, json: norm.json, items: norm.items, compareFields: norm.compareFields, title: norm.title ?? null, headings: norm.headings, extractionConfidence: norm.extractionConfidence };131 const canonical = await store.put(JSON.stringify(canonicalBlob));132 const snapId = newId("snap");133 await db.insert(snapshots).values({134 id: snapId,135 sensorId: sensor.id,136 url: obs.meta.finalUrl || sensor.url,137 capturedAt: obs.fetchedAt,138 httpStatus: st,139 contentType: obs.meta.contentType,140 contentLength: obs.meta.contentLength,141 contentHash: norm.rawHash,142 canonicalHash: norm.canonicalHash,143 semanticHash: norm.semanticHash,144 etag: obs.meta.etag,145 lastModified: obs.meta.lastModified,146 storageKey: raw?.key ?? null,147 canonicalStorageKey: canonical.key,148 parserVersion: PARSER_VERSION,149 fetchDurationMs: obs.meta.durationMs,150 fetchMethod: obs.meta.method,151 mode: norm.mode,152 title: norm.title ?? null,153 publishedAt: norm.publishedAt ?? null,154 extractionConfidence: norm.extractionConfidence,155 extra: norm.extra ?? null,156 });157 await db.update(sensors).set({ lastSnapshotId: snapId }).where(sql`id = ${sensor.id}`);158 await touchUrl(sensor, source, true, snapId);159160 if (!prev) return finish("baseline", { snapshotId: snapId });161162 // ---- Diff -----------------------------------------------------------------------------163 const tDiff = m.stageLatency.startTimer({ stage: "diff" });164 let prevBlob: CanonicalBlob | null = null;165 try {166 prevBlob = prev.canonicalStorageKey ? (JSON.parse(await store.getText(prev.canonicalStorageKey)) as CanonicalBlob) : null;167 } catch (e) {168 log.warn({ sensor: sensor.id, err: (e as Error).message }, "previous canonical blob unreadable");169 }170 if (!prevBlob || prevBlob.mode !== norm.mode) {171 tDiff();172 return finish("changed", { snapshotId: snapId, changed: true });173 }174175 let diff: DiffResult;176 if (norm.mode === "list") {177 const seen = new Set<string>(Array.isArray((sensor.state as { seenKeys?: string[] } | null)?.seenKeys) ? ((sensor.state as { seenKeys: string[] }).seenKeys ?? []) : []);178 const prevItems = prevBlob.items ?? [];179 const curItems = norm.items ?? [];180 diff = diffList(prevItems, curItems, norm.compareFields ?? []);181 // Feeds: items scrolling out of the window are not removals. Anything seen before is not "new".182 if (sensor.type === "RSS" || sensor.type === "ATOM" || sensor.type === "GITHUB_RELEASE" || sensor.type === "REST_API" || sensor.type === "JSON") {183 const staleBefore = Date.now() - 14 * 86400e3;184 diff = {185 ...diff,186 removed: [],187 // never seen before AND not an old item resurfacing at the feed window boundary (backfills, reordering)188 added: diff.added.filter((i) => !seen.has(i.key) && !(typeof i.publishedAt === "string" && i.publishedAt && new Date(i.publishedAt).getTime() < staleBefore)),189 };190 }191 // Statuspage summaries only list active incidents/maintenances: an item leaving the list is a resolution, not a removal.192 if (sensor.type === "STATUSPAGE") diff = { ...diff, removed: [] };193 // Sitemaps/statuspages: a sudden > 50 % shrink is more likely a partial response than mass deletion.194 if (prevItems.length >= 20 && curItems.length < prevItems.length * 0.5) diff = { ...diff, removed: [] };195 } else if (norm.mode === "json") {196 diff = diffJson(prevBlob.json, norm.json);197 } else {198 diff = diffText(prevBlob.text ?? "", norm.text ?? "", `${sensor.id}@${prev.capturedAt.toISOString()}`, `${sensor.id}@${obs.fetchedAt.toISOString()}`);199 }200 tDiff();201 if (diffIsEmpty(diff)) return finish("unchanged", { snapshotId: snapId });202203 // ---- Heuristics + semantic classification ------------------------------------------------204 const tCls = m.stageLatency.startTimer({ stage: "classify" });205 const heuristic = evaluateChange(diff, { sensorType: sensor.type, url: sensor.url, sourceCategories: source.categories, title: norm.title });206 const semantic = classifyChange(diff, { url: sensor.url, sensorType: sensor.type, title: norm.title });207 tCls();208 m.changeClass.inc({ class: semantic.class });209 const thinFlip = (prevBlob.extractionConfidence ?? 1) < 0.5 || norm.extractionConfidence < 0.5;210 const isNoise = NOISE_CLASSES.has(semantic.class) && !heuristic.facts.some((f) => f.kind === "price" && f.before && f.after);211 const meaningfulByRules = heuristic.signal >= config.meaningfulSignal && !thinFlip && !isNoise;212 const changeId = newId("chg");213 const diffBlob = await store.put(diff.kind === "text" ? diff.unified : renderDiff(diff));214 await db.insert(changes).values({ id: changeId, sensorId: sensor.id, oldSnapshotId: prev.id, newSnapshotId: snapId, detectedAt: obs.fetchedAt, kind: diff.kind, diff: summarizeDiff(diff), diffStorageKey: diffBlob.key, signal: heuristic.signal, noiseRatio: Math.max(heuristic.noiseRatio, semantic.noiseRatio), magnitude: heuristic.magnitude, heuristic: { ...(heuristic as unknown as Record<string, unknown>), semantic: { class: semantic.class, confidence: semantic.confidence, reasons: semantic.reasons } }, meaningful: false, changeClass: semantic.class, fieldChanges: semantic.fieldChanges.length ? semantic.fieldChanges : null });215 await db.update(sensors).set({ rawChanges: sensor.rawChanges + 1 }).where(sql`id = ${sensor.id}`);216 await db.execute(sql`insert into url_history (url, at, kind, snapshot_id, change_id) values (${sensor.url}, ${obs.fetchedAt}, 'change', ${snapId}, ${changeId})`);217 await db.execute(sql`update urls set change_count = change_count + 1 where url = ${sensor.url}`);218 await bumpDaily({ raw_changes: 1 });219 await bumpSourceDaily(source.id, { raw_changes: 1 });220 m.changes.inc({ connector: sensor.connector, meaningful: String(meaningfulByRules) });221 await publishChange({ id: changeId, sensorId: sensor.id, sourceId: source.id, kind: diff.kind, signal: heuristic.signal, class: semantic.class, at: obs.fetchedAt.toISOString() });222 log.info({ sensor: sensor.id, kind: diff.kind, signal: heuristic.signal.toFixed(2), type: heuristic.eventType, class: semantic.class, thin: thinFlip }, "change detected");223224 if (!meaningfulByRules) {225 if (isNoise) m.suppressed.inc({ reason: `noise_${semantic.class}` });226 return finish("changed", { snapshotId: snapId, changed: true });227 }228 // Shadow monitoring (Source Factory): evidence and changes are recorded, nothing is published until the sensor is accepted.229 if (sensor.status === "SHADOW") {230 m.suppressed.inc({ reason: "shadow" });231 await db.update(changes).set({ meaningful: true }).where(sql`id = ${changeId}`);232 return finish("changed", { snapshotId: snapId, changed: true });233 }234235 // ---- Event --------------------------------------------------------------------------236 const ev = await createEvent({ sensor, source, obs, norm, prev, snapId, changeId, diff, heuristic, semantic });237 if (!ev) return finish("changed", { snapshotId: snapId, changed: true });238 return finish("event", { snapshotId: snapId, changed: true, eventAt: ev.detectedAt });239}240241async function createEvent(ctx: { sensor: Sensor; source: Source; obs: Observation; norm: NormalizedContent; prev: { id: string; capturedAt: Date; canonicalHash: string }; snapId: string; changeId: string; diff: DiffResult; heuristic: HeuristicResult; semantic: SemanticResult }): Promise<{ id: string; detectedAt: Date } | null> {242 const { sensor, source, obs, norm, prev, snapId, changeId, diff, heuristic, semantic } = ctx;243 const detectedAt = obs.fetchedAt;244 const tEv = m.stageLatency.startTimer({ stage: "event" });245246 // Idempotency (spec §90): the same observation (same sensor, same before/after canonical content) is one event.247 const fingerprint = sha256(`${sensor.id}|${prev.canonicalHash}|${norm.canonicalHash}`);248 const dup = await db.execute<{ id: string }>(sql`select id from events where fingerprint = ${fingerprint} limit 1`);249 if (dup.rows[0]) {250 m.suppressed.inc({ reason: "fingerprint_duplicate" });251 log.info({ sensor: sensor.id, existing: dup.rows[0].id }, "duplicate observation — event already exists");252 tEv();253 return null;254 }255256 const base = describeChange(heuristic, diff, { sourceName: source.name, url: sensor.url, sensorName: sensor.name });257 if (semantic.fieldChanges.length && diff.kind !== "list" && !heuristic.facts.some((f) => f.kind === "price" && f.before && f.after)) {258 base.summary = `${describeFieldChanges(semantic.fieldChanges, 4)}. ${base.summary}`.slice(0, 1500);259 }260 const textForMatching = `${base.title}\n${base.summary}\n${renderDiff(diff).slice(0, 4000)}`;261262 const ent = await resolveEntities({ sourceId: source.id, text: textForMatching, hints: heuristic.keywords });263 const nov = await assessNovelty(`${base.title}\n${base.summary}`, source.id);264 const sourceImportance = sourceImportanceFromTier(sensor.tier, source.importanceWeight * sensor.importanceWeight);265 const firstParty = source.firstParty !== false;266 const anomaly = await sourceAnomaly(source.id);267 let eventType = heuristic.eventType;268 let prelim = computeImportance({ eventType, sourceImportance, entityImportance: ent.importance, novelty: nov.novelty, magnitude: heuristic.magnitude, confirmations: nov.confirmations, unusualness: anomaly });269 // Routine batches (spec §13, §55): "21 new CVEs" from a firehose that fires several times a day is normal270 // operation, not a signal. Damp importance for high-volume list sensors; single items keep full weight.271 const routineBatch = diff.kind === "list" && diff.added.length >= 8 && (await countEvents24h(sensor.id)) >= 3;272 if (routineBatch) {273 prelim.score = Math.round(prelim.score * 0.72 * 10) / 10;274 prelim.components.novelty = Math.min(prelim.components.novelty, 35);275 }276277 // Duplicate suppression: near-identical to something we already published (syndication / re-fetch).278 if (nov.novelty < 12 && nov.nearest) {279 m.suppressed.inc({ reason: "near_duplicate" });280 log.info({ sensor: sensor.id, nearest: nov.nearest.id, sim: nov.nearest.similarity }, "suppressed near-duplicate event");281 tEv();282 return null;283 }284 // Third-party reports of a story we already hold from several sources add little: raise the bar.285 if (!firstParty && nov.novelty < 30 && nov.confirmations >= 3 && prelim.score < 70) {286 m.suppressed.inc({ reason: "redundant_external" });287 tEv();288 return null;289 }290291 let llm: Interpretation | null = null;292 if (llmAvailable() && source.llmEnabled && prelim.score >= config.llm.minImportance) {293 llm = await interpretChange({ sourceName: source.name, sourceCategories: source.categories, url: sensor.url, sensorName: sensor.name, sensorType: sensor.type, heuristic, diff, prelimImportance: prelim.score, title: norm.title });294 if (llm && !llm.meaningful) {295 m.suppressed.inc({ reason: "llm_not_meaningful" });296 log.info({ sensor: sensor.id, type: llm.event_type }, "LLM judged change not meaningful");297 await db.update(changes).set({ heuristic: { ...(heuristic as unknown as Record<string, unknown>), semantic: { class: semantic.class, confidence: semantic.confidence }, llm: { meaningful: false, model: llm.model, title: llm.title } } }).where(sql`id = ${changeId}`);298 tEv();299 return null;300 }301 if (llm) {302 eventType = llm.event_type;303 const spec = eventTypeSpec(eventType);304 prelim = computeImportance({ eventType, sourceImportance, entityImportance: ent.importance, novelty: nov.novelty, magnitude: heuristic.magnitude, confirmations: nov.confirmations, unusualness: anomaly });305 prelim.components.severity = Math.round((spec.severity + llm.severity) / 2);306 prelim.score = Math.round(10 * (0.25 * prelim.components.severity + 0.2 * prelim.components.source + 0.15 * prelim.components.entity + 0.15 * prelim.components.novelty + 0.1 * prelim.components.magnitude + 0.05 * prelim.components.confirmation + 0.05 * prelim.components.userImpact + 0.05 * prelim.components.unusualness)) / 10;307 }308 }309310 const title = (llm?.title ?? base.title).slice(0, 180);311 const summary = (llm?.summary ?? base.summary).slice(0, 1500);312 const spec = eventTypeSpec(eventType);313 const isAnnouncementSensor = ANNOUNCEMENT_SENSORS.has(sensor.type);314 const announcedSim = isAnnouncementSensor ? 1 : recentAnnouncementSimilarity(source.id, `${title}\n${summary}`, 12 * 3600e3);315 // Silent-change bar (spec §23): first-party, silent-eligible type, no matching announcement, real content, importance ≥ bar.316 const silentChange = firstParty && !isAnnouncementSensor && SILENT_ELIGIBLE_TYPES.has(eventType) && (llm ? llm.announced === false : !spec.usuallyAnnounced) && announcedSim < 0.3 && prelim.score >= config.silentMinImportance && !NOISE_CLASSES.has(semantic.class);317318 // Entities named by the LLM that resolve to known aliases319 const extra = llm ? await resolveEntities({ sourceId: source.id, text: llm.entities.join("\n"), hints: [] }) : null;320 const entityIds = [...new Set([...ent.subject, ...ent.mentioned, ...(extra?.mentioned ?? [])])];321322 const confidence = computeConfidence({323 sourceAuthenticity: firstParty ? 1 : 0.6,324 extraction: norm.extractionConfidence,325 diffClarity: 1 - Math.max(heuristic.noiseRatio, semantic.noiseRatio),326 structured: diff.kind !== "text",327 confirmations: nov.confirmations,328 llmAgreement: llm ? (llm.event_type === heuristic.eventType ? 1 : 0.6) * (llm.confidence / 100) : heuristic.signal,329 });330 const evidenceLabel = nov.confirmations > 0 ? "CONFIRMED" : llm && llm.inferred && llm.confidence < 60 ? "INFERRED" : !llm && heuristic.signal < 0.5 ? "UNCONFIRMED" : "OBSERVED";331 const categories = [...new Set([...source.categories, ...Object.entries(FEED_CHANNELS).filter(([, cats]) => cats.some((c) => source.categories.includes(c))).map(([ch]) => ch)])];332 const keywords = [...new Set([...(llm?.keywords ?? []), ...heuristic.keywords])].slice(0, 20);333 const maxDelta = semantic.fieldChanges.reduce((mx, f) => Math.max(mx, Math.abs(f.deltaPct ?? 0)), 0);334 const impact = computeImpact({ eventType, magnitude: heuristic.magnitude, entityImportance: ent.importance, maxDeltaPct: maxDelta, fieldChanges: semantic.fieldChanges.length, firstParty });335 const id = newId("evt");336 const slug = `${slugify(title).slice(0, 70)}-${id.slice(-6)}`;337 const processedAt = new Date();338 const publishedAt = pickPublishedAt(diff, norm);339 const observedFrom = prev.capturedAt;340 const country = source.country ?? null;341 const language = source.language ?? "en";342343 await db.insert(events).values({344 id,345 slug,346 sensorId: sensor.id,347 sourceId: source.id,348 changeId,349 oldSnapshotId: prev.id,350 newSnapshotId: snapId,351 url: sensor.url,352 canonicalUrl: obs.meta.finalUrl || sensor.url,353 eventType,354 title,355 summary,356 whyItMatters: llm?.why_it_matters ?? null,357 importance: prelim.score,358 importanceComponents: prelim.components as unknown as Record<string, number>,359 confidence,360 novelty: nov.novelty,361 impactScore: impact,362 anomalyScore: anomaly,363 categories,364 keywords,365 silentChange,366 evidenceLabel,367 firstParty,368 country,369 language,370 changeClass: semantic.class,371 fieldChanges: semantic.fieldChanges.length ? semantic.fieldChanges : null,372 fingerprint,373 publishedAt,374 observedFrom,375 detectedAt,376 processedAt,377 detectionLatencyMs: publishedAt && detectedAt.getTime() - publishedAt.getTime() < 7 * 86400e3 ? Math.max(0, detectedAt.getTime() - publishedAt.getTime()) : null,378 processingLatencyMs: processedAt.getTime() - detectedAt.getTime(),379 processingVersion: config.processingVersion,380 interpretation: llm ? { model: llm.model, observed: llm.observed, inferred: llm.inferred, who_it_affects: llm.who_it_affects, severity: llm.severity, confidence: llm.confidence, announced: llm.announced, entities: llm.entities } : { model: "heuristics-v2", reasons: [...heuristic.reasons, ...semantic.reasons], facts: heuristic.facts },381 });382 await db.insert(interpretations).values({ eventId: id, version: 1, model: llm?.model ?? "heuristics-v2", payload: { heuristic: heuristic as unknown as Record<string, unknown>, semantic: { class: semantic.class, confidence: semantic.confidence, reasons: semantic.reasons, fieldChanges: semantic.fieldChanges }, llm: llm as unknown as Record<string, unknown> | null } });383 for (const e of entityIds) await db.insert(eventEntities).values({ eventId: id, entityId: e, role: ent.subject.includes(e) ? "subject" : "mentioned" }).onConflictDoNothing();384 await bumpEntityCounters(entityIds, detectedAt, { silent: silentChange, importance: prelim.score });385 await db.update(changes).set({ meaningful: true, eventId: id }).where(sql`id = ${changeId}`);386 await db.update(sensors).set({ meaningfulChanges: sensor.meaningfulChanges + 1 }).where(sql`id = ${sensor.id}`);387 await db.execute(sql`insert into url_history (url, at, kind, snapshot_id, change_id, event_id) values (${sensor.url}, ${detectedAt}, 'event', ${snapId}, ${changeId}, ${id})`);388389 // Provisional signal for clustering; recomputed with the cluster's velocity right after.390 const provisional = computeSignalScore({ importance: prelim.score, confidence, novelty: nov.novelty, velocity: 0, impact, anomaly, confirmations: nov.confirmations, firstParty, silent: silentChange, changeClass: semantic.class, sourceTier: sensor.tier, evidenceLabel });391 const cl = await clusterEvent({ id, sourceId: source.id, sourceName: source.name, sensorId: sensor.id, sensorType: sensor.type, eventType, entityIds, detectedAt, title, summary, importance: prelim.score, signal: provisional.score, categories, firstParty, sourceTier: sensor.tier });392 const signal = computeSignalScore({ importance: prelim.score, confidence, novelty: nov.novelty, velocity: cl.velocity, impact, anomaly, confirmations: Math.max(nov.confirmations, cl.sourceCount - 1), firstParty, silent: silentChange, changeClass: semantic.class, sourceTier: sensor.tier, evidenceLabel });393 if (routineBatch) signal.reasons.push({ sign: "-", text: "routine batch from a high-volume feed", points: -8 });394 await db.update(events).set({ clusterId: cl.clusterId, publishedToFeedAt: new Date(), signalScore: signal.score, velocityScore: cl.velocity, scoreReasons: signal.reasons }).where(sql`id = ${id}`);395 await bumpSourceDaily(source.id, { events: 1 });396397 const entNames = entityIds.length ? await db.execute<{ id: string; name: string; type: string }>(sql`select id, name, type from entities where id = any(${textArray(entityIds)})`) : { rows: [] as { id: string; name: string; type: string }[] };398 const payload: PublishedEvent & Record<string, unknown> = {399 id,400 slug,401 type: eventType,402 group: eventGroupOf(eventType),403 title,404 summary: summary.slice(0, 280),405 importance: prelim.score,406 signal: signal.score,407 confidence,408 novelty: nov.novelty,409 impact,410 velocity: cl.velocity,411 silent: silentChange,412 evidence: evidenceLabel,413 firstParty,414 country,415 language,416 changeClass: semantic.class,417 fieldChanges: semantic.fieldChanges.slice(0, 3),418 source: { id: source.id, name: source.name, domain: source.domain, tier: source.tier },419 sensor: { id: sensor.id, name: sensor.name, type: sensor.type },420 entities: entNames.rows.map((r) => ({ id: r.id, name: r.name, type: r.type })),421 categories,422 url: sensor.url,423 clusterId: cl.clusterId,424 clusterSlug: cl.slug,425 clusterSize: cl.eventCount,426 clusterState: cl.state,427 detectedAt: detectedAt.toISOString(),428 publishedAt: publishedAt?.toISOString() ?? null,429 };430 await publishEvent(payload);431 void evaluateAlerts(payload).catch(() => undefined);432 m.events.inc({ event_type: eventType, silent: String(silentChange) });433 m.processingLatency.observe((Date.now() - detectedAt.getTime()) / 1000);434 await bumpDaily({ events: 1, silent_events: silentChange ? 1 : 0 });435 tEv();436 log.info({ event: id, source: source.id, type: eventType, importance: prelim.score, signal: signal.score, confidence, silent: silentChange, class: semantic.class, cluster: cl.state, llm: llm?.model ?? "heuristics" }, title);437 return { id, detectedAt };438}439440function pickPublishedAt(diff: DiffResult, norm: NormalizedContent): Date | null {441 if (diff.kind === "list") {442 const dates = diff.added.map((i) => i.publishedAt).filter((x): x is string => typeof x === "string" && x.length > 0).map((s) => new Date(s)).filter((d) => !Number.isNaN(d.getTime()));443 if (dates.length) return new Date(Math.max(...dates.map((d) => d.getTime())));444 }445 return norm.publishedAt ?? null;446}447448async function handleMissing(sensor: Sensor, source: Source, obs: Observation, finish: (o: RunOutcome, e?: { error?: string }) => Promise<RunOutcome>): Promise<RunOutcome> {449 const state = (sensor.state ?? {}) as Record<string, unknown>;450 const missing = (state.missing ?? { count: 0, firstAt: obs.fetchedAt.toISOString(), lastAt: null }) as { count: number; firstAt: string; lastAt: string | null; removedEventAt?: string };451 const sepOk = !missing.lastAt || obs.fetchedAt.getTime() - new Date(missing.lastAt).getTime() >= config.deletion.minSeparationMin * 60e3;452 if (sepOk) missing.count += 1;453 missing.lastAt = obs.fetchedAt.toISOString();454 if (sensor.lastSnapshotId && missing.count >= config.deletion.confirmations && !missing.removedEventAt && sensor.status !== "SHADOW") {455 // Confirmed deletion: independent checks separated in time.456 missing.removedEventAt = obs.fetchedAt.toISOString();457 const id = newId("evt");458 const title = `${source.name}: page removed — ${sensor.name}`;459 const summary = `${sensor.url} has returned HTTP ${obs.meta.status} on ${missing.count} checks since ${new Date(missing.firstAt).toISOString()}. The last known version is preserved as a snapshot.`;460 const imp = computeImportance({ eventType: "page_removed", sourceImportance: sourceImportanceFromTier(sensor.tier, source.importanceWeight), entityImportance: 50, novelty: 80, magnitude: 60, confirmations: 0 });461 const slug = `${slugify(title).slice(0, 70)}-${id.slice(-6)}`;462 const firstParty = source.firstParty !== false;463 const fingerprint = sha256(`${sensor.id}|removed|${sensor.lastSnapshotId}`);464 const impact = computeImpact({ eventType: "page_removed", magnitude: 60, entityImportance: 50, firstParty });465 const sig = computeSignalScore({ importance: imp.score, confidence: 85, novelty: 80, velocity: 0, impact, anomaly: 0, confirmations: 0, firstParty, silent: true, sourceTier: sensor.tier, evidenceLabel: "CONFIRMED" });466 await db.insert(events).values({ id, slug, sensorId: sensor.id, sourceId: source.id, oldSnapshotId: sensor.lastSnapshotId, url: sensor.url, eventType: "page_removed", title, summary, importance: imp.score, importanceComponents: imp.components as unknown as Record<string, number>, confidence: 85, novelty: 80, impactScore: impact, signalScore: sig.score, scoreReasons: sig.reasons, categories: source.categories, keywords: ["page removed"], silentChange: true, evidenceLabel: "CONFIRMED", firstParty, country: source.country ?? null, language: source.language ?? "en", changeClass: "meaningful", fingerprint, detectedAt: obs.fetchedAt, processedAt: new Date(), processingVersion: config.processingVersion, interpretation: { model: "rules", checks: missing.count } }).onConflictDoNothing();467 const ent = await resolveEntities({ sourceId: source.id, text: title, hints: [] });468 for (const e of ent.subject) await db.insert(eventEntities).values({ eventId: id, entityId: e, role: "subject" }).onConflictDoNothing();469 await bumpEntityCounters(ent.subject, obs.fetchedAt, { silent: true, importance: imp.score });470 await db.execute(sql`update urls set status = 'removed' where url = ${sensor.url}`);471 await db.execute(sql`insert into url_history (url, at, kind, event_id, note) values (${sensor.url}, ${obs.fetchedAt}, 'removed', ${id}, ${`HTTP ${obs.meta.status}`})`);472 const cl = await clusterEvent({ id, sourceId: source.id, sourceName: source.name, sensorId: sensor.id, sensorType: sensor.type, eventType: "page_removed", entityIds: ent.subject, detectedAt: obs.fetchedAt, title, summary, importance: imp.score, signal: sig.score, categories: source.categories, firstParty, sourceTier: sensor.tier });473 await db.update(events).set({ clusterId: cl.clusterId, publishedToFeedAt: new Date(), velocityScore: cl.velocity }).where(sql`id = ${id}`);474 const payload: PublishedEvent & Record<string, unknown> = { id, slug, type: "page_removed", group: "web", title, summary, importance: imp.score, signal: sig.score, confidence: 85, novelty: 80, silent: true, evidence: "CONFIRMED", firstParty, country: source.country ?? null, source: { id: source.id, name: source.name, domain: source.domain, tier: source.tier }, sensor: { id: sensor.id, name: sensor.name, type: sensor.type }, entities: [], categories: source.categories, url: sensor.url, clusterId: cl.clusterId, clusterSlug: cl.slug, clusterState: cl.state, detectedAt: obs.fetchedAt.toISOString(), publishedAt: null };475 await publishEvent(payload);476 void evaluateAlerts(payload).catch(() => undefined);477 await bumpDaily({ events: 1, silent_events: 1 });478 log.info({ sensor: sensor.id, checks: missing.count }, "page removal confirmed");479 } else {480 await db.execute(sql`update urls set missing_count = missing_count + 1, status = case when status = 'active' then 'pending_removal' else status end where url = ${sensor.url}`);481 }482 await db.update(sensors).set({ state: { ...state, missing } }).where(sql`id = ${sensor.id}`);483 return finish("missing", { error: `HTTP ${obs.meta.status} (${missing.count}/${config.deletion.confirmations} confirmations)` });484}485486async function touchUrl(sensor: Sensor, source: Source, snapshot: boolean, snapshotId?: string): Promise<void> {487 const domain = safeHost(sensor.url);488 await db.execute(sql`insert into urls (url, domain, source_id, sensor_id, first_seen_at, last_seen_at, status, snapshot_count)489 values (${sensor.url}, ${domain}, ${source.id}, ${sensor.id}, now(), now(), 'active', ${snapshot ? 1 : 0})490 on conflict (url) do update set last_seen_at = now(), status = 'active', missing_count = 0, snapshot_count = urls.snapshot_count + ${snapshot ? 1 : 0}, sensor_id = ${sensor.id}`);491 if (snapshot && snapshotId) await db.execute(sql`insert into url_history (url, at, kind, snapshot_id) values (${sensor.url}, now(), 'snapshot', ${snapshotId})`);492}493494function safeHost(u: string): string {495 try {496 return new URL(u).hostname;497 } catch {498 return "";499 }500}501502async function countChanges7d(sensorId: string): Promise<number> {503 const r = await db.execute<{ n: string }>(sql`select count(*)::text as n from changes where sensor_id = ${sensorId} and detected_at >= now() - interval '7 days'`);504 return Number(r.rows[0]?.n ?? 0);505}506async function countEvents24h(sensorId: string): Promise<number> {507 const r = await db.execute<{ n: string }>(sql`select count(*)::text as n from events where sensor_id = ${sensorId} and detected_at >= now() - interval '24 hours'`);508 return Number(r.rows[0]?.n ?? 0);509}510async function countEvents7d(sensorId: string): Promise<number> {511 const r = await db.execute<{ n: string }>(sql`select count(*)::text as n from events where sensor_id = ${sensorId} and detected_at >= now() - interval '7 days'`);512 return Number(r.rows[0]?.n ?? 0);513}514515/** Source activity anomaly at detection time (0–100): last 2 h of raw changes vs the 14-day hourly baseline. */516async function sourceAnomaly(sourceId: string): Promise<number> {517 const r = await db.execute<{ c2h: string; c14d: string }>(sql`518 select (select count(*) from changes c join sensors s on s.id = c.sensor_id where s.source_id = ${sourceId} and c.detected_at >= now() - interval '2 hours')::text as c2h,519 (select count(*) from changes c join sensors s on s.id = c.sensor_id where s.source_id = ${sourceId} and c.detected_at >= now() - interval '14 days')::text as c14d`);520 const cur = Number(r.rows[0]?.c2h ?? 0) / 2;521 const base = Number(r.rows[0]?.c14d ?? 0) / (14 * 24);522 return activityAnomaly(cur, base);523}524