"""The sensor pipeline (spec §17–20, §26, §60–64): one run of one sensor.

    run_sensor(sensor_row, fetcher=…, worker=…) -> RunOutcome

    fetch (conditional) ──▶ NotModified ─▶ observation(not_modified) + interval growth
                       ──▶ failure     ─▶ observation(failure_class) + failures row + policy backoff + status transitions (+ review)
                       ──▶ success     ─▶ archive raw ─▶ connector.extract ─▶ hashes
                                           unchanged ─▶ observation(changed=false) + interval growth
                                           changed   ─▶ snapshot (version, objects, extracted) ─▶ entity reconciliation ─▶ structured_delta
                                                       ─▶ block diff vs previous snapshot ─▶ changes row (significance, kind, status)
                                                       ─▶ sensor counters / validators / adaptive interval / quality ─▶ company + ledgers

Everything after the network happens in ONE transaction per sensor. Raw bytes, normalized text and blocks live in the object store
(content-addressed); rows only reference them. Nothing is ever overwritten: new snapshot versions, status columns, removed_at.
"""
from __future__ import annotations

import asyncio
import json
import logging
import random
import re
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

from companyatlas import archive
from companyatlas.config import settings
from companyatlas.connectors._precision import HTML_JOB_CONNECTORS, apply_precision
from companyatlas.connectors._util import is_engineering, job_fingerprint, norm_name
from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction
from companyatlas.fetch import (
    BlockedError,
    Fetcher,
    FetchError,
    FetchResult,
    NotModified,
    classify_exception,
    file_result,
    text_quality,
)
from companyatlas.ids import new_id
from companyatlas.sdk import connector as connectors
from companyatlas.sdk.diff import DIFF_VERSION, compare
from companyatlas.sdk.models import Block, BlockDiff, Extraction, StructuredDelta
from companyatlas.sdk.normalize import structural_hash, text_hash
from companyatlas.taxonomy import (
    AI_KEYWORDS,
    FAILURE_POLICY,
    SURFACE_IMPORTANCE,
    ChangeKind,
    FailureClass,
    SensorStatus,
    Surface,
    change_kind,
    tier_for_interval,
)
from companyatlas.urls import canonicalize_url, registrable_domain

log = logging.getLogger(__name__)

PIPELINE_VERSION = "pipeline-v1"
DELTA_LIST_LIMIT = 200
EXTRACTED_LIST_LIMIT = 300
EXTRACTED_MAX_BYTES = 900_000
JITTER = 0.1                            # ± on next_run_at
QUALITY_ALPHA = 0.15                    # EMA weight for quality_score updates
HTML_SUSPICIOUS_DROP = 0.7              # HTML listings losing > 70 % of ≥ 10 entities are not trusted for removals
MIN_PREVIOUS_FOR_DROP_GUARD = 10
FETCH_TIMEOUT_FACTOR = 6                # connector.fetch (incl. pagination) may take this × http timeout


@dataclass(slots=True)
class RunOutcome:
    sensor_id: str
    status: str                          # ok | not_modified | unchanged | changed | failed | skipped | redirected
    observation_id: str | None = None
    snapshot_id: str | None = None
    change_id: str | None = None
    failure_class: str | None = None
    error: str | None = None
    significance: float | None = None
    kind: str | None = None
    duration_ms: int = 0
    next_run_at: datetime | None = None
    interval_s: int | None = None
    delta_counts: dict[str, int] = field(default_factory=dict)
    sensor_status: str | None = None

    @property
    def ok(self) -> bool:
        return self.status in ("ok", "not_modified", "unchanged", "changed")


# ------------------------------------------------------------------------------------------------------------ scheduling maths


def _clamp(v: float) -> int:
    return int(max(settings.min_interval_s, min(settings.max_interval_s, v)))


def next_interval_unchanged(current: int, base: int) -> int:
    """Burst decay back to base, then slow growth towards the max (spec §16)."""
    if current < base:
        return _clamp(min(base, current * settings.burst_decay))
    return _clamp(current * settings.stability_growth)


def next_interval_changed(kind: ChangeKind, current: int, base: int) -> int:
    if kind in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL):
        return _clamp(settings.burst_interval_s)
    if kind == ChangeKind.MINOR:
        return _clamp(min(current, base))
    return next_interval_unchanged(current, base)


def next_interval_failed(failure: str, current: int) -> int:
    mult = FAILURE_POLICY.get(failure, FAILURE_POLICY[FailureClass.UNKNOWN])[0]
    return _clamp(current * mult)


def _next_run(now: datetime, interval: int) -> datetime:
    return now + timedelta(seconds=interval * random.uniform(1 - JITTER, 1 + JITTER))


def _quality_update(current: float, target: float) -> float:
    return round(max(0.0, min(100.0, current * (1 - QUALITY_ALPHA) + target * QUALITY_ALPHA)), 2)


def is_ai_title(*parts: str | None) -> bool:
    text = " " + " ".join(p.lower() for p in parts if p) + " "
    return any(k in text for k in AI_KEYWORDS)


# ------------------------------------------------------------------------------------------------------------ helpers


def _structured_hash(ex: Extraction) -> str:
    """Hash of the typed payload only (jobs/people/products/plans/locations/news), order-insensitive."""
    import hashlib

    parts: list[str] = []
    for j in ex.jobs:
        parts.append("job|" + job_fingerprint(j.title, j.location_text, j.external_id, j.url))
    for p in ex.people:
        parts.append(f"person|{norm_name(p.name)}|{(p.title or '').lower()}")
    for pr in ex.products:
        parts.append(f"product|{norm_name(pr.name)}")
    for pl in ex.plans:
        parts.append(f"plan|{norm_name(pl.plan_name)}|{pl.price}|{pl.currency}|{pl.billing_period}|{pl.unit}|{pl.contact_sales}|{'/'.join(pl.features[:25])}")
    for loc in ex.locations:
        parts.append(f"loc|{norm_name(loc.name)}|{loc.city}|{loc.country}")
    for n in ex.news:
        parts.append(f"news|{canonicalize_url(n.url)}")
    if not parts:
        return ""
    return hashlib.sha256("\n".join(sorted(parts)).encode("utf-8")).hexdigest()


def _bounded_extracted(ex: Extraction) -> dict[str, Any]:
    payload = ex.structured_payload()
    for k in ("jobs", "people", "products", "plans", "locations", "news"):
        lst = payload.get(k) or []
        if len(lst) > EXTRACTED_LIST_LIMIT:
            payload[k] = lst[:EXTRACTED_LIST_LIMIT]
            payload.setdefault("truncated", {})[k] = len(lst)
    meta = dict(payload.get("meta") or {})
    if isinstance(meta.get("urls"), list) and len(meta["urls"]) > 500:
        meta["urls"] = meta["urls"][:500]
        meta["urls_truncated"] = True
    payload["meta"] = meta
    raw = jsonb(payload)
    if len(raw) > EXTRACTED_MAX_BYTES:
        for k in ("jobs", "people", "products", "plans", "locations", "news"):
            payload[k] = (payload.get(k) or [])[:50]
        meta.pop("urls", None)
        payload["meta"] = meta
        payload["truncated"] = {**payload.get("truncated", {}), "reason": "size"}
    return payload


