fix(summaries): drop wrong-page summaries; retry a stuck c0mpute node (0.26.1)

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",

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

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<string | undefined> = []
+    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[] = []