Repository Public
1// Condition rules: things that are true for a while rather than events (a transfer still pending, a mandate close to its end,2// a signer that stopped answering, a missing backup…). They are polled; every alert carries a dedupe key naming the episode, so3// a condition that stays true — or a process that restarts while it is true — says it once. Reads only.4import { createHash } from 'node:crypto';5import { getNetwork } from '@spw/core';6import { sql } from '../db.ts';7import { auditChainStatus, collectConnectors, diskFree, lastBackup, type HealthDeps } from '../monitor/health.ts';8import { derivAccount, INTERVAL_MS, type Draft } from './rules.ts';9import { delState, getState, putState, type AlertSettings } from './store.ts';10import { labStatus, type LabStatus } from '../agents/lab.ts'; // v81112export interface ConditionDeps extends HealthDeps {13 backup(): ReturnType<typeof lastBackup>;14 disk(): ReturnType<typeof diskFree>;15 /** v8: health of the OS sandbox of strategy code (default: the real one) */16 lab?(): Promise<LabStatus>;17}18const netName = (id: string): string => { try { return getNetwork(id).name; } catch { return id; } };19const TERMINAL_BAD = ['failed', 'rejected', 'expired', 'replaced', 'draft', 'pending_approval', 'approved'];2021/** "bad for long enough" → 'down' once; back to good after a 'down' → 'up' once. The clock is in the database: a restart changes nothing. */22async function debounce(key: string, bad: boolean | null, now: number, afterMs: number): Promise<{ edge: 'down' | 'up' | null; since: number }> {23 const k = `deb:${key}`, st = await getState<{ since: number; alerted: boolean } | null>(k, null);24 if (bad) {25 if (!st) { await putState(k, { since: now, alerted: false }); return { edge: null, since: now }; }26 if (!st.alerted && now - st.since >= afterMs) { await putState(k, { since: st.since, alerted: true }); return { edge: 'down', since: st.since }; }27 return { edge: null, since: st.since };28 }29 if (st) { await delState(k); return { edge: bad === false && st.alerted ? 'up' : null, since: st.since }; }30 return { edge: null, since: now };31}3233export async function conditionDrafts(d: ConditionDeps, s: AlertSettings): Promise<Draft[]> {34 const now = d.now(), th = s.thresholds, out: Draft[] = [];35 // the alerts layer only looks at what happened after it first ran: no flood of history on the first start36 let epoch = await getState<number | null>('epoch', null);37 if (epoch === null) { epoch = now; await putState('epoch', now); }3839 // 1 — transfers, swaps and bridges still not final40 const stuck = await sql<{ id: string; kind: string; network: string; state: string; phase: string | null; created_at: Date }[]>`41 select id, kind, network, state, bridge->>'phase' as phase, created_at from proposals42 where (kind <> 'bridge' and state in ('signed','broadcast','confirming','reorged') and created_at < ${new Date(now - th.stuckMinutes * 60_000)})43 or (kind = 'bridge' and state <> all(${TERMINAL_BAD}) and coalesce(bridge->>'phase', '') <> 'done' and created_at < ${new Date(now - th.bridgeStuckMinutes * 60_000)})`;44 for (const p of stuck) {45 const state = p.kind === 'bridge' && p.phase && p.state === 'finalized' ? (p.phase === 'attesting' ? '@attesting' : p.phase.startsWith('mint') ? '@minting' : p.phase) : `@${p.state}`;46 out.push({ kind: 'transfer.stuck', params: { what: `@${p.kind === 'swap' ? 'swap' : p.kind === 'bridge' ? 'bridge' : 'transfer'}`, network: netName(p.network), state, minutes: Math.floor((now - p.created_at.getTime()) / 60_000) },47 subject: { type: 'proposal', id: p.id, network: p.network }, dedupe: `stuck:${p.id}` });48 }4950 // 2 — mandates about to expire (72 h, then 24 h)51 const agents = await sql<{ id: string; name: string; state: string; interval: string; last_bar: Date | null; mandate_id: string | null; mandate: { expiresAt?: number } | null }[]>`52 select id, name, state, interval, last_bar, mandate_id, mandate from agents where state in ('PAPER_TRADING', 'LIVE')`;53 for (const a of agents) {54 const exp = (a.mandate?.expiresAt ?? 0) * 1000;55 if (a.state === 'LIVE' && a.mandate_id && exp > now) {56 const left = exp - now, hours = left <= 24 * 3_600_000 ? 24 : left <= 72 * 3_600_000 ? 72 : 0;57 if (hours) out.push({ kind: 'mandate.expiring', severity: hours === 24 ? 'warning' : 'info', params: { agent: a.name, hours }, subject: { type: 'agent', id: a.id }, dedupe: `mandate:${a.mandate_id}:${hours}` });58 }59 // 3 — an active agent that has not completed a bar for too long (stale or inconsistent market data: it holds, and says nothing)60 const iv = INTERVAL_MS[a.interval] ?? 3_600_000;61 if (a.last_bar && now - a.last_bar.getTime() > (th.staleBars + 2) * iv)62 out.push({ kind: 'agent.data_stale', params: { agent: a.name, bars: Math.floor((now - a.last_bar.getTime()) / iv) - 1, interval: a.interval }, subject: { type: 'agent', id: a.id }, dedupe: `stale:${a.id}:${a.last_bar.getTime()}` });63 }6465 // 4 — a live order the venue never confirmed nor denied66 const unknown = await sql<{ account_id: string; client_id: string; instrument: string }[]>`67 select o.account_id, o.client_id, o.instrument from deriv_orders o join deriv_accounts a on a.id = o.account_id68 where o.state = 'unknown' and not o.final and o.updated_at < ${new Date(now - th.unknownOrderSec * 1000)}`;69 for (const o of unknown) out.push({ kind: 'deriv.order_unknown', params: { account: (await derivAccount(sql, o.account_id)).label, clientId: o.client_id, instrument: o.instrument, seconds: th.unknownOrderSec },70 subject: { type: 'account', id: o.account_id }, dedupe: `unknown:${o.account_id}:${o.client_id}` });7172 // 5 — a liquidation reported by a venue (paper liquidations come from the audit chain, labelled simulated)73 const since = new Date(Math.max(epoch, now - 48 * 3_600_000));74 const liq = await sql<{ account_id: string; fill_id: string; instrument: string; size: string; price: string }[]>`75 select f.account_id, f.fill_id, f.instrument, f.size::text as size, f.price::text as price from deriv_fills f join deriv_accounts a on a.id = f.account_id where f.liquidation and f.ts > ${since}`;76 for (const f of liq) out.push({ kind: 'deriv.live.liquidation', params: { account: (await derivAccount(sql, f.account_id)).label, instrument: f.instrument, size: f.size, price: f.price },77 subject: { type: 'account', id: f.account_id }, dedupe: `liq:${f.account_id}:${f.fill_id}` });7879 // 6 — funding payments above the owner's threshold80 const funding = await sql<{ account_id: string; instrument: string; ts: Date; amount: string }[]>`81 select account_id, instrument, ts, amount::text as amount from deriv_funding where ts > ${since} and abs(amount) >= ${th.fundingUsd}`;82 for (const f of funding) { const acct = await derivAccount(sql, f.account_id);83 out.push({ kind: 'deriv.funding', params: { account: acct.label, mode: acct.mode, instrument: f.instrument, amount: Number(f.amount).toFixed(2) }, subject: { type: 'account', id: f.account_id }, dedupe: `funding:${f.account_id}:${f.instrument}:${f.ts.getTime()}` }); }8485 // 7 — signer and every other connector, debounced86 const { connectors, signer } = await collectConnectors(d);87 const sig = await debounce('signer', signer === null, now, Math.min(th.feedDebounceSec, 120) * 1000);88 if (sig.edge === 'down') out.push({ kind: 'security.signer_unreachable', params: { minutes: Math.max(1, Math.round((now - sig.since) / 60_000)) }, dedupe: `signer.down:${sig.since}` });89 if (sig.edge === 'up') out.push({ kind: 'security.signer_back', params: {}, dedupe: `signer.up:${sig.since}` });90 if (signer) {91 const policyHash = createHash('sha256').update(JSON.stringify(signer.policy ?? null)).digest('hex').slice(0, 16), version = (signer.policy as { version?: number } | undefined)?.version ?? null;92 const prev = await getState<{ halted: boolean; policyHash: string } | null>('signer', null);93 if (prev && prev.halted !== signer.halted) out.push({ kind: signer.halted ? 'security.signer_halted' : 'security.signer_unhalted', params: {}, dedupe: `signer.halt:${signer.halted}:${now}` });94 if (prev && prev.policyHash !== policyHash) out.push({ kind: 'security.policy_changed', params: { what: '@signer', detail: version === null ? '' : `version ${version}` }, dedupe: `signer.policy:${policyHash}` });95 if (!prev || prev.halted !== signer.halted || prev.policyHash !== policyHash) await putState('signer', { halted: signer.halted, policyHash });96 }97 for (const c of connectors) {98 if (c.id === 'signer') continue;99 const r = await debounce(`conn:${c.id}`, c.ok === null ? null : !c.ok, now, th.feedDebounceSec * 1000);100 const subject = c.group === 'rpc' ? { type: 'network' as const, network: c.id.slice(4) } : { type: 'none' as const };101 if (r.edge === 'down') out.push({ kind: 'conn.down', params: { name: c.name, minutes: Math.max(1, Math.round((now - r.since) / 60_000)) }, subject, dedupe: `conn.down:${c.id}:${r.since}` });102 if (r.edge === 'up') out.push({ kind: 'conn.up', params: { name: c.name }, subject, dedupe: `conn.up:${c.id}:${r.since}` });103 }104105 // 8 — audit chain106 const audit = await auditChainStatus(now);107 if (!audit.intact) out.push({ kind: 'security.audit_broken', params: { seq: audit.brokenAt ?? 0 }, dedupe: `audit_broken:${audit.brokenAt}` });108109 // 9 — backup, disk, worker loops (a reminder per day while it lasts)110 const dayKey = new Date(now).toISOString().slice(0, 10);111 const b = d.backup();112 if (b && b.lastAt !== null && now - b.lastAt > th.backupHours * 3_600_000) out.push({ kind: 'system.backup_missing', params: { hours: Math.floor((now - b.lastAt) / 3_600_000) }, dedupe: `backup:${dayKey}` });113 const disk = await d.disk();114 if (disk && disk.freePct < th.diskFreePct) out.push({ kind: 'system.disk_low', params: { pct: disk.freePct }, dedupe: `disk:${dayKey}` });115 for (const l of d.loops()) if (l.failures >= 5) out.push({ kind: 'system.worker_failing', params: { loop: l.name, count: l.failures, reason: l.lastError ?? '' }, dedupe: `loop:${l.name}:${Math.floor(now / (6 * 3_600_000))}` });116 // 10 — v8: the OS sandbox of strategy code. Unavailable = no signal and no backtest runs (fail closed): said once per117 // episode (on the second poll in a row that finds it broken), and once when it is back; the development switch that turns confinement off is said once per day while it is set.118 const lab = await (d.lab ?? labStatus)().catch(() => null);119 if (lab) {120 const r = await debounce('lab', lab.mode === 'unavailable', now, 0);121 if (r.edge === 'down') out.push({ kind: 'system.lab_sandbox_unavailable', params: { reason: (lab.reason ?? '').slice(0, 200) }, dedupe: `lab.down:${r.since}` });122 if (r.edge === 'up') out.push({ kind: 'system.lab_sandbox_back', params: {}, dedupe: `lab.up:${r.since}` });123 if (lab.mode === 'off') out.push({ kind: 'system.lab_sandbox_off', params: {}, dedupe: `lab.off:${dayKey}` });124 }125 return out;126}127