def _blocks_json(blocks: list[Block]) -> str:
    return json.dumps([b.to_json() for b in blocks], ensure_ascii=False, default=str)


def _blocks_from_json(raw: str) -> list[Block]:
    out: list[Block] = []
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return out
    for d in data if isinstance(data, list) else []:
        try:
            out.append(Block(key=d["key"], kind=d.get("kind", "other"), text=d.get("text", ""), path=d.get("path", ""), hash=d.get("hash", ""),
                             simhash=int(d.get("simhash") or 0), weight=float(d.get("weight") or 1.0), order=int(d.get("order") or 0), attrs=d.get("attrs") or {}))
        except (KeyError, TypeError, ValueError):
            continue
    return out


def _guess_content_type(path: str) -> str:
    p = path.lower()
    if p.endswith(".json"):
        return "application/json; charset=utf-8"
    if p.endswith((".xml", ".rss", ".atom")):
        return "application/xml; charset=utf-8"
    return "text/html; charset=utf-8"


# ------------------------------------------------------------------------------------------------------------ domain budgets


async def _check_domain_budget(conn: Any, domain: str, now: datetime) -> tuple[bool, datetime | None, str | None]:
    """(allowed, resume_at, reason). Creates the row on first sight; resets `used_today` on a new day."""
    row = await fetch_one(conn, """
        insert into domain_budgets (domain, max_concurrency, requests_per_minute, daily_budget)
        values (:d, :mc, :rpm, :daily)
        on conflict (domain) do update set used_today = case when domain_budgets.budget_day < current_date then 0 else domain_budgets.used_today end,
                                           budget_day = greatest(domain_budgets.budget_day, current_date), updated_at = now()
        returning daily_budget, used_today, blocked_until, block_reason
    """, d=domain, mc=settings.domain_max_concurrency, rpm=settings.default_rate_per_min, daily=settings.domain_daily_budget)
    if row is None:
        return True, None, None
    if row["blocked_until"] and row["blocked_until"] > now:
        return False, row["blocked_until"], row.get("block_reason") or "domain blocked"
    if row["used_today"] >= row["daily_budget"]:
        tomorrow = datetime.combine(datetime.now(UTC).date() + timedelta(days=1), datetime.min.time(), tzinfo=UTC)
        return False, tomorrow, "daily budget exhausted"
    return True, None, None


async def _consume_domain_budget(conn: Any, domain: str, units: int) -> None:
    await execute(conn, "update domain_budgets set used_today = used_today + :u, updated_at = now() where domain = :d", u=units, d=domain)


async def _ledger(conn: Any, company_id: str, connector_id: str, units: int, size_bytes: int) -> None:
    await execute(conn, """
        insert into cost_ledger (day, dimension, key, units) values (current_date, 'fetch', :company, :u)
        on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units
    """, company=company_id, u=units)
    await execute(conn, """
        insert into cost_ledger (day, dimension, key, units) values (current_date, 'fetch', :conn, :u)
        on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units
    """, conn=f"connector:{connector_id}", u=units)
    if size_bytes:
        await execute(conn, """
            insert into cost_ledger (day, dimension, key, units) values (current_date, 'storage_gb', '', :gb)
            on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units
        """, gb=size_bytes / 1e9)


# ------------------------------------------------------------------------------------------------------------ entity reconciliation


def _trustworthy(ex: Extraction, count: int) -> bool:
    return count >= 1 or bool(ex.meta.get("structured"))


def _suspicious_drop(ex: Extraction, previous: int, current: int) -> bool:
    if ex.meta.get("structured"):
        return False
    return previous >= MIN_PREVIOUS_FOR_DROP_GUARD and current < previous * (1 - HTML_SUSPICIOUS_DROP)


async def reconcile_jobs(conn: Any, *, company_id: str, sensor_id: str, ex: Extraction, now: datetime, delta: StructuredDelta) -> None:
    existing = await fetch_all(conn, "select id, fingerprint, status, title from jobs where company_id = :c and sensor_id = :s", c=company_id, s=sensor_id)
    by_fp = {r["fingerprint"]: r for r in existing}
    open_before = sum(1 for r in existing if r["status"] == "open")
    seen: set[str] = set()
    added: list[dict[str, Any]] = []
    for j in ex.jobs:
        fp = job_fingerprint(j.title, j.location_text, j.external_id, j.url)
        if fp in seen:
            continue
        seen.add(fp)
        ai = is_ai_title(j.title, j.department, j.team)
        row = by_fp.get(fp)
        params = {"c": company_id, "s": sensor_id, "fp": fp, "title": j.title[:300], "dept": j.department, "team": j.team, "loc": j.location_text,
                  "city": j.city, "region": j.region, "country": (j.country or None), "remote": j.remote, "et": j.employment_type, "sen": j.seniority,
                  "skills": list(j.skills or [])[:30], "smin": j.salary_min, "smax": j.salary_max, "scur": j.salary_currency, "sper": j.salary_period,
                  "url": j.url, "dh": j.description_hash, "posted": j.posted_at, "ai": ai, "eng": is_engineering(j.title), "raw": jsonb(j.raw or {}),
                  "ext": j.external_id, "now": now}
        if row is None:
            await execute(conn, """
                insert into jobs (id, company_id, sensor_id, external_id, fingerprint, title, department, team, location_text, city, region, country, remote,
                                  employment_type, seniority, skills, salary_min, salary_max, salary_currency, salary_period, url, description_hash, posted_at,
                                  first_seen_at, last_seen_at, status, is_ai, is_engineering, raw)
                values (:id, :c, :s, :ext, :fp, :title, :dept, :team, :loc, :city, :region, :country, :remote, :et, :sen, cast(:skills as text[]), :smin, :smax,
                        :scur, :sper, :url, :dh, :posted, :now, :now, 'open', :ai, :eng, cast(:raw as jsonb))
                on conflict (company_id, fingerprint) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'open',
                        removed_at = null, url = coalesce(excluded.url, jobs.url), is_ai = excluded.is_ai
            """, id=new_id("job"), **params)
            added.append({"title": j.title, "url": j.url, "location_text": j.location_text, "country": j.country, "remote": j.remote, "department": j.department, "is_ai": ai})
        else:
            await execute(conn, """
                update jobs set last_seen_at = :now, status = 'open', removed_at = null, title = :title, location_text = coalesce(:loc, location_text),
                       city = coalesce(:city, city), region = coalesce(:region, region), country = coalesce(:country, country), remote = coalesce(:remote, remote),
                       url = coalesce(:url, url), is_ai = :ai, salary_min = coalesce(:smin, salary_min), salary_max = coalesce(:smax, salary_max),
                       posted_at = coalesce(posted_at, :posted)
                where id = :id
            """, id=row["id"], **params)
            if row["status"] != "open":
                added.append({"title": j.title, "url": j.url, "location_text": j.location_text, "country": j.country, "remote": j.remote, "department": j.department,
                              "is_ai": ai, "relisted": True})
    removed: list[dict[str, Any]] = []
    missing = [r for r in existing if r["status"] == "open" and r["fingerprint"] not in seen]
    if missing and _trustworthy(ex, len(seen)) and not _suspicious_drop(ex, open_before, len(seen)):
        rows = await fetch_all(conn, """
            update jobs set status = 'no_longer_listed', removed_at = :now, last_seen_at = last_seen_at
            where id = any(cast(:ids as text[])) returning title, url, location_text, country, remote, department, is_ai
        """, ids=[r["id"] for r in missing], now=now)
        removed = [dict(r) for r in rows]
    elif missing:
        delta.setdefault("notes", []).append(f"jobs: {len(missing)} missing not marked removed (untrusted extraction)")
    if added or removed or open_before != len(seen):
        delta["jobs"] = {"added": added[:DELTA_LIST_LIMIT], "removed": removed[:DELTA_LIST_LIMIT], "open_before": open_before, "open_after": len(seen),
                         "ai_added": sum(1 for a in added if a.get("is_ai")), "ai_removed": sum(1 for r in removed if r.get("is_ai"))}


