import { 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";
import { getConnector, NormalizeError, PARSER_VERSION, scrapflyAvailable, scrapflyFetch } from "@websensor/connectors";
import { changes, db, eventEntities, events, interpretations, sensorRuns, sensors, snapshots, sql, textArray, type Sensor, type Source } from "@websensor/db";
import { getBlobStore } from "@websensor/store";
import { evaluateAlerts, type PublishedEvent } from "./alerts";
import { assessNovelty, clusterEvent, recentAnnouncementSimilarity } from "./cluster";
import { config, log } from "./config";
import { bumpEntityCounters, resolveEntities } from "./entities";
import { interpretChange, llmAvailable, renderDiff, type Interpretation } from "./interpret";
import { bumpDaily, bumpSourceDaily, m } from "./metrics";
import { publishChange, publishEvent } from "./redis";

export type RunOutcome = "baseline" | "unchanged" | "not_modified" | "changed" | "event" | "error" | "missing" | "rate_limited" | "parse_error";

interface CanonicalBlob {
  mode: NormalizedContent["mode"];
  text?: string;
  json?: unknown;
  items?: { key: string; [k: string]: unknown }[];
  compareFields?: string[];
  title?: string | null;
  headings?: string[];
  extractionConfidence: number;
}

const ANNOUNCEMENT_SENSORS = new Set(["RSS", "ATOM", "STATUSPAGE", "GITHUB_RELEASE", "REST_API"]);

export async function runSensor(sensor: Sensor, source: Source): Promise<RunOutcome> {
  const started = new Date();
  const runId = newId("run");
  const connector = getConnector(sensor.connector);
  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 };
  const stop = m.fetchDuration.startTimer({ connector: sensor.connector });
  let obs: Observation;
  try {
    obs = await connector.fetch(endpoint);
  } catch (e) {
    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: {} } };
  }
  stop();
  // Anti-bot response on a source that permits the Scrapfly fallback (acquisition step 14).
  if (!obs.error && [403, 429, 503].includes(obs.meta.status) && (source.fallback as { scrapfly?: boolean }).scrapfly && scrapflyAvailable() && ["http", "rss", "sitemap"].includes(sensor.connector)) {
    const via = await scrapflyFetch(sensor.id, sensor.url, { renderJs: Boolean((sensor.config as { renderJs?: boolean }).renderJs) });
    log.info({ sensor: sensor.id, direct: obs.meta.status, via: via.error ? via.error.code : via.meta.status }, "scrapfly fallback");
    if (!via.error && via.meta.status < 400 && via.body) {
      obs = via;
      await bumpDaily({ scrapfly_calls: 1 });
    }
  }
  m.bytes.inc({ connector: sensor.connector }, obs.meta.contentLength);
  if (obs.meta.status) m.httpStatus.inc({ status: String(obs.meta.status) });

  const finish = async (outcome: RunOutcome, extra: { error?: string; snapshotId?: string; changed?: boolean; eventAt?: Date; rateLimitedUntil?: Date | null } = {}): Promise<RunOutcome> => {
    const isErr = outcome === "error" || outcome === "parse_error" || outcome === "rate_limited";
    const consecutive = isErr ? sensor.consecutiveErrors + 1 : 0;
    const health = outcome === "rate_limited" ? "RATE_LIMITED" : isErr ? (consecutive >= 5 ? "ERROR" : "DEGRADED") : outcome === "missing" ? "DEGRADED" : "UP";
    // Shadow sensors (Source Factory) keep their lifecycle status until evaluateShadows() decides.
    const status = !sensor.enabled ? "DISABLED" : sensor.status === "SHADOW" ? "SHADOW" : health === "ERROR" || health === "RATE_LIMITED" ? "DEGRADED" : "ACTIVE";
    const changes7d = await countChanges7d(sensor.id);
    const events7d = await countEvents7d(sensor.id);
    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" });
    if (extra.rateLimitedUntil) interval = Math.max(interval, Math.ceil((extra.rateLimitedUntil.getTime() - Date.now()) / 1000));
    const avg = sensor.avgLatencyMs ? Math.round(sensor.avgLatencyMs * 0.8 + obs.meta.durationMs * 0.2) : obs.meta.durationMs;
    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 });
    await db
      .update(sensors)
      .set({
        lastCheckAt: new Date(),
        lastStatus: obs.meta.status || null,
        lastError: isErr || outcome === "missing" ? (extra.error ?? obs.error?.message ?? `HTTP ${obs.meta.status}`).slice(0, 500) : null,
        consecutiveErrors: consecutive,
        health,
        status,
        totalRuns: sensor.totalRuns + 1,
        totalNotModified: sensor.totalNotModified + (outcome === "not_modified" ? 1 : 0),
        avgLatencyMs: avg,
        nextCheckAt: new Date(Date.now() + interval * 1000),
        updatedAt: new Date(),
        ...(extra.changed ? { lastChangeAt: new Date() } : {}),
        ...(extra.eventAt ? { lastEventAt: extra.eventAt } : {}),
        ...(!sensor.validatedAt && !isErr && outcome !== "missing" ? { validatedAt: new Date() } : {}),
        ...(obs.meta.status >= 200 && obs.meta.status < 300 ? { etag: obs.meta.etag ?? sensor.etag, lastModified: obs.meta.lastModified ?? sensor.lastModified } : {}),
      })
      .where(sql`id = ${sensor.id}`);
    m.checks.inc({ connector: sensor.connector, outcome });
    await bumpDaily({ checks: 1, bytes: obs.meta.contentLength, not_modified: outcome === "not_modified" ? 1 : 0, errors: isErr ? 1 : 0 });
    await bumpSourceDaily(source.id, { checks: 1, not_modified: outcome === "not_modified" ? 1 : 0, errors: isErr ? 1 : 0 });
    return outcome;
  };

  // ---- Transport-level outcomes -------------------------------------------------------
  if (obs.error) return finish("error", { error: `${obs.error.code}: ${obs.error.message}` });
  if (obs.notModified) return finish("not_modified");
  const st = obs.meta.status;
  if (st === 429) {
    const ra = Number(obs.meta.headers["retry-after"] ?? 0);
    return finish("rate_limited", { error: "HTTP 429", rateLimitedUntil: new Date(Date.now() + Math.min(6 * 3600e3, (ra > 0 ? ra : 900) * 1000)) });
  }
  if (st === 404 || st === 410) return handleMissing(sensor, source, obs, finish);
  if (st >= 400) return finish("error", { error: `HTTP ${st}` });
  if (!obs.body && obs.meta.method !== "HEAD") return finish("error", { error: "empty body" });

  // ---- Normalize --------------------------------------------------------------------
  let norm: NormalizedContent;
  const tNorm = m.stageLatency.startTimer({ stage: "normalize" });
  try {
    norm = await connector.normalize(endpoint, obs);
  } catch (e) {
    tNorm();
    const msg = e instanceof NormalizeError ? `${e.code}: ${e.message}` : (e as Error).message;
    return finish("parse_error", { error: msg });
  }
  tNorm();
  if (norm.state) await db.update(sensors).set({ state: norm.state }).where(sql`id = ${sensor.id}`);

  // Restore: page was flagged missing but is back.
  const missingState = (sensor.state as { missing?: { count: number; removedEventAt?: string } } | null)?.missing;
  if (missingState) await db.update(sensors).set({ state: { ...(norm.state ?? sensor.state ?? {}), missing: null } }).where(sql`id = ${sensor.id}`);

  // ---- Compare with previous snapshot -------------------------------------------------
  const prev = sensor.lastSnapshotId ? (await db.select().from(snapshots).where(sql`id = ${sensor.lastSnapshotId}`))[0] : undefined;
  if (prev && prev.canonicalHash === norm.canonicalHash) {
    await touchUrl(sensor, source, false);
    return finish("unchanged");
  }

  // Store snapshot (raw + canonical) — evidence is immutable.
  const store = getBlobStore();
  const raw = obs.body ? await store.put(obs.body) : null;
  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 };
  const canonical = await store.put(JSON.stringify(canonicalBlob));
  const snapId = newId("snap");
  await db.insert(snapshots).values({
    id: snapId,
    sensorId: sensor.id,
    url: obs.meta.finalUrl || sensor.url,
    capturedAt: obs.fetchedAt,
    httpStatus: st,
    contentType: obs.meta.contentType,
    contentLength: obs.meta.contentLength,
    contentHash: norm.rawHash,
    canonicalHash: norm.canonicalHash,
    semanticHash: norm.semanticHash,
    etag: obs.meta.etag,
    lastModified: obs.meta.lastModified,
    storageKey: raw?.key ?? null,
    canonicalStorageKey: canonical.key,
    parserVersion: PARSER_VERSION,
    fetchDurationMs: obs.meta.durationMs,
    fetchMethod: obs.meta.method,
    mode: norm.mode,
    title: norm.title ?? null,
    publishedAt: norm.publishedAt ?? null,
    extractionConfidence: norm.extractionConfidence,
    extra: norm.extra ?? null,
  });
  await db.update(sensors).set({ lastSnapshotId: snapId }).where(sql`id = ${sensor.id}`);
  await touchUrl(sensor, source, true, snapId);

  if (!prev) return finish("baseline", { snapshotId: snapId });

  // ---- Diff -----------------------------------------------------------------------------
  const tDiff = m.stageLatency.startTimer({ stage: "diff" });
  let prevBlob: CanonicalBlob | null = null;
  try {
    prevBlob = prev.canonicalStorageKey ? (JSON.parse(await store.getText(prev.canonicalStorageKey)) as CanonicalBlob) : null;
  } catch (e) {
    log.warn({ sensor: sensor.id, err: (e as Error).message }, "previous canonical blob unreadable");
  }
  if (!prevBlob || prevBlob.mode !== norm.mode) {
    tDiff();
    return finish("changed", { snapshotId: snapId, changed: true });
  }

  let diff: DiffResult;
  if (norm.mode === "list") {
    const seen = new Set<string>(Array.isArray((sensor.state as { seenKeys?: string[] } | null)?.seenKeys) ? ((sensor.state as { seenKeys: string[] }).seenKeys ?? []) : []);
    const prevItems = prevBlob.items ?? [];
    const curItems = norm.items ?? [];
    diff = diffList(prevItems, curItems, norm.compareFields ?? []);
    // Feeds: items scrolling out of the window are not removals. Anything seen before is not "new".
    if (sensor.type === "RSS" || sensor.type === "ATOM" || sensor.type === "GITHUB_RELEASE" || sensor.type === "REST_API" || sensor.type === "JSON") {
      const staleBefore = Date.now() - 14 * 86400e3;
      diff = {
        ...diff,
        removed: [],
        // never seen before AND not an old item resurfacing at the feed window boundary (backfills, reordering)
        added: diff.added.filter((i) => !seen.has(i.key) && !(typeof i.publishedAt === "string" && i.publishedAt && new Date(i.publishedAt).getTime() < staleBefore)),
      };
    }
    // Statuspage summaries only list active incidents/maintenances: an item leaving the list is a resolution, not a removal.
    if (sensor.type === "STATUSPAGE") diff = { ...diff, removed: [] };
    // Sitemaps/statuspages: a sudden > 50 % shrink is more likely a partial response than mass deletion.
    if (prevItems.length >= 20 && curItems.length < prevItems.length * 0.5) diff = { ...diff, removed: [] };
  } else if (norm.mode === "json") {
    diff = diffJson(prevBlob.json, norm.json);
  } else {
    diff = diffText(prevBlob.text ?? "", norm.text ?? "", `${sensor.id}@${prev.capturedAt.toISOString()}`, `${sensor.id}@${obs.fetchedAt.toISOString()}`);
  }
  tDiff();
  if (diffIsEmpty(diff)) return finish("unchanged", { snapshotId: snapId });

  // ---- Heuristics + semantic classification ------------------------------------------------
  const tCls = m.stageLatency.startTimer({ stage: "classify" });
  const heuristic = evaluateChange(diff, { sensorType: sensor.type, url: sensor.url, sourceCategories: source.categories, title: norm.title });
  const semantic = classifyChange(diff, { url: sensor.url, sensorType: sensor.type, title: norm.title });
  tCls();
  m.changeClass.inc({ class: semantic.class });
  const thinFlip = (prevBlob.extractionConfidence ?? 1) < 0.5 || norm.extractionConfidence < 0.5;
  const isNoise = NOISE_CLASSES.has(semantic.class) && !heuristic.facts.some((f) => f.kind === "price" && f.before && f.after);
  const meaningfulByRules = heuristic.signal >= config.meaningfulSignal && !thinFlip && !isNoise;
  const changeId = newId("chg");
  const diffBlob = await store.put(diff.kind === "text" ? diff.unified : renderDiff(diff));
  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 });
  await db.update(sensors).set({ rawChanges: sensor.rawChanges + 1 }).where(sql`id = ${sensor.id}`);
  await db.execute(sql`insert into url_history (url, at, kind, snapshot_id, change_id) values (${sensor.url}, ${obs.fetchedAt}, 'change', ${snapId}, ${changeId})`);
  await db.execute(sql`update urls set change_count = change_count + 1 where url = ${sensor.url}`);
  await bumpDaily({ raw_changes: 1 });
  await bumpSourceDaily(source.id, { raw_changes: 1 });
  m.changes.inc({ connector: sensor.connector, meaningful: String(meaningfulByRules) });
  await publishChange({ id: changeId, sensorId: sensor.id, sourceId: source.id, kind: diff.kind, signal: heuristic.signal, class: semantic.class, at: obs.fetchedAt.toISOString() });
  log.info({ sensor: sensor.id, kind: diff.kind, signal: heuristic.signal.toFixed(2), type: heuristic.eventType, class: semantic.class, thin: thinFlip }, "change detected");

  if (!meaningfulByRules) {
    if (isNoise) m.suppressed.inc({ reason: `noise_${semantic.class}` });
    return finish("changed", { snapshotId: snapId, changed: true });
  }
  // Shadow monitoring (Source Factory): evidence and changes are recorded, nothing is published until the sensor is accepted.
  if (sensor.status === "SHADOW") {
    m.suppressed.inc({ reason: "shadow" });
    await db.update(changes).set({ meaningful: true }).where(sql`id = ${changeId}`);
    return finish("changed", { snapshotId: snapId, changed: true });
  }

  // ---- Event --------------------------------------------------------------------------
  const ev = await createEvent({ sensor, source, obs, norm, prev, snapId, changeId, diff, heuristic, semantic });
  if (!ev) return finish("changed", { snapshotId: snapId, changed: true });
  return finish("event", { snapshotId: snapId, changed: true, eventAt: ev.detectedAt });
}

