From 60cffb3585e56614b07b98e48ef542431dbe2c4d Mon Sep 17 00:00:00 2001
From: Rene Fichtmueller <rf@flexoptix.net>
Date: Mon, 6 Jul 2026 11:38:19 +0200
Subject: [PATCH] feat(tip): autonomous data-quality orchestrator (tip-dqo,
 plan-mode)

Close the missing feedback loop. Detection (v_scraper_health, v_data_quality_alerts,
check_switch_quality) was accurate but inert; dispatch was blind fixed-cadence
(boss.send() is never called from any health path). tip-dqo is a separate, read-mostly
control plane that closes DETECT -> PRIORITIZE -> DISPATCH -> VERIFY -> RECHECK against
observation-table freshness (never pg-boss job state), with DB-enforced guardrails.

- sql/129: source_registry (signal->queue map + per-source SLA + guardrail contract +
  global kill-switch), self_heal_dispatch append-only ledger, scraper_run_log,
  acknowledged_exceptions, v_dqo_targets ranked view.
- orchestrator/: control-loop, self-heal-guards (pure, unit-tested), escalation-sink
  (ntfy + no-op fallback), index entrypoint. Reuses the verification-robots dispatch
  idiom (extracted shared dispatchQueues) and the quality views as-is.
- Guardrails: kill-switch, plan-mode default, per-source cooldown/budget/backoff/
  circuit-breaker, load-guard, erik-safe profile cap, delta-gated escalation. Root-cause
  routing escalates code_fix/blocked (e.g. ATGBICS silent stock-write break) to a human
  instead of looping retries; only staleness is auto-dispatched.
- security: remove hardcoded 'tip_dev_2026' DB password fallback across 6 sites ->
  fail-closed requireDbPassword() / dbConnectionString().

SELF_HEAL_PLAN_ONLY defaults true (dispatches nothing; writes plan rows for review).
Verified against the live DB: detects ATGBICS-stock dead/54d -> escalate, Flexoptix-stock
stale/30d -> plan-dispatch; guards + ledger correct; 8/8 guard unit tests pass.
---
 packages/scraper/package.json                 |   5 +-
 .../src/crawler-llm/training-data-writer.ts   |   2 +-
 .../scraper/src/orchestrator/control-loop.ts  | 393 ++++++++++++++++++
 .../src/orchestrator/escalation-sink.ts       |  52 +++
 packages/scraper/src/orchestrator/index.ts    |  85 ++++
 .../src/orchestrator/self-heal-guards.test.ts |  78 ++++
 .../src/orchestrator/self-heal-guards.ts      | 125 ++++++
 .../scraper/src/robots/verification-robots.ts |  64 ++-
 packages/scraper/src/scheduler.ts             |   3 +-
 packages/scraper/src/utils/change-detector.ts |   3 +-
 packages/scraper/src/utils/db-credentials.ts  |  27 ++
 packages/scraper/src/utils/db.ts              |   3 +-
 .../scraper/src/utils/image-downloader.ts     |   3 +-
 scripts/migrate.ts                            |   7 +-
 sql/129-orchestrator-control-plane.sql        | 183 ++++++++
 15 files changed, 1004 insertions(+), 29 deletions(-)
 create mode 100644 packages/scraper/src/orchestrator/control-loop.ts
 create mode 100644 packages/scraper/src/orchestrator/escalation-sink.ts
 create mode 100644 packages/scraper/src/orchestrator/index.ts
 create mode 100644 packages/scraper/src/orchestrator/self-heal-guards.test.ts
 create mode 100644 packages/scraper/src/orchestrator/self-heal-guards.ts
 create mode 100644 packages/scraper/src/utils/db-credentials.ts
 create mode 100644 sql/129-orchestrator-control-plane.sql

diff --git a/packages/scraper/package.json b/packages/scraper/package.json
index c25099d..f7268dd 100644
--- a/packages/scraper/package.json
+++ b/packages/scraper/package.json
@@ -33,7 +33,10 @@
     "scrape:optcore": "tsx src/scrapers/optcore.ts",
     "scrape:news": "tsx src/scrapers/news.ts",
     "scrape:all": "tsx src/index.ts --all",