async def reconcile_named(conn: Any, *, table: str, company_id: str, sensor_id: str, sensor_url: str, items: list[Any], ex: Extraction, now: datetime,
                          delta: StructuredDelta, key: str) -> None:
    """people / products / locations share the same shape: unique (company_id, name_norm), first/last_seen, removed_at, status."""
    listed_status = "listed"
    existing = await fetch_all(conn, f"select id, name_norm, status, title from {table} where company_id = :c and sensor_id = :s"
                               if table == "people" else f"select id, name_norm, status from {table} where company_id = :c and sensor_id = :s",
                               c=company_id, s=sensor_id)
    by_norm = {r["name_norm"]: r for r in existing}
    seen: set[str] = set()
    added: list[dict[str, Any]] = []
    title_changed: list[dict[str, Any]] = []
    for it in items:
        name = getattr(it, "name", None)
        if not name:
            continue
        nn = norm_name(name)
        if not nn or nn in seen:
            continue
        seen.add(nn)
        row = by_norm.get(nn)
        if table == "people":
            payload = {"name": name, "title": it.title, "role_category": it.role_category, "is_executive": it.is_executive}
            if row is None:
                await execute(conn, """
                    insert into people (id, company_id, sensor_id, name, name_norm, title, role_category, is_executive, first_seen_at, last_seen_at, status, source_url)
                    values (:id, :c, :s, :name, :nn, :title, :rc, :ex, :now, :now, 'listed', :url)
                    on conflict (company_id, name_norm) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'listed',
                        removed_at = null, title = coalesce(excluded.title, people.title), role_category = coalesce(excluded.role_category, people.role_category),
                        is_executive = excluded.is_executive
                """, id=new_id("person"), c=company_id, s=sensor_id, name=name[:200], nn=nn, title=it.title, rc=it.role_category, ex=bool(it.is_executive), now=now,
                              url=it.url or sensor_url)
                added.append(payload)
            else:
                if row["status"] != listed_status:
                    added.append({**payload, "relisted": True})
                elif it.title and row.get("title") and norm_name(it.title) != norm_name(row["title"]):
                    title_changed.append({"name": name, "before": row["title"], "after": it.title})
                await execute(conn, """update people set last_seen_at = :now, status = 'listed', removed_at = null, title = coalesce(:title, title),
                                       role_category = coalesce(:rc, role_category), is_executive = :ex where id = :id""",
                              id=row["id"], now=now, title=it.title, rc=it.role_category, ex=bool(it.is_executive))
        elif table == "products":
            payload = {"name": name, "url": it.url}
            if row is None:
                await execute(conn, """
                    insert into products (id, company_id, sensor_id, name, name_norm, category, description, url, first_seen_at, last_seen_at, status)
                    values (:id, :c, :s, :name, :nn, :cat, :desc, :url, :now, :now, 'listed')
                    on conflict (company_id, name_norm) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'listed',
                        removed_at = null, url = coalesce(excluded.url, products.url), description = coalesce(excluded.description, products.description)
                """, id=new_id("product"), c=company_id, s=sensor_id, name=name[:200], nn=nn, cat=it.category, desc=(it.description or None), url=it.url, now=now)
                added.append(payload)
            else:
                if row["status"] != listed_status:
                    added.append({**payload, "relisted": True})
                await execute(conn, "update products set last_seen_at = :now, status = 'listed', removed_at = null, url = coalesce(:url, url) where id = :id",
                              id=row["id"], now=now, url=it.url)
        else:  # locations
            payload = {"name": name, "city": it.city, "country": it.country, "kind": it.kind}
            if row is None:
                await execute(conn, """
                    insert into locations (id, company_id, sensor_id, kind, name, name_norm, city, region, country, first_seen_at, last_seen_at, status, source_url)
                    values (:id, :c, :s, :kind, :name, :nn, :city, :region, :country, :now, :now, 'listed', :url)
                    on conflict (company_id, name_norm) do update set sensor_id = excluded.sensor_id, last_seen_at = excluded.last_seen_at, status = 'listed',
                        removed_at = null, city = coalesce(excluded.city, locations.city), country = coalesce(excluded.country, locations.country)
                """, id=new_id("location"), c=company_id, s=sensor_id, kind=it.kind or "office", name=name[:200], nn=nn, city=it.city, region=it.region,
                              country=(it.country or None), now=now, url=sensor_url)
                added.append(payload)
            else:
                if row["status"] != listed_status:
                    added.append({**payload, "relisted": True})
                await execute(conn, "update locations set last_seen_at = :now, status = 'listed', removed_at = null where id = :id", id=row["id"], now=now)
    removed: list[dict[str, Any]] = []
    missing = [r for r in existing if r["status"] == listed_status and r["name_norm"] not in seen]
    if missing and _trustworthy(ex, len(seen)) and not _suspicious_drop(ex, len(existing), len(seen)):
        cols = "name, title, role_category, is_executive" if table == "people" else ("name, url" if table == "products" else "name, city, country, kind")
        rows = await fetch_all(conn, f"update {table} set status = 'no_longer_listed', removed_at = :now where id = any(cast(:ids as text[])) returning {cols}",
                               ids=[r["id"] for r in missing], now=now)
        removed = [dict(r) for r in rows]
    elif missing:
        delta.setdefault("notes", []).append(f"{key}: {len(missing)} missing not marked removed (untrusted extraction)")
    if added or removed or title_changed:
        entry: dict[str, Any] = {"added": added[:DELTA_LIST_LIMIT], "removed": removed[:DELTA_LIST_LIMIT]}
        if table == "people" and title_changed:
            entry["title_changed"] = title_changed[:DELTA_LIST_LIMIT]
        if table == "locations":
            prior = await fetch_all(conn, "select distinct country from locations where company_id = :c and country is not null and status = 'listed' and first_seen_at < :now",
                                    c=company_id, now=now)
            known = {r["country"] for r in prior}
            new_countries = sorted({a["country"] for a in added if a.get("country") and a["country"] not in known})
            if new_countries:
                entry["new_countries"] = new_countries
        delta[key] = entry


