# Crawl core — discovery, pipeline states, scheduling, failure policies, repair

companies (pending) ─▶ discovery.discover_company ─▶ sensors (pending, staggered next_run_at)
scheduler tick ─▶ claim due sensors (SKIP LOCKED) ─▶ pipeline.run_sensor ─▶ observation ─▶ snapshot ─▶ change (pending) ─▶ intelligence

# Discovery (services/discovery.py, catlas discover|onboard)

Per company, within a ~90 s budget and ≤ ~25 requests: homepage (website variants, redirects; off-domain redirect → domains row kind=redirect, canonical domain updated only when unambiguous and unique) → generic-html extraction of the homepage (nav/footer/main links classified by urls.classify_url, link rel=alternate feeds, JSON-LD sameAs) → robots.txt Sitemap: lines (else /sitemap.xml; index + ≤ 4 children, bounded by discovery_max_sitemap_urls) → ATS detection (links, embedded boards.greenhouse.io/embed/job_board?for=…, jobs.lever.co/…, Ashby, Workable, Workday, Recruitee, Personio, Teamtailor, SmartRecruiters in the homepage + best careers page; the API endpoint is verified with one request) → ≤ 12 common-path probes for missing high-value surfaces (soft-404 and redirect-to-home rejected) → subdomain probes (careers. jobs. news. blog. docs. developer(s). status. investors. ir. shop. — DNS first, ≤ 6 GETs; status. tries /api/v2/summary.json).

Selection: one sensor per surface (best confidence × method weight × verified × depth), plus jobs_board next to careers and feed next to blog/newsroom, capped by discovery_max_sensors_per_company in surface-importance order. Sensor seed values: discovery_confidence, discovery_method (nav | link | sitemap | robots | probe | subdomain | ats | feed), quality_score = 100 × confidence × (0.5 + 0.5 × importance) × reliability, base_interval_s = SURFACE_BASE_INTERVAL_S × tier factor (1: ×0.5, 2: ×0.75, 3: ×1, 4: ×1.5) clamped to [min_interval_s, max_interval_s], tier letter from tier_for_interval, next_run_at random within the first interval (or now with --fetch-now), priority = 0.3 + 0.5 × company importance + 0.2 × surface importance. Company → onboarding_status active / failed (+onboarding_error) / no_website; companies.stats.discovery and source_meta.same_as filled.

onboard_pending claims queue_jobs(kind='discover') with FOR UPDATE SKIP LOCKED (attempts / dead after max_attempts, 30 min × attempts backoff) and then companies.onboarding_status='pending' (marked discovering while claimed).

# Pipeline states (services/pipeline.py, catlas run-sensor)

Objects: raw bytes → archive.put_bytes (observations.object_key), normalized text → text_key, blocks JSON → blocks_key; snapshots.extracted = Extraction.structured_payload() bounded (300 items per list, ≤ 900 KB), extracted_summary counters.

Change classification: sdk.diff.compare(previous blocks, new blocks, surface, texts, structured_delta, history) → significance → taxonomy.change_kind with settings.*_threshold → changes.kind; status='pending' for meaningful / major / critical (consumed by services/events), 'archived' for minor. Pure noise (no typed delta, keep_noise_snapshots=false) writes neither snapshot nor change but updates the sensor hashes so the same noise is not re-detected.

Entity reconciliation (scoped to the sensor): jobs by fingerprint (insert / refresh last_seen_at / no_longer_listed + removed_at), people / products / locations by name_norm, pricing plans versioned (superseded + valid_to, removed), news by canonical URL. Removals happen only when the extraction is trustworthy (≥ 1 entity extracted or a structured response) and never when an HTML listing drops > 70 % of ≥ 10 entities (parse degradation guard → note in structured_delta.notes). structured_delta shape is documented in sdk/models.py.

# Scheduling formula

unchanged / 304:   current < base ? min(base, current × burst_decay) : min(max, current × stability_growth)
meaningful+:       burst_interval_s (then decays back to base, one step per unchanged run)
minor:             min(current, base)
failure:           min(max, current × FAILURE_POLICY[class].multiplier)
next_run_at:       now + interval × U(0.9, 1.1)
quality_score:     EMA(α = 0.15) towards 100 × extraction confidence (success) or 0 (failure)

Scheduler (catlas schedule): every scheduler_tick_s claim status in (active, failing, pending) and next_run_at <= now() not claimed in the last 15 min, ordered by priority desc, next_run_at, limit scheduler_claim_batch, FOR UPDATE SKIP LOCKED; run under asyncio.Semaphore(fetch_concurrency) (per-domain spacing / concurrency in fetch.DomainGovernor); crawl_runs row per tick (claimed, ok, changed, meaningful, failed, not_modified, unchanged, skipped, redirected, tick_ms), heartbeat in settings_kv['scheduler:heartbeat'] (+ per-worker key); onboarding batch every 4 ticks unless --no-onboarding; periodic tasks from services/periodic (interval + cron in settings.tz, APScheduler); SIGTERM/SIGINT release claims. Several schedulers on different machines share one database safely (claims, idempotent runs, no node assumptions). --once runs a single tick.

# Failure policies (taxonomy.FAILURE_POLICY: multiplier, failures-before-failing)

DNS 3.0/3 · TIMEOUT 1.5/4 · HTTP_4XX 2.0/3 · HTTP_5XX 1.5/5 · BOT_CHALLENGE 4.0/2 · PARSING 2.0/3 · SCHEMA 2.0/3 · REDIRECT 2.0/2 · PAGE_REMOVED 4.0/2 · RATE_LIMIT 3.0/4 · ROBOTS 8.0/1 · BLOCKED_DESTINATION 8.0/1 · TOO_LARGE 4.0/2 · UNKNOWN 2.0/3. Status transitions: failing at the class threshold → stale at stale_after_failures (6) → retired (+retired_at) at retire_after_failures (30). ROBOTS → blocked immediately; ROBOTS / repeated BOT_CHALLENGE open a review_queue(kind='blocked_source') item (never bypassed). Domain budgets (domain_budgets: daily_budget, blocked_until) are honoured before any request; used_today resets per day; cost_ledger counts fetch units per company / connector and storage.

# Repair (services/repair.py, catlas repair, periodic every 30 min)

For failing / stale / redirected / repeated PAGE_REMOVED sensors (≤ 6 requests each): retry the URL (recovered → active, next_run_at=now) → candidates for the same surface from the redirect target, the sitemap and the homepage navigation → content-identity check against the last snapshot text (simhash Hamming ≤ 12 or fuzzy ratio ≥ 0.6) → migrate (old sensor retired with config.successor_id, new sensor pending with config.predecessor_id, discovery_method='repair'; history is never deleted) or review_queue(kind='sensor_migration') when uncertain. Each attempt is stamped in config.last_repair_at.

# Operations cheat-sheet

catlas connectors                       # registry + connectors table
catlas discover https://stripe.com      # dry-run surface table (no DB writes)
catlas onboard --company <slug> --fetch-now
catlas schedule --once                  # one tick (tests / cron); catlas schedule for the daemon
catlas run-sensor <id|url> [--file fixture] [--force]
catlas sensors --company <slug> [--status failing] [--surface careers]
catlas repair --dry-run
catlas stats-crawl