From f7efea217854f12b7781e353c49b344d94def80b Mon Sep 17 00:00:00 2001 From: Investor Flow Build Date: Wed, 12 Aug 2026 22:39:23 -0400 Subject: [PATCH] feat: per-source pause/stop controls for queue admin Add operator pause/resume/stop/start for individual adapter sources, persisted in queue_state. Pause freezes draining and retains jobs; stop clears pending/backoff jobs and blocks re-enqueue until start. Wires admin tRPC endpoints plus a clean two-row Actions layout in the queue admin page. --- app/package-lock.json | 4 +- app/server/src/queue/AdapterQueue.ts | 120 ++++++++++- .../src/queue/__tests__/AdapterQueue.test.ts | 66 ++++++ app/server/src/trpc/router.ts | 28 +++ app/src/app/admin/queue/page.tsx | 200 ++++++++++++++---- app/src/components/InfoTooltip.tsx | 2 +- app/src/lib/trpc.ts | 8 + 7 files changed, 388 insertions(+), 40 deletions(-) diff --git a/app/package-lock.json b/app/package-lock.json index 4b95be4..dd21f9c 100644 --- a/app/package-lock.json +++ b/app/package-lock.json @@ -1,11 +1,11 @@ { - "name": "trader-dashboard", + "name": "investor-flow", "version": "0.1.0", "lockfileVersion": 3, "requires": true, "packages": { "": { - "name": "trader-dashboard", + "name": "investor-flow", "version": "0.1.0", "dependencies": { "@radix-ui/react-dialog": "^1.1.17", diff --git a/app/server/src/queue/AdapterQueue.ts b/app/server/src/queue/AdapterQueue.ts index 96339fa..b89bde1 100644 --- a/app/server/src/queue/AdapterQueue.ts +++ b/app/server/src/queue/AdapterQueue.ts @@ -72,6 +72,10 @@ export interface QueueHealthDetailed { lastErrors: Array<{ key: string; error: string | null }>; /** Active source-wide cool-downs (rate-limit pauses). */ sourceCooldowns: SourceCooldownSnapshot[]; + /** Operator-paused sources (drain frozen, jobs retained). */ + pausedSources: string[]; + /** Operator-stopped sources (drain frozen, jobs cleared, enqueue blocked). */ + stoppedSources: string[]; /** Pending jobs broken down by kind (quote/candles/symbol/…). */ pendingByKind: Record; demandSize: number; @@ -135,6 +139,83 @@ export class AdapterQueue implements CacheScheduler { this._db.prepare("INSERT OR REPLACE INTO queue_state (key, value) VALUES ('paused', ?)").run(String(paused)); } + // ---- Per-source control (pause / stop) --------------------------------- + // Persisted in queue_state like cool-downs: 'source_pause:' = '1' and + // 'source_stop:' = '1'. Pause freezes draining but keeps queued jobs; + // stop additionally clears pending/backoff jobs and blocks re-enqueue until + // startSource() re-queues current demand. + + private sourcePauseKey(source: SourceKind | string): string { + return `source_pause:${source}`; + } + + private sourceStopKey(source: SourceKind | string): string { + return `source_stop:${source}`; + } + + isSourcePaused(source: SourceKind | string): boolean { + const row = this._db.prepare('SELECT value FROM queue_state WHERE key=?') + .get(this.sourcePauseKey(source)) as { value: string } | undefined; + return row?.value === '1'; + } + + isSourceStopped(source: SourceKind | string): boolean { + const row = this._db.prepare('SELECT value FROM queue_state WHERE key=?') + .get(this.sourceStopKey(source)) as { value: string } | undefined; + return row?.value === '1'; + } + + /** Whether a source is paused OR stopped — both freeze draining. */ + isSourceControlled(source: SourceKind | string): boolean { + return this.isSourcePaused(source) || this.isSourceStopped(source); + } + + pauseSource(source: SourceKind | string): void { + this._db.prepare('INSERT OR REPLACE INTO queue_state (key, value) VALUES (?, ?)') + .run(this.sourcePauseKey(source), '1'); + } + + resumeSource(source: SourceKind | string): void { + this._db.prepare("DELETE FROM queue_state WHERE key=?").run(this.sourcePauseKey(source)); + } + + /** Stop a source: freeze draining, clear its pending/backoff jobs, block re-enqueue. */ + stopSource(source: SourceKind | string): number { + this._db.prepare('INSERT OR REPLACE INTO queue_state (key, value) VALUES (?, ?)') + .run(this.sourceStopKey(source), '1'); + this._db.prepare("DELETE FROM queue_state WHERE key=?").run(this.sourcePauseKey(source)); + const res = this._db.prepare( + "DELETE FROM adapter_queue WHERE status IN ('pending','backoff') AND substr(key,1,instr(key,':')-1)=?", + ).run(String(source)); + return Number(res.changes); + } + + /** Start a source: clear pause+stop, then re-queue current demand via its schedules. */ + startSource(source: SourceKind | string): void { + this._db.prepare("DELETE FROM queue_state WHERE key=?").run(this.sourceStopKey(source)); + this._db.prepare("DELETE FROM queue_state WHERE key=?").run(this.sourcePauseKey(source)); + // Force matching schedules to fire on the next enqueueDueSchedules tick. + this._db.prepare("UPDATE queue_schedules SET next_enqueue=? WHERE source_kind=? OR source_kind LIKE ? OR source_kind LIKE ?") + .run(new Date().toISOString(), String(source), `${source}-%`, `${source}:%`); + } + + listControlledSources(): { source: string; paused: boolean; stopped: boolean }[] { + const rows = this._db.prepare( + "SELECT key, value FROM queue_state WHERE key LIKE 'source_pause:%' OR key LIKE 'source_stop:%'", + ).all() as Array<{ key: string; value: string }>; + const out = new Map(); + for (const r of rows) { + const isStop = r.key.startsWith('source_stop:'); + const source = r.key.slice(isStop ? 'source_stop:'.length : 'source_pause:'.length); + if (!source || r.value !== '1') continue; + const cur = out.get(source) ?? { source, paused: false, stopped: false }; + if (isStop) cur.stopped = true; + else cur.paused = true; + out.set(source, cur); + } + return [...out.values()]; + } + /** Whether this source is under a rate-limit cool-down (no outbound calls). */ isSourceCoolingDown(source: SourceKind | string, now = Date.now()): boolean { return this.getSourceCooldown(source, now).active; @@ -318,6 +399,13 @@ export class AdapterQueue implements CacheScheduler { const sym = id.split(':')[0]?.toUpperCase(); if (sym && this.isSymbolQuarantined(sym)) return; + // Stopped sources accept no new jobs (schedules + manual triggers alike) + // until startSource() clears the block and re-queues demand. + try { + const { source } = parseCacheKey(key); + if (source && this.isSourceStopped(source)) return; + } catch { /* malformed key — let the normal path decide */ } + const row = this._db.prepare('SELECT status, backoff_until FROM adapter_queue WHERE key=?').get(key) as { status?: string; backoff_until?: string | null } | undefined; if (row) { if (row.status === 'pending' || row.status === 'in_flight') return; @@ -451,13 +539,26 @@ export class AdapterQueue implements CacheScheduler { const used = kindUsed[kind] ?? 0; if (used >= kindBudget) continue; + // Prioritize x timeline jobs so tracked fund captures always make progress. + // Without this, the yfinance backlog (tier 0-3) can starve x timeline jobs + // indefinitely: they sort last and, under backlog pressure, the tier-3 skip + // below drops them because their "sym" is a fund handle, not a symbol. + const isXTimeline = source === 'x' && kind === 'timeline'; + // Backlog pressure: defer non-critical yfinance kinds so quotes drain first. if (backlogPressure && source === 'yfinance' && NON_CRITICAL_YF_KINDS.has(kind)) continue; // Backlog pressure: skip T3 (background) symbols entirely so portfolio/alert symbols drain first. - if (backlogPressure && sym && (tierMap.get(sym) ?? 99) >= 3) continue; + // x timeline jobs are exempt: their sym is a fund handle (never in tierMap), so + // without this they'd always be dropped while the yfinance backlog persists. + if (backlogPressure && sym && (tierMap.get(sym) ?? 99) >= 3 && !isXTimeline) continue; // Source-wide cool-down: skip all jobs for this vendor until the window ends. if (this.isSourceCoolingDown(source, Date.now())) continue; + // Per-source operator control: paused/stopped sources never drain here. + // (Stopped sources additionally have their pending/backoff rows deleted and + // queue() blocks new enqueues, so this is belt-and-braces for in-flight + // recoveries and anything queued just before a stop landed.) + if (this.isSourceControlled(source)) continue; if (family) { const famUsed = familyJobsThisDrain[family] ?? 0; @@ -483,6 +584,7 @@ export class AdapterQueue implements CacheScheduler { if (job.backoff_until && Date.parse(job.backoff_until) > now) continue; if (job.scheduled_for && Date.parse(job.scheduled_for) > now) continue; if (this.isSourceCoolingDown(source, Date.now())) continue; + if (this.isSourceControlled(source)) continue; if (family && this.isVendorFamilyCoolingDown(family, Date.now())) continue; const kindBudget = DRAIN_KIND_BUDGET[kind] ?? DRAIN_KIND_BUDGET._default; if ((kindUsed[kind] ?? 0) >= kindBudget) continue; @@ -752,6 +854,15 @@ this._db.prepare("UPDATE adapter_queue SET status='backoff', last_attempt=?, ret continue; } + // Stopped sources: skip scheduling work entirely until startSource(). + // queue() already drops their jobs, so this just avoids wasted ticks and + // the enqueue bookkeeping. Paused sources still enqueue (jobs sit queued). + if (this.isSourceStopped(cooldownSource)) { + const nextEnqueue = new Date(Date.now() + s.interval_ms).toISOString(); + this._db.prepare("UPDATE queue_schedules SET last_enqueued=?, next_enqueue=? WHERE source_kind=?").run(now, nextEnqueue, s.source_kind); + continue; + } + const symbols = this.demandSymbols(); const d = this._db; @@ -1116,11 +1227,16 @@ this._db.prepare("UPDATE adapter_queue SET status='backoff', last_attempt=?, ret } catch { /* ignore */ } const cools = this.listSourceCooldowns(); + const controlled = this.listControlledSources(); const notes: string[] = []; if (paused) notes.push('queue paused'); for (const c of cools) { notes.push(`${c.source} cooling ${Math.ceil(c.remainingMs / 60_000)}m (hits=${c.consecutiveHits})`); } + for (const c of controlled) { + if (c.stopped) notes.push(`${c.source} stopped`); + else if (c.paused) notes.push(`${c.source} paused`); + } if ((row.q ?? 0) > 100) notes.push(`large backlog pending=${row.q}`); if (candleLagMs != null && candleLagMs > 3 * 86_400_000) notes.push(`SPY candle lag ${Math.round(candleLagMs / 86_400_000)}d`); if (demandSize > DEMAND_SET_SOFT_CAP) notes.push(`demand set ${demandSize} > soft cap ${DEMAND_SET_SOFT_CAP}`); @@ -1140,6 +1256,8 @@ this._db.prepare("UPDATE adapter_queue SET status='backoff', last_attempt=?, ret counts: { pending: row.q ?? 0, in_flight: row.i ?? 0, failed: row.f ?? 0, backoff: row.b ?? 0, done: row.d ?? 0 }, lastErrors: failedJobs, sourceCooldowns: cools, + pausedSources: controlled.filter((c) => c.paused).map((c) => c.source), + stoppedSources: controlled.filter((c) => c.stopped).map((c) => c.source), pendingByKind, demandSize, systemPinCount, diff --git a/app/server/src/queue/__tests__/AdapterQueue.test.ts b/app/server/src/queue/__tests__/AdapterQueue.test.ts index 2937cc1..fe842da 100644 --- a/app/server/src/queue/__tests__/AdapterQueue.test.ts +++ b/app/server/src/queue/__tests__/AdapterQueue.test.ts @@ -254,3 +254,69 @@ test('health() exposes pendingByKind and dataPlaneHealthy', async () => { assert.ok(h.pendingByKind.quote >= 1 || h.queued >= 1); assert.ok(Array.isArray(h.dataPlaneNotes)); }); + +test('pauseSource freezes draining but keeps queued jobs; resume unfreezes', async () => { + const { db, fake, queue } = setup(); + await queue.queue('yfinance:quote:NVDA'); + queue.pauseSource('yfinance'); + assert.equal(queue.isSourcePaused('yfinance'), true); + assert.equal(queue.isSourceControlled('yfinance'), true); + await queue.drain(); + assert.equal(statusOf(db, 'yfinance:quote:NVDA').status, 'pending', 'paused source must not drain'); + assert.equal(fake.calls.length, 0); + queue.resumeSource('yfinance'); + assert.equal(queue.isSourcePaused('yfinance'), false); + await queue.drain(); + assert.equal(statusOf(db, 'yfinance:quote:NVDA').status, 'done'); + assert.equal(fake.calls.length, 1); +}); + +test('stopSource clears pending jobs and blocks new enqueues until start', async () => { + const { db, fake, queue } = setup(); + await queue.queue('yfinance:quote:NVDA'); + await queue.queue('yfinance:quote:AAPL'); + const cleared = queue.stopSource('yfinance'); + assert.equal(cleared, 2, 'stop clears the pending rows'); + assert.equal(count(db, "status='pending'"), 0); + assert.equal(queue.isSourceStopped('yfinance'), true); + // queue() is a no-op for a stopped source + await queue.queue('yfinance:quote:NVDA'); + assert.equal(count(db, "key='yfinance:quote:NVDA'"), 0, 'stopped source must reject new enqueues'); + // drain must not resurrect jobs (there are none), still no fetch + await queue.drain(); + assert.equal(fake.calls.length, 0); + // start re-enables the source + queue.startSource('yfinance'); + assert.equal(queue.isSourceStopped('yfinance'), false); + await queue.queue('yfinance:quote:NVDA'); + await queue.drain(); + assert.equal(statusOf(db, 'yfinance:quote:NVDA').status, 'done'); +}); + +test('stopped source schedules skip enqueue; startSource forces next_enqueue to now', async () => { + const { db, queue } = setup(); + queue.seedDefaultSchedules(); + db.prepare("DELETE FROM adapter_queue").run(); + queue.stopSource('yfinance'); + // Force the yfinance-quote-watched schedule due — stopped source must skip enqueueing. + db.prepare("UPDATE queue_schedules SET next_enqueue=? WHERE source_kind='yfinance-quote-watched'") + .run(new Date(Date.now() - 1000).toISOString()); + await queue.enqueueDueSchedules(); + assert.equal(count(db, "status='pending'"), 0, 'stopped source schedules must not enqueue'); + queue.startSource('yfinance'); + const row = db.prepare("SELECT next_enqueue FROM queue_schedules WHERE source_kind='yfinance-quote-watched'").get() as { next_enqueue: string | null }; + assert.ok(row.next_enqueue, 'startSource must force the schedule to fire'); + assert.ok(Date.parse(row.next_enqueue) <= Date.now() + 5000); +}); + +test('health() reports pausedSources and stoppedSources', async () => { + const { queue } = setup(); + queue.pauseSource('fred'); + queue.stopSource('x'); + const h = queue.health(); + assert.ok(h.pausedSources.includes('fred')); + assert.ok(h.stoppedSources.includes('x')); + const notes = h.dataPlaneNotes.join(' '); + assert.ok(notes.includes('fred paused')); + assert.ok(notes.includes('x stopped')); +}); diff --git a/app/server/src/trpc/router.ts b/app/server/src/trpc/router.ts index f1fad70..7b3a234 100644 --- a/app/server/src/trpc/router.ts +++ b/app/server/src/trpc/router.ts @@ -1707,6 +1707,34 @@ const adminRouter = router({ return { cleared }; }), + queueSourcePause: adminProcedure + .input(z.object({ source: z.string().min(1) })) + .mutation(({ ctx, input }) => { + ctx.queue.pauseSource(input.source); + return { ok: true, source: input.source }; + }), + + queueSourceResume: adminProcedure + .input(z.object({ source: z.string().min(1) })) + .mutation(({ ctx, input }) => { + ctx.queue.resumeSource(input.source); + return { ok: true, source: input.source }; + }), + + queueSourceStop: adminProcedure + .input(z.object({ source: z.string().min(1) })) + .mutation(({ ctx, input }) => { + const cleared = ctx.queue.stopSource(input.source); + return { ok: true, source: input.source, cleared }; + }), + + queueSourceStart: adminProcedure + .input(z.object({ source: z.string().min(1) })) + .mutation(({ ctx, input }) => { + ctx.queue.startSource(input.source); + return { ok: true, source: input.source }; + }), + queueClearDone: adminProcedure .input(z.object({ olderThanHours: z.number().min(1).max(336).optional().default(24) })) .mutation(({ ctx, input }) => { diff --git a/app/src/app/admin/queue/page.tsx b/app/src/app/admin/queue/page.tsx index e3a993c..61053e6 100644 --- a/app/src/app/admin/queue/page.tsx +++ b/app/src/app/admin/queue/page.tsx @@ -2,6 +2,7 @@ import { useEffect, useState, useCallback } from "react"; import { AdminLayout } from "@/components/AdminLayout"; import { api } from "@/lib/trpc"; +import { InfoTooltip } from "@/components/InfoTooltip"; interface QueueItem { source: string; @@ -23,6 +24,8 @@ interface QueueStatus { last_error: string | undefined; counts: Record; lastErrors: Array<{ key: string; error: string | null }>; + pausedSources?: string[]; + stoppedSources?: string[]; } interface ErrorLogEntry { @@ -71,6 +74,18 @@ function formatTimestamp(ts: string | null): string { } } +const SOURCE_DESCRIPTIONS: Record = { + yfinance: "Yahoo Finance market data: quotes, candles, options chains, symbol metadata and fundamentals for every demand symbol.", + "sec-fetch": "SEC EDGAR equity filings (13F/SC 13G) + institutional holders for eligible equity tickers.", + "sec-sc-fetch": "SEC SC filings (13D/G) for the current SC watchlist.", + "sec-tickers": "SEC company-ticker mapping refresh (daily).", + "x": "X/Twitter timelines via the local bird CLI: tracked fund-manager posts and x_accounts feeds that back position captures.", + fred: "FRED macro series warm-up (yields, CPI, employment) used by the macro regime engine.", + "finra-si": "FINRA short-interest biweekly snapshot.", + nasdaq: "NASDAQ short-interest feed.", + reddit: "Reddit market sentiment posts for the tracked subreddits.", +}; + function msToHuman(ms: number): string { if (ms < 60000) return `${Math.round(ms / 1000)}s`; if (ms < 3600000) return `${Math.round(ms / 60000)}m`; @@ -113,6 +128,7 @@ export default function QueuePage() { const [resettingSource, setResettingSource] = useState(null); const [resetResult, setResetResult] = useState>({}); + const [controllingSource, setControllingSource] = useState<{ source: string; action: string } | null>(null); const [newScheduleKind, setNewScheduleKind] = useState("sec-fetch"); const [newScheduleInterval, setNewScheduleInterval] = useState("86400000"); @@ -160,9 +176,11 @@ export default function QueuePage() { }, []); useEffect(() => { - loadData(); - const interval = setInterval(loadData, 5000); - return () => { clearInterval(interval); }; + // Kick off the initial fetch from a callback so setState happens async, + // then poll every 5s (react-hooks/set-state-in-effect). + const boot = setTimeout(() => { void loadData(); }, 0); + const interval = setInterval(() => { void loadData(); }, 5000); + return () => { clearTimeout(boot); clearInterval(interval); }; }, [loadData]); const handlePause = async () => { @@ -297,6 +315,36 @@ export default function QueuePage() { } }; + const handleSourceControl = async (source: string, action: "pause" | "resume" | "stop" | "start") => { + setControllingSource({ source, action }); + setError(null); + try { + switch (action) { + case "pause": + await api.admin.queueSourcePause(source); + setSuccess(`${source} paused — jobs frozen, queue retained`); + break; + case "resume": + await api.admin.queueSourceResume(source); + setSuccess(`${source} resumed`); + break; + case "stop": + const stopRes = await api.admin.queueSourceStop(source); + setSuccess(`${source} stopped — ${stopRes.cleared} job(s) cleared, re-enqueue blocked`); + break; + case "start": + await api.admin.queueSourceStart(source); + setSuccess(`${source} started — re-queuing current demand`); + break; + } + await loadData(); + } catch (e) { + setError(e instanceof Error ? e.message : `Failed to ${action} ${source}`); + } finally { + setControllingSource(null); + } + }; + const handleRetryJob = async (key: string) => { try { await api.admin.queueRetryJob(key); @@ -379,6 +427,10 @@ export default function QueuePage() { const sortedSources = Object.entries(sources).sort(([a], [b]) => a.localeCompare(b)); + const sourceControl: Record = {}; + for (const s of status?.stoppedSources ?? []) sourceControl[s] = "stopped"; + for (const s of status?.pausedSources ?? []) if (!sourceControl[s]) sourceControl[s] = "paused"; + if (loading) { return ; } @@ -462,18 +514,23 @@ export default function QueuePage() {
Pending + Jobs waiting to be picked up by the next drain cycle. In-flight + Jobs currently being fetched. Per source they run one at a time to respect vendor rate limits. Done + Jobs completed successfully and cached. Cleared via the Clear Done button (24h). Backoff + A vendor rate-limit or cool-down. Jobs stall and retry automatically once the window passes. Failed + Permanent failures (quarantine / max retries exceeded). See the Failed Jobs log below for details and manual retry.
@@ -487,20 +544,42 @@ export default function QueuePage() { - - - - - - - - + + + + + + + + - {sortedSources.map(([source, agg]) => ( - - + {sortedSources.map(([source, agg]) => { + const ctrl = sourceControl[source] ?? "running"; + return ( + + @@ -508,32 +587,81 @@ export default function QueuePage() { - ))} + ); + })}
SourcePendingIn-flightDoneBackoffFailedLast AttemptActions + + Source + Vendor data plane. Pause freezes draining but keeps jobs; Stop clears pending jobs and blocks enqueue until Start. + + PendingJobs awaiting the next drain cycle.In-flightJobs actively being fetched right now.DoneCompleted and cached successfully.BackoffRate-limited; auto-retries after cool-down.FailedPermanent failures requiring manual retry.Last AttemptMost recent time any job for this source was attempted. + + Actions + Per-source controls: Pause/Resume freeze draining; Stop clears pending jobs and blocks re-enqueue; Start re-queues demand. Retry forces backoff/failed jobs back to pending. + +
{source}
+ + {source} + {SOURCE_DESCRIPTIONS[source] ?? `Adapter data source: ${source}.`} + + {ctrl !== "running" && ( + + {ctrl} + + )} + 0 ? "text-accent" : "text-fg-faint"}`}>{agg.pending} 0 ? "text-up" : "text-fg-faint"}`}>{agg.inFlight} {agg.done} 0 ? "text-danger" : "text-fg-faint"}`}>{agg.failed} {formatTimestamp(agg.lastAttempt)} -
- {agg.hasBackoff && ( - - )} - {agg.failed > 0 && ( - +
+
+ {ctrl === "stopped" ? ( + + ) : ( + <> + + + + )} +
+ {(agg.hasBackoff || agg.failed > 0) && ( +
+ {agg.hasBackoff && ( + + )} + {agg.failed > 0 && ( + + )} +
)}
diff --git a/app/src/components/InfoTooltip.tsx b/app/src/components/InfoTooltip.tsx index 399e434..40e95ea 100644 --- a/app/src/components/InfoTooltip.tsx +++ b/app/src/components/InfoTooltip.tsx @@ -31,7 +31,7 @@ export function InfoTooltip({ children }: InfoTooltipProps) { ? {open && ( -
+
{children}
)} diff --git a/app/src/lib/trpc.ts b/app/src/lib/trpc.ts index 5f6aa0d..8a3555c 100644 --- a/app/src/lib/trpc.ts +++ b/app/src/lib/trpc.ts @@ -1305,6 +1305,10 @@ export const api = { queueLogs: (key: string, limit: number) => trpcQuery("admin.queueLogs", { key, limit }), queueRetryJob: (key: string) => trpcMutate<{ ok: boolean }>("admin.queueRetryJob", { key }), queueRetrySource: (sourceKind: string) => trpcMutate<{ cleared: number }>("admin.queueRetrySource", { sourceKind }), + queueSourcePause: (source: string) => trpcMutate<{ ok: boolean; source: string }>("admin.queueSourcePause", { source }), + queueSourceResume: (source: string) => trpcMutate<{ ok: boolean; source: string }>("admin.queueSourceResume", { source }), + queueSourceStop: (source: string) => trpcMutate<{ ok: boolean; source: string; cleared: number }>("admin.queueSourceStop", { source }), + queueSourceStart: (source: string) => trpcMutate<{ ok: boolean; source: string }>("admin.queueSourceStart", { source }), auditLog: (limit = 50, offset = 0, actor?: string, action?: string) => trpcQuery>("admin.auditLog", { limit, offset, ...(actor ? { actor } : {}), ...(action ? { action } : {}) }), queueLint: (symbol: string, kind: 'sec-lint-holders' | 'sec-lint-insiders') => @@ -1726,6 +1730,10 @@ export interface QueueStatus { lastErrors: Array<{ key: string; error: string | null }>; /** Active vendor cool-downs after 429 (ADR-0009). */ sourceCooldowns?: SourceCooldown[]; + /** Operator-paused sources (drain frozen, jobs retained). */ + pausedSources?: string[]; + /** Operator-stopped sources (drain frozen, jobs cleared, enqueue blocked). */ + stoppedSources?: string[]; } export interface ScheduleEntry {