async def reconcile_plans(conn: Any, *, company_id: str, sensor_id: str, sensor_url: str, ex: Extraction, now: datetime, delta: StructuredDelta) -> None:
    current = await fetch_all(conn, """select id, plan_norm, plan_name, currency, billing_period, price, unit, features, contact_sales, version_no
                                       from pricing_plans where company_id = :c and sensor_id = :s and status = 'current'""", c=company_id, s=sensor_id)
    by_norm = {r["plan_norm"]: r for r in current}
    seen: set[str] = set()
    added: list[dict[str, Any]] = []
    price_changed: list[dict[str, Any]] = []
    for pl in ex.plans:
        nn = norm_name(pl.plan_name)
        if not nn or nn in seen:
            continue
        seen.add(nn)
        row = by_norm.get(nn)
        feats = list(pl.features or [])[:40]
        same = row is not None and (row["price"] is None) == (pl.price is None) and (row["price"] is None or abs(float(row["price"]) - float(pl.price)) < 1e-9) \
            and (row["currency"] or None) == (pl.currency or None) and (row["billing_period"] or None) == (pl.billing_period or None) \
            and bool(row["contact_sales"]) == bool(pl.contact_sales) and list(row["features"] or []) == feats
        if same:
            await execute(conn, "update pricing_plans set last_seen_at = :now where id = :id", id=row["id"], now=now)
            continue
        version = 1
        if row is not None:
            version = int(row["version_no"]) + 1
            await execute(conn, "update pricing_plans set status = 'superseded', valid_to = :now where id = :id", id=row["id"], now=now)
            if (row["price"] is None) != (pl.price is None) or (row["price"] is not None and abs(float(row["price"]) - float(pl.price)) >= 1e-9):
                before = float(row["price"]) if row["price"] is not None else None
                pct = round((pl.price - before) / before * 100, 2) if (before and pl.price is not None) else None
                price_changed.append({"plan_name": pl.plan_name, "before": before, "after": pl.price, "currency": pl.currency or row["currency"],
                                      "billing_period": pl.billing_period or row["billing_period"], "pct": pct})
        else:
            added.append({"plan_name": pl.plan_name, "price": pl.price, "currency": pl.currency, "billing_period": pl.billing_period})
        await execute(conn, """
            insert into pricing_plans (id, company_id, sensor_id, plan_name, plan_norm, currency, billing_period, price, price_text, unit, features, contact_sales,
                                       version_no, valid_from, first_seen_at, last_seen_at, status, source_url)
            values (:id, :c, :s, :name, :nn, :cur, :bp, :price, :pt, :unit, cast(:feats as jsonb), :cs, :v, :now, :now, :now, 'current', :url)
        """, id=new_id("plan"), c=company_id, s=sensor_id, name=pl.plan_name[:200], nn=nn, cur=pl.currency, bp=pl.billing_period, price=pl.price, pt=pl.price_text,
                      unit=pl.unit, feats=jsonb(feats), cs=bool(pl.contact_sales), v=version, now=now, url=sensor_url)
    removed: list[dict[str, Any]] = []
    missing = [r for r in current if r["plan_norm"] not in seen]
    if missing and _trustworthy(ex, len(seen)):
        rows = await fetch_all(conn, """update pricing_plans set status = 'removed', valid_to = :now where id = any(cast(:ids as text[]))
                                        returning plan_name, price, currency, billing_period""", ids=[r["id"] for r in missing], now=now)
        removed = [dict(r) for r in rows]
    if added or removed or price_changed:
        delta["plans"] = {"added": added, "removed": removed, "price_changed": price_changed}


async def reconcile_news(conn: Any, *, company_id: str, sensor_id: str, ex: Extraction, now: datetime, delta: StructuredDelta) -> None:
    added: list[dict[str, Any]] = []
    seen: set[str] = set()
    for n in ex.news[:500]:
        canon = canonicalize_url(n.url)
        if canon in seen:
            continue
        seen.add(canon)
        row = await fetch_one(conn, """
            insert into news_items (id, company_id, sensor_id, url, canonical_url, title, summary, category, published_at, first_seen_at, language)
            values (:id, :c, :s, :url, :canon, :title, :summary, :cat, :pub, :now, :lang)
            on conflict (company_id, canonical_url) do nothing returning id
        """, id=new_id("news"), c=company_id, s=sensor_id, url=n.url[:2000], canon=canon[:2000], title=n.title[:300], summary=(n.summary or None), cat=n.category,
                              pub=n.published_at, now=now, lang=n.language)
        if row is not None:
            added.append({"title": n.title, "url": n.url, "published_at": n.published_at.isoformat() if n.published_at else None, "category": n.category})
    if added:
        delta["news"] = {"added": added[:DELTA_LIST_LIMIT], "count": len(added)}


async def reconcile(conn: Any, *, company_id: str, sensor: dict[str, Any], ex: Extraction, now: datetime, previous_meta: dict[str, Any] | None) -> StructuredDelta:
    delta: StructuredDelta = {}
    sid, url = str(sensor["id"]), str(sensor["url"])
    surface = str(sensor.get("surface") or "")
    if ex.jobs or surface in (Surface.JOBS_BOARD, Surface.CAREERS):
        await reconcile_jobs(conn, company_id=company_id, sensor_id=sid, ex=ex, now=now, delta=delta)
    if ex.people or surface == Surface.LEADERSHIP:
        await reconcile_named(conn, table="people", company_id=company_id, sensor_id=sid, sensor_url=url, items=ex.people, ex=ex, now=now, delta=delta, key="people")
    if ex.products or surface in (Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS):
        await reconcile_named(conn, table="products", company_id=company_id, sensor_id=sid, sensor_url=url, items=ex.products, ex=ex, now=now, delta=delta, key="products")
    if ex.locations or surface == Surface.LOCATIONS:
        await reconcile_named(conn, table="locations", company_id=company_id, sensor_id=sid, sensor_url=url, items=ex.locations, ex=ex, now=now, delta=delta, key="locations")
    if ex.plans or surface == Surface.PRICING:
        await reconcile_plans(conn, company_id=company_id, sensor_id=sid, sensor_url=url, ex=ex, now=now, delta=delta)
    if ex.news:
        await reconcile_news(conn, company_id=company_id, sensor_id=sid, ex=ex, now=now, delta=delta)
    meta: dict[str, Any] = {"language": ex.language}
    if previous_meta:
        if (previous_meta.get("title") or None) != (ex.title or None) and (previous_meta.get("title") or ex.title):
            meta["title_changed"] = {"before": previous_meta.get("title"), "after": ex.title}
        if (previous_meta.get("description") or None) != (ex.meta.get("description") or None):
            meta["description_changed"] = True
    delta["meta"] = meta
    if int(sensor.get("snapshot_count") or 0) == 0:
        # First snapshot of this sensor: everything it lists pre-dates our observation. Flag it so "new" counts stay honest (spec §145).
        for table in ("jobs", "people", "products", "locations", "news_items", "pricing_plans"):
            await execute(conn, f"update {table} set baseline = true where sensor_id = :sid and first_seen_at = :now", sid=sid, now=now)
        delta["baseline"] = True
    return delta