async 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> {
  const { sensor, source, obs, norm, prev, snapId, changeId, diff, heuristic, semantic } = ctx;
  const detectedAt = obs.fetchedAt;
  const tEv = m.stageLatency.startTimer({ stage: "event" });

  // Idempotency (spec §90): the same observation (same sensor, same before/after canonical content) is one event.
  const fingerprint = sha256(`${sensor.id}|${prev.canonicalHash}|${norm.canonicalHash}`);
  const dup = await db.execute<{ id: string }>(sql`select id from events where fingerprint = ${fingerprint} limit 1`);
  if (dup.rows[0]) {
    m.suppressed.inc({ reason: "fingerprint_duplicate" });
    log.info({ sensor: sensor.id, existing: dup.rows[0].id }, "duplicate observation — event already exists");
    tEv();
    return null;
  }

  const base = describeChange(heuristic, diff, { sourceName: source.name, url: sensor.url, sensorName: sensor.name });
  if (semantic.fieldChanges.length && diff.kind !== "list" && !heuristic.facts.some((f) => f.kind === "price" && f.before && f.after)) {
    base.summary = `${describeFieldChanges(semantic.fieldChanges, 4)}. ${base.summary}`.slice(0, 1500);
  }
  const textForMatching = `${base.title}\n${base.summary}\n${renderDiff(diff).slice(0, 4000)}`;

  const ent = await resolveEntities({ sourceId: source.id, text: textForMatching, hints: heuristic.keywords });
  const nov = await assessNovelty(`${base.title}\n${base.summary}`, source.id);
  const sourceImportance = sourceImportanceFromTier(sensor.tier, source.importanceWeight * sensor.importanceWeight);
  const firstParty = source.firstParty !== false;
  const anomaly = await sourceAnomaly(source.id);
  let eventType = heuristic.eventType;
  let prelim = computeImportance({ eventType, sourceImportance, entityImportance: ent.importance, novelty: nov.novelty, magnitude: heuristic.magnitude, confirmations: nov.confirmations, unusualness: anomaly });
  // Routine batches (spec §13, §55): "21 new CVEs" from a firehose that fires several times a day is normal
  // operation, not a signal. Damp importance for high-volume list sensors; single items keep full weight.
  const routineBatch = diff.kind === "list" && diff.added.length >= 8 && (await countEvents24h(sensor.id)) >= 3;
  if (routineBatch) {
    prelim.score = Math.round(prelim.score * 0.72 * 10) / 10;
    prelim.components.novelty = Math.min(prelim.components.novelty, 35);
  }

  // Duplicate suppression: near-identical to something we already published (syndication / re-fetch).
  if (nov.novelty < 12 && nov.nearest) {
    m.suppressed.inc({ reason: "near_duplicate" });
    log.info({ sensor: sensor.id, nearest: nov.nearest.id, sim: nov.nearest.similarity }, "suppressed near-duplicate event");
    tEv();
    return null;
  }
  // Third-party reports of a story we already hold from several sources add little: raise the bar.
  if (!firstParty && nov.novelty < 30 && nov.confirmations >= 3 && prelim.score < 70) {
    m.suppressed.inc({ reason: "redundant_external" });
    tEv();
    return null;
  }

  let llm: Interpretation | null = null;
  if (llmAvailable() && source.llmEnabled && prelim.score >= config.llm.minImportance) {
    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 });
    if (llm && !llm.meaningful) {
      m.suppressed.inc({ reason: "llm_not_meaningful" });
      log.info({ sensor: sensor.id, type: llm.event_type }, "LLM judged change not meaningful");
      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}`);
      tEv();
      return null;
    }
    if (llm) {
      eventType = llm.event_type;
      const spec = eventTypeSpec(eventType);
      prelim = computeImportance({ eventType, sourceImportance, entityImportance: ent.importance, novelty: nov.novelty, magnitude: heuristic.magnitude, confirmations: nov.confirmations, unusualness: anomaly });
      prelim.components.severity = Math.round((spec.severity + llm.severity) / 2);
      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;
    }
  }

  const title = (llm?.title ?? base.title).slice(0, 180);
  const summary = (llm?.summary ?? base.summary).slice(0, 1500);
  const spec = eventTypeSpec(eventType);
  const isAnnouncementSensor = ANNOUNCEMENT_SENSORS.has(sensor.type);
  const announcedSim = isAnnouncementSensor ? 1 : recentAnnouncementSimilarity(source.id, `${title}\n${summary}`, 12 * 3600e3);
  // Silent-change bar (spec §23): first-party, silent-eligible type, no matching announcement, real content, importance ≥ bar.
  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);

  // Entities named by the LLM that resolve to known aliases
  const extra = llm ? await resolveEntities({ sourceId: source.id, text: llm.entities.join("\n"), hints: [] }) : null;
  const entityIds = [...new Set([...ent.subject, ...ent.mentioned, ...(extra?.mentioned ?? [])])];

  const confidence = computeConfidence({
    sourceAuthenticity: firstParty ? 1 : 0.6,
    extraction: norm.extractionConfidence,
    diffClarity: 1 - Math.max(heuristic.noiseRatio, semantic.noiseRatio),
    structured: diff.kind !== "text",
    confirmations: nov.confirmations,
    llmAgreement: llm ? (llm.event_type === heuristic.eventType ? 1 : 0.6) * (llm.confidence / 100) : heuristic.signal,
  });
  const evidenceLabel = nov.confirmations > 0 ? "CONFIRMED" : llm && llm.inferred && llm.confidence < 60 ? "INFERRED" : !llm && heuristic.signal < 0.5 ? "UNCONFIRMED" : "OBSERVED";
  const categories = [...new Set([...source.categories, ...Object.entries(FEED_CHANNELS).filter(([, cats]) => cats.some((c) => source.categories.includes(c))).map(([ch]) => ch)])];
  const keywords = [...new Set([...(llm?.keywords ?? []), ...heuristic.keywords])].slice(0, 20);
  const maxDelta = semantic.fieldChanges.reduce((mx, f) => Math.max(mx, Math.abs(f.deltaPct ?? 0)), 0);
  const impact = computeImpact({ eventType, magnitude: heuristic.magnitude, entityImportance: ent.importance, maxDeltaPct: maxDelta, fieldChanges: semantic.fieldChanges.length, firstParty });
  const id = newId("evt");
  const slug = `${slugify(title).slice(0, 70)}-${id.slice(-6)}`;
  const processedAt = new Date();
  const publishedAt = pickPublishedAt(diff, norm);
  const observedFrom = prev.capturedAt;
  const country = source.country ?? null;
  const language = source.language ?? "en";

  await db.insert(events).values({
    id,
    slug,
    sensorId: sensor.id,
    sourceId: source.id,
    changeId,
    oldSnapshotId: prev.id,
    newSnapshotId: snapId,
    url: sensor.url,
    canonicalUrl: obs.meta.finalUrl || sensor.url,
    eventType,
    title,
    summary,
    whyItMatters: llm?.why_it_matters ?? null,
    importance: prelim.score,
    importanceComponents: prelim.components as unknown as Record<string, number>,
    confidence,
    novelty: nov.novelty,
    impactScore: impact,
    anomalyScore: anomaly,
    categories,
    keywords,
    silentChange,
    evidenceLabel,
    firstParty,
    country,
    language,
    changeClass: semantic.class,
    fieldChanges: semantic.fieldChanges.length ? semantic.fieldChanges : null,
    fingerprint,
    publishedAt,
    observedFrom,
    detectedAt,
    processedAt,
    detectionLatencyMs: publishedAt && detectedAt.getTime() - publishedAt.getTime() < 7 * 86400e3 ? Math.max(0, detectedAt.getTime() - publishedAt.getTime()) : null,
    processingLatencyMs: processedAt.getTime() - detectedAt.getTime(),
    processingVersion: config.processingVersion,
    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 },
  });
  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 } });
  for (const e of entityIds) await db.insert(eventEntities).values({ eventId: id, entityId: e, role: ent.subject.includes(e) ? "subject" : "mentioned" }).onConflictDoNothing();
  await bumpEntityCounters(entityIds, detectedAt, { silent: silentChange, importance: prelim.score });
  await db.update(changes).set({ meaningful: true, eventId: id }).where(sql`id = ${changeId}`);
  await db.update(sensors).set({ meaningfulChanges: sensor.meaningfulChanges + 1 }).where(sql`id = ${sensor.id}`);
  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})`);

  // Provisional signal for clustering; recomputed with the cluster's velocity right after.
  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 });
  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 });
  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 });
  if (routineBatch) signal.reasons.push({ sign: "-", text: "routine batch from a high-volume feed", points: -8 });
  await db.update(events).set({ clusterId: cl.clusterId, publishedToFeedAt: new Date(), signalScore: signal.score, velocityScore: cl.velocity, scoreReasons: signal.reasons }).where(sql`id = ${id}`);
  await bumpSourceDaily(source.id, { events: 1 });

  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 }[] };
  const payload: PublishedEvent & Record<string, unknown> = {
    id,
    slug,
    type: eventType,
    group: eventGroupOf(eventType),
    title,
    summary: summary.slice(0, 280),
    importance: prelim.score,
    signal: signal.score,
    confidence,
    novelty: nov.novelty,
    impact,
    velocity: cl.velocity,
    silent: silentChange,
    evidence: evidenceLabel,
    firstParty,
    country,
    language,
    changeClass: semantic.class,
    fieldChanges: semantic.fieldChanges.slice(0, 3),
    source: { id: source.id, name: source.name, domain: source.domain, tier: source.tier },
    sensor: { id: sensor.id, name: sensor.name, type: sensor.type },
    entities: entNames.rows.map((r) => ({ id: r.id, name: r.name, type: r.type })),
    categories,
    url: sensor.url,
    clusterId: cl.clusterId,
    clusterSlug: cl.slug,
    clusterSize: cl.eventCount,
    clusterState: cl.state,
    detectedAt: detectedAt.toISOString(),
    publishedAt: publishedAt?.toISOString() ?? null,
  };
  await publishEvent(payload);
  void evaluateAlerts(payload).catch(() => undefined);
  m.events.inc({ event_type: eventType, silent: String(silentChange) });
  m.processingLatency.observe((Date.now() - detectedAt.getTime()) / 1000);
  await bumpDaily({ events: 1, silent_events: silentChange ? 1 : 0 });
  tEv();
  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);
  return { id, detectedAt };
}