-    "robots:verification": "tsx src/robots/verification-robots.ts"
+    "robots:verification": "tsx src/robots/verification-robots.ts",
+    "dqo": "tsx src/orchestrator/index.ts",
+    "dqo:once": "tsx src/orchestrator/index.ts --once",
+    "dqo:test": "node --import tsx --test src/orchestrator/self-heal-guards.test.ts"
   },
   "dependencies": {
     "crawlee": "^3.12.0",
diff --git a/packages/scraper/src/crawler-llm/training-data-writer.ts b/packages/scraper/src/crawler-llm/training-data-writer.ts
index 6d4ddc9..a55a72c 100644
--- a/packages/scraper/src/crawler-llm/training-data-writer.ts
+++ b/packages/scraper/src/crawler-llm/training-data-writer.ts
@@ -69,7 +69,7 @@ export interface CrawlExtraction {
 }
 
 export interface RobotExperience {
-  event: "status" | "tipllm_plan" | "queue_plan" | "queue_enqueued" | "queue_rejected" | "crawler_result";
+  event: "status" | "tipllm_plan" | "queue_plan" | "queue_enqueued" | "queue_rejected" | "crawler_result" | "dqo_dispatch" | "dqo_verify" | "dqo_escalation";
   observed_at: string;
   actor: string;
   profile?: string;
diff --git a/packages/scraper/src/orchestrator/control-loop.ts b/packages/scraper/src/orchestrator/control-loop.ts
new file mode 100644
index 0000000..5da81f9
--- /dev/null
+++ b/packages/scraper/src/orchestrator/control-loop.ts
@@ -0,0 +1,393 @@
+/**
+ * tip-dqo control loop — the closed loop TIP was missing.
+ *
+ *   DETECT      read v_dqo_targets (v_scraper_health joined onto source_registry) — DATA
+ *               freshness of the observation TABLES, never pg-boss job cadence.
+ *   PRIORITIZE  the view is pre-ranked (dead>stale, weighted); split staleness vs
+ *               code_fix/blocked. Only 'staleness' is dispatch-eligible.
+ *   DISPATCH    guardrails (self-heal-guards) + plan-mode + profile cap, then the shared
+ *               verification-robots dispatchQueues() — exactly one boss.send path.
+ *   VERIFY      re-read the source's freshness for pending dispatch rows: did fresh rows
+ *               land? healed / ineffective(+strike). Catches completed-but-wrote-0 jobs.
+ *   ESCALATE    fresh=silent, stale/1-2 strikes=digest, breaker/dead=page (delta-gated by
+ *               acknowledged_exceptions and once-per-24h so the day-1 backlog never spams).
+ *
+ * INVARIANT: VERIFY + DETECT key off observation-table freshness (v_scraper_health), NOT
+ * pgboss.job='completed' — that is the exact blind spot that lets a green-queue/zero-row
+ * scraper look healthy. Do not "optimize" this to read pgboss.job.
+ */
+import { Pool } from "pg";
+import { loadavg } from "os";
+import { requireDbPassword } from "../utils/db-credentials";
+import {
+  dispatchQueues,
+  queuesForProfile,
+  defaultMaxQueues,
+  type RobotProfile,
+} from "../robots/verification-robots";
+import { canDispatch, type GuardConfig, type LedgerFacts } from "./self-heal-guards";
+import { notify } from "./escalation-sink";
+import { writeRobotExperience } from "../crawler-llm/training-data-writer";
+
+const PLAN_ONLY = process.env.SELF_HEAL_PLAN_ONLY !== "false"; // default true — dispatches NOTHING
+const MAX_LOAD = Number.parseFloat(process.env.DQO_MAX_LOAD || "2.5");
+const SETTLE_MINUTES = Number.parseInt(process.env.DQO_VERIFY_SETTLE_MINUTES || "20", 10);
+const ESCALATE_REPAGE_HOURS = 24;
+
+let poolSingleton: Pool | null = null;
+function getPool(): Pool {
+  if (!poolSingleton) {
+    poolSingleton = new Pool({
+      host: process.env.POSTGRES_HOST || "localhost",
+      port: parseInt(process.env.POSTGRES_PORT || "5433"),
+      database: process.env.POSTGRES_DB || "transceiver_db",
+      user: process.env.POSTGRES_USER || "tip",
+      password: requireDbPassword(),
+      max: 2, // read-mostly; kept small so the orchestrator never contends the daemon pool
+      idleTimeoutMillis: 10_000,
+      connectionTimeoutMillis: 5_000,
+    });
+  }
+  return poolSingleton;
+}
+
+export async function closePool(): Promise<void> {
+  if (poolSingleton) {
+    await poolSingleton.end();
+    poolSingleton = null;
+  }
+}
+
+interface Target {
+  source_key: string;
+  vendor: string;
+  source_table: string;
+  status: string;
+  days_since_last: number;
+  rows_30d: number;
+  last_observation: string | null;
+  root_cause_class: "staleness" | "code_fix" | "blocked";
+  queue_names: string[];
+  profile: RobotProfile;
+  dead_days: number;
+  cooldown_minutes: number;
+  budget_per_day: number;
+  max_backoff_hours: number;
+  circuit_breaker_strikes: number;
+  score: number;
+}
+
+function guardConfig(t: Target): GuardConfig {
+  return {
+    cooldown_minutes: t.cooldown_minutes,
+    budget_per_day: t.budget_per_day,
+    max_backoff_hours: t.max_backoff_hours,
+    circuit_breaker_strikes: t.circuit_breaker_strikes,
+  };
+}
+
+// ── kill-switch + global daily ceiling ────────────────────────────────────────
+async function globalGate(pool: Pool): Promise<{ armed: boolean; reason: string; globalBudget: number }> {
+  const { rows } = await pool.query(
+    `select enabled, budget_per_day from source_registry where source_key = '__global__'`,
+  );
+  if (rows.length === 0) return { armed: true, reason: "no __global__ row (default armed)", globalBudget: 9999 };
+  if (!rows[0].enabled) return { armed: false, reason: "kill-switch: __global__.enabled = false", globalBudget: 0 };
+  return { armed: true, reason: "armed", globalBudget: rows[0].budget_per_day };
+}
+
+async function globalDispatchesToday(pool: Pool): Promise<number> {
+  const { rows } = await pool.query(
+    `select count(*)::int n from self_heal_dispatch
+      where mode = 'dispatch' and dispatched_at >= date_trunc('day', now())`,
+  );
+  return rows[0]?.n ?? 0;
+}
+
+// ── DETECT ────────────────────────────────────────────────────────────────────
+export async function detect(pool: Pool): Promise<Target[]> {
+  const { rows } = await pool.query(`select * from v_dqo_targets`);
+  return rows as Target[];
+}
+
+async function ledgerFacts(pool: Pool, sourceKey: string): Promise<LedgerFacts> {
+  const { rows } = await pool.query(
+    `select
+       (select max(dispatched_at) from self_heal_dispatch where source_key = $1 and mode = 'dispatch') as last_dispatch_at,
+       (select coalesce(strikes, 0) from self_heal_dispatch where source_key = $1 and mode = 'dispatch' order by dispatched_at desc limit 1) as last_strikes,
+       (select count(*)::int from self_heal_dispatch where source_key = $1 and mode = 'dispatch' and dispatched_at >= date_trunc('day', now())) as dispatches_today`,
+    [sourceKey],
+  );
+  const r = rows[0] ?? {};
+  return {
+    last_dispatch_at: r.last_dispatch_at ? new Date(r.last_dispatch_at) : null,
+    last_strikes: Number(r.last_strikes ?? 0),
+    dispatches_today: Number(r.dispatches_today ?? 0),
+  };
+}
+
+async function insertLedger(
+  pool: Pool,
+  row: {
+    source_key: string;
+    queues: string[];
+    reason: string;
+    mode: "plan" | "dispatch";
+    rows_before: number | null;
+    last_obs_before: string | null;
+    outcome: "pending" | "skipped" | "escalated";
+    strikes: number;
+    run_id: string;
+  },
+): Promise<void> {
+  await pool.query(
+    `insert into self_heal_dispatch
+       (source_key, queues, reason, mode, rows_before, last_obs_before, outcome, strikes, run_id)
+     values ($1,$2,$3,$4,$5,$6,$7,$8,$9)`,
+    [row.source_key, row.queues, row.reason, row.mode, row.rows_before, row.last_obs_before, row.outcome, row.strikes, row.run_id],
+  );
+}
+
+// ── DISPATCH (one self-heal tick) ─────────────────────────────────────────────
+export interface TickResult {
+  planOnly: boolean;
+  detected: number;
+  planned: number;
+  dispatched: number;
+  skipped: number;
+  escalated: number;
+  lines: string[];
+}
+
+export async function runTick(): Promise<TickResult> {
+  const pool = getPool();
+  const runId = `dqo-self-heal-${new Date().toISOString()}`;
+  const load1 = loadavg()[0];
+  const res: TickResult = { planOnly: PLAN_ONLY, detected: 0, planned: 0, dispatched: 0, skipped: 0, escalated: 0, lines: [] };
+
+  const gate = await globalGate(pool);
+  if (!gate.armed) {
+    res.lines.push(`[HALT] ${gate.reason}`);
+    return res;
+  }
+  const globalToday = await globalDispatchesToday(pool);
+  const globalRemaining = gate.globalBudget - globalToday;
+
+  const targets = await detect(pool);
+  res.detected = targets.length;
+  if (targets.length === 0) {
+    res.lines.push("no stale/dead sources — nothing to do");
+    return res;
+  }
+
+  for (const t of targets) {
+    // Root-cause routing: only staleness is auto-dispatched. code_fix/blocked escalate.
+    if (t.root_cause_class !== "staleness") {
+      await escalateSource(pool, t, `root_cause=${t.root_cause_class} (${t.status} ${t.days_since_last}d) — a re-scrape cannot fix this; human required`, runId);
+      res.escalated++;
+      res.lines.push(`  ESCALATE ${t.source_key} — ${t.root_cause_class} (${t.status} ${t.days_since_last}d)`);
+      continue;
+    }
+
+    // A dead staleness source past dead_days also escalates (retrying won't help a truly dead source).
+    if (t.status === "dead" && t.days_since_last > t.dead_days) {
+      await escalateSource(pool, t, `dead ${t.days_since_last}d > dead_days ${t.dead_days} — escalating`, runId);
+      res.escalated++;
+      res.lines.push(`  ESCALATE ${t.source_key} — dead ${t.days_since_last}d`);
+      continue;
+    }
+
+    const facts = await ledgerFacts(pool, t.source_key);
+    const verdict = canDispatch(facts, guardConfig(t), new Date(), load1);
+
+    // Profile filter + cap on the queues.
+    const eligible = queuesForProfile(t.queue_names, t.profile).slice(0, defaultMaxQueues(t.profile));
+
+    if (eligible.length === 0) {
+      await insertLedger(pool, {
+        source_key: t.source_key, queues: [], reason: `no queue eligible under profile '${t.profile}'`,
+        mode: "plan", rows_before: t.rows_30d, last_obs_before: t.last_observation, outcome: "skipped", strikes: facts.last_strikes, run_id: runId,
+      });
+      res.skipped++;
+      res.lines.push(`  SKIP ${t.source_key} — no queue eligible under '${t.profile}'`);
+      continue;
+    }
+
+    if (!verdict.allow) {
+      await insertLedger(pool, {
+        source_key: t.source_key, queues: eligible, reason: `guard: ${verdict.reason}`,
+        mode: "plan", rows_before: t.rows_30d, last_obs_before: t.last_observation, outcome: "skipped", strikes: facts.last_strikes, run_id: runId,
+      });
+      res.skipped++;
+      res.lines.push(`  SKIP ${t.source_key} — ${verdict.reason}`);
+      continue;
+    }
+
+    if (PLAN_ONLY || globalRemaining <= 0) {
+      const why = PLAN_ONLY ? "plan-mode (SELF_HEAL_PLAN_ONLY)" : `global budget exhausted (${globalToday}/${gate.globalBudget})`;
+      await insertLedger(pool, {
+        source_key: t.source_key, queues: eligible, reason: `WOULD DISPATCH → ${eligible.join(", ")} [${why}]`,
+        mode: "plan", rows_before: t.rows_30d, last_obs_before: t.last_observation, outcome: "pending", strikes: facts.last_strikes, run_id: runId,
+      });
+      res.planned++;
+      res.lines.push(`  PLAN ${t.source_key} → ${eligible.join(", ")} [${why}]`);
+      writeRobotExperience({
+        event: "dqo_dispatch", observed_at: new Date().toISOString(), actor: "tip-dqo",
+        summary: `Plan: would dispatch ${t.source_key} (${t.status} ${t.days_since_last}d)`,
+        input: { source_key: t.source_key, status: t.status, days_since_last: t.days_since_last, eligible, why },
+        decision: { mode: "plan", dispatched: false }, safety_notes: ["Plan mode — no jobs enqueued."],
+      });
+      continue;
+    }
+
+    // ARMED: real dispatch.
+    await dispatchQueues(eligible, { source: "dqo:self-heal", source_key: t.source_key, run_id: runId, reason: `${t.status} ${t.days_since_last}d`, enqueued_at: new Date().toISOString() });
+    await insertLedger(pool, {
+      source_key: t.source_key, queues: eligible, reason: `dispatched (${t.status} ${t.days_since_last}d)`,
+      mode: "dispatch", rows_before: t.rows_30d, last_obs_before: t.last_observation, outcome: "pending", strikes: facts.last_strikes, run_id: runId,
+    });
+    res.dispatched++;
+    res.lines.push(`  DISPATCH ${t.source_key} → ${eligible.join(", ")}`);
+    writeRobotExperience({
+      event: "dqo_dispatch", observed_at: new Date().toISOString(), actor: "tip-dqo",
+      summary: `Dispatched ${t.source_key} (${t.status} ${t.days_since_last}d)`,
+      input: { source_key: t.source_key, eligible }, decision: { mode: "dispatch", dispatched: true },
+      safety_notes: ["Armed dispatch within guardrails."],
+    }, { flush: true });
+  }
+
+  return res;
+}
+
+// ── VERIFY (one verify tick) ──────────────────────────────────────────────────
+export interface VerifyResult {
+  checked: number;
+  healed: number;
+  ineffective: number;
+  lines: string[];
+}
+
+export async function runVerifyTick(): Promise<VerifyResult> {
+  const pool = getPool();
+  const res: VerifyResult = { checked: 0, healed: 0, ineffective: 0, lines: [] };
+
+  // pending REAL dispatches past the settle window
+  const { rows: pending } = await pool.query(
+    `select d.id, d.source_key, d.rows_before, d.last_obs_before, d.strikes, d.run_id,
+            sr.vendor, sr.source_table, sr.dead_days, sr.circuit_breaker_strikes
+       from self_heal_dispatch d
+       join source_registry sr on sr.source_key = d.source_key
+      where d.mode = 'dispatch' and d.outcome = 'pending'
+        and d.dispatched_at < now() - ($1 || ' minutes')::interval`,
+    [String(SETTLE_MINUTES)],
+  );
+  // No pending dispatches → skip the expensive v_scraper_health scan entirely.
+  if (pending.length === 0) return res;
+
+  // Compute the (expensive) freshness view ONCE and index it, instead of re-running the
+  // whole aggregate per pending source. Keyed off observation TABLES, never pgboss.job.
+  const { rows: healthRows } = await pool.query(
+    `select vendor, source_table, rows_30d, last_observation from v_scraper_health`,
+  );
+  const health = new Map<string, { rows_30d: number | null; last_observation: string | null }>();
+  for (const hr of healthRows) {
+    health.set(`${hr.vendor}|${hr.source_table}`, { rows_30d: hr.rows_30d, last_observation: hr.last_observation });
+  }
+
+  for (const p of pending) {
+    res.checked++;
+    const h = health.get(`${p.vendor}|${p.source_table}`);
+    const rowsAfter: number | null = h?.rows_30d ?? null;
+    const lastObsAfter: string | null = h?.last_observation ?? null;
+    const advanced = lastObsAfter && p.last_obs_before && new Date(lastObsAfter) > new Date(p.last_obs_before);
+    const grew = rowsAfter != null && p.rows_before != null && rowsAfter > p.rows_before;
+
+    if (advanced || grew) {
+      await pool.query(
+        `update self_heal_dispatch set outcome='healed', checked_at=now(), rows_after=$2, last_obs_after=$3, strikes=0 where id=$1`,
+        [p.id, rowsAfter, lastObsAfter],
+      );
+      res.healed++;
+      res.lines.push(`  HEALED ${p.source_key} (rows ${p.rows_before}→${rowsAfter})`);
+      writeRobotExperience({
+        event: "dqo_verify", observed_at: new Date().toISOString(), actor: "tip-dqo",
+        summary: `Healed ${p.source_key}`, input: { source_key: p.source_key, rows_before: p.rows_before, rows_after: rowsAfter },
+        decision: { outcome: "healed" },
+      });
+    } else {
+      const newStrikes = Number(p.strikes ?? 0) + 1;
+      await pool.query(
+        `update self_heal_dispatch set outcome='ineffective', checked_at=now(), rows_after=$2, last_obs_after=$3, strikes=$4 where id=$1`,
+        [p.id, rowsAfter, lastObsAfter, newStrikes],
+      );
+      res.ineffective++;
+      res.lines.push(`  INEFFECTIVE ${p.source_key} (no new rows; strikes→${newStrikes})`);
+      writeRobotExperience({
+        event: "dqo_verify", observed_at: new Date().toISOString(), actor: "tip-dqo",
+        summary: `Ineffective dispatch ${p.source_key} — no fresh rows landed`,
+        input: { source_key: p.source_key, rows_before: p.rows_before, rows_after: rowsAfter, strikes: newStrikes },
+        decision: { outcome: "ineffective" },
+      });
+      if (newStrikes >= Number(p.circuit_breaker_strikes ?? 3)) {
+        await escalateByKey(pool, p.source_key, `circuit-breaker open: ${newStrikes} ineffective dispatches — retrying wastes resources, human required`, p.run_id);
+        res.lines.push(`  ESCALATE ${p.source_key} — breaker open (${newStrikes} strikes)`);
+      }
+    }
+  }
+
+  return res;
+}
+
+// ── ESCALATE (delta-gated, once-per-24h per source) ───────────────────────────
+async function alreadyEscalatedRecently(pool: Pool, sourceKey: string): Promise<boolean> {
+  const { rows } = await pool.query(
+    `select 1 from self_heal_dispatch
+      where source_key = $1 and outcome = 'escalated'
+        and dispatched_at > now() - ($2 || ' hours')::interval limit 1`,
+    [sourceKey, String(ESCALATE_REPAGE_HOURS)],
+  );
+  return rows.length > 0;
+}
+
+async function isAcknowledged(pool: Pool, sourceKey: string): Promise<boolean> {
+  const { rows } = await pool.query(
+    `select 1 from acknowledged_exceptions
+      where (source_key = $1 or source_key is null)
+        and (ack_until is null or ack_until > now()) limit 1`,
+    [sourceKey],
+  );
+  return rows.length > 0;
+}
+
+async function escalateSource(pool: Pool, t: Target, reason: string, runId: string): Promise<void> {
+  await escalateByKey(pool, t.source_key, reason, runId, { rows: t.rows_30d, lastObs: t.last_observation, status: t.status });
+}
+
+async function escalateByKey(
+  pool: Pool,
+  sourceKey: string,
+  reason: string,
+  runId: string,
+  ctx?: { rows?: number; lastObs?: string | null; status?: string },
+): Promise<void> {
+  // delta-gate: suppress if acknowledged, and never re-page the same source within 24h
+  if (await isAcknowledged(pool, sourceKey)) return;
+  if (await alreadyEscalatedRecently(pool, sourceKey)) return;
+
+  await insertLedger(pool, {
+    source_key: sourceKey, queues: [], reason, mode: "plan",
+    rows_before: ctx?.rows ?? null, last_obs_before: ctx?.lastObs ?? null, outcome: "escalated", strikes: 0, run_id: runId,
+  });
+  await notify("page", `TIP data-quality: ${sourceKey}`, reason);
+  writeRobotExperience({
+    event: "dqo_escalation", observed_at: new Date().toISOString(), actor: "tip-dqo",
+    summary: `Escalated ${sourceKey}: ${reason}`, input: { source_key: sourceKey, ...ctx },
+    decision: { escalated: true }, safety_notes: ["Paged once; suppressed for 24h + honors acknowledged_exceptions."],
+  }, { flush: true });
+}
+
+export async function runOnce(): Promise<{ tick: TickResult; verify: VerifyResult }> {
+  const tick = await runTick();
+  const verify = await runVerifyTick();
+  return { tick, verify };
+}
diff --git a/packages/scraper/src/orchestrator/escalation-sink.ts b/packages/scraper/src/orchestrator/escalation-sink.ts
new file mode 100644
index 0000000..1becc7e
--- /dev/null
+++ b/packages/scraper/src/orchestrator/escalation-sink.ts
@@ -0,0 +1,52 @@
+/**
+ * Notification sink for the tip-dqo escalation ladder.
+ *
+ * Default channel: self-hosted ntfy on Erik (no external secret, fits "Gitea = EVERYTHING").
+ * Set NTFY_URL (e.g. https://ntfy.fichtmueller.org) and optionally NTFY_TOPIC (default
+ * "tip-dqo"). If NTFY_URL is unset, notify() is a logged no-op so the loop still runs and
+ * the escalation is recorded in the ledger — the channel can be wired later without a
+ * code change.
+ */
+
+export type EscalationLevel = "digest" | "page";
+
+export interface NotifyResult {
+  sent: boolean;
+  channel: "ntfy" | "noop" | "error";
+  detail?: string;
+}
+
+export async function notify(
+  level: EscalationLevel,
+  title: string,
+  body: string,
+): Promise<NotifyResult> {
+  const base = process.env.NTFY_URL?.replace(/\/+$/, "");
+  const topic = process.env.NTFY_TOPIC || "tip-dqo";
+
+  if (!base) {
+    console.warn(`[dqo:escalation:noop] (${level}) ${title} — ${body} [set NTFY_URL to enable]`);
+    return { sent: false, channel: "noop" };
+  }
+
+  try {
+    const resp = await fetch(`${base}/${topic}`, {
+      method: "POST",
+      headers: {
+        Title: title,
+        Priority: level === "page" ? "high" : "default",
+        Tags: level === "page" ? "rotating_light" : "warning",
+      },
+      body,
+      signal: AbortSignal.timeout(10_000),
+    });
+    if (!resp.ok) {
+      return { sent: false, channel: "error", detail: `HTTP ${resp.status}` };
+    }
+    return { sent: true, channel: "ntfy" };
+  } catch (err) {
+    const detail = err instanceof Error ? err.message : String(err);
+    console.error(`[dqo:escalation] ntfy failed: ${detail}`);
+    return { sent: false, channel: "error", detail };
+  }
+}
diff --git a/packages/scraper/src/orchestrator/index.ts b/packages/scraper/src/orchestrator/index.ts
new file mode 100644
index 0000000..2e6faa1
--- /dev/null
+++ b/packages/scraper/src/orchestrator/index.ts
@@ -0,0 +1,85 @@
+/**
+ * tip-dqo — TIP Autonomous Data-Quality Orchestrator (entrypoint).
+ *
+ * A SEPARATE always-on process (PM2 app `tip-dqo`), deliberately NOT grafted into the
+ * tip-scraper-daemon event loop, so its blast radius is isolated and it never contends
+ * the daemon's pg Pool. Read-mostly: its only writes are to its own ledger/registry tables.
+ *
+ *   Default (no args)  : self-heal tick every DQO_SELF_HEAL_MINUTES (30), verify tick every
+ *                        DQO_VERIFY_MINUTES (15). Runs until SIGINT/SIGTERM.
+ *   --once             : run one self-heal + verify tick, print the plan, exit. Used to
+ *                        PROVE the loop against the live DB without scheduling anything.
+ *
+ * SELF_HEAL_PLAN_ONLY defaults to "true" (dispatches NOTHING; writes plan rows a human
+ * reviews). Set SELF_HEAL_PLAN_ONLY=false only to arm live dispatch.
+ */
+import { config } from "dotenv";
+import { join } from "path";
+import { runOnce, runTick, runVerifyTick, closePool } from "./control-loop";
+
+config({ path: join(__dirname, "..", "..", "..", "..", ".env") });
+
+const SELF_HEAL_MINUTES = Number.parseInt(process.env.DQO_SELF_HEAL_MINUTES || "30", 10);
+const VERIFY_MINUTES = Number.parseInt(process.env.DQO_VERIFY_MINUTES || "15", 10);
+const PLAN_ONLY = process.env.SELF_HEAL_PLAN_ONLY !== "false";
+
+function stamp(): string {
+  return new Date().toISOString().replace("T", " ").slice(0, 19);
+}
+
+async function once(): Promise<void> {
+  console.log(`\n=== tip-dqo one-shot (${PLAN_ONLY ? "PLAN-ONLY" : "ARMED"}) ${stamp()} ===`);
+  const { tick, verify } = await runOnce();
+  console.log(`\n[self-heal] detected=${tick.detected} planned=${tick.planned} dispatched=${tick.dispatched} skipped=${tick.skipped} escalated=${tick.escalated}`);
+  for (const l of tick.lines) console.log(l);
+  console.log(`\n[verify] checked=${verify.checked} healed=${verify.healed} ineffective=${verify.ineffective}`);
+  for (const l of verify.lines) console.log(l);
+  console.log("");
+  await closePool();
+}
+
+async function loop(): Promise<void> {
+  console.log(`=== tip-dqo starting (${PLAN_ONLY ? "PLAN-ONLY" : "ARMED"}) — self-heal ${SELF_HEAL_MINUTES}m / verify ${VERIFY_MINUTES}m ===`);
+
+  const selfHeal = async () => {
+    try {
+      const r = await runTick();
+      console.log(`[${stamp()}] self-heal: detected=${r.detected} planned=${r.planned} dispatched=${r.dispatched} skipped=${r.skipped} escalated=${r.escalated}`);
+      for (const l of r.lines) console.log(l);
+    } catch (err) {
+      console.error(`[${stamp()}] self-heal error:`, err instanceof Error ? err.message : err);
+    }
+  };
+  const verify = async () => {
+    try {
+      const r = await runVerifyTick();
+      if (r.checked > 0) {
+        console.log(`[${stamp()}] verify: checked=${r.checked} healed=${r.healed} ineffective=${r.ineffective}`);
+        for (const l of r.lines) console.log(l);
+      }
+    } catch (err) {
+      console.error(`[${stamp()}] verify error:`, err instanceof Error ? err.message : err);
+    }
+  };
+
+  await selfHeal(); // run once on boot
+  const t1 = setInterval(selfHeal, SELF_HEAL_MINUTES * 60_000);
+  const t2 = setInterval(verify, VERIFY_MINUTES * 60_000);
+
+  const shutdown = async (sig: string) => {
+    console.log(`\n[${stamp()}] ${sig} — shutting down tip-dqo`);
+    clearInterval(t1);
+    clearInterval(t2);
+    await closePool().catch(() => {});
+    process.exit(0);
+  };
+  process.on("SIGINT", () => void shutdown("SIGINT"));
+  process.on("SIGTERM", () => void shutdown("SIGTERM"));
+}
+
+const isOnce = process.argv.includes("--once");
+(isOnce ? once() : loop()).catch(async (err) => {
+  console.error("tip-dqo fatal:", err);
+  await closePool().catch(() => {});
+  process.exit(1);
+});
diff --git a/packages/scraper/src/orchestrator/self-heal-guards.test.ts b/packages/scraper/src/orchestrator/self-heal-guards.test.ts
new file mode 100644
index 0000000..ff0c283
--- /dev/null
+++ b/packages/scraper/src/orchestrator/self-heal-guards.test.ts
@@ -0,0 +1,78 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import {
+  isLoadAcceptable,
+  minIntervalOk,
+  budgetOk,
+  backoffElapsed,
+  breakerTripped,
+  canDispatch,
+  type GuardConfig,
+  type LedgerFacts,
+} from "./self-heal-guards";
+
+const CFG: GuardConfig = {
+  cooldown_minutes: 180,
+  budget_per_day: 4,
+  max_backoff_hours: 24,
+  circuit_breaker_strikes: 3,
+};
+const NOW = new Date("2026-07-06T12:00:00Z");
+const fresh: LedgerFacts = { last_dispatch_at: null, last_strikes: 0, dispatches_today: 0 };
+
+test("load guard", () => {
+  assert.equal(isLoadAcceptable(1.2).allow, true);
+  assert.equal(isLoadAcceptable(3.0).allow, false);
+  assert.equal(isLoadAcceptable(2.5).allow, true); // boundary: not greater than
+});
+
+test("min-interval blocks within cooldown, allows after", () => {
+  const recent = { ...fresh, last_dispatch_at: new Date("2026-07-06T11:00:00Z") }; // 60min ago
+  assert.equal(minIntervalOk(recent, CFG, NOW).allow, false);
+  const old = { ...fresh, last_dispatch_at: new Date("2026-07-06T08:00:00Z") }; // 240min ago
+  assert.equal(minIntervalOk(old, CFG, NOW).allow, true);
+  assert.equal(minIntervalOk(fresh, CFG, NOW).allow, true); // never dispatched
+});
+
+test("daily budget", () => {
+  assert.equal(budgetOk({ ...fresh, dispatches_today: 3 }, CFG).allow, true);
+  assert.equal(budgetOk({ ...fresh, dispatches_today: 4 }, CFG).allow, false);
+  assert.equal(budgetOk({ ...fresh, dispatches_today: 9 }, CFG).allow, false);
+});
+
+test("exponential backoff scales with strikes and caps", () => {
+  // 2 strikes → 4h backoff. Dispatched 1h ago → still backing off.
+  const s2recent = { last_dispatch_at: new Date("2026-07-06T11:00:00Z"), last_strikes: 2, dispatches_today: 0 };
+  assert.equal(backoffElapsed(s2recent, CFG, NOW).allow, false);
+  // 2 strikes → 4h backoff. Dispatched 5h ago → elapsed.
+  const s2old = { last_dispatch_at: new Date("2026-07-06T07:00:00Z"), last_strikes: 2, dispatches_today: 0 };
+  assert.equal(backoffElapsed(s2old, CFG, NOW).allow, true);
+  // 0 strikes → no backoff
+  assert.equal(backoffElapsed(fresh, CFG, NOW).allow, true);
+  // huge strikes → capped at max_backoff_hours (24h), dispatched 25h ago → elapsed
+  const capped = { last_dispatch_at: new Date("2026-07-05T11:00:00Z"), last_strikes: 10, dispatches_today: 0 };
+  assert.equal(backoffElapsed(capped, CFG, NOW).allow, true);
+});
+
+test("circuit breaker opens at threshold", () => {
+  assert.equal(breakerTripped({ ...fresh, last_strikes: 2 }, CFG).allow, true);
+  assert.equal(breakerTripped({ ...fresh, last_strikes: 3 }, CFG).allow, false);
+  assert.equal(breakerTripped({ ...fresh, last_strikes: 5 }, CFG).allow, false);
+});
+
+test("canDispatch: clean source under low load passes", () => {
+  assert.equal(canDispatch(fresh, CFG, NOW, 1.0).allow, true);
+});
+
+test("canDispatch: breaker is reported before other brakes", () => {
+  const tripped = { last_dispatch_at: new Date("2026-07-06T11:59:00Z"), last_strikes: 3, dispatches_today: 9 };
+  const v = canDispatch(tripped, CFG, NOW, 5.0);
+  assert.equal(v.allow, false);
+  assert.match(v.reason, /breaker OPEN/);
+});
+
+test("canDispatch: high load blocks an otherwise-clean source", () => {
+  const v = canDispatch(fresh, CFG, NOW, 9.0);
+  assert.equal(v.allow, false);
+  assert.match(v.reason, /load/);
+});
diff --git a/packages/scraper/src/orchestrator/self-heal-guards.ts b/packages/scraper/src/orchestrator/self-heal-guards.ts
new file mode 100644
index 0000000..ce47547
--- /dev/null
+++ b/packages/scraper/src/orchestrator/self-heal-guards.ts
@@ -0,0 +1,125 @@
+/**
+ * Cost-safety guardrails for the tip-dqo self-heal loop — PURE, network-free, unit-testable.
+ *
+ * The controller must NEVER hammer a source. These functions turn ledger facts +
+ * the source's guardrail config into a typed {allow, reason} verdict, so every skip is
+ * explainable and logged (never a silent no-op). The DB reads live in control-loop.ts;
+ * the DECISIONS live here and are tested with node:test against fixture facts.
+ *
+ * Nine compounding brakes (this module implements 5 of them; the rest — kill-switch,
+ * plan-mode, root-cause routing, profile cap — live in control-loop.ts):
+ *   min-interval · daily budget · exponential backoff · circuit-breaker · load-guard
+ */
+
+export interface GuardVerdict {
+  allow: boolean;
+  reason: string;
+}
+
+/** The per-source guardrail contract, straight from source_registry. */
+export interface GuardConfig {
+  cooldown_minutes: number;
+  budget_per_day: number;
+  max_backoff_hours: number;
+  circuit_breaker_strikes: number;
+}
+
+/** Facts read from the self_heal_dispatch ledger for one source. */
+export interface LedgerFacts {
+  /** dispatched_at of the most recent mode='dispatch' row, or null if never dispatched. */
+  last_dispatch_at: Date | null;
+  /** strikes as of the most recent verified dispatch (running no-progress count). */
+  last_strikes: number;
+  /** count of mode='dispatch' rows for this source since local midnight. */
+  dispatches_today: number;
+}
+
+const HOUR_MS = 3_600_000;
+const MIN_MS = 60_000;
+
+/**
+ * Load guard — do not dispatch when the box is already busy. Pure: caller passes the
+ * 1-minute load average (os.loadavg()[0]) so this stays testable. Mirrors the existing
+ * scheduler.ts isLoadAcceptable so self-heal never fights the load guard.
+ */
+export function isLoadAcceptable(load1: number, maxLoad = 2.5): GuardVerdict {
+  if (load1 > maxLoad) {
+    return { allow: false, reason: `load ${load1.toFixed(2)} > ${maxLoad}` };
+  }
+  return { allow: true, reason: `load ${load1.toFixed(2)} <= ${maxLoad}` };
+}
+
+/** Per-source minimum interval — a source cannot be re-fired within its cooldown window. */
+export function minIntervalOk(facts: LedgerFacts, cfg: GuardConfig, now: Date): GuardVerdict {
+  if (!facts.last_dispatch_at) return { allow: true, reason: "no prior dispatch" };
+  const elapsedMin = (now.getTime() - facts.last_dispatch_at.getTime()) / MIN_MS;
+  if (elapsedMin < cfg.cooldown_minutes) {
+    return {
+      allow: false,
+      reason: `cooldown: ${Math.round(elapsedMin)}min < ${cfg.cooldown_minutes}min since last dispatch`,
+    };
+  }
+  return { allow: true, reason: `cooldown elapsed (${Math.round(elapsedMin)}min)` };
+}
+
+/** Daily budget — no more than budget_per_day dispatches for this source since midnight. */
+export function budgetOk(facts: LedgerFacts, cfg: GuardConfig): GuardVerdict {
+  if (facts.dispatches_today >= cfg.budget_per_day) {
+    return {
+      allow: false,
+      reason: `budget: ${facts.dispatches_today}/${cfg.budget_per_day} dispatches used today`,
+    };
+  }
+  return { allow: true, reason: `budget ${facts.dispatches_today}/${cfg.budget_per_day}` };
+}
+
+/**
+ * Exponential backoff — a repeatedly-ineffective source backs off 2^strikes hours
+ * (capped at max_backoff_hours) instead of retrying every tick.
+ */
+export function backoffElapsed(facts: LedgerFacts, cfg: GuardConfig, now: Date): GuardVerdict {
+  if (facts.last_strikes <= 0 || !facts.last_dispatch_at) {
+    return { allow: true, reason: "no backoff (0 strikes)" };
+  }
+  const backoffHours = Math.min(Math.pow(2, facts.last_strikes), cfg.max_backoff_hours);
+  const readyAt = new Date(facts.last_dispatch_at.getTime() + backoffHours * HOUR_MS);
+  if (now < readyAt) {
+    return {
+      allow: false,
+      reason: `backoff: ${facts.last_strikes} strikes → wait ${backoffHours}h (ready ${readyAt.toISOString()})`,
+    };
+  }
+  return { allow: true, reason: `backoff elapsed (${backoffHours}h after ${facts.last_strikes} strikes)` };
+}
+
+/** Circuit-breaker — at N strikes the source is removed from the retry pool (escalated instead). */
+export function breakerTripped(facts: LedgerFacts, cfg: GuardConfig): GuardVerdict {
+  if (facts.last_strikes >= cfg.circuit_breaker_strikes) {
+    return {
+      allow: false,
+      reason: `breaker OPEN: ${facts.last_strikes} >= ${cfg.circuit_breaker_strikes} strikes`,
+    };
+  }
+  return { allow: true, reason: `breaker closed (${facts.last_strikes}/${cfg.circuit_breaker_strikes})` };
+}
+
+/**
+ * Combine all guardrails into one verdict. Returns the FIRST failing brake so the skip
+ * reason is specific. load1 is the current 1-min load average.
+ */
+export function canDispatch(
+  facts: LedgerFacts,
+  cfg: GuardConfig,
+  now: Date,
+  load1: number,
+): GuardVerdict {
+  const checks = [
+    breakerTripped(facts, cfg),   // breaker first: an open breaker means "escalate, don't retry"
+    budgetOk(facts, cfg),
+    minIntervalOk(facts, cfg, now),
+    backoffElapsed(facts, cfg, now),
+    isLoadAcceptable(load1),
+  ];
+  const failed = checks.find((c) => !c.allow);
+  return failed ?? { allow: true, reason: "all guardrails passed" };
+}
diff --git a/packages/scraper/src/robots/verification-robots.ts b/packages/scraper/src/robots/verification-robots.ts
index ac067ee..b79c1a4 100644
--- a/packages/scraper/src/robots/verification-robots.ts
+++ b/packages/scraper/src/robots/verification-robots.ts
@@ -20,11 +20,12 @@ import { join } from "path";
 import PgBoss from "pg-boss";
 import { pool } from "../utils/db";
 import { writeRobotExperience } from "../crawler-llm/training-data-writer";
+import { dbConnectionString } from "../utils/db-credentials";
 
 config({ path: join(__dirname, "..", "..", "..", "..", ".env") });
 
 type RobotWave = "details-fast-lane" | "priority-vendors" | "all";
-type RobotProfile = "erik-safe" | "pi-fetch" | "proxmox-heavy";
+export type RobotProfile = "erik-safe" | "pi-fetch" | "proxmox-heavy";
 
 interface VendorBlockerRow {
   vendor: string;
@@ -49,7 +50,7 @@ interface TipLlmResearchPlanResponse {
   error?: string;
 }
 
-const connectionString = `postgres://${process.env.POSTGRES_USER || "tip"}:${process.env.POSTGRES_PASSWORD || "tip_dev_2026"}@${process.env.POSTGRES_HOST || "localhost"}:${process.env.POSTGRES_PORT || "5433"}/${process.env.POSTGRES_DB || "transceiver_db"}`;
+const connectionString = dbConnectionString();
 const TIP_API_URL = process.env.TIP_API_URL || "http://127.0.0.1:3201";
 const TIP_API_TOKEN = process.env.TIP_API_TOKEN || process.env.TIP_TOKEN || "";
 
@@ -64,7 +65,7 @@ const DETAILS_FAST_LANE_QUEUES = [
   "maintenance:reconcile-verification",
 ];
 
-const VENDOR_QUEUE_MAP: Array<{ match: RegExp; queues: string[] }> = [
+export const VENDOR_QUEUE_MAP: Array<{ match: RegExp; queues: string[] }> = [
   { match: /juniper/i, queues: ["scrape:catalog:juniper-oem", "scrape:catalog:juniper-mx-oem", "scrape:catalog:juniper-qfx-oem", "scrape:compat:juniper", "discover:vendor:juniper"] },
   { match: /cisco/i, queues: ["scrape:catalog:cisco-nexus-oem", "scrape:catalog:cisco-catalyst-oem", "scrape:catalog:cisco-asr-oem", "scrape:compat:cisco", "discover:vendor:cisco-tmg"] },
   { match: /fs\.?com/i, queues: ["scrape:pricing:fs", "discover:vendor:fs-com"] },
@@ -83,7 +84,7 @@ const VENDOR_QUEUE_MAP: Array<{ match: RegExp; queues: string[] }> = [
   { match: /ciena/i, queues: ["scrape:catalog:ciena-oem", "scrape:catalog:ciena-waveserver-oem"] },
 ];
 
-const HEAVY_QUEUES = new Set([
+export const HEAVY_QUEUES = new Set([
   "scrape:pricing:fs",
   "scrape:pricing:10gtek",
   "scrape:pricing:prolabs",
@@ -93,7 +94,7 @@ const HEAVY_QUEUES = new Set([
   "scrape:compat:edgecore",
 ]);
 
-const ERIK_SAFE_QUEUES = new Set([
+export const ERIK_SAFE_QUEUES = new Set([
   "scrape:pricing:flexoptix",
   "scrape:pricing:fibermall",
   "scrape:pricing:atgbics",
@@ -104,11 +105,11 @@ const ERIK_SAFE_QUEUES = new Set([
   "maintenance:reconcile-verification",
 ]);
 
-function isDiscoveryQueue(queue: string): boolean {
+export function isDiscoveryQueue(queue: string): boolean {
   return queue.startsWith("discover:");
 }
 
-function queuesForProfile(queues: string[], profile: RobotProfile): string[] {
+export function queuesForProfile(queues: string[], profile: RobotProfile): string[] {
   if (profile === "proxmox-heavy") return queues;
 
   if (profile === "erik-safe") {
@@ -118,7 +119,7 @@ function queuesForProfile(queues: string[], profile: RobotProfile): string[] {
   return queues.filter((queue) => !HEAVY_QUEUES.has(queue) && !isDiscoveryQueue(queue));
 }
 
-function defaultMaxQueues(profile: RobotProfile): number {
+export function defaultMaxQueues(profile: RobotProfile): number {
   if (profile === "erik-safe") return 3;
   if (profile === "pi-fetch") return 10;
   return 30;
@@ -132,7 +133,7 @@ function unique(items: string[]): string[] {
   return [...new Set(items)];
 }
 
-async function createBoss(): Promise<PgBoss> {
+export async function createBoss(): Promise<PgBoss> {
   const boss = new PgBoss({
     connectionString,
     retryLimit: 1,
@@ -145,6 +146,32 @@ async function createBoss(): Promise<PgBoss> {
   return boss;
 }
 
+/**
+ * Single shared dispatch primitive: create each queue if needed, then send one job per
+ * queue with the standard safety options (retryLimit 1, 2h expiry). Used by BOTH
+ * enqueueRobotWave and the tip-dqo orchestrator so there is exactly ONE boss.send path
+ * with identical safety options across the whole platform.
+ */
+export async function dispatchQueues(
+  queues: string[],
+  data: Record<string, unknown>,
+): Promise<void> {
+  if (queues.length === 0) return;
+  const boss = await createBoss();
+  try {
+    for (const queue of queues) {
+      await boss.createQueue(queue).catch(() => {});
+      await (boss as unknown as { send: (name: string, data?: object, options?: object) => Promise<string | null> }).send(
+        queue,
+        data,
+        { retryLimit: 1, expireInSeconds: 7200 },
+      );
+    }
+  } finally {
+    await boss.stop();
+  }
+}
+
 export async function getVerificationStatus(): Promise<{ summary: Record<string, string>; vendors: VendorBlockerRow[] }> {
   const summaryResult = await pool.query(`
     SELECT
@@ -260,19 +287,12 @@ export async function enqueueRobotWave(
     return queues;
   }
 
-  const boss = await createBoss();
-  try {
-    for (const queue of queues) {
-      await boss.createQueue(queue).catch(() => {});
-      await (boss as unknown as { send: (name: string, data?: object, options?: object) => Promise<string | null> }).send(
-        queue,
-        { source: "verification-robots", wave, run_id: runId, enqueued_at: new Date().toISOString() },
-        { retryLimit: 1, expireInSeconds: 7200 },
-      );
-    }
-  } finally {
-    await boss.stop();
-  }
+  await dispatchQueues(queues, {
+    source: "verification-robots",
+    wave,
+    run_id: runId,
+    enqueued_at: new Date().toISOString(),
+  });
 
   console.log("\nJobs enqueued.");
   writeRobotExperience({
diff --git a/packages/scraper/src/scheduler.ts b/packages/scraper/src/scheduler.ts
index bb683f8..0765727 100644
--- a/packages/scraper/src/scheduler.ts
+++ b/packages/scraper/src/scheduler.ts
@@ -22,6 +22,7 @@ import PgBoss from "pg-boss";
 import { config } from "dotenv";
 import { join } from "path";
 import { loadavg } from "os";
+import { dbConnectionString } from "./utils/db-credentials";
 
 // withIsolatedStorage removed — all Crawlee scrapers now use makeCrawleeConfig()
 // for instance-level storage isolation. See packages/scraper/src/utils/crawlee-config.ts
@@ -42,7 +43,7 @@ function isLoadAcceptable(maxLoad = 2.5): boolean {
 
 config({ path: join(__dirname, "..", "..", "..", ".env") });
 
-const connectionString = `postgres://${process.env.POSTGRES_USER || "tip"}:${process.env.POSTGRES_PASSWORD || "tip_dev_2026"}@${process.env.POSTGRES_HOST || "localhost"}:${process.env.POSTGRES_PORT || "5433"}/${process.env.POSTGRES_DB || "transceiver_db"}`;
+const connectionString = dbConnectionString();
 
 type EquivalenceProduct = {
   part_number?: string | null;
diff --git a/packages/scraper/src/utils/change-detector.ts b/packages/scraper/src/utils/change-detector.ts
index f4e9023..117ed7c 100644
--- a/packages/scraper/src/utils/change-detector.ts
+++ b/packages/scraper/src/utils/change-detector.ts
@@ -5,13 +5,14 @@
  * and generates alerts for price changes, new products, stock changes.
  */
 import { Pool } from "pg";
+import { requireDbPassword } from "./db-credentials";
 
 const pool = new Pool({
   host: process.env.POSTGRES_HOST || "localhost",
   port: parseInt(process.env.POSTGRES_PORT || "5433"),
   database: process.env.POSTGRES_DB || "transceiver_db",
   user: process.env.POSTGRES_USER || "tip",
-  password: process.env.POSTGRES_PASSWORD || "tip_dev_2026",
+  password: requireDbPassword(),
   max: 3,
 });
 
diff --git a/packages/scraper/src/utils/db-credentials.ts b/packages/scraper/src/utils/db-credentials.ts
new file mode 100644
index 0000000..36b30cd
--- /dev/null
+++ b/packages/scraper/src/utils/db-credentials.ts
@@ -0,0 +1,27 @@
+/**
+ * Central DB credential resolution — fail-closed.
+ *
+ * Replaces the hardcoded `|| "tip_dev_2026"` password fallback that was scattered
+ * across the scraper package. A known default password in source is a credential leak
+ * (rules/common/security.md); if the env var is unset we THROW rather than silently
+ * connecting with a guessable password. Production always sets POSTGRES_PASSWORD, so
+ * this is a no-op at runtime there and only closes the hole.
+ */
+
+export function requireDbPassword(): string {
+  const pw = process.env.POSTGRES_PASSWORD || process.env.TIP_DB_PASS;
+  if (!pw) {
+    throw new Error(
+      "POSTGRES_PASSWORD (or TIP_DB_PASS) must be set — refusing to fall back to a hardcoded default password.",
+    );
+  }
+  return pw;
+}
+
+export function dbConnectionString(): string {
+  const user = process.env.POSTGRES_USER || "tip";
+  const host = process.env.POSTGRES_HOST || "localhost";
+  const port = process.env.POSTGRES_PORT || "5433";
+  const db = process.env.POSTGRES_DB || "transceiver_db";
+  return `postgres://${user}:${encodeURIComponent(requireDbPassword())}@${host}:${port}/${db}`;
+}
diff --git a/packages/scraper/src/utils/db.ts b/packages/scraper/src/utils/db.ts
index c0cd439..e76c4eb 100644
--- a/packages/scraper/src/utils/db.ts
+++ b/packages/scraper/src/utils/db.ts
@@ -3,6 +3,7 @@ import type { PoolConfig } from "pg";
 import { config } from "dotenv";
 import { join } from "path";
 import { contentHash } from "./hash";
+import { requireDbPassword } from "./db-credentials";
 
 config({ path: join(__dirname, "..", "..", "..", "..", ".env") });
 
@@ -14,7 +15,7 @@ const poolConfig: PoolConfig = !hasExplicitDbConfig && process.env.DATABASE_URL
     port: parseInt(process.env.POSTGRES_PORT || process.env.TIP_DB_PORT || "5433"),
     database: process.env.POSTGRES_DB || process.env.TIP_DB_NAME || "transceiver_db",
     user: process.env.POSTGRES_USER || process.env.TIP_DB_USER || "tip",
-    password: process.env.POSTGRES_PASSWORD || process.env.TIP_DB_PASS || "tip_dev_2026",
+    password: requireDbPassword(),
     ssl: false,
   };
 
diff --git a/packages/scraper/src/utils/image-downloader.ts b/packages/scraper/src/utils/image-downloader.ts
index 8cbba67..f7e53fc 100644
--- a/packages/scraper/src/utils/image-downloader.ts
+++ b/packages/scraper/src/utils/image-downloader.ts
@@ -6,13 +6,14 @@
  */
 import { Pool } from "pg";
 import { createHash } from "crypto";
+import { requireDbPassword } from "./db-credentials";
 
 const pool = new Pool({
   host: process.env.POSTGRES_HOST || "localhost",
   port: parseInt(process.env.POSTGRES_PORT || "5433"),
   database: process.env.POSTGRES_DB || "transceiver_db",
   user: process.env.POSTGRES_USER || "tip",
-  password: process.env.POSTGRES_PASSWORD || "tip_dev_2026",
+  password: requireDbPassword(),
   max: 3,
 });
 
diff --git a/scripts/migrate.ts b/scripts/migrate.ts
index 268ce90..7a6cbac 100644
--- a/scripts/migrate.ts
+++ b/scripts/migrate.ts
@@ -5,12 +5,17 @@ import { config } from "dotenv";
 
 config();
 
+const dbPassword = process.env.POSTGRES_PASSWORD || process.env.TIP_DB_PASS;
+if (!dbPassword) {
+  throw new Error("POSTGRES_PASSWORD (or TIP_DB_PASS) must be set — refusing a hardcoded default.");
+}
+
 const pool = new Pool({
   host: process.env.POSTGRES_HOST || "localhost",
   port: parseInt(process.env.POSTGRES_PORT || "5432"),
   database: process.env.POSTGRES_DB || "transceiver_db",
   user: process.env.POSTGRES_USER || "tip",
-  password: process.env.POSTGRES_PASSWORD || "tip_dev_2026",
+  password: dbPassword,
 });
 
 async function migrate(): Promise<void> {
diff --git a/sql/129-orchestrator-control-plane.sql b/sql/129-orchestrator-control-plane.sql
new file mode 100644
index 0000000..0e993e9
--- /dev/null
+++ b/sql/129-orchestrator-control-plane.sql
@@ -0,0 +1,183 @@
+-- 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;