def _delta_counts(delta: StructuredDelta) -> dict[str, int]:
    out: dict[str, int] = {}
    for k, v in delta.items():
        if isinstance(v, dict):
            for kk in ("added", "removed", "price_changed", "title_changed", "new_countries"):
                if isinstance(v.get(kk), list) and v[kk]:
                    out[f"{k}_{kk}"] = len(v[kk])
    return out


def _clean_label(value: str | None) -> str | None:
    if value is None:
        return None
    v = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", value)
    v = re.sub(r"\s+", " ", v).strip()
    return v or None


def _drop_corrupt_entities(ex: Extraction, *, connector_id: str | None = None) -> None:
    """Remove typed items whose labels carry replacement characters or no letters at all (mis-decoded or empty extractions), then apply the
    shared precision rules (`connectors/_precision`) as the last line of defence: CTA/nav anchors, cookie categories, marketing headings,
    swapped person cards. Job rules apply to HTML-scraped listings only — structured boards (ATS) are trusted."""
    def ok(label: str | None) -> bool:
        return bool(label) and "\ufffd" not in label and re.search(r"[^\W\d_]", label) is not None

    for j in ex.jobs:
        j.title = _clean_label(j.title) or j.title
    ex.jobs = [j for j in ex.jobs if ok(j.title)]
    for p in ex.people:
        p.name = _clean_label(p.name) or p.name
    ex.people = [p for p in ex.people if ok(p.name)]
    for p in ex.products:
        p.name = _clean_label(p.name) or p.name
    ex.products = [p for p in ex.products if ok(p.name)]
    for p in ex.plans:
        p.plan_name = _clean_label(p.plan_name) or p.plan_name
    ex.plans = [p for p in ex.plans if ok(p.plan_name)]
    for loc in ex.locations:
        loc.name = _clean_label(loc.name) or loc.name
    ex.locations = [loc for loc in ex.locations if ok(loc.name)]
    for n in ex.news:
        n.title = _clean_label(n.title) or n.title
    ex.news = [n for n in ex.news if ok(n.title) and "\ufffd" not in (n.url or "")]
    if ex.title:
        ex.title = _clean_label(ex.title)
    dropped = apply_precision(ex, html_jobs=connector_id in HTML_JOB_CONNECTORS)
    if dropped:
        log.debug("precision rules dropped items", extra={"connector_id": connector_id, "dropped": dropped})


# ------------------------------------------------------------------------------------------------------------ the run


