-- sql/128-orchestrator-control-plane.sql -- The DATA plane for the TIP Autonomous Data-Quality Orchestrator ("tip-dqo"). -- -- TIP already has a world-class DETECTION layer (sql/123-126: v_scraper_health, -- v_data_quality_alerts, check_switch_quality) and a blind fixed-cadence dispatcher -- (scheduler.ts: boss.send() is never called from any health path). This migration -- adds the tables that let a controller close the loop DETECT -> PRIORITIZE -> -- DISPATCH -> VERIFY -> RECHECK against DATA freshness, with DB-ENFORCED guardrails -- (the ledger is the single source of truth for cooldown/budget/backoff/breaker, so -- the loop physically cannot hammer a source even under concurrent ticks). -- -- Read-only against the existing quality views; additive only (no existing object is -- altered). Idempotent (IF NOT EXISTS / CREATE OR REPLACE / ON CONFLICT DO NOTHING). -- Applied by scripts/migrate.ts as the next numbered migration after 127. -- ── 1) source_registry — the signal->runner map + per-source SLA + guardrail contract. -- One row per (vendor, source_table) the loop may act on, PLUS the '__global__' -- kill-switch / global-budget row. root_cause_class is LOAD-BEARING: only -- 'staleness' rows are ever auto-dispatched; 'code_fix'/'blocked' rows escalate to a -- human instead of looping retries (re-scraping cannot fix a silent write-path bug). create table if not exists source_registry ( source_key text primary key, -- 'atgbics/stock_observations' | '__global__' vendor text, -- matches v_scraper_health.vendor source_table text, -- matches v_scraper_health.source_table queue_names text[] not null default '{}', -- pg-boss queues to dispatch value_weight numeric not null default 1, -- business priority multiplier warn_days int not null default 8, dead_days int not null default 30, cooldown_minutes int not null default 180, -- >= pricing cadence (2h) so no in-flight re-fire budget_per_day int not null default 4, max_backoff_hours int not null default 24, circuit_breaker_strikes int not null default 3, root_cause_class text not null default 'staleness' check (root_cause_class in ('staleness','code_fix','blocked')), profile text not null default 'erik-safe', enabled boolean not null default true, notes text, updated_at timestamptz not null default now() ); -- ── 2) self_heal_dispatch — the append-only dispatch LEDGER. Single source of truth for -- every guardrail (cooldown/budget/backoff/breaker) and the plan/dispatch audit trail. -- strikes carries the running no-progress count as of this row's verification. create table if not exists self_heal_dispatch ( id bigserial primary key, source_key text not null, queues text[] not null default '{}', reason text, mode text not null check (mode in ('plan','dispatch')), rows_before bigint, last_obs_before date, dispatched_at timestamptz not null default now(), checked_at timestamptz, rows_after bigint, last_obs_after date, outcome text not null default 'pending' check (outcome in ('pending','healed','ineffective','escalated','skipped')), strikes int not null default 0, run_id text ); create index if not exists idx_self_heal_dispatch_source_time on self_heal_dispatch (source_key, dispatched_at desc); create index if not exists idx_self_heal_dispatch_pending on self_heal_dispatch (outcome, dispatched_at) where outcome = 'pending'; -- ── 3) scraper_run_log — the ran-but-wrote-0 detector. Lets VERIFY distinguish a silent -- write-path break (job 'completed', 0 rows written — the ATGBICS-stock signature) -- from a legitimately-no-new-rows run. Built here; scrapers can emit into it over time. -- VERIFY v1 falls back to v_scraper_health freshness when a source has no run_log rows. create table if not exists scraper_run_log ( id bigserial primary key, vendor text, table_name text, run_at timestamptz not null default now(), rows_written int not null default 0, source text, -- who triggered it (e.g. 'dqo:self-heal', 'cron') run_id text ); create index if not exists idx_scraper_run_log_vendor_time on scraper_run_log (vendor, table_name, run_at desc); -- ── 4) acknowledged_exceptions — delta-gate suppression so day-1 known-open failures -- (528 speed<=0, ~3050 units_sold, 78% coding template garbage) NEVER page. The -- escalation ladder only fires for entities NOT covered by an active ack. create table if not exists acknowledged_exceptions ( id bigserial primary key, check_name text, -- v_data_quality_alerts.category or a check_switch_quality name source_key text, -- optional: scope the ack to one source reason text not null, task_ref text, -- Gitea issue / spawned-task id ack_until timestamptz, -- null = indefinite created_at timestamptz not null default now() ); -- ── 5) v_dqo_targets — the single ranked SELECT the controller reads for DETECT+PRIORITIZE. -- Joins the live freshness signal (v_scraper_health, keyed off observation TABLES, not -- pg-boss job cadence) onto the registry, ranks dead-before-stale weighted by business -- value and how far past dead_days. Guardrail filtering (cooldown/budget/backoff/breaker) -- is applied in the controller against the ledger, not here, to keep the view simple. create or replace view v_dqo_targets as select sr.source_key, h.vendor, h.source_table, h.status, h.days_since_last, h.rows_30d, h.last_observation, sr.root_cause_class, sr.queue_names, sr.profile, sr.value_weight, sr.warn_days, sr.dead_days, sr.cooldown_minutes, sr.budget_per_day, sr.max_backoff_hours, sr.circuit_breaker_strikes, round( (case h.status when 'dead' then 3 when 'stale' then 2 else 1 end)::numeric * sr.value_weight * (1 + coalesce(h.days_since_last, 0)::numeric / nullif(sr.dead_days, 0)) , 2) as score from v_scraper_health h join source_registry sr on sr.vendor = h.vendor and sr.source_table = h.source_table where h.status in ('stale', 'dead') and sr.enabled and sr.source_key <> '__global__' order by (case h.status when 'dead' then 3 when 'stale' then 2 else 1 end) desc, score desc; -- ── 6) SEED — the global kill-switch/budget row + the known EXPECTED_VENDORS sources. -- root_cause_class is HAND-CURATED from the 2026-07-06 gap analysis, not guessed: -- ATGBICS/stock = code_fix (price is the single most-alive source yet stock writes -- 0 rows/54d = silent write-path break; escalate, never loop) -- Flexoptix/stock= staleness (price+catalog sync work; a re-scrape can plausibly heal it -- — the first source to arm out of plan-mode) -- Every other source defaults to 'staleness'. Extra (vendor, source_table) rows that do -- not match a live v_scraper_health row are harmless — the join just yields no target. insert into source_registry (source_key, vendor, source_table, queue_names, value_weight, root_cause_class, profile, notes) values -- global kill-switch + global daily dispatch ceiling (belt-and-suspenders over per-source budget) ('__global__', null, null, '{}', 1, 'staleness', 'erik-safe', 'Global control row: enabled=false disarms the entire loop; budget_per_day is the global daily dispatch ceiling.'), -- stock sources (higher value_weight — stock freshness is core to sourcing) ('atgbics/stock_observations', 'ATGBICS', 'stock_observations', '{scrape:pricing:atgbics}', 1.5, 'code_fix', 'erik-safe', 'Silent stock write-path break: price fresh, stock 0 rows/54d. Escalate to a human; a re-scrape cannot fix a code bug.'), ('flexoptix/stock_observations', 'Flexoptix', 'stock_observations', '{scrape:pricing:flexoptix}', 1.5, 'staleness', 'erik-safe', 'First source to arm out of plan-mode: lowest-risk staleness, erik-safe queue.'), ('fibermall/stock_observations', 'FiberMall', 'stock_observations', '{scrape:pricing:fibermall}', 1.5, 'staleness', 'erik-safe', null), ('qsfptek/stock_observations', 'QSFPTEK', 'stock_observations', '{scrape:pricing:qsfptek}', 1.5, 'staleness', 'erik-safe', 'The only currently-live stock writer — the reference for a working stock-write path.'), ('naddod/stock_observations', 'NADDOD', 'stock_observations', '{scrape:pricing:naddod}', 1.5, 'staleness', 'erik-safe', null), ('gbics/stock_observations', 'GBICS', 'stock_observations', '{scrape:pricing:gbics}', 1.5, 'staleness', 'erik-safe', 'Queue not in ERIK_SAFE_QUEUES — needs pi-fetch/proxmox profile to actually dispatch.'), ('sfpcables/stock_observations', 'SFPcables', 'stock_observations', '{scrape:pricing:sfpcables}', 1.5, 'staleness', 'erik-safe', 'Queue not in ERIK_SAFE_QUEUES — needs pi-fetch/proxmox profile to actually dispatch.'), -- price sources ('atgbics/price_observations', 'ATGBICS', 'price_observations', '{scrape:pricing:atgbics}', 1.0, 'staleness', 'erik-safe', null), ('flexoptix/price_observations', 'Flexoptix', 'price_observations', '{scrape:pricing:flexoptix}', 1.0, 'staleness', 'erik-safe', null), ('fibermall/price_observations', 'FiberMall', 'price_observations', '{scrape:pricing:fibermall}', 1.0, 'staleness', 'erik-safe', null), ('qsfptek/price_observations', 'QSFPTEK', 'price_observations', '{scrape:pricing:qsfptek}', 1.0, 'staleness', 'erik-safe', null), ('naddod/price_observations', 'NADDOD', 'price_observations', '{scrape:pricing:naddod}', 1.0, 'staleness', 'erik-safe', null), ('gbics/price_observations', 'GBICS', 'price_observations', '{scrape:pricing:gbics}', 1.0, 'staleness', 'erik-safe', 'Queue not in ERIK_SAFE_QUEUES — needs pi-fetch/proxmox profile to actually dispatch.'), ('sfpcables/price_observations', 'SFPcables', 'price_observations', '{scrape:pricing:sfpcables}', 1.0, 'staleness', 'erik-safe', 'Queue not in ERIK_SAFE_QUEUES — needs pi-fetch/proxmox profile to actually dispatch.'), ('10gtek/price_observations', '10Gtek', 'price_observations', '{scrape:pricing:10gtek}', 1.0, 'staleness', 'proxmox-heavy', 'scrape:pricing:10gtek is a HEAVY queue — only the proxmox-heavy profile dispatches it.') on conflict (source_key) do nothing; -- set the global daily dispatch ceiling on the __global__ row (soak default: 20/day) update source_registry set budget_per_day = 20 where source_key = '__global__'; -- Proof (read anytime): -- select * from v_dqo_targets; -- what the loop would act on right now -- select enabled from source_registry where source_key='__global__'; -- kill-switch state -- select source_key, outcome, mode, dispatched_at from self_heal_dispatch order by id desc limit 20;