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.
This commit is contained in:
Rene Fichtmueller
parent
d9c37af642
commit
60cffb3585
@ -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",
|
||||
|
||||
@ -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;
|
||||
|
||||
393
packages/scraper/src/orchestrator/control-loop.ts
Normal file
393
packages/scraper/src/orchestrator/control-loop.ts
Normal file
@ -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 };
|
||||
}
|
||||
52
packages/scraper/src/orchestrator/escalation-sink.ts
Normal file
52
packages/scraper/src/orchestrator/escalation-sink.ts
Normal file
@ -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 };
|
||||
}
|
||||
}
|
||||
85
packages/scraper/src/orchestrator/index.ts
Normal file
85
packages/scraper/src/orchestrator/index.ts
Normal file
@ -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);
|
||||
});
|
||||
78
packages/scraper/src/orchestrator/self-heal-guards.test.ts
Normal file
78
packages/scraper/src/orchestrator/self-heal-guards.test.ts
Normal file
@ -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/);
|
||||
});
|
||||
125
packages/scraper/src/orchestrator/self-heal-guards.ts
Normal file
125
packages/scraper/src/orchestrator/self-heal-guards.ts
Normal file
@ -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" };
|
||||
}
|
||||
@ -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({
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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,
|
||||
});
|
||||
|
||||
|
||||
27
packages/scraper/src/utils/db-credentials.ts
Normal file
27
packages/scraper/src/utils/db-credentials.ts
Normal file
@ -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}`;
|
||||
}
|
||||
@ -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,
|
||||
};
|
||||
|
||||
|
||||
@ -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,
|
||||
});
|
||||
|
||||
|
||||
@ -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> {
|
||||
|
||||
183
sql/129-orchestrator-control-plane.sql
Normal file
183
sql/129-orchestrator-control-plane.sql
Normal file
@ -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;
|
||||
Loading…
x
Reference in New Issue
Block a user