async def run_sensor(sensor: dict[str, Any], *, fetcher: Fetcher, worker: str = "local", force: bool = False, result: FetchResult | None = None,
                     collection_method: str = "live") -> RunOutcome:
    """One complete run. Never raises for fetch/extract problems (they become observations); DB errors propagate."""
    t0 = time.perf_counter()
    sid = str(sensor["id"])
    now = datetime.now(UTC)
    outcome = RunOutcome(sensor_id=sid, status="failed")
    domain = str(sensor.get("domain") or registrable_domain(str(sensor["url"])))
    company_id = str(sensor["company_id"])
    cfg = dict(sensor.get("config") or {})
    connector = connectors.get(str(sensor["connector_id"]))

    # ---- admission: domain budget / block (own short transaction)
    if result is None:
        async with transaction() as conn:
            allowed, resume_at, reason = await _check_domain_budget(conn, domain, now)
            if not allowed:
                resume = (resume_at or now + timedelta(hours=1)) + timedelta(seconds=random.uniform(60, 900))
                await execute(conn, "update sensors set next_run_at = :n, claimed_by = null, claimed_at = null, updated_at = now() where id = :id", n=resume, id=sid)
                outcome.status, outcome.error, outcome.next_run_at = "skipped", reason, resume
                outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
                return outcome
            company = await fetch_one(conn, "select id, canonical_domain, tier, importance, slug from companies where id = :id", id=company_id)
    else:
        async with transaction() as conn:
            company = await fetch_one(conn, "select id, canonical_domain, tier, importance, slug from companies where id = :id", id=company_id)
    ctx = connectors.ConnectorContext(company=company or {"id": company_id})

    # ---- network (outside any transaction)
    fetched: FetchResult | None = result
    failure: tuple[str, str, int | None] | None = None
    not_modified = False
    fetch_ms = 0
    if fetched is None:
        req_sensor = {**sensor, "etag": None, "last_modified": None} if force else sensor
        tf = time.perf_counter()
        try:
            fetched = await asyncio.wait_for(connector.fetch(ctx, req_sensor, fetcher), settings.http_timeout_s * FETCH_TIMEOUT_FACTOR)
        except NotModified as nm:
            not_modified = True
            fetch_ms = nm.duration_ms
        except (FetchError, BlockedError) as exc:
            failure = (str(exc.failure), str(exc)[:500], exc.status)
        except TimeoutError:
            failure = (str(FailureClass.TIMEOUT), "connector fetch timed out", None)
        except Exception as exc:  # noqa: BLE001
            failure = (str(classify_exception(exc)), f"{exc.__class__.__name__}: {exc}"[:500], None)
        fetch_ms = fetch_ms or int((time.perf_counter() - tf) * 1000)

    # ---- redirect to another registrable domain (keeps the observation, parks the sensor)
    if fetched is not None and result is None and registrable_domain(fetched.final_url) != registrable_domain(str(sensor["url"])):
        failure = (str(FailureClass.REDIRECT), f"redirected off-domain to {fetched.final_url}", fetched.status)
        cfg["redirect_url"] = fetched.final_url

    async with transaction() as conn:
        base_interval = int(sensor.get("base_interval_s") or 86400)
        current = int(sensor.get("current_interval_s") or base_interval)
        quality = float(sensor.get("quality_score") or 50.0)
        obs_id = new_id("observation")
        outcome.observation_id = obs_id

        if not_modified:
            interval = next_interval_unchanged(current, base_interval)
            nxt = _next_run(now, interval)
            await execute(conn, """
                insert into observations (id, sensor_id, company_id, fetched_at, status_code, duration_ms, transport, final_url, not_modified, changed,
                                          connector_version, collection_method, worker)
                values (:id, :sid, :cid, :now, 304, :ms, 'http', :url, true, false, :cv, :cm, :worker)
            """, id=obs_id, sid=sid, cid=company_id, now=now, ms=fetch_ms, url=sensor["url"], cv=connector.connector_id, cm=collection_method, worker=worker)
            quality = _quality_update(quality, 100.0)
            await execute(conn, """
                update sensors set status = 'active', last_run_at = :now, last_success_at = :now, last_status = 304, last_failure_class = null, last_error = null,
                       consecutive_failures = 0, consecutive_unchanged = consecutive_unchanged + 1, observation_count = observation_count + 1,
                       current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q, claimed_by = null, claimed_at = null, updated_at = now()
                where id = :id
            """, now=now, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality, id=sid)
            await execute(conn, "update companies set last_observed_at = :now, first_observed_at = coalesce(first_observed_at, :now) where id = :id", now=now, id=company_id)
            await _consume_domain_budget(conn, domain, 1)
            await _ledger(conn, company_id, connector.connector_id, 1, 0)
            outcome.status, outcome.next_run_at, outcome.interval_s, outcome.sensor_status = "not_modified", nxt, interval, "active"
            outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
            return outcome

        if failure is not None:
            await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,
                                  connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,
                                  collection_method=collection_method, final_url=fetched.final_url if fetched else None)
            await _consume_domain_budget(conn, domain, 1)
            await _ledger(conn, company_id, connector.connector_id, 1, 0)
            outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
            return outcome

        assert fetched is not None
        pages = int(fetched.headers.get("x-companyatlas-pages", "1") or 1)
        object_key, _stored, _created = archive.put_bytes(fetched.content)

        # ---- extract
        try:
            ex = connector.extract(sensor, fetched)
        except Exception as exc:  # noqa: BLE001
            failure = (str(FailureClass.PARSING), f"extract failed: {exc.__class__.__name__}: {exc}"[:500], fetched.status)
            await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,
                                  connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,
                                  collection_method=collection_method, final_url=fetched.final_url, object_key=object_key, size=len(fetched.content),
                                  content_type=fetched.content_type)
            await _consume_domain_budget(conn, domain, pages)
            await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))
            outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
            return outcome

        # ---- corruption guard (spec: never fabricate): mis-decoded bodies never become snapshots, entities or events
        quality = text_quality(ex.text) if ex.text else 1.0
        if quality < 0.99 or (ex.title and "\ufffd" in ex.title):
            failure = (str(FailureClass.PARSING), f"extracted text is corrupt (quality {quality:.3f}, content-type {fetched.content_type[:40]!r})", fetched.status)
            await _record_failure(conn, sensor=sensor, cfg=cfg, company_id=company_id, obs_id=obs_id, failure=failure, now=now, fetch_ms=fetch_ms,
                                  connector_id=connector.connector_id, worker=worker, current=current, quality=quality, outcome=outcome,
                                  collection_method=collection_method, final_url=fetched.final_url, object_key=object_key, size=len(fetched.content),
                                  content_type=fetched.content_type)
            await _consume_domain_budget(conn, domain, pages)
            await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))
            outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
            return outcome
        _drop_corrupt_entities(ex, connector_id=connector.connector_id)
        ex.normalized_hash = ex.normalized_hash or text_hash(ex.text)
        ex.structured_hash = ex.structured_hash or _structured_hash(ex)
        struct_hash = structural_hash(ex.blocks)
        unchanged = (not force and sensor.get("last_normalized_hash") == ex.normalized_hash and (cfg.get("last_structured_hash") or "") == ex.structured_hash
                     and sensor.get("last_snapshot_id") is not None)
        etag, last_mod = fetched.etag, fetched.last_modified

        await execute(conn, """
            insert into observations (id, sensor_id, company_id, fetched_at, status_code, duration_ms, transport, final_url, redirects, not_modified, changed,
                                      content_hash, normalized_hash, structural_hash, object_key, size_bytes, content_type, connector_version, collection_method, worker)
            values (:id, :sid, :cid, :now, :st, :ms, :tr, :url, :redir, false, :changed, :ch, :nh, :sh, :ok, :size, :ct, :cv, :cm, :worker)
        """, id=obs_id, sid=sid, cid=company_id, now=now, st=fetched.status, ms=fetch_ms, tr=fetched.transport, url=fetched.final_url, redir=fetched.redirects,
                      changed=not unchanged, ch=fetched.sha256, nh=ex.normalized_hash, sh=struct_hash, ok=object_key, size=len(fetched.content),
                      ct=fetched.content_type[:100], cv=connector.connector_id, cm=collection_method, worker=worker)
        await _consume_domain_budget(conn, domain, pages)
        await _ledger(conn, company_id, connector.connector_id, pages, len(fetched.content))
        extraction_conf = 100.0 if (ex.meta.get("structured") or len(ex.text) > 200) else 60.0

        if unchanged:
            interval = next_interval_unchanged(current, base_interval)
            nxt = _next_run(now, interval)
            quality = _quality_update(quality, extraction_conf)
            await execute(conn, """
                update sensors set status = 'active', last_run_at = :now, last_success_at = :now, last_status = :st, last_failure_class = null, last_error = null,
                       consecutive_failures = 0, consecutive_unchanged = consecutive_unchanged + 1, observation_count = observation_count + 1,
                       current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q, etag = coalesce(:etag, etag),
                       last_modified = coalesce(:lm, last_modified), last_content_hash = :ch, claimed_by = null, claimed_at = null, updated_at = now()
                where id = :id
            """, now=now, st=fetched.status, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality, etag=etag, lm=last_mod, ch=fetched.sha256, id=sid)
            await execute(conn, "update companies set last_observed_at = :now, first_observed_at = coalesce(first_observed_at, :now) where id = :id", now=now, id=company_id)
            outcome.status, outcome.next_run_at, outcome.interval_s, outcome.sensor_status = "unchanged", nxt, interval, "active"
            outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
            return outcome

        # ---- changed (or first run / forced): snapshot + reconciliation + diff
        prev_id = sensor.get("last_snapshot_id")
        prev = await fetch_one(conn, "select id, version_no, blocks_key, text_key, title, extracted from snapshots where id = :id", id=prev_id) if prev_id else None
        prev_blocks: list[Block] = []
        prev_text = ""
        if prev is not None:
            try:
                if prev.get("blocks_key") and archive.exists(prev["blocks_key"]):
                    prev_blocks = _blocks_from_json(archive.get_text(prev["blocks_key"]))
                if prev.get("text_key") and archive.exists(prev["text_key"]):
                    prev_text = archive.get_text(prev["text_key"])
            except OSError:
                log.warning("previous snapshot objects unreadable", extra={"sensor_id": sid, "snapshot_id": prev_id})
        prev_meta = None
        if prev is not None:
            pm = (prev.get("extracted") or {}).get("meta") if isinstance(prev.get("extracted"), dict) else {}
            prev_meta = {"title": prev.get("title"), "description": (pm or {}).get("description")}

        delta = await reconcile(conn, company_id=company_id, sensor=sensor, ex=ex, now=now, previous_meta=prev_meta)
        history = {"consecutive_unchanged": sensor.get("consecutive_unchanged") or 0, "observation_count": sensor.get("observation_count") or 0,
                   "change_count": sensor.get("change_count") or 0}
        diff: BlockDiff = compare(prev_blocks, ex.blocks, surface=str(sensor["surface"]), before_text=prev_text, after_text=ex.text,
                                  structured_delta=delta, history=history) if prev is not None else BlockDiff()
        kind = change_kind(diff.significance, noise=settings.noise_threshold, meaningful=settings.meaningful_threshold, major=settings.major_threshold,
                           critical=settings.critical_threshold) if prev is not None else ChangeKind.NOISE
        typed = any(k in delta and any(isinstance(v, list) and v for v in delta[k].values()) for k in ("jobs", "people", "products", "plans", "locations", "news")
                    if isinstance(delta.get(k), dict))
        keep_snapshot = prev is None or kind != ChangeKind.NOISE or typed or settings.keep_noise_snapshots
        snap_id: str | None = None
        change_id: str | None = None
        version_no = int(sensor.get("snapshot_count") or 0) + 1
        if keep_snapshot:
            snap_id = new_id("snapshot")
            text_key, _s, _c = archive.put_text(ex.text)
            blocks_key, _s, _c = archive.put_text(_blocks_json(ex.blocks))
            extracted = _bounded_extracted(ex)
            await execute(conn, """
                insert into snapshots (id, sensor_id, company_id, observation_id, previous_snapshot_id, version_no, fetched_at, content_hash, normalized_hash,
                                       structural_hash, object_key, text_key, blocks_key, extracted, extracted_summary, title, language, size_bytes, text_length,
                                       block_count, content_type, connector_version, collection_method)
                values (:id, :sid, :cid, :oid, :prev, :v, :now, :ch, :nh, :sh, :ok, :tk, :bk, cast(:ex as jsonb), cast(:sum as jsonb), :title, :lang, :size, :tl, :bc,
                        :ct, :cv, :cm)
            """, id=snap_id, sid=sid, cid=company_id, oid=obs_id, prev=prev_id, v=version_no, now=now, ch=fetched.sha256, nh=ex.normalized_hash, sh=struct_hash,
                          ok=object_key, tk=text_key, bk=blocks_key, ex=jsonb(extracted), sum=jsonb(ex.summary()), title=(ex.title or None), lang=ex.language,
                          size=len(fetched.content), tl=len(ex.text), bc=len(ex.blocks), ct=fetched.content_type[:100], cv=connector.connector_id, cm=collection_method)
            if prev is not None:
                change_id = new_id("change")
                status = "pending" if kind in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL) else "archived"
                await execute(conn, """
                    insert into changes (id, sensor_id, company_id, surface, snapshot_before, snapshot_after, detected_at, significance, kind, blocks_added,
                                         blocks_removed, blocks_modified, blocks_moved, text_delta_ratio, similarity, diff, structured_delta, status, diff_version)
                    values (:id, :sid, :cid, :surface, :before, :after, :now, :sig, :kind, :ba, :br, :bm, :bmv, :tdr, :sim, cast(:diff as jsonb), cast(:sd as jsonb),
                            :status, :dv)
                """, id=change_id, sid=sid, cid=company_id, surface=str(sensor["surface"]), before=prev_id, after=snap_id, now=now, sig=diff.significance,
                              kind=str(kind), ba=len(diff.added), br=len(diff.removed), bm=len(diff.modified), bmv=len(diff.moved), tdr=diff.text_delta_ratio,
                              sim=diff.similarity, diff=jsonb(diff.to_json()), sd=jsonb(delta), status=status, dv=DIFF_VERSION)
        meaningful = prev is not None and kind in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL)
        if prev is None:
            interval = current
        else:
            interval = next_interval_changed(kind, current, base_interval)
        nxt = _next_run(now, interval)
        quality = _quality_update(quality, extraction_conf)
        cfg["last_structured_hash"] = ex.structured_hash
        cfg.pop("redirect_url", None)
        await execute(conn, """
            update sensors set status = 'active', last_run_at = :now, last_success_at = :now, last_status = :st, last_failure_class = null, last_error = null,
                   consecutive_failures = 0, consecutive_unchanged = case when :meaningful then 0 else consecutive_unchanged + 1 end,
                   observation_count = observation_count + 1, snapshot_count = case when :kept then snapshot_count + 1 else snapshot_count end,
                   change_count = case when :haschange then change_count + 1 else change_count end,
                   meaningful_change_count = case when :meaningful then meaningful_change_count + 1 else meaningful_change_count end,
                   last_change_at = case when :haschange then :now else last_change_at end,
                   last_meaningful_change_at = case when :meaningful then :now else last_meaningful_change_at end,
                   last_content_hash = :ch, last_normalized_hash = :nh, last_structural_hash = :sh, last_snapshot_id = coalesce(:snap, last_snapshot_id),
                   etag = :etag, last_modified = :lm, current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q,
                   config = config || cast(:cfg as jsonb), claimed_by = null, claimed_at = null, updated_at = now()
            where id = :id
        """, now=now, st=fetched.status, meaningful=meaningful, kept=keep_snapshot, haschange=change_id is not None, ch=fetched.sha256, nh=ex.normalized_hash,
                      sh=struct_hash, snap=snap_id, etag=etag, lm=last_mod, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality,
                      cfg=jsonb({"last_structured_hash": ex.structured_hash, "redirect_url": None}), id=sid)
        await execute(conn, """
            update companies set last_observed_at = :now, first_observed_at = coalesce(first_observed_at, :now),
                   last_change_at = case when :meaningful then :now else last_change_at end,
                   stats = stats || jsonb_build_object('observations', coalesce((stats->>'observations')::int, 0) + 1,
                                                        'snapshots', coalesce((stats->>'snapshots')::int, 0) + case when :kept then 1 else 0 end,
                                                        'changes', coalesce((stats->>'changes')::int, 0) + case when :haschange then 1 else 0 end,
                                                        'meaningful_changes', coalesce((stats->>'meaningful_changes')::int, 0) + case when :meaningful then 1 else 0 end),
                   updated_at = now()
            where id = :id
        """, now=now, meaningful=meaningful, kept=keep_snapshot, haschange=change_id is not None, id=company_id)
        outcome.status = "changed" if prev is not None else "ok"
        outcome.snapshot_id, outcome.change_id = snap_id, change_id
        outcome.significance = diff.significance if prev is not None else None
        outcome.kind = str(kind) if prev is not None else None
        outcome.next_run_at, outcome.interval_s, outcome.sensor_status = nxt, interval, "active"
        outcome.delta_counts = _delta_counts(delta)
        outcome.duration_ms = int((time.perf_counter() - t0) * 1000)
        log.info("sensor run", extra={"sensor_id": sid, "surface": sensor.get("surface"), "status": outcome.status, "kind": outcome.kind, "significance": outcome.significance,
                                      "delta": outcome.delta_counts, "interval_s": interval, "ms": outcome.duration_ms})
        return outcome


