Add pipeline backlog/throughput dashboard to admin Logs tab
Admins had no visibility into how many articles were queued for AI synthesis or waiting out the hold-before-publish window, nor how fast Ollama could clear that backlog. GET /api/admin/pipeline-stats reports a live backlog snapshot (items awaiting embedding, clusters on hold vs. ready, items still held) computed straight from the DB with no AI calls, plus real Ollama generate() throughput (tokens/sec, in-flight call) tracked from actual requests, and estimates minutes-to-clear from recent generate() call durations. Surfaced as a stat-tile dashboard atop the Logs tab.
This commit is contained in:
@@ -11,6 +11,8 @@ import { totalStorageBytes } from '../storage/media/index.js';
|
|||||||
import { OllamaProvider } from '../inference/ollama-provider.js';
|
import { OllamaProvider } from '../inference/ollama-provider.js';
|
||||||
import { pollSourceNow } from '../ingestion/poller.js';
|
import { pollSourceNow } from '../ingestion/poller.js';
|
||||||
import { logger, listLogs } from '../storage/db/logs.js';
|
import { logger, listLogs } from '../storage/db/logs.js';
|
||||||
|
import * as backlogStats from '../queue/backlogStats.js';
|
||||||
|
import * as ollamaStats from '../inference/stats.js';
|
||||||
import * as telegramClient from '../telegram/client.js';
|
import * as telegramClient from '../telegram/client.js';
|
||||||
import { geocodeLocation } from '../weather/client.js';
|
import { geocodeLocation } from '../weather/client.js';
|
||||||
import { pollWeatherNow } from '../weather/poller.js';
|
import { pollWeatherNow } from '../weather/poller.js';
|
||||||
@@ -347,4 +349,37 @@ export async function registerAdminRoutes(app: FastifyInstance) {
|
|||||||
limit: limit ? Number(limit) : undefined
|
limit: limit ? Number(limit) : undefined
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Backlog/throughput dashboard for the Logs tab — backlog counts are recomputed live
|
||||||
|
// from the DB on every request (cheap: no AI calls, see backlogStats.ts), while Ollama
|
||||||
|
// throughput/in-flight status comes from a rolling in-memory sample of recent
|
||||||
|
// generate() calls (see inference/stats.ts) since that can only be observed as calls
|
||||||
|
// actually happen, not recomputed on demand.
|
||||||
|
app.get('/api/admin/pipeline-stats', async () => {
|
||||||
|
const settings = settingsDb.getSettings();
|
||||||
|
const backlog = backlogStats.getBacklogSnapshot(settings);
|
||||||
|
const throughput = ollamaStats.getThroughput();
|
||||||
|
const inFlight = ollamaStats.getInFlight();
|
||||||
|
const { lastDirectCycle, lastSynthesisCycle } = backlogStats.getLastCycles();
|
||||||
|
|
||||||
|
// Estimate is deliberately conservative: only clusters that actually need an LLM
|
||||||
|
// call (2+ items — see backlogStats.ts) count toward it, and it's null (rather than
|
||||||
|
// a misleading guess) until at least one real generate() call has completed, since
|
||||||
|
// there's no token-speed data to estimate from yet.
|
||||||
|
const estimatedMinutesToClear =
|
||||||
|
backlog.clusters.readyNowNeedingSynthesis === 0
|
||||||
|
? 0
|
||||||
|
: throughput.avgGenerateDurationMs !== null
|
||||||
|
? Math.ceil((backlog.clusters.readyNowNeedingSynthesis * throughput.avgGenerateDurationMs) / 60_000)
|
||||||
|
: null;
|
||||||
|
|
||||||
|
return {
|
||||||
|
timestamp: new Date().toISOString(),
|
||||||
|
ollama: { inFlight, ...throughput },
|
||||||
|
backlog,
|
||||||
|
estimatedMinutesToClear,
|
||||||
|
lastDirectCycle,
|
||||||
|
lastSynthesisCycle
|
||||||
|
};
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import { Agent } from 'undici';
|
import { Agent } from 'undici';
|
||||||
import type { InferenceProvider } from './provider.js';
|
import type { InferenceProvider } from './provider.js';
|
||||||
|
import * as stats from './stats.js';
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Node's global fetch (undici) defaults to a 5-minute headers/body timeout — fine for
|
* Node's global fetch (undici) defaults to a 5-minute headers/body timeout — fine for
|
||||||
@@ -51,28 +52,52 @@ export class OllamaProvider implements InferenceProvider {
|
|||||||
|
|
||||||
async generate(
|
async generate(
|
||||||
prompt: string,
|
prompt: string,
|
||||||
opts: { model?: string; system?: string; numCtx?: number; numPredict?: number } = {}
|
opts: { model?: string; system?: string; numCtx?: number; numPredict?: number; label?: string } = {}
|
||||||
): Promise<string> {
|
): Promise<string> {
|
||||||
const res = await fetch(`${this.base()}/api/generate`, {
|
const startedAt = Date.now();
|
||||||
method: 'POST',
|
stats.recordGenerateStart(opts.label ?? 'synthesis');
|
||||||
headers: { 'Content-Type': 'application/json' },
|
try {
|
||||||
body: JSON.stringify({
|
const res = await fetch(`${this.base()}/api/generate`, {
|
||||||
model: opts.model,
|
method: 'POST',
|
||||||
prompt,
|
headers: { 'Content-Type': 'application/json' },
|
||||||
system: opts.system,
|
body: JSON.stringify({
|
||||||
stream: false,
|
model: opts.model,
|
||||||
options: {
|
prompt,
|
||||||
num_ctx: opts.numCtx ?? DEFAULT_NUM_CTX,
|
system: opts.system,
|
||||||
num_predict: opts.numPredict ?? DEFAULT_NUM_PREDICT
|
stream: false,
|
||||||
}
|
options: {
|
||||||
}),
|
num_ctx: opts.numCtx ?? DEFAULT_NUM_CTX,
|
||||||
// Not in the ambient RequestInit type this project resolves to, but Node's global
|
num_predict: opts.numPredict ?? DEFAULT_NUM_PREDICT
|
||||||
// fetch (built on undici) honors it at runtime — see noTimeoutDispatcher above.
|
}
|
||||||
dispatcher: noTimeoutDispatcher
|
}),
|
||||||
} as RequestInit);
|
// Not in the ambient RequestInit type this project resolves to, but Node's global
|
||||||
if (!res.ok) throw new Error(`Ollama generate failed: ${res.status} ${await res.text()}`);
|
// fetch (built on undici) honors it at runtime — see noTimeoutDispatcher above.
|
||||||
const data = (await res.json()) as { response: string };
|
dispatcher: noTimeoutDispatcher
|
||||||
return data.response;
|
} as RequestInit);
|
||||||
|
if (!res.ok) throw new Error(`Ollama generate failed: ${res.status} ${await res.text()}`);
|
||||||
|
const data = (await res.json()) as {
|
||||||
|
response: string;
|
||||||
|
eval_count?: number;
|
||||||
|
eval_duration?: number;
|
||||||
|
prompt_eval_count?: number;
|
||||||
|
prompt_eval_duration?: number;
|
||||||
|
total_duration?: number;
|
||||||
|
};
|
||||||
|
// Ollama reports these *_duration fields in nanoseconds — dividing eval_count by
|
||||||
|
// (eval_duration/1e9) gives generation tokens/sec, and total_duration/1e6 gives
|
||||||
|
// wall-clock milliseconds (falling back to a local measurement if a given Ollama
|
||||||
|
// version's response ever omits it).
|
||||||
|
stats.recordGenerateEnd({
|
||||||
|
genTokensPerSec: data.eval_count && data.eval_duration ? data.eval_count / (data.eval_duration / 1e9) : null,
|
||||||
|
promptTokensPerSec:
|
||||||
|
data.prompt_eval_count && data.prompt_eval_duration ? data.prompt_eval_count / (data.prompt_eval_duration / 1e9) : null,
|
||||||
|
totalDurationMs: data.total_duration ? data.total_duration / 1e6 : Date.now() - startedAt
|
||||||
|
});
|
||||||
|
return data.response;
|
||||||
|
} catch (err) {
|
||||||
|
stats.recordGenerateEnd(null);
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async embed(text: string, opts: { model?: string } = {}): Promise<number[]> {
|
async embed(text: string, opts: { model?: string } = {}): Promise<number[]> {
|
||||||
|
|||||||
@@ -1,5 +1,8 @@
|
|||||||
export interface InferenceProvider {
|
export interface InferenceProvider {
|
||||||
generate(prompt: string, opts?: { model?: string; system?: string; numCtx?: number; numPredict?: number }): Promise<string>;
|
generate(
|
||||||
|
prompt: string,
|
||||||
|
opts?: { model?: string; system?: string; numCtx?: number; numPredict?: number; label?: string }
|
||||||
|
): Promise<string>;
|
||||||
embed(text: string, opts?: { model?: string }): Promise<number[]>;
|
embed(text: string, opts?: { model?: string }): Promise<number[]>;
|
||||||
listModels(): Promise<string[]>;
|
listModels(): Promise<string[]>;
|
||||||
isReachable(): Promise<boolean>;
|
isReachable(): Promise<boolean>;
|
||||||
|
|||||||
@@ -0,0 +1,57 @@
|
|||||||
|
/**
|
||||||
|
* In-memory-only tracking of Ollama generate() throughput and in-flight status, for the
|
||||||
|
* admin "Logs" dashboard (see queue/backlogStats.ts, api/admin.ts's GET
|
||||||
|
* /api/admin/pipeline-stats). Deliberately not persisted to disk — a restart losing a
|
||||||
|
* few minutes of rolling samples is fine, since the next few generate() calls rebuild it.
|
||||||
|
*/
|
||||||
|
|
||||||
|
const MAX_SAMPLES = 20;
|
||||||
|
|
||||||
|
export interface GenerateSample {
|
||||||
|
/** Generation speed (tokens/sec) from Ollama's eval_count/eval_duration — null if the response omitted them. */
|
||||||
|
genTokensPerSec: number | null;
|
||||||
|
/** Prompt-processing speed (tokens/sec) from prompt_eval_count/prompt_eval_duration — usually the dominant cost on CPU-only inference. */
|
||||||
|
promptTokensPerSec: number | null;
|
||||||
|
totalDurationMs: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
const samples: GenerateSample[] = [];
|
||||||
|
let inFlight: { label: string; startedAt: number } | null = null;
|
||||||
|
|
||||||
|
/** Call immediately before issuing a generate() request. */
|
||||||
|
export function recordGenerateStart(label: string): void {
|
||||||
|
inFlight = { label, startedAt: Date.now() };
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Call in a finally block after the request settles — pass null on failure/abort. */
|
||||||
|
export function recordGenerateEnd(sample: GenerateSample | null): void {
|
||||||
|
inFlight = null;
|
||||||
|
if (!sample) return;
|
||||||
|
samples.push(sample);
|
||||||
|
if (samples.length > MAX_SAMPLES) samples.shift();
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getInFlight(): { label: string; elapsedMs: number } | null {
|
||||||
|
return inFlight ? { label: inFlight.label, elapsedMs: Date.now() - inFlight.startedAt } : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function average(nums: number[]): number | null {
|
||||||
|
if (nums.length === 0) return null;
|
||||||
|
return nums.reduce((a, b) => a + b, 0) / nums.length;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface ThroughputStats {
|
||||||
|
sampleCount: number;
|
||||||
|
avgGenTokensPerSec: number | null;
|
||||||
|
avgPromptTokensPerSec: number | null;
|
||||||
|
avgGenerateDurationMs: number | null;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getThroughput(): ThroughputStats {
|
||||||
|
return {
|
||||||
|
sampleCount: samples.length,
|
||||||
|
avgGenTokensPerSec: average(samples.map((s) => s.genTokensPerSec).filter((n): n is number => n !== null)),
|
||||||
|
avgPromptTokensPerSec: average(samples.map((s) => s.promptTokensPerSec).filter((n): n is number => n !== null)),
|
||||||
|
avgGenerateDurationMs: average(samples.map((s) => s.totalDurationMs))
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -137,7 +137,8 @@ export async function synthesizeArticle(
|
|||||||
): Promise<SynthesisResult> {
|
): Promise<SynthesisResult> {
|
||||||
const prompt = buildPrompt(items, sourceNames);
|
const prompt = buildPrompt(items, sourceNames);
|
||||||
const system = SYSTEM_PROMPT_BASE + styleAddendum(settings);
|
const system = SYSTEM_PROMPT_BASE + styleAddendum(settings);
|
||||||
const raw = await provider.generate(prompt, { model, system, numCtx: DEFAULT_NUM_CTX, numPredict: DEFAULT_NUM_PREDICT });
|
const label = `Merging ${items.length} source${items.length === 1 ? '' : 's'}: "${items[0]?.title.slice(0, 60) ?? ''}"`;
|
||||||
|
const raw = await provider.generate(prompt, { model, system, numCtx: DEFAULT_NUM_CTX, numPredict: DEFAULT_NUM_PREDICT, label });
|
||||||
return parseResult(raw);
|
return parseResult(raw);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -175,7 +176,8 @@ export async function synthesizeRecap(
|
|||||||
model,
|
model,
|
||||||
system: RECAP_SYSTEM_PROMPT_BASE + styleAddendum(settings),
|
system: RECAP_SYSTEM_PROMPT_BASE + styleAddendum(settings),
|
||||||
numCtx: DEFAULT_NUM_CTX,
|
numCtx: DEFAULT_NUM_CTX,
|
||||||
numPredict: DEFAULT_NUM_PREDICT
|
numPredict: DEFAULT_NUM_PREDICT,
|
||||||
|
label: `Recapping event: "${eventName.slice(0, 60)}"`
|
||||||
});
|
});
|
||||||
return parseResult(raw);
|
return parseResult(raw);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,114 @@
|
|||||||
|
import * as contentItemsDb from '../storage/db/contentItems.js';
|
||||||
|
import * as sourcesDb from '../storage/db/sources.js';
|
||||||
|
import * as categoriesDb from '../storage/db/categories.js';
|
||||||
|
import { clusterItems } from '../pipeline/clustering.js';
|
||||||
|
import type { ContentItem, GlobalSettings, Source } from '../storage/db/types.js';
|
||||||
|
|
||||||
|
interface CycleRecord {
|
||||||
|
at: string;
|
||||||
|
published: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
let lastDirectCycle: CycleRecord | null = null;
|
||||||
|
let lastSynthesisCycle: CycleRecord | null = null;
|
||||||
|
|
||||||
|
/** Called by priorityQueue.ts at the end of runDirectPublishCycle. */
|
||||||
|
export function recordDirectPublishCycle(published: number): void {
|
||||||
|
lastDirectCycle = { at: new Date().toISOString(), published };
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Called by priorityQueue.ts at the end of runSynthesisCycle. */
|
||||||
|
export function recordSynthesisCycle(published: number): void {
|
||||||
|
lastSynthesisCycle = { at: new Date().toISOString(), published };
|
||||||
|
}
|
||||||
|
|
||||||
|
export function getLastCycles(): { lastDirectCycle: CycleRecord | null; lastSynthesisCycle: CycleRecord | null } {
|
||||||
|
return { lastDirectCycle, lastSynthesisCycle };
|
||||||
|
}
|
||||||
|
|
||||||
|
function inAiDisabledCategory(item: ContentItem, disabledNames: Set<string>, sourcesById: Map<string, Source>): boolean {
|
||||||
|
const source = sourcesById.get(item.sourceId);
|
||||||
|
for (const cat of source?.category ?? []) {
|
||||||
|
if (disabledNames.has(cat.split(':')[0].trim().toLowerCase())) return true;
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface BacklogSnapshot {
|
||||||
|
totalUnclusteredItems: number;
|
||||||
|
/** Items that need no AI at all (YouTube/Nitter/Telegram sources, or "No AI" categories) — publish on the next direct-publish tick. */
|
||||||
|
directEligibleItems: number;
|
||||||
|
/** Mergeable items that haven't been embedded yet (embed() failed/pending, or just ingested since the last synthesis tick). */
|
||||||
|
awaitingEmbeddingItems: number;
|
||||||
|
clusters: {
|
||||||
|
total: number;
|
||||||
|
/** Cleared the hold-before-publish window — will publish on the next synthesis tick. */
|
||||||
|
readyNow: number;
|
||||||
|
/** Of readyNow, clusters with 2+ items — these are the ones that actually need an LLM generate() call (single-item clusters publish verbatim, no AI). */
|
||||||
|
readyNowNeedingSynthesis: number;
|
||||||
|
/** Still waiting out the hold-before-publish window. */
|
||||||
|
onHold: number;
|
||||||
|
itemsOnHold: number;
|
||||||
|
earliestHoldRemainingMs: number | null;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Read-only snapshot of the current backlog for the admin dashboard — mirrors the same
|
||||||
|
* categorization runDirectPublishCycle/runSynthesisCycle use (priorityQueue.ts), but never
|
||||||
|
* calls the AI itself: items with no embedding yet are just counted, not embedded, and
|
||||||
|
* clustering only runs over items that already have one (cosine similarity over stored
|
||||||
|
* vectors — no network call). Cheap enough to call on every dashboard refresh.
|
||||||
|
*/
|
||||||
|
export function getBacklogSnapshot(settings: GlobalSettings): BacklogSnapshot {
|
||||||
|
const items = contentItemsDb.unclusteredItemsExcludingSources([]);
|
||||||
|
const sourcesById = new Map(sourcesDb.listSources().map((s) => [s.id, s]));
|
||||||
|
const categories = categoriesDb.listCategories();
|
||||||
|
|
||||||
|
const directPublishSourceIds = new Set(
|
||||||
|
[...sourcesById.values()].filter((s) => s.type === 'youtube' || s.type === 'nitter' || s.type === 'telegram').map((s) => s.id)
|
||||||
|
);
|
||||||
|
const aiDisabledCategoryNames = new Set(categories.filter((c) => c.disableAi).map((c) => c.name.toLowerCase()));
|
||||||
|
|
||||||
|
const directEligible: ContentItem[] = [];
|
||||||
|
const mergeable: ContentItem[] = [];
|
||||||
|
for (const item of items) {
|
||||||
|
if (directPublishSourceIds.has(item.sourceId) || inAiDisabledCategory(item, aiDisabledCategoryNames, sourcesById)) {
|
||||||
|
directEligible.push(item);
|
||||||
|
} else {
|
||||||
|
mergeable.push(item);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const awaitingEmbedding = mergeable.filter((item) => !item.embedding);
|
||||||
|
const embedded = mergeable.filter((item) => item.embedding);
|
||||||
|
|
||||||
|
const clusters = clusterItems(embedded, settings.mergeStrictness);
|
||||||
|
const holdMs = settings.holdBeforePublishMinutes * 60_000;
|
||||||
|
|
||||||
|
let readyNow = 0;
|
||||||
|
let readyNowNeedingSynthesis = 0;
|
||||||
|
let onHold = 0;
|
||||||
|
let itemsOnHold = 0;
|
||||||
|
let earliestHoldRemainingMs: number | null = null;
|
||||||
|
|
||||||
|
for (const cluster of clusters) {
|
||||||
|
const earliestFetch = Math.min(...cluster.items.map((i) => new Date(i.fetchedAt).getTime()));
|
||||||
|
const remaining = holdMs - (Date.now() - earliestFetch);
|
||||||
|
if (remaining > 0) {
|
||||||
|
onHold++;
|
||||||
|
itemsOnHold += cluster.items.length;
|
||||||
|
earliestHoldRemainingMs = earliestHoldRemainingMs === null ? remaining : Math.min(earliestHoldRemainingMs, remaining);
|
||||||
|
} else {
|
||||||
|
readyNow++;
|
||||||
|
if (cluster.items.length > 1) readyNowNeedingSynthesis++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return {
|
||||||
|
totalUnclusteredItems: items.length,
|
||||||
|
directEligibleItems: directEligible.length,
|
||||||
|
awaitingEmbeddingItems: awaitingEmbedding.length,
|
||||||
|
clusters: { total: clusters.length, readyNow, readyNowNeedingSynthesis, onHold, itemsOnHold, earliestHoldRemainingMs }
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -7,6 +7,7 @@ import { embedPendingItems } from '../pipeline/embedding.js';
|
|||||||
import { clusterItems } from '../pipeline/clustering.js';
|
import { clusterItems } from '../pipeline/clustering.js';
|
||||||
import { publishCluster, publishDirect } from '../pipeline/publish.js';
|
import { publishCluster, publishDirect } from '../pipeline/publish.js';
|
||||||
import { logger } from '../storage/db/logs.js';
|
import { logger } from '../storage/db/logs.js';
|
||||||
|
import * as backlogStats from './backlogStats.js';
|
||||||
import type { GlobalSettings, ContentItem, TrackedEvent, Source } from '../storage/db/types.js';
|
import type { GlobalSettings, ContentItem, TrackedEvent, Source } from '../storage/db/types.js';
|
||||||
|
|
||||||
function partition<T>(items: T[], predicate: (item: T) => boolean): [T[], T[]] {
|
function partition<T>(items: T[], predicate: (item: T) => boolean): [T[], T[]] {
|
||||||
@@ -123,7 +124,10 @@ export async function runPassthroughCycle(settings: GlobalSettings): Promise<num
|
|||||||
export async function runDirectPublishCycle(settings: GlobalSettings): Promise<number> {
|
export async function runDirectPublishCycle(settings: GlobalSettings): Promise<number> {
|
||||||
const activeEvents = eventsDb.listActiveEvents();
|
const activeEvents = eventsDb.listActiveEvents();
|
||||||
const items = contentItemsDb.unclusteredItemsExcludingSources([]);
|
const items = contentItemsDb.unclusteredItemsExcludingSources([]);
|
||||||
if (items.length === 0) return 0;
|
if (items.length === 0) {
|
||||||
|
backlogStats.recordDirectPublishCycle(0);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
const sourcesById = new Map(sourcesDb.listSources().map((s) => [s.id, s]));
|
const sourcesById = new Map(sourcesDb.listSources().map((s) => [s.id, s]));
|
||||||
const categories = categoriesDb.listCategories();
|
const categories = categoriesDb.listCategories();
|
||||||
@@ -152,7 +156,9 @@ export async function runDirectPublishCycle(settings: GlobalSettings): Promise<n
|
|||||||
'Direct publish failed'
|
'Direct publish failed'
|
||||||
);
|
);
|
||||||
|
|
||||||
return publishedTypeDirect + publishedCategoryDirect;
|
const total = publishedTypeDirect + publishedCategoryDirect;
|
||||||
|
backlogStats.recordDirectPublishCycle(total);
|
||||||
|
return total;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -167,7 +173,10 @@ export async function runDirectPublishCycle(settings: GlobalSettings): Promise<n
|
|||||||
export async function runSynthesisCycle(provider: InferenceProvider, settings: GlobalSettings): Promise<number> {
|
export async function runSynthesisCycle(provider: InferenceProvider, settings: GlobalSettings): Promise<number> {
|
||||||
const activeEvents = eventsDb.listActiveEvents();
|
const activeEvents = eventsDb.listActiveEvents();
|
||||||
const items = contentItemsDb.unclusteredItemsExcludingSources([]);
|
const items = contentItemsDb.unclusteredItemsExcludingSources([]);
|
||||||
if (items.length === 0) return 0;
|
if (items.length === 0) {
|
||||||
|
backlogStats.recordSynthesisCycle(0);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
// One fetch of the full source list per cycle, reused below for both the
|
// One fetch of the full source list per cycle, reused below for both the
|
||||||
// direct-publish exclusion and each item's category/rank lookups — avoids a
|
// direct-publish exclusion and each item's category/rank lookups — avoids a
|
||||||
@@ -245,5 +254,7 @@ export async function runSynthesisCycle(provider: InferenceProvider, settings: G
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
backlogStats.recordSynthesisCycle(published);
|
||||||
|
|
||||||
return published;
|
return published;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,7 +13,8 @@ import type {
|
|||||||
AdminStockTicker,
|
AdminStockTicker,
|
||||||
AdminBookmark,
|
AdminBookmark,
|
||||||
Poe2BrowseEntry,
|
Poe2BrowseEntry,
|
||||||
AdminPoe2Entry
|
AdminPoe2Entry,
|
||||||
|
PipelineStats
|
||||||
} from './adminTypes';
|
} from './adminTypes';
|
||||||
|
|
||||||
async function request<T>(path: string, options: RequestInit = {}, fetchFn: typeof fetch = fetch): Promise<T> {
|
async function request<T>(path: string, options: RequestInit = {}, fetchFn: typeof fetch = fetch): Promise<T> {
|
||||||
@@ -176,6 +177,8 @@ export const getLogs = (filters: { level?: 'info' | 'warn' | 'error'; limit?: nu
|
|||||||
return request<LogEntry[]>(`/api/admin/logs${qs ? `?${qs}` : ''}`, {}, fetchFn);
|
return request<LogEntry[]>(`/api/admin/logs${qs ? `?${qs}` : ''}`, {}, fetchFn);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export const getPipelineStats = (fetchFn?: typeof fetch) => request<PipelineStats>('/api/admin/pipeline-stats', {}, fetchFn);
|
||||||
|
|
||||||
// Weather — config/cache lives on AdminSettings.weather (see updateSettings above); this
|
// Weather — config/cache lives on AdminSettings.weather (see updateSettings above); this
|
||||||
// is just the geocoding lookup used to resolve a typed city name to lat/lon.
|
// is just the geocoding lookup used to resolve a typed city name to lat/lon.
|
||||||
export const geocodeLocation = (query: string, fetchFn?: typeof fetch) =>
|
export const geocodeLocation = (query: string, fetchFn?: typeof fetch) =>
|
||||||
|
|||||||
@@ -198,6 +198,33 @@ export interface TelegramStatus {
|
|||||||
phone: string | null;
|
phone: string | null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface PipelineStats {
|
||||||
|
timestamp: string;
|
||||||
|
ollama: {
|
||||||
|
inFlight: { label: string; elapsedMs: number } | null;
|
||||||
|
sampleCount: number;
|
||||||
|
avgGenTokensPerSec: number | null;
|
||||||
|
avgPromptTokensPerSec: number | null;
|
||||||
|
avgGenerateDurationMs: number | null;
|
||||||
|
};
|
||||||
|
backlog: {
|
||||||
|
totalUnclusteredItems: number;
|
||||||
|
directEligibleItems: number;
|
||||||
|
awaitingEmbeddingItems: number;
|
||||||
|
clusters: {
|
||||||
|
total: number;
|
||||||
|
readyNow: number;
|
||||||
|
readyNowNeedingSynthesis: number;
|
||||||
|
onHold: number;
|
||||||
|
itemsOnHold: number;
|
||||||
|
earliestHoldRemainingMs: number | null;
|
||||||
|
};
|
||||||
|
};
|
||||||
|
estimatedMinutesToClear: number | null;
|
||||||
|
lastDirectCycle: { at: string; published: number } | null;
|
||||||
|
lastSynthesisCycle: { at: string; published: number } | null;
|
||||||
|
}
|
||||||
|
|
||||||
export interface LogEntry {
|
export interface LogEntry {
|
||||||
id: number;
|
id: number;
|
||||||
timestamp: string;
|
timestamp: string;
|
||||||
|
|||||||
@@ -1,10 +1,11 @@
|
|||||||
<script lang="ts">
|
<script lang="ts">
|
||||||
import { onMount, onDestroy } from 'svelte';
|
import { onMount, onDestroy } from 'svelte';
|
||||||
import type { LogEntry } from '$lib/adminTypes';
|
import type { LogEntry, PipelineStats } from '$lib/adminTypes';
|
||||||
import { getLogs } from '$lib/adminApi';
|
import { getLogs, getPipelineStats } from '$lib/adminApi';
|
||||||
|
|
||||||
let { logs: initial }: { logs: LogEntry[] } = $props();
|
let { logs: initial }: { logs: LogEntry[] } = $props();
|
||||||
let logs = $state([...initial]);
|
let logs = $state([...initial]);
|
||||||
|
let stats = $state<PipelineStats | null>(null);
|
||||||
let filter = $state<'all' | 'info' | 'warn' | 'error'>('all');
|
let filter = $state<'all' | 'info' | 'warn' | 'error'>('all');
|
||||||
let autoRefresh = $state(true);
|
let autoRefresh = $state(true);
|
||||||
let loading = $state(false);
|
let loading = $state(false);
|
||||||
@@ -13,7 +14,12 @@
|
|||||||
async function refresh() {
|
async function refresh() {
|
||||||
loading = true;
|
loading = true;
|
||||||
try {
|
try {
|
||||||
logs = await getLogs(filter === 'all' ? {} : { level: filter });
|
const [nextLogs, nextStats] = await Promise.all([
|
||||||
|
getLogs(filter === 'all' ? {} : { level: filter }),
|
||||||
|
getPipelineStats().catch(() => stats) // stats endpoint failing shouldn't block the log list
|
||||||
|
]);
|
||||||
|
logs = nextLogs;
|
||||||
|
stats = nextStats;
|
||||||
} finally {
|
} finally {
|
||||||
loading = false;
|
loading = false;
|
||||||
}
|
}
|
||||||
@@ -25,6 +31,7 @@
|
|||||||
}
|
}
|
||||||
|
|
||||||
onMount(() => {
|
onMount(() => {
|
||||||
|
refresh();
|
||||||
timer = setInterval(() => {
|
timer = setInterval(() => {
|
||||||
if (autoRefresh) refresh();
|
if (autoRefresh) refresh();
|
||||||
}, 5000);
|
}, 5000);
|
||||||
@@ -35,8 +42,98 @@
|
|||||||
const d = new Date(iso);
|
const d = new Date(iso);
|
||||||
return d.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit', second: '2-digit' });
|
return d.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit', second: '2-digit' });
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function formatAgo(iso: string | null): string {
|
||||||
|
if (!iso) return 'never';
|
||||||
|
const seconds = Math.round((Date.now() - new Date(iso).getTime()) / 1000);
|
||||||
|
if (seconds < 5) return 'just now';
|
||||||
|
if (seconds < 60) return `${seconds}s ago`;
|
||||||
|
const minutes = Math.round(seconds / 60);
|
||||||
|
if (minutes < 60) return `${minutes}m ago`;
|
||||||
|
return `${Math.round(minutes / 60)}h ago`;
|
||||||
|
}
|
||||||
|
|
||||||
|
function formatDuration(ms: number): string {
|
||||||
|
if (ms < 1000) return `${Math.round(ms)}ms`;
|
||||||
|
if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s`;
|
||||||
|
return `${Math.floor(ms / 60_000)}m ${Math.round((ms % 60_000) / 1000)}s`;
|
||||||
|
}
|
||||||
|
|
||||||
|
function formatEta(minutes: number | null): string {
|
||||||
|
if (minutes === null) return 'collecting data…';
|
||||||
|
if (minutes === 0) return 'caught up';
|
||||||
|
if (minutes < 60) return `~${minutes}m`;
|
||||||
|
return `~${Math.floor(minutes / 60)}h ${minutes % 60}m`;
|
||||||
|
}
|
||||||
</script>
|
</script>
|
||||||
|
|
||||||
|
{#if stats}
|
||||||
|
<div class="dashboard">
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Backlog</span>
|
||||||
|
<span class="tile-value">{stats.backlog.totalUnclusteredItems}</span>
|
||||||
|
<span class="tile-sub">item{stats.backlog.totalUnclusteredItems === 1 ? '' : 's'} not yet published</span>
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Awaiting embedding</span>
|
||||||
|
<span class="tile-value">{stats.backlog.awaitingEmbeddingItems}</span>
|
||||||
|
<span class="tile-sub">need an embed() call before they can cluster</span>
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Held for publishing</span>
|
||||||
|
<span class="tile-value">{stats.backlog.clusters.itemsOnHold}</span>
|
||||||
|
<span class="tile-sub">
|
||||||
|
{stats.backlog.clusters.onHold} cluster{stats.backlog.clusters.onHold === 1 ? '' : 's'} on hold-before-publish
|
||||||
|
{#if stats.backlog.clusters.earliestHoldRemainingMs !== null}
|
||||||
|
· earliest clears in {formatDuration(stats.backlog.clusters.earliestHoldRemainingMs)}
|
||||||
|
{/if}
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Awaiting synthesis</span>
|
||||||
|
<span class="tile-value">{stats.backlog.clusters.readyNowNeedingSynthesis}</span>
|
||||||
|
<span class="tile-sub">multi-source cluster{stats.backlog.clusters.readyNowNeedingSynthesis === 1 ? '' : 's'} ready, needs an AI merge</span>
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Estimated to clear</span>
|
||||||
|
<span class="tile-value">{formatEta(stats.estimatedMinutesToClear)}</span>
|
||||||
|
<span class="tile-sub">
|
||||||
|
{#if stats.ollama.avgGenerateDurationMs !== null}
|
||||||
|
based on {stats.ollama.sampleCount} recent generate call{stats.ollama.sampleCount === 1 ? '' : 's'}, avg {formatDuration(stats.ollama.avgGenerateDurationMs)} each
|
||||||
|
{:else}
|
||||||
|
no completed generate calls yet
|
||||||
|
{/if}
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Ollama right now</span>
|
||||||
|
{#if stats.ollama.inFlight}
|
||||||
|
<span class="tile-value live">Synthesizing</span>
|
||||||
|
<span class="tile-sub" title={stats.ollama.inFlight.label}>{stats.ollama.inFlight.label} · {formatDuration(stats.ollama.inFlight.elapsedMs)} elapsed</span>
|
||||||
|
{:else}
|
||||||
|
<span class="tile-value">Idle</span>
|
||||||
|
<span class="tile-sub">
|
||||||
|
{#if stats.ollama.avgGenTokensPerSec !== null}
|
||||||
|
~{stats.ollama.avgGenTokensPerSec.toFixed(1)} gen tok/s, ~{stats.ollama.avgPromptTokensPerSec?.toFixed(1) ?? '?'} prompt tok/s
|
||||||
|
{:else}
|
||||||
|
no throughput data yet
|
||||||
|
{/if}
|
||||||
|
</span>
|
||||||
|
{/if}
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Last direct-publish tick</span>
|
||||||
|
<span class="tile-value">{stats.lastDirectCycle ? stats.lastDirectCycle.published : '—'}</span>
|
||||||
|
<span class="tile-sub">{formatAgo(stats.lastDirectCycle?.at ?? null)}</span>
|
||||||
|
</div>
|
||||||
|
<div class="tile">
|
||||||
|
<span class="tile-label">Last synthesis tick</span>
|
||||||
|
<span class="tile-value">{stats.lastSynthesisCycle ? stats.lastSynthesisCycle.published : '—'}</span>
|
||||||
|
<span class="tile-sub">{formatAgo(stats.lastSynthesisCycle?.at ?? null)}</span>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
{/if}
|
||||||
|
|
||||||
<div class="toolbar">
|
<div class="toolbar">
|
||||||
<div class="filters">
|
<div class="filters">
|
||||||
<button class="pill" class:active={filter === 'all'} onclick={() => setFilter('all')}>All</button>
|
<button class="pill" class:active={filter === 'all'} onclick={() => setFilter('all')}>All</button>
|
||||||
@@ -70,6 +167,41 @@
|
|||||||
</div>
|
</div>
|
||||||
|
|
||||||
<style>
|
<style>
|
||||||
|
.dashboard {
|
||||||
|
display: grid;
|
||||||
|
grid-template-columns: repeat(auto-fill, minmax(180px, 1fr));
|
||||||
|
gap: 10px;
|
||||||
|
margin-bottom: 16px;
|
||||||
|
}
|
||||||
|
.tile {
|
||||||
|
display: flex;
|
||||||
|
flex-direction: column;
|
||||||
|
gap: 4px;
|
||||||
|
background: var(--surface-1);
|
||||||
|
border: 0.5px solid var(--border);
|
||||||
|
border-radius: 12px;
|
||||||
|
padding: 12px 14px;
|
||||||
|
}
|
||||||
|
.tile-label {
|
||||||
|
font-size: 11px;
|
||||||
|
text-transform: uppercase;
|
||||||
|
letter-spacing: 0.02em;
|
||||||
|
color: var(--text-muted);
|
||||||
|
}
|
||||||
|
.tile-value {
|
||||||
|
font-size: 20px;
|
||||||
|
font-weight: 600;
|
||||||
|
color: var(--text-primary);
|
||||||
|
}
|
||||||
|
.tile-value.live {
|
||||||
|
color: var(--text-accent);
|
||||||
|
}
|
||||||
|
.tile-sub {
|
||||||
|
font-size: 11px;
|
||||||
|
color: var(--text-secondary);
|
||||||
|
line-height: 1.4;
|
||||||
|
}
|
||||||
|
|
||||||
.toolbar {
|
.toolbar {
|
||||||
display: flex;
|
display: flex;
|
||||||
align-items: center;
|
align-items: center;
|
||||||
|
|||||||
Reference in New Issue
Block a user