diff --git a/backend/package-lock.json b/backend/package-lock.json index 3bff8b6..120652a 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -15,7 +15,8 @@ "fastify": "^5.10.0", "jsdom": "^29.1.1", "rss-parser": "^3.13.0", - "telegram": "^2.26.22" + "telegram": "^2.26.22", + "undici": "^7.28.0" }, "devDependencies": { "@types/jsdom": "^28.0.3", diff --git a/backend/package.json b/backend/package.json index 9eca3ab..fb0285e 100644 --- a/backend/package.json +++ b/backend/package.json @@ -18,7 +18,8 @@ "fastify": "^5.10.0", "jsdom": "^29.1.1", "rss-parser": "^3.13.0", - "telegram": "^2.26.22" + "telegram": "^2.26.22", + "undici": "^7.28.0" }, "devDependencies": { "@types/jsdom": "^28.0.3", diff --git a/backend/src/api/admin.ts b/backend/src/api/admin.ts index cf2b1f3..0ec1887 100644 --- a/backend/src/api/admin.ts +++ b/backend/src/api/admin.ts @@ -2,15 +2,19 @@ import type { FastifyInstance } from 'fastify'; import * as settingsDb from '../storage/db/settings.js'; import * as sourcesDb from '../storage/db/sources.js'; import * as eventsDb from '../storage/db/events.js'; +import * as articlesDb from '../storage/db/articles.js'; import * as categoriesDb from '../storage/db/categories.js'; import * as stocksDb from '../storage/db/stocks.js'; import * as bookmarksDb from '../storage/db/bookmarks.js'; import * as poe2WatchlistDb from '../storage/db/poe2Watchlist.js'; -import { clearSourceContent, reissueSourceContent, clearAllArticles, clearAllMedia } from '../storage/contentCascade.js'; +import { clearSourceContent, reissueSourceContent, reissueArticle, clearAllArticles, clearAllMedia } from '../storage/contentCascade.js'; import { totalStorageBytes } from '../storage/media/index.js'; import { OllamaProvider } from '../inference/ollama-provider.js'; +import { publishEventRecap } from '../pipeline/publish.js'; import { pollSourceNow } from '../ingestion/poller.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 { geocodeLocation } from '../weather/client.js'; import { pollWeatherNow } from '../weather/poller.js'; @@ -65,9 +69,14 @@ export async function registerAdminRoutes(app: FastifyInstance) { // --- Categories (add/remove — reordering/privacy is via PATCH /settings above) --- app.post('/api/admin/categories', async (req, reply) => { - const { name, isPrivate, isSpillover } = req.body as { name?: string; isPrivate?: boolean; isSpillover?: boolean }; + const { name, isPrivate, isSpillover, disableAi } = req.body as { + name?: string; + isPrivate?: boolean; + isSpillover?: boolean; + disableAi?: boolean; + }; if (!name || !name.trim()) return reply.code(400).send({ error: 'name required' }); - const created = categoriesDb.createCategory(name.trim(), !!isPrivate, !!isSpillover); + const created = categoriesDb.createCategory(name.trim(), !!isPrivate, !!isSpillover, !!disableAi); return reply.code(201).send(created); }); @@ -144,6 +153,17 @@ export async function registerAdminRoutes(app: FastifyInstance) { return reissueSourceContent(id); }); + // Fixes one specific bad article (e.g. a degenerate/empty AI synthesis — see + // synthesis.ts's assertNonEmpty) by deleting it and requeuing every item it merged, + // regardless of how many different sources contributed — reissueSourceContent above + // deliberately won't touch a multi-source article at all. + app.post('/api/admin/articles/:id/reissue', async (req, reply) => { + const { id } = req.params as { id: string }; + const result = reissueArticle(id); + if (!result) return reply.code(404).send({ error: 'not found' }); + return result; + }); + // --- Tracked events --- app.get('/api/admin/events', async () => eventsDb.listEvents()); @@ -165,6 +185,39 @@ export async function registerAdminRoutes(app: FastifyInstance) { return reply.code(204).send(); }); + // Forces one tracked item's recap to run right now, ignoring its recapIntervalHours + // cadence entirely (even if recaps are turned off for it) — for "I want a wrap-up + // right now" rather than waiting out the timer. Still summarizes the same real + // window eventsRecap.ts would (everything published since lastRecapAt, or the last + // 24h if it's never recapped) rather than some arbitrary admin-chosen range, and + // still requires that window to actually contain something — an AI call with zero + // source material to summarize would just hallucinate content it wasn't given. + app.post('/api/admin/events/:id/recap-now', async (req, reply) => { + const { id } = req.params as { id: string }; + const event = eventsDb.getEvent(id); + if (!event) return reply.code(404).send({ error: 'not found' }); + if (event.sourceIds.length === 0) { + return { published: false, reason: 'No sources assigned to this item yet.' }; + } + + const since = event.lastRecapAt ?? new Date(Date.now() - 24 * 3600_000).toISOString(); + const constituents = articlesDb.articlesForEventSince(event.id, since); + if (constituents.length === 0) { + return { published: false, reason: 'Nothing new published under this item since its last recap.' }; + } + + const settings = settingsDb.getSettings(); + const provider = new OllamaProvider(settings.aiServiceHost, settings.aiServicePort); + try { + const article = await publishEventRecap(provider, settings, event, constituents); + eventsDb.markRecapped(event.id); + logger.info('events', `Manually forced recap for "${event.name}" from ${constituents.length} article(s)`); + return { published: true, title: article.title }; + } catch (err) { + return reply.code(502).send({ error: `Recap failed: ${(err as Error).message}` }); + } + }); + // --- Models / AI service (fetched live from the configured Ollama host) --- app.get('/api/admin/models', async (_req, reply) => { const settings = settingsDb.getSettings(); @@ -186,6 +239,20 @@ export async function registerAdminRoutes(app: FastifyInstance) { return { connected, host: settings.aiServiceHost, port: settings.aiServicePort, ramGB: null, gpu: null }; }); + // Detects the selected synthesis model's own max context length (when Ollama exposes + // it) so the Models tab's num_ctx/num_predict sliders can be bounded by what the + // model actually supports, instead of an arbitrary fixed cap. contextLength is null + // when undetectable (older Ollama version, unusual model format, unreachable) — the + // frontend falls back to a generous default range in that case rather than blocking. + app.get('/api/admin/model-context', async (req, reply) => { + const { model } = req.query as { model?: string }; + if (!model) return reply.code(400).send({ error: 'model query param required' }); + const settings = settingsDb.getSettings(); + const provider = new OllamaProvider(settings.aiServiceHost, settings.aiServicePort); + const contextLength = await provider.getModelContextLength(model); + return { contextLength }; + }); + // --- Telegram account (Connections tab — see telegram/client.ts and credentials.ts. // API ID/hash and the resulting login session are stored encrypted at rest; none of // these routes ever echo them back to the client.) --- @@ -342,4 +409,37 @@ export async function registerAdminRoutes(app: FastifyInstance) { 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 + }; + }); } diff --git a/backend/src/api/public.ts b/backend/src/api/public.ts index e2a0898..1dc876b 100644 --- a/backend/src/api/public.ts +++ b/backend/src/api/public.ts @@ -45,11 +45,25 @@ export async function registerPublicRoutes(app: FastifyInstance) { return tagsDb.listActiveTags(); }); + app.get('/api/tag/:slug', async (req, reply) => { + const { slug } = req.params as { slug: string }; + const tag = tagsDb.getTagBySlug(slug); + if (!tag) return reply.code(404).send({ error: 'not found' }); + return tag; + }); + app.get('/api/events', async () => { - // Public fields only — sourceIds, keywords etc. stay admin-only. - return eventsDb - .listEvents() - .map((e) => ({ id: e.id, name: e.name, active: e.active, recapIntervalHours: e.recapIntervalHours, isSpillover: e.isSpillover })); + // Public fields only — sourceIds, keywords etc. stay admin-only. lastRecapAt is + // safe to expose (just a timestamp, no source/keyword detail) and lets the + // tracked-event page show when the next AI recap is due. + return eventsDb.listEvents().map((e) => ({ + id: e.id, + name: e.name, + active: e.active, + recapIntervalHours: e.recapIntervalHours, + lastRecapAt: e.lastRecapAt, + isSpillover: e.isSpillover + })); }); // Drives the site nav — admin-editable (add/remove/reorder) via /api/admin/categories, diff --git a/backend/src/inference/ollama-provider.ts b/backend/src/inference/ollama-provider.ts index b735c5f..247f878 100644 --- a/backend/src/inference/ollama-provider.ts +++ b/backend/src/inference/ollama-provider.ts @@ -1,4 +1,39 @@ +import { Agent } from 'undici'; 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 + * ordinary HTTP calls, but a real problem for /api/generate on CPU-only inference: a + * near-full context window can legitimately take longer than that just for prompt + * processing on the reference hardware (i5-6600K, no GPU, ~17 tokens/sec). Once + * synthesis prompts started carrying full article bodies instead of short blurbs, every + * generate() call past a few thousand tokens got killed at exactly 5m0s — visible in + * Ollama's own log as the request being cancelled, not a genuine model/server error — + * so no cluster could ever finish synthesizing. No timeout at all here; Ollama's own + * process is the natural backstop, not a clock tuned for hardware this doesn't run on. + */ +const noTimeoutDispatcher = new Agent({ headersTimeout: 0, bodyTimeout: 0 }); + +/** + * Default context window / max-generation length requested from Ollama when a caller + * doesn't specify its own. Ollama otherwise falls back to whatever the model's + * Modelfile/runner defaults to (observed as low as 4096 tokens for qwen2.5:7b-instruct + * here, well under that model's 32768-token training context) and SILENTLY truncates + * any prompt that doesn't fit — dropping the middle of the prompt with no error + * surfaced anywhere. Explicitly setting num_ctx/num_predict on every request makes the + * limit deliberate and stable instead of whatever Ollama happens to pick. + * + * 8192 is sized for CPU-only inference (the reference box is an i5-6600K running + * Ollama in Docker, no GPU, ~17 tokens/sec prompt processing) — RAM is not the + * constraint (48GB available; the KV cache for 8192 tokens is well under 1GB), but + * prompt-processing time scales with context, so this trades headroom against + * per-request latency rather than maxing out the model's full 32768-token capacity. + * Callers that build prompts (see pipeline/synthesis.ts) size their own content to fit + * within this budget up front, rather than relying on Ollama to truncate for them. + */ +export const DEFAULT_NUM_CTX = 8192; +export const DEFAULT_NUM_PREDICT = 700; /** * Talks to a self-hosted Ollama instance over HTTP. Address is a normal backend @@ -15,28 +50,71 @@ export class OllamaProvider implements InferenceProvider { return `${this.host}:${this.port}`; } - async generate(prompt: string, opts: { model?: string; system?: string } = {}): Promise { - const res = await fetch(`${this.base()}/api/generate`, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - model: opts.model, - prompt, - system: opts.system, - stream: false - }) - }); - if (!res.ok) throw new Error(`Ollama generate failed: ${res.status} ${await res.text()}`); - const data = (await res.json()) as { response: string }; - return data.response; + async generate( + prompt: string, + opts: { model?: string; system?: string; numCtx?: number; numPredict?: number; label?: string } = {} + ): Promise { + const startedAt = Date.now(); + stats.recordGenerateStart(opts.label ?? 'synthesis'); + try { + const res = await fetch(`${this.base()}/api/generate`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + model: opts.model, + prompt, + system: opts.system, + stream: false, + options: { + num_ctx: opts.numCtx ?? DEFAULT_NUM_CTX, + num_predict: opts.numPredict ?? DEFAULT_NUM_PREDICT + } + }), + // Not in the ambient RequestInit type this project resolves to, but Node's global + // fetch (built on undici) honors it at runtime — see noTimeoutDispatcher above. + dispatcher: noTimeoutDispatcher + } 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 { const res = await fetch(`${this.base()}/api/embeddings`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ model: opts.model, prompt: text }) - }); + body: JSON.stringify({ model: opts.model, prompt: text }), + // Ollama serves one inference request at a time (n_slots = 1) — an embed call + // queued behind a slow generate() call waits for that same slot, and on this + // CPU-only hardware a generate() call can easily run past 5 minutes. Without + // this, that wait alone was enough to trip the same default fetch timeout + // generate() had (see noTimeoutDispatcher above), silently dropping the item + // from embedPendingItems — it never got clustered, so a single-source item + // unlucky enough to be embedded while Ollama was busy never published at all, + // retried every cycle with the same result for as long as Ollama stayed busy. + dispatcher: noTimeoutDispatcher + } as RequestInit); if (!res.ok) throw new Error(`Ollama embed failed: ${res.status} ${await res.text()}`); const data = (await res.json()) as { embedding: number[] }; return data.embedding; @@ -57,4 +135,32 @@ export class OllamaProvider implements InferenceProvider { return false; } } + + /** + * Ollama's /api/show returns a model_info object whose keys are prefixed by the + * model's own architecture name (e.g. "qwen2.context_length", "llama.context_length") + * rather than one fixed field — there's no single stable key across model families. + * Scanning for whichever key ends in ".context_length" avoids hardcoding a list of + * known architectures that will inevitably miss a future/uncommon one. Returns null + * (rather than throwing) on any failure — the admin-facing slider falls back to a + * generous default cap when this can't be determined, rather than blocking the whole + * Models tab on one unreliable, best-effort lookup. + */ + async getModelContextLength(model: string): Promise { + try { + const res = await fetch(`${this.base()}/api/show`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model, name: model }), + signal: AbortSignal.timeout(5000) + }); + if (!res.ok) return null; + const data = (await res.json()) as { model_info?: Record }; + const entry = Object.entries(data.model_info ?? {}).find(([key]) => key.endsWith('.context_length')); + const value = entry?.[1]; + return typeof value === 'number' && value > 0 ? value : null; + } catch { + return null; + } + } } diff --git a/backend/src/inference/provider.ts b/backend/src/inference/provider.ts index d994b1d..9e08016 100644 --- a/backend/src/inference/provider.ts +++ b/backend/src/inference/provider.ts @@ -1,6 +1,11 @@ export interface InferenceProvider { - generate(prompt: string, opts?: { model?: string; system?: string }): Promise; + generate( + prompt: string, + opts?: { model?: string; system?: string; numCtx?: number; numPredict?: number; label?: string } + ): Promise; embed(text: string, opts?: { model?: string }): Promise; listModels(): Promise; isReachable(): Promise; + /** The model's own reported max context length (training/architecture limit), or null if the server doesn't expose it — used to bound the admin-facing num_ctx slider (Models tab) so it can't be set past what the model actually supports. */ + getModelContextLength(model: string): Promise; } diff --git a/backend/src/inference/stats.ts b/backend/src/inference/stats.ts new file mode 100644 index 0000000..4f1d68a --- /dev/null +++ b/backend/src/inference/stats.ts @@ -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)) + }; +} diff --git a/backend/src/pipeline/publish.ts b/backend/src/pipeline/publish.ts index ab19350..3366230 100644 --- a/backend/src/pipeline/publish.ts +++ b/backend/src/pipeline/publish.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto'; import type { InferenceProvider } from '../inference/provider.js'; import type { Cluster } from './clustering.js'; -import { synthesizeArticle, synthesizeRecap } from './synthesis.js'; +import { synthesizeArticle, synthesizeRecap, extractTags } from './synthesis.js'; import { selectBestImage, faviconUrlFor } from './image-selection.js'; import { downloadAndStore, promoteToPublished, storeMediaBuffer } from '../storage/media/index.js'; import { downloadMessageMedia, downloadChannelAvatar } from '../telegram/client.js'; @@ -36,10 +36,23 @@ function anyPushesToTopStories(items: ContentItem[]): boolean { return items.some((item) => sources.getSource(item.sourceId)?.pushToTopStories ?? false); } -/** Takes the first line of the synthesized body as a working title until a dedicated title-generation step exists. */ -function deriveTitle(body: string): string { - const firstLine = body.split('\n')[0]; - return firstLine.length > 100 ? firstLine.slice(0, 97) + '…' : firstLine; +/** Embeds each label and resolves/dedupes it against existing tags — shared by every publish path that has tagLabels in hand (from a full synthesis call or the lightweight extractTags), so the dedup behavior stays identical regardless of how the labels were produced. */ +async function resolveTagIds( + provider: InferenceProvider, + tagLabels: string[], + settings: GlobalSettings, + logSource: string +): Promise { + const tagIds: string[] = []; + for (const label of tagLabels) { + try { + const embedding = await provider.embed(label, { model: settings.selectedModels.embedding }); + tagIds.push(tags.resolveOrCreateTag(label, embedding, settings.tagDedupThreshold).id); + } catch (err) { + logger.error(logSource, `Tag embedding failed for "${label}": ${(err as Error).message}`); + } + } + return tagIds; } /** @@ -225,19 +238,19 @@ async function resolveQuotedTweet( } /** - * Publishes a single item as-is, with no AI calls at all — used when the AI service - * isn't reachable (e.g. Ollama hasn't been set up yet, per the "assume it arrives - * after the backend launches" requirement). No rewriting, no tag extraction, no - * embedding. This is deliberately a lesser version of the real pipeline: once Ollama - * is available, newly-ingested items get the full embed/cluster/synthesize treatment - * and can be linked as follow-ups to these passthrough articles via the normal - * tag-based thread detection — but these earlier articles aren't retroactively - * rewritten or merged with anything after the fact. + * Publishes a single item as-is — the title/body are never rewritten or merged (see + * priorityQueue.ts for why: single-source clusters, YouTube/Nitter/Telegram items, and + * AI-disabled categories all route here specifically to avoid that risk). When a + * provider is given (AI is actually reachable and this item isn't in an AI-disabled + * category), it still gets tags via a lightweight standalone extraction call — every + * published article should be taggable/discoverable via /tag/[slug], not just the + * AI-merged ones. Passing no provider (Ollama unreachable, or the item's category has + * AI turned off entirely) skips tagging too — same as the old "no tags yet" behavior. */ export async function publishDirect( item: ContentItem, settings: GlobalSettings, - opts: { eventId?: string } = {} + opts: { eventId?: string; provider?: InferenceProvider } = {} ): Promise { const category = uniqueCategories([item]); const storedMediaIds: string[] = []; @@ -311,6 +324,16 @@ export async function publishDirect( }; } + let tagIds: string[] = []; + if (opts.provider) { + try { + const tagLabels = await extractTags(opts.provider, settings.selectedModels.synthesis, item, settings); + tagIds = await resolveTagIds(opts.provider, tagLabels, settings, 'synthesis'); + } catch (err) { + logger.error('synthesis', `Tag extraction failed for "${item.title}": ${(err as Error).message}`); + } + } + const article = await articles.insertArticle({ title: item.title, body: item.body || item.summary, @@ -335,7 +358,7 @@ export async function publishDirect( publishedAt: item.publishedAt, updatedAt: item.publishedAt, mergeConfidence: 1.0, - tags: [], // no LLM available to extract tags yet — backfilling these later is a reasonable future improvement + tags: tagIds, threadId: randomUUID(), previousArticleId: null, nextArticleId: null, @@ -349,8 +372,9 @@ export async function publishDirect( /** * Publishing is always automatic — there's no draft/review state (see schema doc). - * A cluster of size 1 publishes as-is via the same path; synthesizeArticle lightly - * rewrites rather than merges when there's only one source. + * Callers should route a size-1 cluster to publishDirect instead — there's nothing to + * merge, so an LLM rewrite would only add risk (hallucinated attribution, altered + * facts) for no synthesis benefit. See priorityQueue.ts's runSynthesisCycle. */ export async function publishCluster( provider: InferenceProvider, @@ -360,18 +384,10 @@ export async function publishCluster( ): Promise { const items = cluster.items; - const { body, tagLabels } = await synthesizeArticle(provider, settings.selectedModels.synthesis, items); + const sourceNames = new Map(items.map((item) => [item.sourceId, sources.getSource(item.sourceId)?.name ?? 'Unknown source'])); + const { title, body, tagLabels } = await synthesizeArticle(provider, settings.selectedModels.synthesis, items, sourceNames, settings); - const resolvedTags = []; - for (const label of tagLabels) { - try { - const embedding = await provider.embed(label, { model: settings.selectedModels.embedding }); - resolvedTags.push(tags.resolveOrCreateTag(label, embedding, settings.tagDedupThreshold)); - } catch (err) { - logger.error('synthesis', `Tag embedding failed for "${label}": ${(err as Error).message}`); - } - } - const tagIds = resolvedTags.map((t) => t.id); + const tagIds = await resolveTagIds(provider, tagLabels, settings, 'synthesis'); const { heroImage, storedMediaId } = await resolveHeroImage(items, items[0]?.link ?? ''); const videoItem = items.find((i) => i.videos.length > 0); @@ -416,7 +432,7 @@ export async function publishCluster( const now = new Date().toISOString(); const article = articles.insertArticle({ - title: deriveTitle(body), + title, body, heroImage, video, @@ -459,24 +475,15 @@ export async function publishEventRecap( event: TrackedEvent, constituents: MergedArticle[] ): Promise { - const { body, tagLabels } = await synthesizeRecap(provider, settings.selectedModels.synthesis, event.name, constituents); - - const resolvedTags = []; - for (const label of tagLabels) { - try { - const embedding = await provider.embed(label, { model: settings.selectedModels.embedding }); - resolvedTags.push(tags.resolveOrCreateTag(label, embedding, settings.tagDedupThreshold)); - } catch (err) { - logger.error('events', `Tag embedding failed for "${label}": ${(err as Error).message}`); - } - } + const { title, body, tagLabels } = await synthesizeRecap(provider, settings.selectedModels.synthesis, event, constituents, settings); + const tagIds = await resolveTagIds(provider, tagLabels, settings, 'events'); const category = [...new Set(constituents.flatMap((a) => a.category))]; const heroImage = constituents.find((a) => a.heroImage)?.heroImage ?? null; const now = new Date().toISOString(); return articles.insertArticle({ - title: `${event.name}: recap`, + title: title || `${event.name}: recap`, body, heroImage, video: null, @@ -490,7 +497,7 @@ export async function publishEventRecap( publishedAt: now, updatedAt: now, mergeConfidence: 1.0, - tags: resolvedTags.map((t) => t.id), + tags: tagIds, threadId: randomUUID(), previousArticleId: null, nextArticleId: null, diff --git a/backend/src/pipeline/synthesis.ts b/backend/src/pipeline/synthesis.ts index d52a539..d83d880 100644 --- a/backend/src/pipeline/synthesis.ts +++ b/backend/src/pipeline/synthesis.ts @@ -1,62 +1,231 @@ import type { InferenceProvider } from '../inference/provider.js'; -import type { ContentItem, MergedArticle } from '../storage/db/types.js'; +import type { ContentItem, GlobalSettings, MergedArticle, TrackedEvent } from '../storage/db/types.js'; +import { logger } from '../storage/db/logs.js'; +const TITLE_DELIMITER = '---TITLE---'; const TAG_DELIMITER = '---TAGS---'; -const RECAP_SYSTEM_PROMPT = `You are a neutral news synthesis assistant. Given a chronological list of articles already published about an ongoing tracked event, write a single recap article that: -- Summarizes what has happened across the period covered, in chronological order -- Highlights the most significant developments rather than restating every article -- Stays neutral and factual, without editorializing -- Is 3-5 short paragraphs +// Small/quantized models don't always reproduce a literal delimiter exactly — extra +// dashes, an inserted blank line, different case (seen in production with the tag +// delimiter: "---\n\nTAGS---" instead of "---TAGS---", which an exact-string split +// missed entirely, leaking the raw delimiter text into the published body). Splitting +// on a loose regex instead tolerates that variance. +const TITLE_DELIMITER_RE = /-{2,}\s*TITLE\s*-{2,}/i; +const TAG_DELIMITER_RE = /-{2,}\s*TAGS\s*-{2,}/i; -After the recap, on a new line, write exactly "${TAG_DELIMITER}" followed by 2-4 short comma-separated topic/entity tags (e.g. proper nouns, named events) that this recap is about. If nothing salient qualifies, leave the tag line empty.`; +// Ollama truncates prompts that don't fit its context window by keeping a small prefix +// and dropping everything else in the middle — silently, with no error, and with no +// regard for which sources end up cut (see ollama-provider.ts for the incident that +// prompted this). Rather than relying on that, prompts here are sized to fit the +// admin-configured num_ctx/num_predict (Models tab) up front: each source/article gets +// an equal character budget, cut only when the whole prompt would otherwise overflow, so +// every source stays at least partially represented (and attributable) instead of some +// being dropped outright. ~4 chars/token is a rough heuristic (no tokenizer available +// here) — good enough for a safety margin, not meant to be exact. +const CHARS_PER_TOKEN = 4; +const RESERVED_OVERHEAD_TOKENS = 300; // system prompt + per-entry headers/formatting +const MIN_ENTRY_CHARS = 300; // floor so a huge cluster/recap doesn't shrink every entry to nothing -const SYSTEM_PROMPT = `You are a neutral news synthesis assistant. Given summaries from multiple news sources describing the same event, write a single original article that: -- Attributes specific claims to the outlet that reported them (e.g. "Reuters reported...", "AP notes...") -- Does not copy phrasing verbatim from any source -- Stays neutral and factual, without editorializing -- Is 2-4 short paragraphs +/** Character budget for prompt *input* — leaves numPredict's worth of the context window free for the model's own response, per the admin's configured num_ctx/num_predict (GlobalSettings.synthesisNumCtx/synthesisNumPredict). */ +function maxInputChars(numCtx: number, numPredict: number): number { + return Math.max(0, (numCtx - numPredict - RESERVED_OVERHEAD_TOKENS) * CHARS_PER_TOKEN); +} -If only one source is provided, lightly rewrite it in your own words rather than merging. +function capEntryText(text: string, budgetChars: number): string { + return text.length > budgetChars ? text.slice(0, budgetChars) + '…' : text; +} -After the article, on a new line, write exactly "${TAG_DELIMITER}" followed by 2-4 short comma-separated topic/entity tags (e.g. proper nouns, named events) that this article is about. If nothing salient qualifies, leave the tag line empty.`; +const TAG_EXTRACTION_SYSTEM_PROMPT = `You are a tagging assistant. Given a news item's title and summary, respond with ONLY 2-4 short comma-separated topic/entity tags (e.g. proper nouns, named people, places, organizations, or named events) that this item is about — nothing else, no commentary, no leading text. If nothing salient qualifies, respond with an empty line.`; + +/** Short response — a handful of tags, not prose — so this doesn't need DEFAULT_NUM_PREDICT's full budget. */ +const TAG_EXTRACTION_NUM_PREDICT = 40; + +function parseTagLabels(raw: string): string[] { + return raw + .split(',') + .map((t) => t.trim()) + .filter((t) => t.length > 0 && t.length < 60); +} + +/** + * Lightweight standalone tag extraction for a single item — unlike synthesizeArticle, + * this doesn't rewrite or attribute anything, so it's safe to run even for items that + * publish verbatim via publishDirect (single-source clusters, or format-based direct + * publishes like YouTube/Nitter/Telegram — see priorityQueue.ts). Every published + * article should end up with tags regardless of whether it went through a full AI + * merge, and this is the minimal AI call that makes that possible without triggering + * the rewrite/attribution risk a full synthesizeArticle call would add for no benefit. + */ +export async function extractTags( + provider: InferenceProvider, + model: string, + item: Pick, + settings: GlobalSettings +): Promise { + const summary = capEntryText(item.body || item.summary, maxInputChars(settings.synthesisNumCtx, TAG_EXTRACTION_NUM_PREDICT)); + const prompt = `Title: ${item.title}\nSummary: ${summary}`; + const raw = await provider.generate(prompt, { + model, + system: TAG_EXTRACTION_SYSTEM_PROMPT, + numCtx: settings.synthesisNumCtx, + numPredict: TAG_EXTRACTION_NUM_PREDICT, + label: `Extracting tags: "${item.title.slice(0, 60)}"` + }); + return parseTagLabels(raw); +} + +const RECAP_SYSTEM_PROMPT_BASE = `You are a neutral news synthesis assistant. Given a chronological list of articles already published about an ongoing tracked event, write your response in exactly three parts, in this order: + +1. A short, specific headline for this recap (a single line, ideally under 12 words, no surrounding quotation marks, no trailing period). +2. On a new line, write exactly "${TITLE_DELIMITER}", then the recap: + - Write a full, comprehensive news article covering the period — not a short summary or a bare list of bullet points. Use as many paragraphs and as much length as the material actually warrants; do not artificially cut it short. + - Organize it in chronological order, but group and connect related developments into a coherent narrative rather than restating each source article one at a time + - Give real weight and detail to the most significant developments; minor ones can be covered more briefly, but nothing significant should be dropped for the sake of brevity + - Stays neutral and factual, without editorializing +3. On a new line after the recap, write exactly "${TAG_DELIMITER}" followed by 2-4 short comma-separated topic/entity tags (e.g. proper nouns, named events) that this recap is about. If nothing salient qualifies, leave the tag line empty.`; + +const SYSTEM_PROMPT_BASE = `You are a neutral news synthesis assistant. Given summaries from multiple news sources describing the same event, write your response in exactly three parts, in this order: + +1. A short, specific headline for this story (a single line, ideally under 12 words, no surrounding quotation marks, no trailing period, no site/outlet name). +2. On a new line, write exactly "${TITLE_DELIMITER}", then the article: + - Attributes specific claims to the outlet that reported them, using each source's exact name as given below (e.g. if a source is labeled "Source 1 (Reuters)", write "Reuters reported..."). Never invent, guess, or substitute an outlet name that isn't one of the source names actually given below. + - Does not copy phrasing verbatim from any source + - Stays neutral and factual, without editorializing + - Is 2-4 short paragraphs +3. On a new line after the article, write exactly "${TAG_DELIMITER}" followed by 2-4 short comma-separated topic/entity tags (e.g. proper nouns, named events) that this article is about. If nothing salient qualifies, leave the tag line empty.`; + +// Admin-selectable presets (Merge tab, "Writing style") — appended to whichever base +// prompt applies. 'default' adds nothing: the base prompts above already describe the +// original neutral wire-service tone this pipeline shipped with. +const STYLE_PRESETS: Record = { + default: '', + casual: 'Write in a casual, conversational tone, like a knowledgeable friend catching you up on what happened — contractions and plain language are fine. Still stay factual and keep outlet attribution accurate.', + formal: 'Write in a formal, measured register — precise language, no contractions, no colloquialisms.' +}; + +/** Admin-configurable tone: a preset plus optional free-text instructions, both from GlobalSettings — the only two knobs that affect HOW the model writes, as opposed to WHAT gets clustered/published. Appended to the base prompt, never replacing its structural rules (attribution, paragraph count, tag format). Applies only to regular same-story merges — recaps have their own independent style knob, see recapStyleAddendum below. */ +function styleAddendum(settings: GlobalSettings): string { + const preset = STYLE_PRESETS[settings.synthesisStylePreset] ?? ''; + const custom = settings.synthesisCustomInstructions.trim(); + const lines = [preset, custom].filter(Boolean); + if (lines.length === 0) return ''; + return `\n\nAdditional style instructions from the site admin (follow these without breaking the rules above):\n${lines.join('\n')}`; +} + +/** + * Per-tracked-item recap style — deliberately independent of the global synthesisStylePreset + * above (set in the Merge tab), since a recap's tone/scope is a very different kind of + * knob: it's set once per tracked item (the "More" section on its own edit panel, next to + * its recap cadence), not globally for every merge on the site. Reuses the same preset + * strings for consistency, but reads from the event's own fields instead of GlobalSettings. + */ +function recapStyleAddendum(event: TrackedEvent): string { + const preset = STYLE_PRESETS[event.recapStylePreset] ?? ''; + const custom = event.recapCustomInstructions.trim(); + const lines = [preset, custom].filter(Boolean); + if (lines.length === 0) return ''; + return `\n\nAdditional style instructions from the site admin for this recap (follow these without breaking the rules above):\n${lines.join('\n')}`; +} export interface SynthesisResult { + title: string; body: string; tagLabels: string[]; } -function buildPrompt(items: ContentItem[]): string { - return items - .map((item, i) => `Source ${i + 1} (${item.sourceId}):\nTitle: ${item.title}\nSummary: ${item.summary}`) - .join('\n\n'); +/** Only used when the model doesn't follow the requested title/delimiter format at all — a real headline beats a truncated sentence fragment, but publishing with no title at all is worse than either. */ +function fallbackTitle(body: string): string { + const firstLine = body.split('\n')[0]; + return firstLine.length > 100 ? firstLine.slice(0, 97) + '…' : firstLine; +} + +function buildPrompt(items: ContentItem[], sourceNames: Map, numCtx: number, numPredict: number): string { + const budgetPerItem = Math.max(MIN_ENTRY_CHARS, Math.floor(maxInputChars(numCtx, numPredict) / items.length)); + let truncated = 0; + const entries = items.map((item, i) => { + // Same fallback publishDirect uses (publish.ts) — body is the full article text + // when the feed supplies it (e.g. RSS ), summary is a ~500-char + // blurb. Using summary alone starved the model of real content to synthesize + // from, so a single-source cluster just echoed the blurb back nearly verbatim. + const full = item.body || item.summary; + const text = capEntryText(full, budgetPerItem); + if (text !== full) truncated++; + // The label here (not item.sourceId, an opaque internal id the model can't use) + // is the only real outlet name the model ever sees — without it, a small model + // has nothing to attribute to and falls back to copying the illustrative outlet + // names out of its own system prompt instructions instead (seen in production: + // a single-source item fabricating "Reuters reported..."/"AP notes..." wholesale). + const name = sourceNames.get(item.sourceId) ?? 'Unknown source'; + return `Source ${i + 1} (${name}):\nTitle: ${item.title}\nSummary: ${text}`; + }); + if (truncated > 0) { + logger.warn('synthesis', `Trimmed ${truncated}/${items.length} source article${truncated === 1 ? '' : 's'} to fit the model's context window`); + } + return entries.join('\n\n'); } function parseResult(raw: string): SynthesisResult { - const [body, tagSection] = raw.split(TAG_DELIMITER); - const tagLabels = (tagSection ?? '') - .split(',') - .map((t) => t.trim()) - .filter((t) => t.length > 0 && t.length < 60); + const [beforeTags, tagSection] = raw.split(TAG_DELIMITER_RE); + const tagLabels = parseTagLabels(tagSection ?? ''); - return { body: body.trim(), tagLabels }; + const titleSplit = (beforeTags ?? raw).split(TITLE_DELIMITER_RE); + const titlePart = titleSplit[0]; + // join() rather than titleSplit[1] in case the delimiter text somehow appears again + // inside the body itself — keeps that content rather than silently dropping it. + const bodyPart = titleSplit.length > 1 ? titleSplit.slice(1).join('') : undefined; + // If the title delimiter never showed up, the model didn't follow the requested + // format — treat the whole thing as body rather than mistaking the article itself + // for a "title", and fall back to the old truncated-first-line heuristic. + const body = (bodyPart ?? titlePart).trim(); + const title = bodyPart !== undefined ? titlePart.trim() : fallbackTitle(body); + + return { title, body, tagLabels }; +} + +/** + * A quantized/small model occasionally reproduces just the requested delimiter + * scaffold ("---TITLE---\n\n---TAGS---") with no real headline or article text in + * between — a structurally "valid" response by parseResult's own logic (delimiters + * found, nothing crashed) but empty in substance. Left unchecked this published a + * blank article (empty title/body, still with real sources/hero image attached) once + * in production. Treating an empty body as a hard failure lets the caller's existing + * catch-and-retry logic (see priorityQueue.ts's runSynthesisCycle) leave the cluster + * unclustered for the next cycle instead of ever inserting one of these. + */ +function assertNonEmpty(result: SynthesisResult, context: string): SynthesisResult { + if (!result.body.trim()) { + throw new Error(`Model returned an empty article body for ${context}`); + } + return result; } export async function synthesizeArticle( provider: InferenceProvider, model: string, - items: ContentItem[] + items: ContentItem[], + sourceNames: Map, + settings: GlobalSettings ): Promise { - const prompt = buildPrompt(items); - const raw = await provider.generate(prompt, { model, system: SYSTEM_PROMPT }); - return parseResult(raw); + const { synthesisNumCtx: numCtx, synthesisNumPredict: numPredict } = settings; + const prompt = buildPrompt(items, sourceNames, numCtx, numPredict); + const system = SYSTEM_PROMPT_BASE + styleAddendum(settings); + 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, numPredict, label }); + return assertNonEmpty(parseResult(raw), `"${items[0]?.title.slice(0, 60) ?? ''}"`); } -function buildRecapPrompt(eventName: string, articles: MergedArticle[]): string { - const entries = articles - .map((article, i) => `Article ${i + 1} (published ${article.publishedAt}):\nTitle: ${article.title}\n${article.body}`) - .join('\n\n'); - return `Tracked event: ${eventName}\n\n${entries}`; +function buildRecapPrompt(eventName: string, articles: MergedArticle[], numCtx: number, numPredict: number): string { + const budgetPerArticle = Math.max(MIN_ENTRY_CHARS, Math.floor(maxInputChars(numCtx, numPredict) / articles.length)); + let truncated = 0; + const entries = articles.map((article, i) => { + const body = capEntryText(article.body, budgetPerArticle); + if (body !== article.body) truncated++; + return `Article ${i + 1} (published ${article.publishedAt}):\nTitle: ${article.title}\n${body}`; + }); + if (truncated > 0) { + logger.warn('events', `Trimmed ${truncated}/${articles.length} recap article bod${truncated === 1 ? 'y' : 'ies'} to fit the model's context window`); + } + return `Tracked event: ${eventName}\n\n${entries.join('\n\n')}`; } /** @@ -70,10 +239,18 @@ function buildRecapPrompt(eventName: string, articles: MergedArticle[]): string export async function synthesizeRecap( provider: InferenceProvider, model: string, - eventName: string, - articles: MergedArticle[] + event: TrackedEvent, + articles: MergedArticle[], + settings: GlobalSettings ): Promise { - const prompt = buildRecapPrompt(eventName, articles); - const raw = await provider.generate(prompt, { model, system: RECAP_SYSTEM_PROMPT }); - return parseResult(raw); + const { synthesisNumCtx: numCtx, synthesisNumPredict: numPredict } = settings; + const prompt = buildRecapPrompt(event.name, articles, numCtx, numPredict); + const raw = await provider.generate(prompt, { + model, + system: RECAP_SYSTEM_PROMPT_BASE + recapStyleAddendum(event), + numCtx, + numPredict, + label: `Recapping event: "${event.name.slice(0, 60)}"` + }); + return assertNonEmpty(parseResult(raw), `event recap "${event.name.slice(0, 60)}"`); } diff --git a/backend/src/queue/backlogStats.ts b/backend/src/queue/backlogStats.ts new file mode 100644 index 0000000..ff69d4f --- /dev/null +++ b/backend/src/queue/backlogStats.ts @@ -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, sourcesById: Map): 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 } + }; +} diff --git a/backend/src/queue/priorityQueue.ts b/backend/src/queue/priorityQueue.ts index 1eded36..a1cc0ec 100644 --- a/backend/src/queue/priorityQueue.ts +++ b/backend/src/queue/priorityQueue.ts @@ -7,6 +7,7 @@ import { embedPendingItems } from '../pipeline/embedding.js'; import { clusterItems } from '../pipeline/clustering.js'; import { publishCluster, publishDirect } from '../pipeline/publish.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'; function partition(items: T[], predicate: (item: T) => boolean): [T[], T[]] { @@ -45,23 +46,37 @@ function primaryCategoryRank(item: ContentItem, rankByName: Map, return best; } +/** True if any of the item's source's categories (same leading-segment match as primaryCategoryRank) has AI disabled. */ +function inAiDisabledCategory(item: ContentItem, disabledNames: Set, sourcesById: Map): boolean { + const source = sourcesById.get(item.sourceId); + for (const cat of source?.category ?? []) { + const leading = cat.split(':')[0].trim().toLowerCase(); + if (disabledNames.has(leading)) return true; + } + return false; +} + /** * Shared by both the passthrough (no-AI) and synthesis direct-publish paths — same * publish-then-tag-then-log/error shape, differing only in how the success/failure - * message describes why the item skipped merging. + * message describes why the item skipped merging. `provider`, when given, still gets + * these articles tagged (via publishDirect's lightweight extraction) without rewriting + * anything — omit it entirely for items whose category has AI turned off, or when + * Ollama isn't reachable at all (see call sites). */ async function publishItemsDirect( items: ContentItem[], settings: GlobalSettings, activeEvents: TrackedEvent[], describeSuccess: (item: ContentItem) => string, - failureLabel: string + failureLabel: string, + provider?: InferenceProvider ): Promise { let published = 0; for (const item of items) { try { const eventId = claimedEventId(item, activeEvents) ?? undefined; - const article = await publishDirect(item, settings, { eventId }); + const article = await publishDirect(item, settings, { eventId, provider }); contentItemsDb.assignCluster([item.id], article.id); published++; logger.info('synthesis', `Published "${article.title}" directly (${describeSuccess(item)})`); @@ -78,7 +93,10 @@ async function publishItemsDirect( * pipeline, this doesn't wait out the hold-before-publish window: that window exists to * give corroborating sources time to arrive before an AI merge locks in, which doesn't * apply here since there's no merging happening at all — each item is just itself. - * Still respects category priority. + * Still respects category priority. Harmless overlap with runDirectPublishCycle (which + * runs regardless of reachability) — an item already published by one is simply gone + * from the other's next "unclustered" query, since assignCluster lands before either + * moves on to its next item. */ export async function runPassthroughCycle(settings: GlobalSettings): Promise { const activeEvents = eventsDb.listActiveEvents(); @@ -97,43 +115,97 @@ export async function runPassthroughCycle(settings: GlobalSettings): Promise { + const activeEvents = eventsDb.listActiveEvents(); + const items = contentItemsDb.unclusteredItemsExcludingSources([]); + if (items.length === 0) { + backlogStats.recordDirectPublishCycle(0); + return 0; + } + + 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 [typeDirectItems, remaining] = partition(items, (item) => directPublishSourceIds.has(item.sourceId)); + + const aiDisabledCategoryNames = new Set(categories.filter((c) => c.disableAi).map((c) => c.name.toLowerCase())); + const [categoryDirectItems] = partition(remaining, (item) => inAiDisabledCategory(item, aiDisabledCategoryNames, sourcesById)); + + const publishedTypeDirect = await publishItemsDirect( + typeDirectItems, + settings, + activeEvents, + (item) => sourcesById.get(item.sourceId)?.type ?? 'unknown', + 'Direct publish failed', + provider + ); + + const publishedCategoryDirect = await publishItemsDirect( + categoryDirectItems, + settings, + activeEvents, + () => 'AI disabled for category', + 'Direct publish failed' + ); + + const total = publishedTypeDirect + publishedCategoryDirect; + backlogStats.recordDirectPublishCycle(total); + return total; +} + +/** + * One pass of the synthesis queue: cluster whatever's unclustered (excluding items + * runDirectPublishCycle already owns — see there), ordered by admin-defined category + * priority, and publish clusters that have cleared the hold-before-publish window. + * Items claimed by an active tracked event (belonging to one of its sources and + * matching its keyword filter, if any) publish exactly like everything else — + * individually or merged with same-story coverage — just tagged with the event's id so + * they're browsable under it and eligible for eventsRecap.ts's periodic AI wrap-up. */ export async function runSynthesisCycle(provider: InferenceProvider, settings: GlobalSettings): Promise { const activeEvents = eventsDb.listActiveEvents(); 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 - // direct-publish partition and each item's category/type lookups — avoids a + // direct-publish exclusion and each item's category/rank lookups — avoids a // separate sourcesDb.getSource() round-trip per item. const sourcesById = new Map(sourcesDb.listSources().map((s) => [s.id, s])); + const categories = categoriesDb.listCategories(); + const rankByName = new Map(categories.map((c) => [c.name.toLowerCase(), c.priorityRank])); - // YouTube videos, Nitter tweets, and Telegram messages never get LLM-merged with - // anything else — each is always its own article, same shape whether the AI service - // is up or not. Route them straight to publishDirect, same as the no-AI passthrough path. + // YouTube/Nitter/Telegram items and AI-disabled-category items are runDirectPublishCycle's + // job (its own guarded tick, so a slow merge backlog here never blocks them) — excluded + // here too since a batch just ingested this instant may still be unclustered when this + // runs before that cycle's own pass gets to it. const directPublishSourceIds = new Set( [...sourcesById.values()].filter((s) => s.type === 'youtube' || s.type === 'nitter' || s.type === 'telegram').map((s) => s.id) ); - const [directItems, mergeableItems] = partition(items, (item) => directPublishSourceIds.has(item.sourceId)); - - const publishedDirect = await publishItemsDirect( - directItems, - settings, - activeEvents, - (item) => sourcesById.get(item.sourceId)?.type ?? 'unknown', - 'Direct publish failed' + const aiDisabledCategoryNames = new Set(categories.filter((c) => c.disableAi).map((c) => c.name.toLowerCase())); + const mergeableItems = items.filter( + (item) => !directPublishSourceIds.has(item.sourceId) && !inAiDisabledCategory(item, aiDisabledCategoryNames, sourcesById) ); - const categories = categoriesDb.listCategories(); - const rankByName = new Map(categories.map((c) => [c.name.toLowerCase(), c.priorityRank])); - const ranked = mergeableItems .map((item) => ({ item, rank: primaryCategoryRank(item, rankByName, sourcesById) })) .sort((a, b) => a.rank - b.rank) @@ -161,7 +233,14 @@ export async function runSynthesisCycle(provider: InferenceProvider, settings: G // in practice a cluster's items are all near-duplicate coverage of the same // story, so they'd all match the same event's filter anyway when they match at all. const eventId = cluster.items.map((i) => claimedEventId(i, activeEvents)).find((id) => id !== null) ?? undefined; - const article = await publishCluster(provider, settings, cluster, { eventId }); + // A single-item cluster has nothing to merge — publish the source's own text + // verbatim instead of asking the LLM to "lightly rewrite" it, which only risked + // introducing errors (or fabricated attribution — see synthesis.ts) with no + // actual synthesis to justify the risk. + const article = + cluster.items.length === 1 + ? await publishDirect(cluster.items[0], settings, { eventId, provider }) + : await publishCluster(provider, settings, cluster, { eventId }); contentItemsDb.assignCluster( cluster.items.map((i) => i.id), cluster.id @@ -184,5 +263,7 @@ export async function runSynthesisCycle(provider: InferenceProvider, settings: G ); } - return published + publishedDirect; + backlogStats.recordSynthesisCycle(published); + + return published; } diff --git a/backend/src/queue/scheduler.ts b/backend/src/queue/scheduler.ts index 76dd754..b790bfe 100644 --- a/backend/src/queue/scheduler.ts +++ b/backend/src/queue/scheduler.ts @@ -1,5 +1,5 @@ import { pollDueSources } from '../ingestion/poller.js'; -import { runSynthesisCycle, runPassthroughCycle } from './priorityQueue.js'; +import { runSynthesisCycle, runPassthroughCycle, runDirectPublishCycle } from './priorityQueue.js'; import { runEventRecaps } from './eventsRecap.js'; import { runRetentionSweep } from './retention.js'; import { OllamaProvider } from '../inference/ollama-provider.js'; @@ -10,28 +10,77 @@ import { pollStocksNow } from '../stocks/poller.js'; import { pollPoe2Now } from '../poe2/poller.js'; const POLL_TICK_MS = 60_000; // checks which sources are due every minute; each source's own interval governs actual fetch frequency +const DIRECT_PUBLISH_TICK_MS = 60_000; const SYNTHESIS_TICK_MS = 60_000; const RETENTION_TICK_MS = 60 * 60_000; // hourly const WEATHER_TICK_MS = 45 * 60_000; const STOCKS_TICK_MS = 15 * 60_000; // per admin spec — stock prices move faster than weather const POE2_TICK_MS = 60 * 60_000; // poe.ninja's own overview data doesn't refresh faster than hourly, so polling more often than this just re-fetches the same numbers +/** + * Runs fn on every tick, but skips a tick outright if the previous one is still in + * flight instead of overlapping it. Matters most for the synthesis tick: an item stays + * "unclustered" (cluster_id IS NULL — see contentItems.unclusteredItemsExcludingSources) + * until AFTER its cluster finishes synthesizing and publishing, so a generate() call + * that runs past the next tick (easily minutes, on CPU-only inference — see + * ollama-provider.ts) used to let the same item get picked up and republished as a + * fresh, differently-worded article by an overlapping cycle, repeatedly, until the + * first cycle's assignCluster() finally landed. Node is single-threaded, so the only + * source of "concurrent" runs here is exactly this interval overlap. + * + * Each call gets its own independent `running` flag/timer — the direct-publish and + * synthesis ticks are deliberately two separate calls to this (not one shared guard) + * precisely so a slow AI-merge backlog on one never blocks the other's fast, + * no-AI-needed items from publishing on schedule. They operate on disjoint item sets + * (see priorityQueue.ts), so there's no risk of the two racing each other into a + * duplicate publish the way an overlapping call to the *same* fn would. + */ +function everyTickSkippingOverlap(ms: number, fn: () => Promise) { + let running = false; + setInterval(() => { + if (running) return; + running = true; + fn().finally(() => { + running = false; + }); + }, ms); +} + export function startScheduler() { const provider = () => { const s = settingsDb.getSettings(); return new OllamaProvider(s.aiServiceHost, s.aiServicePort); }; - setInterval(async () => { + everyTickSkippingOverlap(POLL_TICK_MS, async () => { try { const ingested = await pollDueSources(); if (ingested > 0) logger.info('scheduler', `Poll tick: ingested ${ingested} new item(s)`); } catch (err) { logger.error('scheduler', `Poll tick failed: ${(err as Error).message}`); } - }, POLL_TICK_MS); + }); - setInterval(async () => { + everyTickSkippingOverlap(DIRECT_PUBLISH_TICK_MS, async () => { + try { + const settings = settingsDb.getSettings(); + const p = provider(); + // This tick runs regardless of Ollama's reachability (nothing here rewrites or + // merges), but tagging direct-published items (see runDirectPublishCycle/ + // publishDirect) does need a working AI service — only offer the provider + // through when it's actually reachable, so an unconfigured Ollama doesn't spam + // the log with a failed tag-extraction attempt on every single item, every tick. + const reachable = await p.isReachable(); + const published = await runDirectPublishCycle(settings, reachable ? p : undefined); + if (published > 0) { + logger.info('scheduler', `Direct-publish tick: published ${published} article(s)`); + } + } catch (err) { + logger.error('scheduler', `Direct-publish tick failed: ${(err as Error).message}`); + } + }); + + everyTickSkippingOverlap(SYNTHESIS_TICK_MS, async () => { try { const settings = settingsDb.getSettings(); const p = provider(); @@ -55,16 +104,16 @@ export function startScheduler() { } catch (err) { logger.error('scheduler', `Synthesis tick failed: ${(err as Error).message}`); } - }, SYNTHESIS_TICK_MS); + }); - setInterval(() => { + everyTickSkippingOverlap(RETENTION_TICK_MS, async () => { try { runRetentionSweep(settingsDb.getSettings()); logger.info('retention', 'Retention sweep completed'); } catch (err) { logger.error('retention', `Retention tick failed: ${(err as Error).message}`); } - }, RETENTION_TICK_MS); + }); // Immediate first call for all three — unlike RSS sources (whose "due" check makes a // brand-new source eligible on the very next 1-minute tick), weather/stocks/poe2 have @@ -98,5 +147,8 @@ export function startScheduler() { pollPoe2Now().catch((err) => logger.error('poe2', `Poll tick failed: ${err.message}`)); }, POE2_TICK_MS); - logger.info('scheduler', 'Started: poll every 1m, synthesis every 1m, retention every 1h, weather every 45m, stocks every 15m, poe2 every 1h'); + logger.info( + 'scheduler', + 'Started: poll every 1m, direct-publish every 1m, synthesis every 1m, retention every 1h, weather every 45m, stocks every 15m, poe2 every 1h' + ); } diff --git a/backend/src/storage/contentCascade.ts b/backend/src/storage/contentCascade.ts index 3757245..15ef491 100644 --- a/backend/src/storage/contentCascade.ts +++ b/backend/src/storage/contentCascade.ts @@ -86,6 +86,29 @@ export function reissueSourceContent(sourceId: string): ReissueResult { return { articlesDeleted, itemsRequeued: requeueIds.size }; } +/** + * Deletes one specific article (and its media) and requeues every content item that + * contributed to it — unlike reissueSourceContent, this works regardless of how many + * different sources the article merged together, since it's scoped to the article + * itself rather than "everything from source X". Exists for exactly the failure mode + * synthesis.ts's assertNonEmpty guards against going forward: a bad synthesis call + * that already made it into a published (garbage) article before that guard existed, + * where the source-scoped reissue tools can't help because the article spans sources. + * Returns null if the article doesn't exist. + */ +export function reissueArticle(articleId: string): ReissueResult | null { + const article = articlesDb.getArticle(articleId); + if (!article) return null; + + const itemIds = article.sources.map((s) => s.itemId); + deleteMediaByArticleId(article.id); + articlesDb.deleteArticle(article.id); + contentItemsDb.resetClusterForItems(itemIds); + + logger.info('admin', `Reissuing article ${articleId}: deleted, ${itemIds.length} item(s) requeued`); + return { articlesDeleted: 1, itemsRequeued: itemIds.length }; +} + /** Wipes every published article and its media, keeping raw ingested items intact so they can be re-synthesized fresh. */ export function clearAllArticles(): number { const articles = articlesDb.allArticlesNewestFirst(); diff --git a/backend/src/storage/db/categories.ts b/backend/src/storage/db/categories.ts index 2397610..42fb7d3 100644 --- a/backend/src/storage/db/categories.ts +++ b/backend/src/storage/db/categories.ts @@ -9,7 +9,8 @@ function rowToCategory(row: any): Category { priorityRank: row.priority_rank, isDefault: !!row.is_default, isPrivate: !!row.is_private, - isSpillover: !!row.is_spillover + isSpillover: !!row.is_spillover, + disableAi: !!row.disable_ai }; } @@ -24,18 +25,20 @@ export function listPrivateCategoryNames(): string[] { return rows.map((r) => r.name); } -export function setCategoryOrder(order: { id: string; priorityRank: number; isPrivate: boolean; isSpillover: boolean }[]) { - const stmt = db.prepare('UPDATE categories SET priority_rank = ?, is_private = ?, is_spillover = ? WHERE id = ?'); - for (const c of order) stmt.run(c.priorityRank, c.isPrivate ? 1 : 0, c.isSpillover ? 1 : 0, c.id); +export function setCategoryOrder( + order: { id: string; priorityRank: number; isPrivate: boolean; isSpillover: boolean; disableAi: boolean }[] +) { + const stmt = db.prepare('UPDATE categories SET priority_rank = ?, is_private = ?, is_spillover = ?, disable_ai = ? WHERE id = ?'); + for (const c of order) stmt.run(c.priorityRank, c.isPrivate ? 1 : 0, c.isSpillover ? 1 : 0, c.disableAi ? 1 : 0, c.id); } -export function createCategory(name: string, isPrivate = false, isSpillover = false): Category { +export function createCategory(name: string, isPrivate = false, isSpillover = false, disableAi = false): Category { const id = `cat-${name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/(^-|-$)/g, '')}-${randomUUID().slice(0, 6)}`; const maxRank = db.prepare('SELECT COALESCE(MAX(priority_rank), 0) as m FROM categories').get() as { m: number }; db.prepare( - 'INSERT INTO categories (id, name, priority_rank, is_default, is_private, is_spillover) VALUES (?, ?, ?, 0, ?, ?)' - ).run(id, name, maxRank.m + 1, isPrivate ? 1 : 0, isSpillover ? 1 : 0); - return { id, name, priorityRank: maxRank.m + 1, isDefault: false, isPrivate, isSpillover }; + 'INSERT INTO categories (id, name, priority_rank, is_default, is_private, is_spillover, disable_ai) VALUES (?, ?, ?, 0, ?, ?, ?)' + ).run(id, name, maxRank.m + 1, isPrivate ? 1 : 0, isSpillover ? 1 : 0, disableAi ? 1 : 0); + return { id, name, priorityRank: maxRank.m + 1, isDefault: false, isPrivate, isSpillover, disableAi }; } export function deleteCategory(id: string) { diff --git a/backend/src/storage/db/events.ts b/backend/src/storage/db/events.ts index e2b12ab..ba6b1e8 100644 --- a/backend/src/storage/db/events.ts +++ b/backend/src/storage/db/events.ts @@ -14,6 +14,8 @@ function rowToEvent(row: any): TrackedEvent { isSpillover: !!row.is_spillover, retentionOverrideDays: row.retention_override_days, lastRecapAt: row.last_recap_at, + recapStylePreset: row.recap_style_preset, + recapCustomInstructions: row.recap_custom_instructions, createdAt: row.created_at }; } @@ -52,8 +54,8 @@ export function createEvent(input: Partial): TrackedEvent { const id = `evt-${randomUUID()}`; const now = new Date().toISOString(); db.prepare( - `INSERT INTO tracked_events (id, name, description, source_ids, keywords, recap_interval_hours, active, is_spillover, retention_override_days, last_recap_at, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)` + `INSERT INTO tracked_events (id, name, description, source_ids, keywords, recap_interval_hours, active, is_spillover, retention_override_days, last_recap_at, recap_style_preset, recap_custom_instructions, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?)` ).run( id, input.name ?? 'Untitled event', @@ -64,6 +66,8 @@ export function createEvent(input: Partial): TrackedEvent { input.active === false ? 0 : 1, input.isSpillover ? 1 : 0, input.retentionOverrideDays ?? null, + input.recapStylePreset ?? 'default', + input.recapCustomInstructions ?? '', now ); return getEvent(id)!; @@ -74,7 +78,7 @@ export function updateEvent(id: string, patch: Partial): TrackedEv if (!existing) return null; const merged = { ...existing, ...patch }; db.prepare( - `UPDATE tracked_events SET name=?, description=?, source_ids=?, keywords=?, recap_interval_hours=?, active=?, is_spillover=?, retention_override_days=?, last_recap_at=? WHERE id=?` + `UPDATE tracked_events SET name=?, description=?, source_ids=?, keywords=?, recap_interval_hours=?, active=?, is_spillover=?, retention_override_days=?, last_recap_at=?, recap_style_preset=?, recap_custom_instructions=? WHERE id=?` ).run( merged.name, merged.description, @@ -85,6 +89,8 @@ export function updateEvent(id: string, patch: Partial): TrackedEv merged.isSpillover ? 1 : 0, merged.retentionOverrideDays, merged.lastRecapAt, + merged.recapStylePreset, + merged.recapCustomInstructions, id ); return getEvent(id); diff --git a/backend/src/storage/db/index.ts b/backend/src/storage/db/index.ts index 2de9caa..d16ad59 100644 --- a/backend/src/storage/db/index.ts +++ b/backend/src/storage/db/index.ts @@ -135,6 +135,8 @@ export function migrate() { is_spillover INTEGER NOT NULL DEFAULT 0, retention_override_days INTEGER, last_recap_at TEXT, + recap_style_preset TEXT NOT NULL DEFAULT 'default', -- default | casual | formal — this item's own recap tone, independent of the global Merge-tab style (see pipeline/synthesis.ts) + recap_custom_instructions TEXT NOT NULL DEFAULT '', -- free-text addendum for this item's recap prompt specifically created_at TEXT NOT NULL ); @@ -153,7 +155,8 @@ export function migrate() { priority_rank INTEGER NOT NULL, is_default INTEGER NOT NULL DEFAULT 0, is_private INTEGER NOT NULL DEFAULT 0, - is_spillover INTEGER NOT NULL DEFAULT 0 -- collapsed into the nav's "More »" overflow page instead of its own tab + is_spillover INTEGER NOT NULL DEFAULT 0, -- collapsed into the nav's "More »" overflow page instead of its own tab + disable_ai INTEGER NOT NULL DEFAULT 0 -- skip clustering/synthesis for this category's items; publish each one directly ); CREATE TABLE IF NOT EXISTS logs ( @@ -185,6 +188,10 @@ export function migrate() { fxtwitter_base_url TEXT NOT NULL DEFAULT 'https://api.fxtwitter.com', nitter_instance_url TEXT NOT NULL DEFAULT 'https://nitter.net', -- admin's preferred instance, prefills new Nitter sources (Connections tab) telegram_media_mode TEXT NOT NULL DEFAULT 'self-host', -- self-host | proxy (no "direct" — Telegram has no public hotlinkable media URL) + synthesis_style_preset TEXT NOT NULL DEFAULT 'default', -- default | casual | formal — see pipeline/synthesis.ts's STYLE_PRESETS + synthesis_custom_instructions TEXT NOT NULL DEFAULT '', -- free-text addendum appended to the synthesis system prompt, on top of the preset + synthesis_num_ctx INTEGER NOT NULL DEFAULT 8192, -- admin-tunable context window (Models tab) — see inference/ollama-provider.ts's DEFAULT_NUM_CTX + synthesis_num_predict INTEGER NOT NULL DEFAULT 700, -- admin-tunable max response length — too low silently truncates output mid-sentence widget_weather_enabled INTEGER NOT NULL DEFAULT 1, widget_stocks_enabled INTEGER NOT NULL DEFAULT 1, widget_bookmarks_enabled INTEGER NOT NULL DEFAULT 1, @@ -321,6 +328,9 @@ export function migrate() { if (!hasColumn('categories', 'is_spillover')) { db.exec('ALTER TABLE categories ADD COLUMN is_spillover INTEGER NOT NULL DEFAULT 0'); } + if (!hasColumn('categories', 'disable_ai')) { + db.exec('ALTER TABLE categories ADD COLUMN disable_ai INTEGER NOT NULL DEFAULT 0'); + } if (!hasColumn('content_items', 'telegram_message')) { db.exec('ALTER TABLE content_items ADD COLUMN telegram_message TEXT'); } @@ -339,6 +349,12 @@ export function migrate() { if (!hasColumn('tracked_events', 'recap_interval_hours')) { db.exec('ALTER TABLE tracked_events ADD COLUMN recap_interval_hours INTEGER'); } + if (!hasColumn('tracked_events', 'recap_style_preset')) { + db.exec("ALTER TABLE tracked_events ADD COLUMN recap_style_preset TEXT NOT NULL DEFAULT 'default'"); + } + if (!hasColumn('tracked_events', 'recap_custom_instructions')) { + db.exec("ALTER TABLE tracked_events ADD COLUMN recap_custom_instructions TEXT NOT NULL DEFAULT ''"); + } if (!hasColumn('merged_articles', 'is_recap')) { db.exec('ALTER TABLE merged_articles ADD COLUMN is_recap INTEGER NOT NULL DEFAULT 0'); } @@ -390,6 +406,18 @@ export function migrate() { stmt.run(`stk-${symbol.replace(/[^a-z0-9]+/gi, '-')}`, label, symbol, i + 1, new Date().toISOString()); }); } + if (!hasColumn('global_settings', 'synthesis_style_preset')) { + db.exec("ALTER TABLE global_settings ADD COLUMN synthesis_style_preset TEXT NOT NULL DEFAULT 'default'"); + } + if (!hasColumn('global_settings', 'synthesis_custom_instructions')) { + db.exec("ALTER TABLE global_settings ADD COLUMN synthesis_custom_instructions TEXT NOT NULL DEFAULT ''"); + } + if (!hasColumn('global_settings', 'synthesis_num_ctx')) { + db.exec('ALTER TABLE global_settings ADD COLUMN synthesis_num_ctx INTEGER NOT NULL DEFAULT 8192'); + } + if (!hasColumn('global_settings', 'synthesis_num_predict')) { + db.exec('ALTER TABLE global_settings ADD COLUMN synthesis_num_predict INTEGER NOT NULL DEFAULT 700'); + } // Seed default categories if none exist yet. "News" sits right under "Top stories" — // general news sources belong here, not on "Top stories" itself, which isn't a real diff --git a/backend/src/storage/db/settings.ts b/backend/src/storage/db/settings.ts index 604422e..73206d3 100644 --- a/backend/src/storage/db/settings.ts +++ b/backend/src/storage/db/settings.ts @@ -16,6 +16,10 @@ function rowToSettings(row: any): GlobalSettings { fxtwitterBaseUrl: row.fxtwitter_base_url, nitterInstanceUrl: row.nitter_instance_url, telegramMediaMode: row.telegram_media_mode, + synthesisStylePreset: row.synthesis_style_preset, + synthesisCustomInstructions: row.synthesis_custom_instructions, + synthesisNumCtx: row.synthesis_num_ctx, + synthesisNumPredict: row.synthesis_num_predict, widgets: { weather: !!row.widget_weather_enabled, stocks: !!row.widget_stocks_enabled, @@ -80,6 +84,8 @@ export function updateSettings(patch: Partial): GlobalSettings { ai_service_host=$ai_service_host, ai_service_port=$ai_service_port, selected_models=$selected_models, nitter_media_mode=$nitter_media_mode, fxtwitter_base_url=$fxtwitter_base_url, nitter_instance_url=$nitter_instance_url, telegram_media_mode=$telegram_media_mode, + synthesis_style_preset=$synthesis_style_preset, synthesis_custom_instructions=$synthesis_custom_instructions, + synthesis_num_ctx=$synthesis_num_ctx, synthesis_num_predict=$synthesis_num_predict, widget_weather_enabled=$widget_weather_enabled, widget_stocks_enabled=$widget_stocks_enabled, widget_bookmarks_enabled=$widget_bookmarks_enabled, widget_poe2_enabled=$widget_poe2_enabled, widget_order=$widget_order, @@ -105,6 +111,10 @@ export function updateSettings(patch: Partial): GlobalSettings { $fxtwitter_base_url: merged.fxtwitterBaseUrl, $nitter_instance_url: merged.nitterInstanceUrl, $telegram_media_mode: merged.telegramMediaMode, + $synthesis_style_preset: merged.synthesisStylePreset, + $synthesis_custom_instructions: merged.synthesisCustomInstructions, + $synthesis_num_ctx: merged.synthesisNumCtx, + $synthesis_num_predict: merged.synthesisNumPredict, $widget_weather_enabled: merged.widgets.weather ? 1 : 0, $widget_stocks_enabled: merged.widgets.stocks ? 1 : 0, $widget_bookmarks_enabled: merged.widgets.bookmarks ? 1 : 0, diff --git a/backend/src/storage/db/tags.ts b/backend/src/storage/db/tags.ts index 455f47a..728e89a 100644 --- a/backend/src/storage/db/tags.ts +++ b/backend/src/storage/db/tags.ts @@ -28,6 +28,12 @@ export function listActiveTags(): Tag[] { return rows.map(rowToTag); } +/** By slug rather than id — that's what tag chips link by (see publish.ts/frontend tag pages). Matches regardless of active/expired status: an old article's tag chip should still resolve to its (now possibly expired) tag rather than 404 just because nothing new has used it lately. */ +export function getTagBySlug(slug: string): Tag | null { + const row = db.prepare('SELECT * FROM tags WHERE slug = ?').get(slug); + return row ? rowToTag(row) : null; +} + function cosineSimilarity(a: number[], b: number[]): number { if (a.length === 0 || b.length === 0 || a.length !== b.length) return 0; let dot = 0, diff --git a/backend/src/storage/db/types.ts b/backend/src/storage/db/types.ts index 82836f4..ad597f7 100644 --- a/backend/src/storage/db/types.ts +++ b/backend/src/storage/db/types.ts @@ -185,6 +185,10 @@ export interface TrackedEvent { isSpillover: boolean; retentionOverrideDays: number | null; lastRecapAt: string | null; + /** Tone preset for this item's own recap, independent of the global Merge-tab synthesis style — see pipeline/synthesis.ts's STYLE_PRESETS. 'default' adds nothing on top of the base recap prompt. */ + recapStylePreset: 'default' | 'casual' | 'formal'; + /** Free-text instructions appended to the recap system prompt for this item specifically — e.g. "focus on military developments", "write as a full narrative, not bullet points". Empty string means no addendum. */ + recapCustomInstructions: string; createdAt: string; } @@ -206,6 +210,8 @@ export interface Category { isPrivate: boolean; /** Grouped into the nav's "More »" overflow page instead of getting its own top-level tab — see +layout.svelte and /more. */ isSpillover: boolean; + /** Skips clustering/AI synthesis for this category's items — each one publishes directly (own article, own source's text), same as YouTube/Nitter/Telegram items always do. See priorityQueue.ts's runSynthesisCycle. */ + disableAi: boolean; } export interface WeatherHourEntry { @@ -283,6 +289,14 @@ export interface GlobalSettings { nitterInstanceUrl: string; /** How Telegram message media (attached photos/videos, channel avatars) is served — see pipeline/publish.ts's resolveTelegramMedia. No "direct" option: Telegram has no public hotlinkable media URL, bytes only come from the authenticated MTProto session. */ telegramMediaMode: 'self-host' | 'proxy'; + /** Tone preset applied to every AI-synthesized article/recap (see pipeline/synthesis.ts's STYLE_PRESETS) — 'default' is the original neutral wire-service tone with no addendum. Never applies to single-source items, which always publish verbatim without going through the AI at all. */ + synthesisStylePreset: 'default' | 'casual' | 'formal'; + /** Free-text instructions appended to the synthesis system prompt alongside the style preset — e.g. "keep it under 3 sentences per paragraph". Empty string means no addendum. */ + synthesisCustomInstructions: string; + /** Total context window (prompt + response) requested from Ollama for every synthesis/recap/tag-extraction call — see inference/ollama-provider.ts's DEFAULT_NUM_CTX for why this is ever explicit at all, and the Models tab for the admin-facing slider (bounded by the selected synthesis model's own reported max, when Ollama exposes it). */ + synthesisNumCtx: number; + /** Max tokens the model is allowed to generate per synthesis/recap call — too low silently truncates the output mid-sentence rather than erroring (this is what a "cut off" recap/article means). Recaps in particular need real headroom: they're asked to summarize many source articles into several paragraphs, unlike a same-story merge. */ + synthesisNumPredict: number; /** Per-widget enable flags — see admin/settings' consolidated "Widgets" tab. Weather/Stocks/PoE2's backend pollers (scheduler.ts) are gated on these too, not just sidebar visibility; Bookmarks has no poller so its flag only affects the sidebar. */ widgets: { weather: boolean; diff --git a/frontend/src/lib/adminApi.ts b/frontend/src/lib/adminApi.ts index 304d197..22bbeb1 100644 --- a/frontend/src/lib/adminApi.ts +++ b/frontend/src/lib/adminApi.ts @@ -13,7 +13,10 @@ import type { AdminStockTicker, AdminBookmark, Poe2BrowseEntry, - AdminPoe2Entry + AdminPoe2Entry, + PipelineStats, + ModelContextInfo, + ForceRecapResult } from './adminTypes'; async function request(path: string, options: RequestInit = {}, fetchFn: typeof fetch = fetch): Promise { @@ -66,10 +69,16 @@ export const updateSettings = (patch: Partial, fetchFn?: typeof f request('/api/admin/settings', { method: 'PATCH', body: JSON.stringify(patch) }, fetchFn); // Categories -export const createCategory = (name: string, isPrivate = false, isSpillover = false, fetchFn?: typeof fetch) => +export const createCategory = ( + name: string, + isPrivate = false, + isSpillover = false, + disableAi = false, + fetchFn?: typeof fetch +) => request( '/api/admin/categories', - { method: 'POST', body: JSON.stringify({ name, isPrivate, isSpillover }) }, + { method: 'POST', body: JSON.stringify({ name, isPrivate, isSpillover, disableAi }) }, fetchFn ); @@ -97,6 +106,11 @@ export const pollSourceNow = (id: string, fetchFn?: typeof fetch) => export const reissueSourceContent = (id: string, fetchFn?: typeof fetch) => request<{ articlesDeleted: number; itemsRequeued: number }>(`/api/admin/sources/${id}/reissue`, { method: 'POST' }, fetchFn); +// Fixes one specific bad article regardless of how many sources it merged — unlike +// reissueSourceContent above, which deliberately won't touch a multi-source article. +export const reissueArticle = (id: string, fetchFn?: typeof fetch) => + request<{ articlesDeleted: number; itemsRequeued: number }>(`/api/admin/articles/${id}/reissue`, { method: 'POST' }, fetchFn); + // Content clearing — wipe articles/media/a source's raw items so they can be repopulated fresh. export const clearSourceContent = (id: string, fetchFn?: typeof fetch) => request<{ itemsDeleted: number; articlesDeleted: number }>(`/api/admin/content/sources/${id}`, { method: 'DELETE' }, fetchFn); @@ -120,6 +134,12 @@ export const updateEvent = (id: string, patch: Partial, fetch export const deleteEvent = (id: string, fetchFn?: typeof fetch) => request(`/api/admin/events/${id}`, { method: 'DELETE' }, fetchFn); +// Runs this item's recap immediately, ignoring its recapIntervalHours cadence — still +// summarizes the same real window (everything since lastRecapAt) and still requires +// there to actually be something new to summarize (see the backend route). +export const forceRecap = (id: string, fetchFn?: typeof fetch) => + request(`/api/admin/events/${id}/recap-now`, { method: 'POST' }, fetchFn); + // Models / AI service export const getModels = (fetchFn?: typeof fetch) => request('/api/admin/models', {}, fetchFn); @@ -127,6 +147,9 @@ export const getModels = (fetchFn?: typeof fetch) => export const getAiStatus = (fetchFn?: typeof fetch) => request('/api/admin/ai-status', {}, fetchFn); +export const getModelContext = (model: string, fetchFn?: typeof fetch) => + request(`/api/admin/model-context?model=${encodeURIComponent(model)}`, {}, fetchFn); + // Telegram account (Connections tab) — API ID/hash and the resulting login session are // stored encrypted at rest server-side (see backend telegram/credentials.ts); none of // these ever come back from the server, only status flags. @@ -170,6 +193,8 @@ export const getLogs = (filters: { level?: 'info' | 'warn' | 'error'; limit?: nu return request(`/api/admin/logs${qs ? `?${qs}` : ''}`, {}, fetchFn); }; +export const getPipelineStats = (fetchFn?: typeof fetch) => request('/api/admin/pipeline-stats', {}, fetchFn); + // 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. export const geocodeLocation = (query: string, fetchFn?: typeof fetch) => diff --git a/frontend/src/lib/adminTypes.ts b/frontend/src/lib/adminTypes.ts index a703596..f7041c9 100644 --- a/frontend/src/lib/adminTypes.ts +++ b/frontend/src/lib/adminTypes.ts @@ -14,6 +14,7 @@ export interface CategoryPriority { isDefault: boolean; isPrivate: boolean; isSpillover: boolean; + disableAi: boolean; } export interface WeatherHourEntry { @@ -139,6 +140,12 @@ export interface AdminSettings { fxtwitterBaseUrl: string; nitterInstanceUrl: string; telegramMediaMode: 'self-host' | 'proxy'; + synthesisStylePreset: 'default' | 'casual' | 'formal'; + synthesisCustomInstructions: string; + /** Total context window (prompt + response) requested from Ollama for every synthesis/recap/tag-extraction call. */ + synthesisNumCtx: number; + /** Max tokens the model may generate per call — too low silently truncates output mid-sentence. */ + synthesisNumPredict: number; widgets: AdminWidgetsEnabled; widgetOrder: ('weather' | 'stocks' | 'bookmarks' | 'poe2')[]; retention: RetentionSettings; @@ -173,6 +180,19 @@ export interface AdminTrackedEvent { active: boolean; isSpillover: boolean; retentionOverrideDays: number | null; + /** This item's own recap tone, independent of the global Merge-tab synthesis style. */ + recapStylePreset: 'default' | 'casual' | 'formal'; + /** Free-text instructions appended to this item's recap prompt specifically. */ + recapCustomInstructions: string; +} + +/** Response from POST /api/admin/events/:id/recap-now. */ +export interface ForceRecapResult { + published: boolean; + /** Set when published is true. */ + title?: string; + /** Set when published is false — why nothing was generated (no sources assigned, nothing new since last recap). */ + reason?: string; } export interface ModelCatalog { @@ -181,6 +201,11 @@ export interface ModelCatalog { synthesis: string[]; } +/** Response from GET /api/admin/model-context — the selected model's own reported max context length, or null if Ollama doesn't expose it for this model/version. */ +export interface ModelContextInfo { + contextLength: number | null; +} + export interface AiStatus { connected: boolean; host: string; @@ -195,6 +220,33 @@ export interface TelegramStatus { 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 { id: number; timestamp: string; diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index dba92e5..387dd23 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -34,6 +34,10 @@ export function getTags(fetchFn?: typeof fetch): Promise { return get('/api/tags', fetchFn); } +export function getTagBySlug(slug: string, fetchFn?: typeof fetch): Promise { + return get(`/api/tag/${slug}`, fetchFn); +} + export function getEvents(fetchFn?: typeof fetch): Promise { return get('/api/events', fetchFn); } diff --git a/frontend/src/lib/components/admin/EventsTab.svelte b/frontend/src/lib/components/admin/EventsTab.svelte index fd9a0e3..80d5c91 100644 --- a/frontend/src/lib/components/admin/EventsTab.svelte +++ b/frontend/src/lib/components/admin/EventsTab.svelte @@ -1,6 +1,7 @@ +{#if stats} +
+
+ Backlog + {stats.backlog.totalUnclusteredItems} + item{stats.backlog.totalUnclusteredItems === 1 ? '' : 's'} not yet published +
+
+ Awaiting embedding + {stats.backlog.awaitingEmbeddingItems} + need an embed() call before they can cluster +
+
+ Held for publishing + {stats.backlog.clusters.itemsOnHold} + + {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} + +
+
+ Awaiting synthesis + {stats.backlog.clusters.readyNowNeedingSynthesis} + multi-source cluster{stats.backlog.clusters.readyNowNeedingSynthesis === 1 ? '' : 's'} ready, needs an AI merge +
+
+ Estimated to clear + {formatEta(stats.estimatedMinutesToClear)} + + {#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} + +
+
+ Ollama right now + {#if stats.ollama.inFlight} + Synthesizing + {stats.ollama.inFlight.label} · {formatDuration(stats.ollama.inFlight.elapsedMs)} elapsed + {:else} + Idle + + {#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} + + {/if} +
+
+ Last direct-publish tick + {stats.lastDirectCycle ? stats.lastDirectCycle.published : '—'} + {formatAgo(stats.lastDirectCycle?.at ?? null)} +
+
+ Last synthesis tick + {stats.lastSynthesisCycle ? stats.lastSynthesisCycle.published : '—'} + {formatAgo(stats.lastSynthesisCycle?.at ?? null)} +
+
+{/if} +
@@ -70,6 +167,41 @@
diff --git a/frontend/src/routes/tag/[slug]/+page.ts b/frontend/src/routes/tag/[slug]/+page.ts new file mode 100644 index 0000000..04f9471 --- /dev/null +++ b/frontend/src/routes/tag/[slug]/+page.ts @@ -0,0 +1,22 @@ +import { error } from '@sveltejs/kit'; +import type { PageLoad } from './$types'; +import { getFeed, getTagBySlug } from '$lib/api'; + +const PAGE_SIZE = 15; + +// Mirrors /category/[name] and /event/[id] — a tag chip links by slug, so the slug is +// resolved to the real tag (id + label) via a dedicated backend lookup (GET +// /api/tag/:slug) rather than a preloaded list, since tags aren't loaded by the root +// layout the way categories/events are. +export const load: PageLoad = async ({ params, fetch }) => { + let tag; + try { + tag = await getTagBySlug(params.slug, fetch); + } catch { + throw error(404, 'Tag not found'); + } + + const filters = { tag: tag.id }; + const initial = await getFeed({ ...filters, limit: PAGE_SIZE }, fetch); + return { initial, filters, tag, pageSize: PAGE_SIZE }; +};