async def _record_failure(conn: Any, *, sensor: dict[str, Any], cfg: dict[str, Any], company_id: str, obs_id: str, failure: tuple[str, str, int | None], now: datetime,
                          fetch_ms: int, connector_id: str, worker: str, current: int, quality: float, outcome: RunOutcome, collection_method: str,
                          final_url: str | None = None, object_key: str | None = None, size: int | None = None, content_type: str | None = None) -> None:
    fclass, message, status_code = failure
    sid = str(sensor["id"])
    await execute(conn, """
        insert into observations (id, sensor_id, company_id, fetched_at, status_code, duration_ms, transport, final_url, not_modified, changed, failure_class, error,
                                  object_key, size_bytes, content_type, connector_version, collection_method, worker)
        values (:id, :sid, :cid, :now, :st, :ms, 'http', :url, false, false, :fc, :err, :ok, :size, :ct, :cv, :cm, :worker)
    """, id=obs_id, sid=sid, cid=company_id, now=now, st=status_code, ms=fetch_ms, url=final_url or sensor["url"], fc=fclass, err=message, ok=object_key, size=size,
                  ct=(content_type or "")[:100] or None, cv=connector_id, cm=collection_method, worker=worker)
    await execute(conn, "insert into failures (id, sensor_id, company_id, at, failure_class, status_code, message, url) values (:id, :sid, :cid, :now, :fc, :st, :msg, :url)",
                  id=new_id("failure"), sid=sid, cid=company_id, now=now, fc=fclass, st=status_code, msg=message[:500], url=sensor["url"])
    failures = int(sensor.get("consecutive_failures") or 0) + 1
    _mult, threshold = FAILURE_POLICY.get(fclass, FAILURE_POLICY[FailureClass.UNKNOWN])
    prev_status = str(sensor.get("status") or "active")
    status = prev_status if prev_status in (SensorStatus.PAUSED, SensorStatus.RETIRED) else SensorStatus.ACTIVE
    review_kind: str | None = None
    if fclass == FailureClass.REDIRECT and cfg.get("redirect_url"):
        status, review_kind = SensorStatus.REDIRECTED, "sensor_migration"
    elif fclass == FailureClass.ROBOTS:
        status, review_kind = SensorStatus.BLOCKED, "blocked_source"
    elif fclass == FailureClass.BOT_CHALLENGE:
        status = SensorStatus.FAILING if failures >= threshold else status
        review_kind = "blocked_source" if failures >= threshold else None
    elif failures >= settings.retire_after_failures:
        status = SensorStatus.RETIRED
    elif failures >= settings.stale_after_failures:
        status = SensorStatus.STALE
    elif failures >= threshold:
        status = SensorStatus.FAILING
    interval = next_interval_failed(fclass, current)
    nxt = _next_run(now, interval)
    quality = _quality_update(quality, 0.0)
    retire = status == SensorStatus.RETIRED
    await execute(conn, """
        update sensors set status = :status, last_run_at = :now, last_status = :st, last_failure_class = :fc, last_error = :err, consecutive_failures = :cf,
               consecutive_unchanged = 0, observation_count = observation_count + 1, current_interval_s = :iv, tier = :tier, next_run_at = :nxt, quality_score = :q,
               config = config || cast(:cfg as jsonb), retired_at = case when :retire then coalesce(retired_at, :now) else retired_at end,
               claimed_by = null, claimed_at = null, updated_at = now()
        where id = :id
    """, status=str(status), now=now, st=status_code, fc=fclass, err=message[:500], cf=failures, iv=interval, tier=tier_for_interval(interval), nxt=nxt, q=quality,
                  cfg=jsonb({k: v for k, v in cfg.items() if k in ("redirect_url",)}), retire=retire, id=sid)
    if review_kind:
        existing = await fetch_one(conn, "select id from review_queue where kind = :k and ref_id = :r and status = 'open'", k=review_kind, r=sid)
        if existing is None:
            await execute(conn, """insert into review_queue (id, kind, ref_id, company_id, payload) values (:id, :k, :r, :cid, cast(:p as jsonb))""",
                          id=new_id("review"), k=review_kind, r=sid, cid=company_id,
                          p=jsonb({"sensor_id": sid, "url": sensor["url"], "surface": sensor.get("surface"), "failure_class": fclass, "message": message[:300],
                                   "redirect_url": cfg.get("redirect_url"), "consecutive_failures": failures}))
    await execute(conn, "update companies set last_observed_at = :now where id = :id", now=now, id=company_id)
    outcome.status = "redirected" if status == SensorStatus.REDIRECTED else "failed"
    outcome.failure_class, outcome.error, outcome.next_run_at, outcome.interval_s, outcome.sensor_status = fclass, message, nxt, interval, str(status)


