From 53837e6506122809763cf084b766773ae0d167a3 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 2 Nov 2026 12:16:36 +0000 Subject: [PATCH] fix(summaries): drop summaries of the wrong page; retry a stuck c0mpute node (0.26.1) - The TronBrowser relay handed concurrent reads each other's page (fixed in tronbrowser.dev #176), so different results got the same summary. Summaries written since 2026-11-02 11:00 UTC are dropped once and rewritten. - 'No live node serves qwen2.5:7b' now falls to the best general model Infernet lists as live (never a fine-tune or abliterated one), then any node. A node that takes the job and never answers gets ~55% of the budget, then one retry on another node, instead of failing at 90 s. Co-Authored-By: Claude Opus 5.5 --- package.json | 2 +- src/summarize.ts | 61 ++++++++++++++++++++++++++++++++++--- test/summarize.test.ts | 69 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 126 insertions(+), 6 deletions(-) diff --git a/package.json b/package.json index 8b95e96..dabf294 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "pitworld", - "version": "0.26.0", + "version": "0.26.1", "mcpName": "world.pit/pit", "description": "pit.world MCP server: private web search and Moshpit name lookups for AI agents (stdio, no API key). Also a web app, API, CLI and TUI.", "license": "MIT", diff --git a/src/summarize.ts b/src/summarize.ts index 2f2e45a..7fd93ea 100644 --- a/src/summarize.ts +++ b/src/summarize.ts @@ -49,7 +49,7 @@ export class C0mputeError extends Error { } } -export type C0mputeOptions = { baseUrl?: string; apiKey?: string; model?: string; timeoutMs?: number; fetch?: Fetch } +export type C0mputeOptions = { baseUrl?: string; apiKey?: string; model?: string; timeoutMs?: number; fetch?: Fetch; /** Share of timeoutMs the first node gets before one retry (default 0.55). */ firstTryShare?: number } /** * Infernet's chat API, the same way OutreachGraph calls it (packages/ai/src/c0mpute.ts): @@ -63,7 +63,46 @@ export type C0mputeOptions = { baseUrl?: string; apiKey?: string; model?: string export async function c0mputeChat(system: string, user: string, options: C0mputeOptions = {}): Promise<{ text: string; model: string }> { const base = (options.baseUrl ?? DEFAULT_C0MPUTE_URL).replace(/\/+$/, '') const fetchImpl = options.fetch ?? fetch - const signal = AbortSignal.timeout(options.timeoutMs ?? SUMMARY_TIMEOUT_MS) + const total = options.timeoutMs ?? SUMMARY_TIMEOUT_MS + const overall = AbortSignal.timeout(total) + const started = Date.now() + // A node that takes the job and never answers must not eat the whole budget: the first try gets + // a bit over half of it, then one retry on whichever node takes it next gets the rest. + const firstTry = Math.round(total * (options.firstTryShare ?? 0.55)) + let lastError: unknown + for (let attempt = 0; attempt < 2; attempt++) { + const left = total - (Date.now() - started) + if (left < 5_000) break + const signal = AbortSignal.any([overall, AbortSignal.timeout(attempt === 0 ? Math.min(firstTry, left) : left)]) + try { + return await c0mputeAttempt(base, fetchImpl, system, user, options, signal) + } catch (error) { + lastError = error + // Refused outright (bad key, bad request): another try gets the same answer. + if (error instanceof C0mputeError && error.status >= 400 && error.status < 500 && error.status !== 429) break + } + } + if (lastError instanceof C0mputeError) throw lastError + throw new C0mputeError(0, lastError instanceof Error ? lastError.message : String(lastError ?? 'no answer')) +} + +/** General models to fall back to, best first, when a node for the preferred one is not live. Never a fine-tune or an "abliterated" model. */ +export const FALLBACK_MODELS = ['qwen2.5:7b', 'qwen3:4b', 'gemma3:4b', 'qwen2.5:3b', 'qwen2.5:1.5b'] + +/** The best general model from Infernet's list of live ones, other than the one just refused. */ +export function pickModel(available: unknown, refused?: string): string | undefined { + if (!Array.isArray(available)) return undefined + return FALLBACK_MODELS.find((m) => m !== refused && available.includes(m)) +} + +async function c0mputeAttempt( + base: string, + fetchImpl: Fetch, + system: string, + user: string, + options: C0mputeOptions, + signal: AbortSignal, +): Promise<{ text: string; model: string }> { const submit = (modelName: string | undefined) => fetchImpl(`${base}/api/chat`, { method: 'POST', @@ -82,8 +121,14 @@ export async function c0mputeChat(system: string, user: string, options: C0mpute }) try { let submitted = await submit(options.model) - // No live node serves the preferred model right now: any free node will do. - if (submitted.status === 503 && options.model) submitted = await submit(undefined) + // No live node serves the preferred model right now: the best general model Infernet lists as + // live, then any free node. + if (submitted.status === 503 && options.model) { + const refusal = (await submitted.json().catch(() => ({}))) as { detail?: { availableModels?: unknown } } + const next = pickModel(refusal.detail?.availableModels, options.model) + submitted = await submit(next) + if (submitted.status === 503 && next) submitted = await submit(undefined) + } const job = (await submitted.json().catch(() => ({}))) as { jobId?: string; source?: string; model?: string | null; error?: string } if (!submitted.ok || !job.jobId) throw new C0mputeError(submitted.status, job.error ?? 'no job was created') if (job.source !== 'p2p') throw new C0mputeError(429, `no c0mpute node took the job (source ${job.source ?? 'unknown'})`) @@ -294,6 +339,14 @@ export class SummaryStore { // written from the engines' snippet alone, so they are rewritten from the page on the next request. const columns = this.db.query('PRAGMA table_info(summaries)').all() as Array<{ name: string }> if (!columns.some((c) => c.name === 'source')) this.db.exec('ALTER TABLE summaries ADD COLUMN source TEXT') + // One-time: from 2026-11-02 11:00 UTC the TronBrowser relay could hand concurrent reads each other's + // page (fixed in tronbrowser.dev #176), so summaries written since may describe another result. + // They are dropped once and rewritten on the next request. + this.db.exec('CREATE TABLE IF NOT EXISTS store_migrations (name TEXT PRIMARY KEY, at INTEGER NOT NULL)') + if (!this.db.query("SELECT 1 FROM store_migrations WHERE name = 'relay-crosstalk-2026-11-02'").get()) { + this.db.query('DELETE FROM summaries WHERE created_at >= ?').run(Date.UTC(2026, 10, 2, 11, 0)) + this.db.query("INSERT INTO store_migrations (name, at) VALUES ('relay-crosstalk-2026-11-02', ?)").run(Date.now()) + } } get(url: string): { summary: string; model: string; createdAt: number; source: string | null } | null { const row = this.db.query('SELECT summary, model, created_at, source FROM summaries WHERE url = ?').get(url) as { summary: string; model: string; created_at: number; source: string | null } | null diff --git a/test/summarize.test.ts b/test/summarize.test.ts index 5cec686..4bbf5dc 100644 --- a/test/summarize.test.ts +++ b/test/summarize.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from 'bun:test' import { Database } from 'bun:sqlite' import { tmpdir } from 'node:os' -import { c0mputeChat, cleanSummary, fetchPageText, htmlText, readPage, readStream, relayPageText, Summarizer, SummaryStore, summaryKey } from '../src/summarize.ts' +import { c0mputeChat, pickModel, cleanSummary, fetchPageText, htmlText, readPage, readStream, relayPageText, Summarizer, SummaryStore, summaryKey } from '../src/summarize.ts' import { createApp } from '../src/server.ts' import { Cache } from '../src/search.ts' @@ -31,6 +31,73 @@ function infernet(sent: any[], options: { source?: string; noModel?: boolean; su }) as typeof fetch } +describe('c0mpute when nodes are missing or stuck', () => { + const done = () => sse([['meta', { model: 'qwen3:4b' }], ['done', { text: 'Fine.' }]]) + + it('picks the best general model Infernet lists as live, never a fine-tune', () => { + expect(pickModel(['hf.co/x/saas-agent', 'jaahas/qwen3.5-uncensored:2b', 'gemma3:4b', 'qwen3:4b'], 'qwen2.5:7b')).toBe('qwen3:4b') + expect(pickModel(['qwen2.5:7b'], 'qwen2.5:7b')).toBeUndefined() + expect(pickModel(undefined)).toBeUndefined() + }) + + it('on "no live node serves it" asks for a listed model, then any node', async () => { + const asked: Array = [] + const f = (async (input: any, init?: RequestInit) => { + const url = String(input) + if (url.endsWith('/api/chat')) { + const body = JSON.parse(String(init?.body)) + asked.push(body.modelName) + if (body.modelName) return Response.json({ error: 'No live node serves it', detail: { availableModels: ['gemma3:4b', 'qwen3:4b'] } }, { status: 503 }) + return Response.json({ jobId: 'job-1', source: 'p2p' }) + } + return done() + }) as typeof fetch + const r = await c0mputeChat('s', 'u', { model: 'qwen2.5:7b', fetch: f }) + expect(asked).toEqual(['qwen2.5:7b', 'qwen3:4b', undefined]) + expect(r.text).toBe('Fine.') + }) + + it('a node that takes the job and never answers is retried once on another', async () => { + let jobs = 0 + const f = (async (input: any, init?: RequestInit) => { + const url = String(input) + if (url.endsWith('/api/chat')) return Response.json({ jobId: `job-${++jobs}`, source: 'p2p' }) + if (url.endsWith('/job-1')) { + // A stuck node: the stream stays open until the caller gives up. + return new Response(new ReadableStream({ start(c) { init?.signal?.addEventListener('abort', () => c.error(new Error('aborted'))) } })) + } + return done() + }) as typeof fetch + const t0 = Date.now() + const r = await c0mputeChat('s', 'u', { fetch: f, timeoutMs: 12_000, firstTryShare: 0.25 }) + expect(r.text).toBe('Fine.') + expect(jobs).toBe(2) + expect(Date.now() - t0).toBeLessThan(6_000) + }) + + it('does not retry a refused key', async () => { + let calls = 0 + const f = (async () => (calls++, Response.json({ error: 'bad key' }, { status: 401 }))) as unknown as typeof fetch + await expect(c0mputeChat('s', 'u', { fetch: f })).rejects.toThrow('401') + expect(calls).toBe(1) + }) + + it('drops summaries written while the page reader mixed up pages, once', () => { + const path = `${tmpdir()}/pit-crosstalk-${process.pid}-${Date.now()}.db` + const old = new Database(path) + old.exec('CREATE TABLE summaries (url TEXT PRIMARY KEY, summary TEXT NOT NULL, model TEXT NOT NULL, created_at INTEGER NOT NULL, source TEXT)') + old.query('INSERT INTO summaries VALUES (?, ?, ?, ?, ?)').run('before.example', 'ok', 'm', Date.UTC(2026, 10, 2, 10), 'page') + old.query('INSERT INTO summaries VALUES (?, ?, ?, ?, ?)').run('during.example', 'wrong page', 'm', Date.UTC(2026, 10, 2, 12), 'page') + old.close() + const store = new SummaryStore(path) + expect(store.get('before.example')?.summary).toBe('ok') + expect(store.get('during.example')).toBeNull() + // Later summaries survive a restart. + store.set('later.example', 'right page', 'm') + expect(new SummaryStore(path).get('later.example')?.summary).toBe('right page') + }) +}) + describe('c0mpute client', () => { it('sends p2p_only, the trusted key and the model, and reads the stream', async () => { const sent: any[] = []