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(