# ------------------------------------------------------------------------------------------------------------ convenience entry points


async def load_sensor(sensor_ref: str) -> dict[str, Any] | None:
    async with transaction() as conn:
        row = await fetch_one(conn, "select * from sensors where id = :ref", ref=sensor_ref)
        if row is None:
            row = await fetch_one(conn, "select * from sensors where canonical_url = :c order by created_at limit 1", c=canonicalize_url(sensor_ref))
        if row is None:
            row = await fetch_one(conn, "select * from sensors where url = :u order by created_at limit 1", u=sensor_ref)
    return row


async def run_sensor_ids(ids: list[str], *, fetcher: Fetcher, worker: str = "local", force: bool = False, concurrency: int | None = None) -> list[RunOutcome]:
    sem = asyncio.Semaphore(max(1, concurrency or settings.fetch_concurrency))
    out: list[RunOutcome] = []

    async def one(sid: str) -> None:
        row = await load_sensor(sid)
        if row is None:
            out.append(RunOutcome(sensor_id=sid, status="failed", error="sensor not found"))
            return
        async with sem:
            try:
                out.append(await run_sensor(row, fetcher=fetcher, worker=worker, force=force))
            except Exception as exc:
                log.exception("sensor run crashed", extra={"sensor_id": sid})
                out.append(RunOutcome(sensor_id=sid, status="failed", error=f"{exc.__class__.__name__}: {exc}"[:300]))
                async with transaction() as conn:
                    await execute(conn, "update sensors set claimed_by = null, claimed_at = null, next_run_at = now() + interval '30 minutes' where id = :id", id=sid)

    await asyncio.gather(*(one(s) for s in ids))
    return out


async def run_sensor_by_url_with_file(sensor_ref: str, path: str, *, worker: str = "file", force: bool = False, content_type: str | None = None) -> RunOutcome:
    """Run a sensor against a fixture file instead of the network (`catlas run-sensor --file`)."""
    row = await load_sensor(sensor_ref)
    if row is None:
        raise LookupError(f"sensor {sensor_ref!r} not found")
    res = file_result(path, url=str(row["url"]), content_type=content_type or _guess_content_type(path))
    return await run_sensor(row, fetcher=Fetcher(), worker=worker, force=force, result=res)


def surface_importance(surface: str) -> float:
    return float(SURFACE_IMPORTANCE.get(surface, 0.3))


__all__ = ["PIPELINE_VERSION", "RunOutcome", "is_ai_title", "load_sensor", "next_interval_changed", "next_interval_failed", "next_interval_unchanged", "reconcile",
           "run_sensor", "run_sensor_by_url_with_file", "run_sensor_ids"]