function pickPublishedAt(diff: DiffResult, norm: NormalizedContent): Date | null {
  if (diff.kind === "list") {
    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()));
    if (dates.length) return new Date(Math.max(...dates.map((d) => d.getTime())));
  }
  return norm.publishedAt ?? null;
}

async function handleMissing(sensor: Sensor, source: Source, obs: Observation, finish: (o: RunOutcome, e?: { error?: string }) => Promise<RunOutcome>): Promise<RunOutcome> {
  const state = (sensor.state ?? {}) as Record<string, unknown>;
  const missing = (state.missing ?? { count: 0, firstAt: obs.fetchedAt.toISOString(), lastAt: null }) as { count: number; firstAt: string; lastAt: string | null; removedEventAt?: string };
  const sepOk = !missing.lastAt || obs.fetchedAt.getTime() - new Date(missing.lastAt).getTime() >= config.deletion.minSeparationMin * 60e3;
  if (sepOk) missing.count += 1;
  missing.lastAt = obs.fetchedAt.toISOString();
  if (sensor.lastSnapshotId && missing.count >= config.deletion.confirmations && !missing.removedEventAt && sensor.status !== "SHADOW") {
    // Confirmed deletion: independent checks separated in time.
    missing.removedEventAt = obs.fetchedAt.toISOString();
    const id = newId("evt");
    const title = `${source.name}: page removed — ${sensor.name}`;
    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.`;
    const imp = computeImportance({ eventType: "page_removed", sourceImportance: sourceImportanceFromTier(sensor.tier, source.importanceWeight), entityImportance: 50, novelty: 80, magnitude: 60, confirmations: 0 });
    const slug = `${slugify(title).slice(0, 70)}-${id.slice(-6)}`;
    const firstParty = source.firstParty !== false;
    const fingerprint = sha256(`${sensor.id}|removed|${sensor.lastSnapshotId}`);
    const impact = computeImpact({ eventType: "page_removed", magnitude: 60, entityImportance: 50, firstParty });
    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" });
    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();
    const ent = await resolveEntities({ sourceId: source.id, text: title, hints: [] });
    for (const e of ent.subject) await db.insert(eventEntities).values({ eventId: id, entityId: e, role: "subject" }).onConflictDoNothing();
    await bumpEntityCounters(ent.subject, obs.fetchedAt, { silent: true, importance: imp.score });
    await db.execute(sql`update urls set status = 'removed' where url = ${sensor.url}`);
    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}`})`);
    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 });
    await db.update(events).set({ clusterId: cl.clusterId, publishedToFeedAt: new Date(), velocityScore: cl.velocity }).where(sql`id = ${id}`);
    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 };
    await publishEvent(payload);
    void evaluateAlerts(payload).catch(() => undefined);
    await bumpDaily({ events: 1, silent_events: 1 });
    log.info({ sensor: sensor.id, checks: missing.count }, "page removal confirmed");
  } else {
    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}`);
  }
  await db.update(sensors).set({ state: { ...state, missing } }).where(sql`id = ${sensor.id}`);
  return finish("missing", { error: `HTTP ${obs.meta.status} (${missing.count}/${config.deletion.confirmations} confirmations)` });
}

