From 2af32a77d82eb6395ad8a8ccd81245066cede8c4 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 21:14:33 +0000 Subject: [PATCH] Bound wiki-agent context use: output caps + distill-then-act for oversized threads On 2026-07-23 the wiki sweep skipped two long threads with "maximum context length is 163840 tokens ... you requested 65536 output tokens". Diagnosis found three compounding problems, none of them "the model is too small": 1. The headless tool loop never set max_completion_tokens, so the serving backend reserved its own 65536-token default output budget out of the window - for an agent whose real output is tool calls plus a two-sentence summary. runHeadlessAgent now takes maxTokens and the wiki fleet (article agent, record extraction, librarian) caps every round at 8192, reclaiming ~57k tokens of input room. 2. Thread slices were fed to the model unbounded, and the enforced ceiling is not the registry's contextWindow: the same model id (deepseek-v4-flash) rejected at 163840 on Jul 23 and accepted 1M-scale requests a week later. New agents/_accumulator.ts ports fnord's accumulator pattern as distill-then-act: transcripts estimated over a pinned conservative working window (96k tokens) are chunked and folded into an accumulated notes buffer by read-only distill completions, and the normal tool loop then runs once over the notes with the full toolbox, so write dedup discipline is untouched. A context-length 400 on a direct run retries through the same path; a 400 during distillation backs off the chunk budget (0.2 steps, 0.6 floor) before giving up. Context-length errors that survive distillation are flagged deterministic and skip the thread on the first failure with an honest reason instead of burning the transient 3-strike budget. 3. The content-filter fallback model (arcee-trinity-large-thinking) no longer exists in Venice's catalog, so the fallback path could never succeed. Re-pointed to venice-uncensored-1-2 (supports function calling; non-reasoning, so the fallback attempt keeps reasoning_effort off the wire). Threads that fit - 410 of 412 in the live DB - run byte-identical to before; the distill path is a backstop for the tail. Covered by supabase/functions/tests/accumulator.test.ts plus new behavior tests for the mid-run recovery, the deterministic-skip surface, and the per-round output cap. CLAUDE.md gains the "always set maxTokens / never trust contextWindow" lesson so the next sub-call doesn't rediscover it. --- CLAUDE.md | 14 + docs/dev/wiki.md | 51 ++- docs/user/wiki.md | 2 +- supabase/functions/tests/accumulator.test.ts | 224 ++++++++++++ .../functions/tests/wiki_behavior.test.ts | 119 ++++++- .../functions/venice/agents/_accumulator.ts | 333 ++++++++++++++++++ supabase/functions/venice/agents/_run.ts | 14 + supabase/functions/venice/agents/wiki.ts | 166 +++++++-- .../functions/venice/agents/wiki_librarian.ts | 5 + .../functions/venice/agents/wiki_records.ts | 108 +++++- 10 files changed, 984 insertions(+), 52 deletions(-) create mode 100644 supabase/functions/tests/accumulator.test.ts create mode 100644 supabase/functions/venice/agents/_accumulator.ts diff --git a/CLAUDE.md b/CLAUDE.md index 8eb090d1..fee6ba1a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -415,6 +415,20 @@ map). When adding an agent sub-call: rather than treating truncated output as a valid empty result - `completeJsonObjectWithMeta` in `venice/agents/_curation_helpers.ts` exists for this. +- ALWAYS set `max_completion_tokens` (`maxTokens`) explicitly, + including on `runHeadlessAgent` tool loops. When the field is + absent from the wire body, the serving backend reserves its own + default output budget out of the context window - observed at + 65536 tokens - and long inputs then 400 with "maximum context + length exceeded" even though the expected output is tiny. +- Do not treat the registry's `contextWindow` + (src/lib/models/index.ts) as an enforced contract. The ceiling + belongs to whatever backend is serving the model id and it moves: + deepseek-v4-flash enforced 163840 on 2026-07-23 and accepted + 1M-scale requests a week later. Code that must fit inside a + window budgets against a pinned conservative constant (see + `WORKING_CONTEXT_TOKENS` in `venice/agents/_accumulator.ts`) and + handles the context-length 400 reactively. ## User-facing documentation diff --git a/docs/dev/wiki.md b/docs/dev/wiki.md index fdcc1dcb..fb253d1c 100644 --- a/docs/dev/wiki.md +++ b/docs/dev/wiki.md @@ -149,9 +149,19 @@ Edge function (`supabase/functions/venice/`): and fallback constants, and the per-run tunables: model `deepseek-v4-flash` (hardcoded mirror of `agentModel('wiki')` - `AGENT_MODELS` is a static role->model map, not a per-user tier), - fallback `arcee-trinity-large-thinking`, `reasoningEffort: - 'medium'`, claim TTL 600s, failure cap 3, sweep bound 3 - threads/tick. Test-only invariants exported via `__test`. + fallback `venice-uncensored-1-2`, `reasoningEffort: + 'medium'`, output cap 8192 tokens/round, claim TTL 600s, failure + cap 3, sweep bound 3 threads/tick. Test-only invariants exported + via `__test`. +- `agents/_accumulator.ts` - distill-then-act support for oversized + transcripts (port of fnord's accumulator pattern): the pinned + `WORKING_CONTEXT_TOKENS` budget, `transcriptFitsDirect`, + `distillTranscript` (chunked notes accumulation w/ context-length + backoff), `isContextLengthError`, and + `renderDistilledNotesBlock`. Shared by `wiki.ts` and + `wiki_records.ts`; see "Context-window handling" under the sweep + section for the full flow. Unit-tested at + `supabase/functions/tests/accumulator.test.ts`. - `agents/wiki_manual.ts` - the per-article manual agent (the "Ask agent to update" flow). Exports `runWikiManualUpdate(adminClient, userId, { articleId, instructions })`: reads the persisted article @@ -748,7 +758,7 @@ conversation. - `wiki_skip_fallback_attempted boolean not null default false` - true when the per-thread skip was stamped after the agent already retried with the uncensored fallback model - (`arcee-trinity-large-thinking` per + (`venice-uncensored-1-2` per `CONTENT_FILTER_FALLBACK_MODEL` in the edge agent). The eligibility predicate's OR clause uses this to re-eligibilise legacy content-classifier skips (rows skipped before the fallback @@ -1042,8 +1052,10 @@ the route, which pg_net ignores but the dev shim prints. **Content-classifier fallback (in `runWikiAgentOnThread`).** Before the per-thread failure counter ever increments for a classifier rejection, the agent itself retries the tool loop once against -`CONTENT_FILTER_FALLBACK_MODEL` (`arcee-trinity-large-thinking`, -which does not run the same input classifier). Only the +`CONTENT_FILTER_FALLBACK_MODEL` (`venice-uncensored-1-2`, +which does not run the same input classifier; it must support +function calling and is a non-reasoning model, so the fallback +attempt keeps `reasoning_effort` off the wire). Only the content-filter sentinel (`"Input text data may contain inappropriate content"`, matched as a substring of the error message) triggers the retry; any other error (network blip, 500, parse failure) returns @@ -1056,6 +1068,33 @@ and when the counter eventually advances the pointer, so the eligibility predicate's OR clause can't loop the same thread back into the queue. +**Context-window handling (distill-then-act, `_accumulator.ts`).** +Both per-thread agents (article + record extraction) bound what they +send to Venice. Every tool-loop round carries an explicit +`max_completion_tokens` (8192); without it, the serving backend +reserves its own default output budget - observed at 65536 tokens - +out of the context window, which on 2026-07-23 starved two long +threads of input room and skipped them ("maximum context length is +163840 tokens"). The registry's `contextWindow` for the wiki models +says 1M, but the enforced ceiling demonstrably moves between +backends serving the same model id, so the agents budget against the +conservative `WORKING_CONTEXT_TOKENS` (96k) in +`agents/_accumulator.ts` instead. A transcript estimated over that +window is **distilled before the tool loop**: the accumulator +renders the slice to text (tool traffic excerpted), chunks it, and +folds each chunk into an accumulated notes buffer via one completion +per chunk - fnord's accumulator pattern - then the normal tool loop +runs once over the notes instead of the raw transcript, so all +writes still happen with the full toolbox and dedup discipline +(chunk passes are read-only by design). A context-length 400 from a +direct run triggers the same distill path reactively; a 400 during +distillation backs off the chunk budget (0.2 steps, 0.6 floor) +before giving up. A context-length error that survives all of that +is marked `deterministic` on the run outcome and the sweep skips the +thread on the FIRST failure (`p_max_failures = 1`) with the honest +reason, instead of burning the transient 3-strike budget on an error +that repeats identically. + This differs from the journal flow, which uses an atomic `upsert_journal_entry_and_mark_thread` RPC because the entry write and the pointer advance must happen in lockstep. The wiki agent's diff --git a/docs/user/wiki.md b/docs/user/wiki.md index e53743e4..9513b963 100644 --- a/docs/user/wiki.md +++ b/docs/user/wiki.md @@ -353,7 +353,7 @@ Sometimes the agent can't process a conversation. The most common reason is Venice's content classifier rejecting the conversation body as inappropriate. When that happens, the agent automatically retries the run against an **uncensored fallback model** (currently -`arcee-trinity-large-thinking`) which doesn't run the same +`venice-uncensored-1-2`) which doesn't run the same classifier, so most conversations the default model balks at get processed transparently on the retry. You don't have to do anything; it just works on the next sweep. diff --git a/supabase/functions/tests/accumulator.test.ts b/supabase/functions/tests/accumulator.test.ts new file mode 100644 index 00000000..47af467e --- /dev/null +++ b/supabase/functions/tests/accumulator.test.ts @@ -0,0 +1,224 @@ +// Unit coverage for the distill-then-act accumulator +// (venice/agents/_accumulator.ts): the token estimate + fits-direct +// gate, the chunker's whole-block / hard-split behavior, message +// rendering (tool-traffic excerpting), the distill loop's buffer +// threading, the context-length backoff, and the empty-completion +// guard. All offline - the completion seam is scripted, no network. + +import { assertEquals, assertRejects, assertStringIncludes } from '@std/assert'; +import { + __test, + distillTranscript, + estimateWireTokens, + isContextLengthError, + renderDistilledNotesBlock, + transcriptFitsDirect, + WORKING_CONTEXT_TOKENS, +} from '../venice/agents/_accumulator.ts'; +import type { VeniceWireMessage } from '../venice/agents/_recall_helpers.ts'; +import type { ToolCompletionResult } from '../venice/tools/_venice_complete.ts'; + +const { nextChunk, renderMessage, TOOL_RESULT_EXCERPT_CHARS } = __test; + +function completion(partial: Partial): ToolCompletionResult { + return { + text: '', + reasoning: '', + citations: [], + finishReason: 'stop', + usage: null, + toolCalls: [], + ...partial, + }; +} + +const CTX_ERROR = new Error( + "Venice chat/completions 400: {\"error\":{\"message\":\"This model's maximum context length is 163840 tokens.", +); + +Deno.test('isContextLengthError matches the upstream sentinel and nothing else', () => { + assertEquals(isContextLengthError(CTX_ERROR), true); + assertEquals(isContextLengthError(new Error('Venice chat/completions 500: gateway error')), false); + assertEquals( + isContextLengthError( + new Error('Venice chat/completions 400: {"error":"Input text data may contain inappropriate content."}'), + ), + false, + ); + assertEquals(isContextLengthError('maximum context length'), false); // not an Error + assertEquals(isContextLengthError(null), false); +}); + +Deno.test('estimateWireTokens counts content plus tool_calls JSON', () => { + const messages: VeniceWireMessage[] = [ + { role: 'user', content: 'a'.repeat(400) }, + { + role: 'assistant', + content: '', + tool_calls: [{ id: '1', type: 'function', function: { name: 'x', arguments: '{}' } }], + }, + ]; + const tokens = estimateWireTokens(messages); + // 400 chars of content plus the tool_calls JSON, at 4 chars/token. + const toolJson = JSON.stringify(messages[1].tool_calls).length; + assertEquals(tokens, Math.ceil((400 + toolJson) / 4)); +}); + +Deno.test('transcriptFitsDirect gates on the working context window', () => { + const small: VeniceWireMessage[] = [{ role: 'user', content: 'hi' }]; + assertEquals(transcriptFitsDirect(small), true); + const big: VeniceWireMessage[] = [ + { role: 'user', content: 'x'.repeat((WORKING_CONTEXT_TOKENS + 1) * 4) }, + ]; + assertEquals(transcriptFitsDirect(big), false); +}); + +Deno.test('nextChunk takes whole blocks up to the budget', () => { + const { chunk, consumed, remainder } = nextChunk(['aaaa', 'bbbb', 'cccc'], 10); + // 4 + 1 + 4 + 1 = 10 fits two blocks; the third would overflow. + assertEquals(consumed, 2); + assertEquals(chunk, 'aaaa\nbbbb'); + assertEquals(remainder, null); +}); + +Deno.test('nextChunk hard-splits a single oversized block and returns the tail', () => { + const { chunk, consumed, remainder } = nextChunk(['x'.repeat(25), 'next'], 10); + assertEquals(consumed, 1); + assertEquals(chunk, 'x'.repeat(10)); + assertEquals(remainder, 'x'.repeat(15)); +}); + +Deno.test('renderMessage keeps prose whole and excerpts tool traffic', () => { + const prose = renderMessage({ role: 'user', content: 'tell me about my dog' }); + assertEquals(prose, '[user]\ntell me about my dog'); + + const bigResult = renderMessage({ + role: 'tool', + content: 'r'.repeat(TOOL_RESULT_EXCERPT_CHARS + 500), + name: 'web_search', + }); + assertStringIncludes(bigResult, '[tool result (web_search)]'); + assertStringIncludes(bigResult, '500 more chars omitted'); + + const withCalls = renderMessage({ + role: 'assistant', + content: 'checking', + tool_calls: [ + { id: '1', type: 'function', function: { name: 'wiki_search', arguments: '{"query":"dog"}' } }, + ], + }); + assertStringIncludes(withCalls, '(called wiki_search with {"query":"dog"})'); +}); + +Deno.test('distillTranscript threads the notes buffer across chunks', async () => { + // Two blocks sized so each fills most of a chunk budget - forces two + // distill passes with the first pass's notes visible to the second. + const blockChars = (WORKING_CONTEXT_TOKENS - 20_000) * 4; + const messages: VeniceWireMessage[] = [ + { role: 'user', content: 'a'.repeat(blockChars) }, + { role: 'user', content: 'b'.repeat(blockChars) }, + ]; + const seenNotes: string[] = []; + let pass = 0; + const notes = await distillTranscript({ + apiKey: 'k', + model: 'm', + messages, + focus: 'FOCUS-MARKER', + // deno-lint-ignore require-await + complete: async (opts) => { + pass += 1; + const system = String(opts.messages[0].content); + assertStringIncludes(system, 'FOCUS-MARKER'); + const user = String(opts.messages[1].content); + seenNotes.push(user.split('# Next transcript chunk')[0]); + return completion({ text: `NOTES-${pass}` }); + }, + }); + assertEquals(pass, 2); + assertEquals(notes, 'NOTES-2'); + assertStringIncludes(seenNotes[0], '(none yet - this is the first chunk)'); + // Pass 2 builds on pass 1's notes, not on a reset buffer. + assertStringIncludes(seenNotes[1], 'NOTES-1'); +}); + +Deno.test('distillTranscript backs off the chunk budget on a context-length rejection', async () => { + const messages: VeniceWireMessage[] = [ + { role: 'user', content: 'z'.repeat(WORKING_CONTEXT_TOKENS * 4 * 2) }, + ]; + const chunkSizes: number[] = []; + let first = true; + const notes = await distillTranscript({ + apiKey: 'k', + model: 'm', + messages, + focus: 'f', + // deno-lint-ignore require-await + complete: async (opts) => { + const user = String(opts.messages[1].content); + chunkSizes.push(user.length); + if (first) { + first = false; + throw CTX_ERROR; + } + return completion({ text: 'ok' }); + }, + }); + assertEquals(notes, 'ok'); + // The retry after the rejection sent a strictly smaller chunk. + assertEquals(chunkSizes.length >= 2, true); + assertEquals(chunkSizes[1] < chunkSizes[0], true); +}); + +Deno.test('distillTranscript gives up when backoff hits the floor', async () => { + const messages: VeniceWireMessage[] = [ + { role: 'user', content: 'z'.repeat(WORKING_CONTEXT_TOKENS * 4) }, + ]; + let calls = 0; + await assertRejects( + () => + distillTranscript({ + apiKey: 'k', + model: 'm', + messages, + focus: 'f', + // deno-lint-ignore require-await + complete: async () => { + calls += 1; + throw CTX_ERROR; + }, + }), + Error, + 'maximum context length', + ); + // Fractions tried: 1.0, 0.8, 0.6 - then 0.4 would cross the floor, + // so the third rejection surfaces. + assertEquals(calls, 3); +}); + +Deno.test('distillTranscript rejects an empty distill pass instead of wiping the notes', async () => { + const messages: VeniceWireMessage[] = [{ role: 'user', content: 'hello' }]; + await assertRejects( + () => + distillTranscript({ + apiKey: 'k', + model: 'm', + messages, + focus: 'f', + // A truncated reasoning pass: the whole output budget went to + // thinking and content came back empty. + // deno-lint-ignore require-await + complete: async () => completion({ text: '', finishReason: 'length' }), + }), + Error, + 'finish_reason=length', + ); +}); + +Deno.test('renderDistilledNotesBlock wraps the notes and names them a distillation', () => { + const block = renderDistilledNotesBlock('THE NOTES'); + assertStringIncludes(block, ''); + assertStringIncludes(block, 'THE NOTES'); + assertStringIncludes(block, 'too large to include verbatim'); + assertStringIncludes(block, ''); +}); diff --git a/supabase/functions/tests/wiki_behavior.test.ts b/supabase/functions/tests/wiki_behavior.test.ts index a64bc215..22cdcf84 100644 --- a/supabase/functions/tests/wiki_behavior.test.ts +++ b/supabase/functions/tests/wiki_behavior.test.ts @@ -96,10 +96,12 @@ Deno.test('retry: classifier rejection on the primary retries the uncensored fal tables: HAPPY_TABLES, }); + const efforts: Array = []; const result = await retryWikiThread(admin, 'u', 't-1', { // deno-lint-ignore require-await complete: async (opts) => { models.push(opts.model); + efforts.push(opts.reasoningEffort); if (opts.model === 'deepseek-v4-flash') throw FILTER_ERROR; return completion({ text: 'Fallback ran, no edits warranted.' }); }, @@ -115,7 +117,10 @@ Deno.test('retry: classifier rejection on the primary retries the uncensored fal } // Order matters: primary first, fallback second. A reversed order // would mean the agent skipped the configured model entirely. - assertEquals(models, ['deepseek-v4-flash', 'arcee-trinity-large-thinking']); + assertEquals(models, ['deepseek-v4-flash', 'venice-uncensored-1-2']); + // The fallback is a non-reasoning model: reasoning_effort must stay + // off its wire body (some providers 400 on the unknown field). + assertEquals(efforts, ['medium', undefined]); // Success advances the pointer + clears the skip marker. assertEquals( rpcCalls.some((c) => c.name === 'manual_advance_wiki_pointer'), @@ -172,15 +177,15 @@ Deno.test('retry: both attempts failing surfaces the FALLBACK error', async () = complete: async (opts) => { models.push(opts.model); if (opts.model === 'deepseek-v4-flash') throw FILTER_ERROR; - throw new Error('arcee timeout'); + throw new Error('fallback timeout'); }, }); assertEquals(result.kind, 'error'); // The primary's classifier rejection is no longer the headline once // we have moved past it - the fallback's failure is what stopped us. - if (result.kind === 'error') assertStringIncludes(result.error, 'arcee timeout'); - assertEquals(models, ['deepseek-v4-flash', 'arcee-trinity-large-thinking']); + if (result.kind === 'error') assertStringIncludes(result.error, 'fallback timeout'); + assertEquals(models, ['deepseek-v4-flash', 'venice-uncensored-1-2']); }); Deno.test('retry: no terminal assistant message is a no-op that never reaches Venice', async () => { @@ -244,6 +249,112 @@ Deno.test('retry: an already-claimed thread is busy and never reaches Venice', a ); }); +// A completion request is a distill pass when it carries no tools - +// the tool loop always sends the toolbox wire list, the distill +// completions never do. +function isDistillCall(opts: { tools?: readonly unknown[] }): boolean { + return !opts.tools || opts.tools.length === 0; +} + +const CTX_ERROR = new Error( + "Venice chat/completions 400: {\"error\":{\"message\":\"This model's maximum context length is 163840 tokens. However, you requested 8192 output tokens and your prompt contains at least 160000 input tokens\"}}", +); + +Deno.test('retry: a mid-run context-length rejection distills the transcript and retries, without the uncensored fallback', async () => { + const calls: Array<'act' | 'distill'> = []; + const models = new Set(); + const { admin } = makeAdmin({ + rpc: (name) => + name === 'claim_wiki_thread_for_retry' + ? { data: true, error: null } + : name === 'compute_wiki_terminal_msg_id' + ? { data: 'a1', error: null } + : { data: null, error: null }, + tables: HAPPY_TABLES, + }); + + const result = await retryWikiThread(admin, 'u', 't-1', { + // deno-lint-ignore require-await + complete: async (opts) => { + models.add(opts.model); + if (isDistillCall(opts)) { + calls.push('distill'); + return completion({ text: 'DISTILLED NOTES' }); + } + calls.push('act'); + // First act attempt: the backend says the transcript is too big. + if (calls.filter((c) => c === 'act').length === 1) throw CTX_ERROR; + // Second act attempt must be reading notes, not the transcript. + const first = opts.messages[0]; + assertStringIncludes(String(first.content), ''); + assertStringIncludes(String(first.content), 'DISTILLED NOTES'); + return completion({ text: 'Processed from notes.' }); + }, + }); + + assertEquals(result.kind, 'ok'); + if (result.kind === 'ok') assertEquals(result.reasoning, 'Processed from notes.'); + assertEquals(calls, ['act', 'distill', 'act']); + // The context-length path stays on the primary model - the + // uncensored fallback is for classifier rejections only. + assertEquals([...models], ['deepseek-v4-flash']); +}); + +Deno.test('retry: a context-length rejection that survives distillation surfaces as an error', async () => { + const { admin, rpcCalls } = makeAdmin({ + rpc: (name) => + name === 'claim_wiki_thread_for_retry' + ? { data: true, error: null } + : name === 'compute_wiki_terminal_msg_id' + ? { data: 'a1', error: null } + : { data: null, error: null }, + tables: HAPPY_TABLES, + }); + + const result = await retryWikiThread(admin, 'u', 't-1', { + // Every completion - act and distill alike - hits the ceiling. + // deno-lint-ignore require-await + complete: async () => { + throw CTX_ERROR; + }, + }); + + assertEquals(result.kind, 'error'); + if (result.kind === 'error') assertStringIncludes(result.error, 'maximum context length'); + // No pointer advance: the thread stays visible in the Skipped panel. + assertEquals( + rpcCalls.some((c) => c.name === 'manual_advance_wiki_pointer'), + false, + ); +}); + +Deno.test('retry: agent rounds carry an explicit max_completion_tokens cap', async () => { + const caps: Array = []; + const { admin } = makeAdmin({ + rpc: (name) => + name === 'claim_wiki_thread_for_retry' + ? { data: true, error: null } + : name === 'compute_wiki_terminal_msg_id' + ? { data: 'a1', error: null } + : { data: null, error: null }, + tables: HAPPY_TABLES, + }); + + const result = await retryWikiThread(admin, 'u', 't-1', { + // deno-lint-ignore require-await + complete: async (opts) => { + caps.push(opts.maxTokens); + return completion({ text: 'done' }); + }, + }); + + assertEquals(result.kind, 'ok'); + // An absent cap makes the serving backend reserve its own default + // output budget (observed 65536 tokens) out of the context window - + // the root cause of the 2026-07-23 skipped threads. + assertEquals(caps, [8_192]); +}); + Deno.test('retry: a pointer-advance failure after a successful run surfaces as an error', async () => { const { admin } = makeAdmin({ rpc: (name) => { diff --git a/supabase/functions/venice/agents/_accumulator.ts b/supabase/functions/venice/agents/_accumulator.ts new file mode 100644 index 00000000..b59de69a --- /dev/null +++ b/supabase/functions/venice/agents/_accumulator.ts @@ -0,0 +1,333 @@ +// Distill-then-act support for headless agents whose input transcript +// may not fit the serving backend's context window. Port of the +// accumulator pattern from fnord (lib/ai/accumulator.ex): when a +// transcript is too large to hand to the tool-loop agent verbatim, we +// render it to text, split it into chunks, and run one completion per +// chunk that folds the chunk into an accumulated notes buffer. The +// notes - not the raw transcript - then feed the agent's normal tool +// loop, so all writes still happen in one pass with the full toolbox +// and the usual dedup discipline. Chunk passes are read-only by +// design: letting them write would mean tool calls issued from +// partial knowledge of the conversation. +// +// Consumers: agents/wiki.ts and agents/wiki_records.ts (the two +// sweep agents that feed whole thread slices to a model). The other +// slice consumers (reflection, summary, samskara_evaluation, +// thread_topics) still send unbounded slices and can adopt this +// module when their inputs start hitting the same wall. + +import { toolComplete } from '../tools/_venice_complete.ts'; +import type { AgentCompleteFn } from './_run.ts'; +import type { VeniceWireMessage } from './_recall_helpers.ts'; + +// The char->token estimate used across nak (see MAX_RECALL_CHARS in +// _recall_helpers.ts: "~50k tokens at 4 chars/token"). Conservative +// for English prose; CJK-heavy content runs closer to 1 char/token, +// which the working-window margin below absorbs. +const CHARS_PER_TOKEN = 4; + +/** + * The context window we BUDGET against, deliberately far below the + * 1M the model registry claims for the wiki fleet's models. The + * registry number is unverified metadata, not a contract: on + * 2026-07-23 the backend serving deepseek-v4-flash enforced a 163840 + * ceiling (two threads were skipped with "maximum context length is + * 163840 tokens"), and a probe a week later accepted requests sized + * for 1M+. Budgeting against a number that moves under us means 400s; + * budgeting conservatively costs at most a few extra distill passes + * on rare oversized threads. 96k also keeps headroom under the lowest + * ceiling observed in production for the act pass's tool schemas and + * mid-loop tool-result growth. + */ +export const WORKING_CONTEXT_TOKENS = 96_000; + +// Output budget for each distill completion. The notes for one chunk +// are a few paragraphs; thousands of tokens of headroom because a +// reasoning model's thinking pass spends from the same budget (see +// CLAUDE.md "Venice sub-completions on reasoning models"). +const DISTILL_MAX_OUTPUT_TOKENS = 8_192; + +// Mirror of fnord's backoff knobs: on a context-length 400, shrink +// the chunk budget by BACKOFF_STEP of the working window and retry; +// below BACKOFF_FLOOR give up. Deliberate divergence from fnord: the +// shrunken fraction persists across subsequent chunks instead of +// resetting per chunk - a ceiling tight enough to reject one chunk +// will reject the next identically, and resetting guarantees a +// failed call per chunk. +const BACKOFF_FLOOR = 0.6; +const BACKOFF_STEP = 0.2; + +// Below this chunk budget the distill loop errors out rather than +// crawling through a transcript a paragraph at a time - if overhead +// (prompt + accumulated notes + output reserve) eats the window down +// to this, something is structurally wrong. +const MIN_CHUNK_CHARS = 8_000; + +// Tool traffic inside the transcript is mostly environmental (search +// dumps, article bodies the chat already read); the distill notes +// target what the USER said and decided. Excerpting keeps a single +// web_search result from displacing three user turns from a chunk. +const TOOL_RESULT_EXCERPT_CHARS = 2_000; +const TOOL_CALL_ARGS_EXCERPT_CHARS = 500; + +/** + * Substring sentinel for the upstream "prompt too large" rejection. + * Observed shape (truncated to VeniceError's 200-char body slice): + * + * Venice chat/completions 400: {"error":{"message":"This model's + * maximum context length is 163840 tokens. However, you requested + * 65536 output tokens and your prompt contains at least 98305 input + * tokens, ... + * + * Matching the phrase rather than the status keeps it narrow: content + * -classifier 400s and malformed-request 400s carry different bodies + * and must stay on their own error paths. + */ +const CONTEXT_LENGTH_SENTINEL = 'maximum context length'; + +/** + * True when `err` is the deterministic "input does not fit the + * model's context window" rejection. Callers branch on this to + * distill-and-retry (or to skip with an honest reason) instead of + * burning transient-failure retries on an error that cannot clear. + */ +export function isContextLengthError(err: unknown): boolean { + if (!(err instanceof Error)) return false; + return err.message.includes(CONTEXT_LENGTH_SENTINEL); +} + +/** + * Rough token estimate for the wire-shape transcript the act pass + * would send: message content plus tool_calls JSON, at the standard + * chars-per-token ratio. + */ +export function estimateWireTokens(messages: readonly VeniceWireMessage[]): number { + let chars = 0; + for (const m of messages) { + chars += m.content?.length ?? 0; + if (m.tool_calls && m.tool_calls.length > 0) { + chars += JSON.stringify(m.tool_calls).length; + } + } + return Math.ceil(chars / CHARS_PER_TOKEN); +} + +/** + * Whether a transcript is small enough to hand to the tool loop + * verbatim. Over-budget transcripts go through distillTranscript + * first; under-budget ones run exactly as they always have. + */ +export function transcriptFitsDirect(messages: readonly VeniceWireMessage[]): boolean { + return estimateWireTokens(messages) <= WORKING_CONTEXT_TOKENS; +} + +function excerpt(text: string, cap: number): string { + if (text.length <= cap) return text; + return `${text.slice(0, cap)}\n[... ${text.length - cap} more chars omitted from this distillation]`; +} + +/** + * Render one wire message as a plain-text block for chunking. Tool + * results and tool-call arguments are excerpted (see the constants + * above); user and assistant prose is kept whole. + */ +function renderMessage(m: VeniceWireMessage): string { + const lines: string[] = []; + if (m.role === 'tool') { + lines.push(`[tool result${m.name ? ` (${m.name})` : ''}]`); + if (m.content) lines.push(excerpt(m.content, TOOL_RESULT_EXCERPT_CHARS)); + } else { + lines.push(`[${m.role}]`); + if (m.content) lines.push(m.content); + for (const call of m.tool_calls ?? []) { + lines.push( + `(called ${call.function.name} with ${excerpt(call.function.arguments, TOOL_CALL_ARGS_EXCERPT_CHARS)})`, + ); + } + } + return lines.join('\n'); +} + +const DISTILL_SYSTEM_PROMPT = `You are distilling a long conversation transcript into working notes for another agent. You process the transcript in sequential chunks; each request shows your accumulated notes so far and the next chunk. + +Update the notes: +1. Fold in everything from the current chunk that matters for the goal below. +2. Build on the existing notes, preserving their structure. Revise an entry only when the new chunk contradicts or refines it. Never drop an entry. +3. Record concrete details: dates, names, quantities, decisions, reversals. Quote short key statements verbatim when the wording carries weight. +4. If the chunk ends mid-topic, mark the dangling thread with so a later chunk can complete it. + +Respond ONLY with the full updated notes - no preamble, no commentary.`; + +/** + * Take the next chunk from the pending rendered blocks without + * consuming them - the caller commits the consumption only after the + * chunk's completion succeeds, so a backoff retry recomputes from + * untouched state. + * + * Whole blocks are preferred; a single block larger than the budget + * is hard-split, with `remainder` carrying the tail back to the head + * of the queue. + */ +function nextChunk( + pending: readonly string[], + budgetChars: number, +): { chunk: string; consumed: number; remainder: string | null } { + const parts: string[] = []; + let used = 0; + let consumed = 0; + for (const block of pending) { + // +1 for the joining newline. + if (used + block.length + 1 > budgetChars && consumed > 0) break; + if (block.length > budgetChars && consumed === 0) { + // Oversized single block: split it at the budget. + return { + chunk: block.slice(0, budgetChars), + consumed: 1, + remainder: block.slice(budgetChars), + }; + } + parts.push(block); + used += block.length + 1; + consumed += 1; + } + return { chunk: parts.join('\n'), consumed, remainder: null }; +} + +export interface DistillTranscriptOptions { + apiKey: string; + /** Concrete Venice model id; callers pass the same id the act pass runs on. */ + model: string; + /** The transcript to distill - the slice WITHOUT the agent's final prompt turn. */ + messages: readonly VeniceWireMessage[]; + /** + * What the notes must capture, stated for the distill model. Agent- + * specific: the wiki agent wants user-revealing facts and article + * subjects, the records agent wants dated events. + */ + focus: string; + /** + * reasoning_effort for the distill calls. Callers pass 'low' for + * reasoning models (extraction over in-context evidence does not + * earn a thinking budget - see CLAUDE.md) and omit it entirely for + * non-reasoning models, where the field can 400. + */ + reasoningEffort?: 'low' | 'medium' | 'high'; + /** Test seam; defaults to toolComplete (the live Venice call). */ + complete?: AgentCompleteFn; + /** Best-effort progress line, e.g. into the run's edge logger. */ + onInfo?: (message: string) => void; +} + +/** + * Distill an oversized transcript into an accumulated notes string. + * Throws on failure - including a context-length rejection that + * backoff cannot clear - so the caller's normal error path handles + * it; a thrown context-length error from HERE means the thread is + * unprocessable at any chunking, which is skip-with-honest-reason + * territory. + */ +export async function distillTranscript(opts: DistillTranscriptOptions): Promise { + const complete = opts.complete ?? toolComplete; + const info = opts.onInfo ?? ((): void => {}); + const systemPrompt = `${DISTILL_SYSTEM_PROMPT}\n\n# Goal\n${opts.focus}`; + + const pending = opts.messages.map(renderMessage); + let buffer = ''; + let frac = 1.0; + let pass = 0; + + while (pending.length > 0) { + // Everything that rides along with the chunk spends from the same + // window: the prompt, the accumulated notes, and the reserved + // output budget. + const overheadTokens = Math.ceil((systemPrompt.length + buffer.length) / CHARS_PER_TOKEN) + + DISTILL_MAX_OUTPUT_TOKENS; + const budgetChars = (Math.floor(WORKING_CONTEXT_TOKENS * frac) - overheadTokens) * + CHARS_PER_TOKEN; + if (budgetChars < MIN_CHUNK_CHARS) { + throw new Error( + `distillation cannot fit a workable chunk (budget ${budgetChars} chars at fraction ${frac.toFixed(1)})`, + ); + } + + const { chunk, consumed, remainder } = nextChunk(pending, budgetChars); + const userPrompt = `# Notes so far\n${ + buffer || '(none yet - this is the first chunk)' + }\n\n# Next transcript chunk\n${chunk}`; + + let text: string; + let finishReason: string | null; + try { + const completion = await complete({ + apiKey: opts.apiKey, + model: opts.model, + messages: [ + { role: 'system', content: systemPrompt }, + { role: 'user', content: userPrompt }, + ], + maxTokens: DISTILL_MAX_OUTPUT_TOKENS, + reasoningEffort: opts.reasoningEffort, + retryRateLimit: true, + }); + text = completion.text.trim(); + finishReason = completion.finishReason; + } catch (err) { + if (isContextLengthError(err) && frac - BACKOFF_STEP >= BACKOFF_FLOOR) { + frac -= BACKOFF_STEP; + info( + `distill chunk rejected for context length; backing off chunk budget to ${ + Math.round(frac * 100) + }% of the working window`, + ); + continue; // nothing was consumed; recompute the chunk smaller. + } + throw err; + } + + if (!text) { + // A truncated reasoning pass can eat the whole output budget and + // return empty content (finish_reason 'length'); folding that in + // would silently wipe the notes accumulated so far. + throw new Error( + `distill pass returned no text (finish_reason=${finishReason ?? 'unknown'})`, + ); + } + + buffer = text; + pending.splice(0, consumed); + if (remainder !== null) pending.unshift(remainder); + pass += 1; + info(`distill pass ${pass} folded ${consumed} message block(s); ${pending.length} remaining`); + } + + return buffer; +} + +/** + * The stand-in block the act pass sends in place of the raw + * transcript, so the agent model knows it is reading a distillation + * rather than the conversation itself. + */ +export function renderDistilledNotesBlock(notes: string): string { + return [ + '', + 'The conversation was too large to include verbatim. The notes below were', + 'distilled from the full transcript by a prior pass; treat them as your', + 'view of the conversation.', + '', + notes, + '', + ].join('\n'); +} + +// Test-only surface: the chunking and rendering internals, exercised +// in supabase/functions/tests/accumulator.test.ts without a network. +export const __test = { + nextChunk, + renderMessage, + DISTILL_SYSTEM_PROMPT, + MIN_CHUNK_CHARS, + TOOL_RESULT_EXCERPT_CHARS, + BACKOFF_FLOOR, + BACKOFF_STEP, +}; diff --git a/supabase/functions/venice/agents/_run.ts b/supabase/functions/venice/agents/_run.ts index 633b01ce..243cf429 100644 --- a/supabase/functions/venice/agents/_run.ts +++ b/supabase/functions/venice/agents/_run.ts @@ -266,6 +266,19 @@ export interface RunHeadlessAgentOptions { * lease - wants a budget under the platform's kill threshold. */ budgetMs?: number; + /** + * Per-round max_completion_tokens, forwarded to every completion the + * loop makes. Callers should always set this: when the field is + * absent from the wire body, the serving backend reserves ITS OWN + * default output budget out of the context window - observed at + * 65536 tokens - and a long transcript then 400s with "maximum + * context length exceeded" even though the agent only ever emits + * tool calls plus a short operator summary. Headless-agent output is + * small by design; a few thousand tokens of headroom above the + * expected reply (reasoning models spend from this budget for their + * thinking pass too) reclaims the rest of the window for input. + */ + maxTokens?: number; /** Optional reasoning_effort knob, forwarded verbatim to Venice. */ reasoningEffort?: 'low' | 'medium' | 'high'; /** Optional Venice-specific disable_thinking kill switch. */ @@ -396,6 +409,7 @@ export async function runHeadlessAgent( // helper builds carries a real string field for every row. messages: messages.map((m) => ({ ...m, content: m.content ?? '' })), tools: wireList, + maxTokens: opts.maxTokens, reasoningEffort: opts.reasoningEffort, disableThinking: opts.disableThinking, // Headless agents (wiki, reflection, the librarians) run in the diff --git a/supabase/functions/venice/agents/wiki.ts b/supabase/functions/venice/agents/wiki.ts index 71e55c3b..392b369a 100644 --- a/supabase/functions/venice/agents/wiki.ts +++ b/supabase/functions/venice/agents/wiki.ts @@ -82,6 +82,12 @@ import { renderUserProfileBlock, type WikiUserProfile, } from './_wiki_profile.ts'; +import { + distillTranscript, + isContextLengthError, + renderDistilledNotesBlock, + transcriptFitsDirect, +} from './_accumulator.ts'; // Mirror of agentModel('wiki').id in src/lib/models/index.ts. // AGENT_MODELS is a static role->model map, NOT one of the per-user @@ -111,12 +117,49 @@ const CONTENT_FILTER_SENTINEL = * bodies it doesn't like even before the model gets a chance to read * them - on a wiki run that means the agent can't process the * conversation no matter how many retries we throw at it. - * arcee-trinity-large-thinking does not run that classifier, so a - * single retry against it unblocks the conversation. We retry exactly - * once: if the fallback also fails, the failure path records it and - * the per-thread counter eventually advances the pointer. + * venice-uncensored-1-2 does not run that classifier, so a single + * retry against it unblocks the conversation. We retry exactly once: + * if the fallback also fails, the failure path records it and the + * per-thread counter eventually advances the pointer. + * + * The fallback must support function calling (the agent is a tool + * loop) and is NOT a reasoning model, so the fallback attempt keeps + * reasoning_effort off the wire. The previous slot holder + * (arcee-trinity-large-thinking) vanished from Venice's catalog, + * which made every fallback attempt fail with an unknown-model error + * - when re-pointing this constant, verify the replacement id exists + * in GET /models and reports supportsFunctionCalling. + */ +const CONTENT_FILTER_FALLBACK_MODEL = 'venice-uncensored-1-2'; + +/** + * Per-round output cap for the agent's tool loop. Without an explicit + * max_completion_tokens the serving backend reserves its own default + * output budget - observed at 65536 tokens - out of the context + * window, which on 2026-07-23 (163840-token backend ceiling) left + * only 98304 tokens for input and skipped two long threads. The + * agent's real output per round is tool calls plus a two-sentence + * operator summary; 8192 leaves the reasoning pass generous headroom + * (it spends from this same budget) while reclaiming the rest of the + * window for the transcript. */ -const CONTENT_FILTER_FALLBACK_MODEL = 'arcee-trinity-large-thinking'; +const WIKI_MAX_COMPLETION_TOKENS = 8_192; + +/** + * What the distill pass must capture when a transcript is too large + * to feed the tool loop verbatim (see _accumulator.ts). Framed + * around the autonomous prompt's prime directive: aspects of the + * USER, not a summary of the discussion. + */ +const WIKI_DISTILL_FOCUS = + 'You are preparing notes for an agent that maintains an encyclopedia ' + + 'about the user. Capture what this conversation reveals about the user: ' + + 'durable facts, preferences and opinions, projects and skills, ' + + 'relationships, decisions and reversals, and dated events (keep the ' + + 'dates). Note subjects the user engaged with in enough depth to deserve ' + + 'an encyclopedia article, and note topics that came up only in passing ' + + 'so the agent can see they were shallow. Ignore generic Q&A that ' + + 'reveals nothing about the user personally.'; // Mirror of the browser wiki manager's WORKER_DEFAULTS // (src/lib/agents/wiki/manager.ts pre-cutover): 10-minute claim TTL @@ -1037,7 +1080,17 @@ function normaliseReasoning(finalText: string): string { type WikiRunOutcome = | { kind: 'done'; toolCalls: number; reasoning: string; messageCount: number } | { kind: 'empty-slice' } - | { kind: 'error'; error: string }; + | { + kind: 'error'; + error: string; + /** + * True when retrying cannot change the result (a context-length + * rejection that survived the distill path). The sweep skips such + * threads on the first failure instead of burning the transient + * 3-strike budget on an error that will repeat identically. + */ + deterministic?: boolean; + }; /** * Run the wiki agent's tool loop against one claimed thread. Shared @@ -1059,7 +1112,8 @@ async function runWikiAgentOnThread( log: EdgeLogger, complete?: AgentCompleteFn, ): Promise { - let convo: VeniceWireMessage[]; + let transcript: VeniceWireMessage[]; + let finalTurn: string; let messageCount: number; let apiKey: string; try { @@ -1072,7 +1126,7 @@ async function runWikiAgentOnThread( apiKey = key; const profile = await loadWikiProfile(adminClient, userId); - convo = slice.map(messageToVenice); + transcript = slice.map(messageToVenice); // Surface the thread's live attachment filenames so the (text-tier) // model knows what it can inspect with analyze_image and attach with // record_file_attach - the raw slice carries no attachment metadata. @@ -1080,11 +1134,10 @@ async function runWikiAgentOnThread( // Wiki instruction as the final user turn - the "switch modes" // idiom. The model sees the whole prior conversation in its // native shape and reads this as "now do this different task." + // Kept separate from the transcript so the distill path can swap + // the conversation for notes while sending the same instruction. const prompt = buildWikiAutonomousPrompt({ userProfile: profile }); - convo.push({ - role: 'user', - content: attachmentsNote ? `${attachmentsNote}\n\n${prompt}` : prompt, - }); + finalTurn = attachmentsNote ? `${attachmentsNote}\n\n${prompt}` : prompt; } catch (err) { // History fetch / prompt build failed before any Venice call. No // fallback applies; surface as a normal agent error. @@ -1102,16 +1155,19 @@ async function runWikiAgentOnThread( const attempt = async ( model: string, + // undefined = keep reasoning_effort off the wire entirely; the + // uncensored fallback is a non-reasoning model and some providers + // 400 on the unrecognised field. + reasoningEffort: 'low' | 'medium' | 'high' | undefined, ): Promise< | { kind: 'ok'; result: RunHeadlessAgentResult } | { kind: 'error'; error: unknown } > => { - log.debug(`asking ${model} about thread ${threadId} (${messageCount} messages)`); - try { - const result = await runHeadlessAgent( + const run = (messages: VeniceWireMessage[]): Promise => + runHeadlessAgent( { model, - messages: convo, + messages, toolbox: buildWikiToolbox(), baseCtx, apiKey, @@ -1120,26 +1176,70 @@ async function runWikiAgentOnThread( // runHeadlessAgent run to its own maxRounds backstop. signal: new AbortController().signal, complete, - // 'medium', not 'low': production traffic showed the agent - // surface-pattern-matching its way through conversations - - // extracting every named entity into a separate article - // instead of stopping to ask "what aspect of the user does - // this conversation actually reveal?". Medium gives the - // model budget to apply the prime-directive framing before - // dispatching tool calls. - reasoningEffort: 'medium', + maxTokens: WIKI_MAX_COMPLETION_TOKENS, + reasoningEffort, }, // parentDepth 0: the wiki agent is a top-level agent (depth 1), // same as reflection. 0, ); - return { kind: 'ok', result }; + // Distilled shape: one user turn carrying the notes block plus the + // same final instruction the direct shape ends on. Distills with + // THIS attempt's model so the content-filter fallback path also + // clears classifier rejections on the distill completions - at the + // cost of re-distilling when the fallback fires on an oversized + // thread (rare crossing of two rare conditions). + const distilled = async (): Promise => { + const notes = await distillTranscript({ + apiKey, + model, + messages: transcript, + focus: WIKI_DISTILL_FOCUS, + // 'low' for reasoning models: distillation is extraction over + // evidence already in context (see CLAUDE.md on sub-completion + // budgets). undefined stays undefined for the non-reasoning + // fallback. + reasoningEffort: reasoningEffort === undefined ? undefined : 'low', + complete, + onInfo: (m) => log.debug(`thread ${threadId}: ${m}`), + }); + return [{ role: 'user', content: `${renderDistilledNotesBlock(notes)}\n\n${finalTurn}` }]; + }; + log.debug(`asking ${model} about thread ${threadId} (${messageCount} messages)`); + try { + if (!transcriptFitsDirect(transcript)) { + log.info( + `thread ${threadId} transcript exceeds the working context window; ` + + `distilling before the tool loop`, + ); + return { kind: 'ok', result: await run(await distilled()) }; + } + try { + return { + kind: 'ok', + result: await run([...transcript, { role: 'user', content: finalTurn }]), + }; + } catch (err) { + // The estimate said the transcript fits, but the backend + // rejected it anyway - a tighter ceiling than the pinned + // working window, or mid-loop growth from accumulated tool + // results. Same remedy either way: distill and retry once. + if (!isContextLengthError(err)) throw err; + log.warn(`thread ${threadId} hit the context ceiling mid-run; retrying distilled`); + return { kind: 'ok', result: await run(await distilled()) }; + } } catch (err) { return { kind: 'error', error: err }; } }; - const primary = await attempt(WIKI_MODEL); + // 'medium', not 'low': production traffic showed the agent + // surface-pattern-matching its way through conversations - + // extracting every named entity into a separate article instead of + // stopping to ask "what aspect of the user does this conversation + // actually reveal?". Medium gives the model budget to apply the + // prime-directive framing before dispatching tool calls. + const primary = await attempt(WIKI_MODEL, 'medium'); if (primary.kind === 'ok') { return { kind: 'done', @@ -1155,6 +1255,10 @@ async function runWikiAgentOnThread( primary.error instanceof Error ? primary.error.message : String(primary.error), + // A context-length error surviving attempt()'s distill path + // means the thread cannot fit at any chunking - retrying next + // sweep would repeat the identical failure. + deterministic: isContextLengthError(primary.error), }; } @@ -1162,7 +1266,8 @@ async function runWikiAgentOnThread( `content-classifier rejection on thread ${threadId}; ` + `retrying with ${CONTENT_FILTER_FALLBACK_MODEL}`, ); - const fallback = await attempt(CONTENT_FILTER_FALLBACK_MODEL); + // reasoning_effort omitted: the fallback is a non-reasoning model. + const fallback = await attempt(CONTENT_FILTER_FALLBACK_MODEL, undefined); if (fallback.kind === 'ok') { log.info( `fallback ${CONTENT_FILTER_FALLBACK_MODEL} cleared content-filter ` + @@ -1181,6 +1286,7 @@ async function runWikiAgentOnThread( fallback.error instanceof Error ? fallback.error.message : String(fallback.error), + deterministic: isContextLengthError(fallback.error), }; } @@ -1210,12 +1316,15 @@ async function recordWikiFailureOrSkip( terminalMsgId: string, reason: string, userId: string, + // 1 for deterministic failures (skip immediately - the retry would + // fail identically); MAX_FAILURES_PER_THREAD for transient ones. + maxFailures: number = MAX_FAILURES_PER_THREAD, ): Promise<'released' | 'skipped' | 'claim-lost'> { const { data, error } = await adminClient.rpc('record_wiki_failure_or_skip', { p_thread_id: threadId, p_holder_id: holderId, p_msg_id: terminalMsgId, - p_max_failures: MAX_FAILURES_PER_THREAD, + p_max_failures: maxFailures, p_reason: reason, p_user_id: userId, }); @@ -1360,6 +1469,7 @@ export async function runWikiSweepTick( terminalMsgId, outcome.error, userId, + outcome.deterministic ? 1 : MAX_FAILURES_PER_THREAD, ); } catch (rpcErr) { // Counter bookkeeping failed. The original agent error is diff --git a/supabase/functions/venice/agents/wiki_librarian.ts b/supabase/functions/venice/agents/wiki_librarian.ts index 0404d95f..fe049142 100644 --- a/supabase/functions/venice/agents/wiki_librarian.ts +++ b/supabase/functions/venice/agents/wiki_librarian.ts @@ -1382,6 +1382,11 @@ async function runLibrarianReview(args: LibrarianReviewArgs): Promise { - let convo: VeniceWireMessage[]; + let transcript: VeniceWireMessage[]; + let finalTurn: string; let messageCount: number; let apiKey: string; try { @@ -357,15 +392,16 @@ async function runExtractionOnThread( if (!key) return { kind: 'error', error: 'no Venice key configured (app_config unseeded)' }; apiKey = key; - convo = slice.map(messageToVenice); + transcript = slice.map(messageToVenice); // Surface the thread's live attachment filenames so the (text-tier) // model knows what it can inspect with analyze_image and attach with // record_file_attach - the raw slice carries no attachment metadata. + // Kept separate from the transcript so the distill path can swap + // the conversation for notes while sending the same instruction. const attachmentsNote = await loadThreadAttachmentsNote(adminClient, threadId); - convo.push({ - role: 'user', - content: attachmentsNote ? `${attachmentsNote}\n\n${WIKI_RECORDS_PROMPT}` : WIKI_RECORDS_PROMPT, - }); + finalTurn = attachmentsNote + ? `${attachmentsNote}\n\n${WIKI_RECORDS_PROMPT}` + : WIKI_RECORDS_PROMPT; } catch (err) { return { kind: 'error', error: err instanceof Error ? err.message : String(err) }; } @@ -376,21 +412,59 @@ async function runExtractionOnThread( threadId, }; - try { - log.debug(`asking ${WIKI_RECORDS_MODEL} about thread ${threadId} (${messageCount} messages)`); - const result = await runHeadlessAgent( + const run = (messages: VeniceWireMessage[]): ReturnType => + runHeadlessAgent( { model: WIKI_RECORDS_MODEL, - messages: convo, + messages, toolbox: buildWikiRecordsToolbox(), baseCtx, apiKey, signal: new AbortController().signal, complete, + maxTokens: WIKI_RECORDS_MAX_COMPLETION_TOKENS, reasoningEffort: 'medium', }, 0, ); + // Distilled shape: one user turn carrying the notes block plus the + // same final instruction the direct shape ends on. + const distilled = async (): Promise => { + const notes = await distillTranscript({ + apiKey, + model: WIKI_RECORDS_MODEL, + messages: transcript, + focus: WIKI_RECORDS_DISTILL_FOCUS, + // 'low': distillation is extraction over evidence already in + // context (see CLAUDE.md on sub-completion budgets). + reasoningEffort: 'low', + complete, + onInfo: (m) => log.debug(`thread ${threadId}: ${m}`), + }); + return [{ role: 'user', content: `${renderDistilledNotesBlock(notes)}\n\n${finalTurn}` }]; + }; + + try { + log.debug(`asking ${WIKI_RECORDS_MODEL} about thread ${threadId} (${messageCount} messages)`); + let result; + if (!transcriptFitsDirect(transcript)) { + log.info( + `thread ${threadId} transcript exceeds the working context window; ` + + `distilling before the tool loop`, + ); + result = await run(await distilled()); + } else { + try { + result = await run([...transcript, { role: 'user', content: finalTurn }]); + } catch (err) { + // The estimate said the transcript fits, but the backend + // rejected it - a tighter ceiling than the pinned working + // window, or mid-loop growth from accumulated tool results. + if (!isContextLengthError(err)) throw err; + log.warn(`thread ${threadId} hit the context ceiling mid-run; retrying distilled`); + result = await run(await distilled()); + } + } return { kind: 'done', toolCalls: result.toolCalls, @@ -398,7 +472,11 @@ async function runExtractionOnThread( messageCount, }; } catch (err) { - return { kind: 'error', error: err instanceof Error ? err.message : String(err) }; + return { + kind: 'error', + error: err instanceof Error ? err.message : String(err), + deterministic: isContextLengthError(err), + }; } } @@ -426,12 +504,15 @@ async function recordExtractionFailureOrSkip( terminalMsgId: string, reason: string, userId: string, + // 1 for deterministic failures (skip immediately - the retry would + // fail identically); MAX_FAILURES_PER_THREAD for transient ones. + maxFailures: number = MAX_FAILURES_PER_THREAD, ): Promise<'released' | 'skipped' | 'claim-lost'> { const { data, error } = await adminClient.rpc('record_wiki_record_failure_or_skip', { p_thread_id: threadId, p_holder_id: holderId, p_msg_id: terminalMsgId, - p_max_failures: MAX_FAILURES_PER_THREAD, + p_max_failures: maxFailures, p_reason: reason, p_user_id: userId, }); @@ -549,6 +630,7 @@ export async function runWikiRecordsSweepTick( terminalMsgId, outcome.error, userId, + outcome.deterministic ? 1 : MAX_FAILURES_PER_THREAD, ); } catch (rpcErr) { log.warn(