async function touchUrl(sensor: Sensor, source: Source, snapshot: boolean, snapshotId?: string): Promise<void> {
  const domain = safeHost(sensor.url);
  await db.execute(sql`insert into urls (url, domain, source_id, sensor_id, first_seen_at, last_seen_at, status, snapshot_count)
    values (${sensor.url}, ${domain}, ${source.id}, ${sensor.id}, now(), now(), 'active', ${snapshot ? 1 : 0})
    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}`);
  if (snapshot && snapshotId) await db.execute(sql`insert into url_history (url, at, kind, snapshot_id) values (${sensor.url}, now(), 'snapshot', ${snapshotId})`);
}

function safeHost(u: string): string {
  try {
    return new URL(u).hostname;
  } catch {
    return "";
  }
}

async function countChanges7d(sensorId: string): Promise<number> {
  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'`);
  return Number(r.rows[0]?.n ?? 0);
}
async function countEvents24h(sensorId: string): Promise<number> {
  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'`);
  return Number(r.rows[0]?.n ?? 0);
}
async function countEvents7d(sensorId: string): Promise<number> {
  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'`);
  return Number(r.rows[0]?.n ?? 0);
}

/** Source activity anomaly at detection time (0–100): last 2 h of raw changes vs the 14-day hourly baseline. */
async function sourceAnomaly(sourceId: string): Promise<number> {
  const r = await db.execute<{ c2h: string; c14d: string }>(sql`
    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,
           (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`);
  const cur = Number(r.rows[0]?.c2h ?? 0) / 2;
  const base = Number(r.rows[0]?.c14d ?? 0) / (14 * 24);
  return activityAnomaly(cur, base);
}