From 1f7ffe1e7f8b36da332a137b533bc9e649daa0c3 Mon Sep 17 00:00:00 2001 From: MemOS AutoDev Date: Tue, 21 Jul 2026 12:11:58 +0800 Subject: [PATCH 1/3] fix(openclaw): route classifyTopic + arbitrateTopicSplit per provider (#1611) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Summarizer.classifyTopic` and `Summarizer.arbitrateTopicSplit` always dispatched to the OpenAI implementation regardless of the summarizer's configured provider. When configured against an Anthropic-only endpoint (e.g. Kimi Code's `/coding/v1/messages`), every call returned 404 because the OpenAI helper appended `/chat/completions` to a URL that does not exist — wasting tokens, polluting logs, adding event-loop latency. Fix (Option A from the issue): - Add native classifyTopic{Anthropic,Gemini,Bedrock} and arbitrateTopicSplit{Anthropic,Gemini,Bedrock} in the three provider files, mirroring the shape of the existing judgeNewTopic* helpers. - Re-export TOPIC_CLASSIFIER_PROMPT / TOPIC_ARBITRATION_PROMPT from openai.ts so all providers share prompt strings and parseTopicClassifyResult remains the single JSON parser — no behavioral drift across providers. - Rewire callTopicClassifier / callTopicArbitration switch statements so the anthropic / gemini / bedrock arms route to their native implementations instead of falling through to the OpenAI helpers. - Bedrock helpers require cfg.endpoint (throws `Bedrock topic-classifier|arbitration requires 'endpoint'`) matching every other Bedrock helper in the file. - Mirror every edit into packages/memos-core/src/ingest/providers/ so the byte-similar sibling tree cannot silently regress the bug after the next sync. Tests: new tests/topic-classifier-dispatch.test.ts (8 cases) stubs globalThis.fetch and asserts each provider hits the correct URL + body shape; includes a regression case where an Anthropic 404 surfaces an `Anthropic topic-classifier failed` error rather than the old `OpenAI topic-classifier failed` — the exact string from issue #1611. TDD baseline: 7/8 fail against buggy code. Post-fix: 8/8 pass. Existing tests/topic-judge-minimax-1315.test.ts (5/5) continues to pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/ingest/providers/anthropic.ts | 97 +++++++ .../src/ingest/providers/bedrock.ts | 99 +++++++ .../src/ingest/providers/gemini.ts | 96 +++++++ .../src/ingest/providers/index.ts | 14 +- .../src/ingest/providers/openai.ts | 4 +- .../tests/topic-classifier-dispatch.test.ts | 270 ++++++++++++++++++ .../src/ingest/providers/anthropic.ts | 97 +++++++ .../src/ingest/providers/bedrock.ts | 99 +++++++ .../memos-core/src/ingest/providers/gemini.ts | 96 +++++++ .../memos-core/src/ingest/providers/index.ts | 14 +- .../memos-core/src/ingest/providers/openai.ts | 4 +- 11 files changed, 876 insertions(+), 14 deletions(-) create mode 100644 apps/memos-local-openclaw/tests/topic-classifier-dispatch.test.ts diff --git a/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts b/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts index e9845d334..4a3bdae3d 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts @@ -336,6 +336,103 @@ import { DEDUP_JUDGE_PROMPT, parseDedupResult } from "./openai"; import type { DedupResult } from "./openai"; export type { DedupResult } from "./openai"; +// ─── Structured Topic Classifier / Arbitration ─── +// +// Native Anthropic Messages transport for the topic classifier + arbitration. +// The prompt strings (TOPIC_CLASSIFIER_PROMPT / TOPIC_ARBITRATION_PROMPT) and +// the parser (parseTopicClassifyResult) are re-imported from openai.ts so all +// providers stay behaviorally identical modulo the wire format. + +import { + TOPIC_CLASSIFIER_PROMPT, + TOPIC_ARBITRATION_PROMPT, + parseTopicClassifyResult, +} from "./openai"; +import type { TopicClassifyResult } from "./openai"; +export type { TopicClassifyResult } from "./openai"; + +export async function classifyTopicAnthropic( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; + const model = cfg.model ?? "claude-3-haiku-20240307"; + const headers: Record = { + "Content-Type": "application/json", + "x-api-key": cfg.apiKey ?? "", + "anthropic-version": "2023-06-01", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(endpoint, { + method: "POST", + headers, + body: JSON.stringify({ + model, + max_tokens: 60, + temperature: 0, + system: TOPIC_CLASSIFIER_PROMPT, + messages: [{ role: "user", content: userContent }], + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Anthropic topic-classifier failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; + const raw = json.content.find((c) => c.type === "text")?.text?.trim() ?? ""; + log.debug(`Topic classifier raw: "${raw}"`); + return parseTopicClassifyResult(raw, log); +} + +export async function arbitrateTopicSplitAnthropic( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; + const model = cfg.model ?? "claude-3-haiku-20240307"; + const headers: Record = { + "Content-Type": "application/json", + "x-api-key": cfg.apiKey ?? "", + "anthropic-version": "2023-06-01", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(endpoint, { + method: "POST", + headers, + body: JSON.stringify({ + model, + max_tokens: 60, + temperature: 0, + system: TOPIC_ARBITRATION_PROMPT, + messages: [{ role: "user", content: userContent }], + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Anthropic topic-arbitration failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; + const answer = json.content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; + log.debug(`Topic arbitration result: "${answer}"`); + return answer.startsWith("NEW") ? "NEW" : "SAME"; +} + export async function judgeDedupAnthropic( newSummary: string, candidates: Array<{ index: number; summary: string; chunkId: string }>, diff --git a/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts b/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts index 1fcd0b359..695ae1f73 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts @@ -344,6 +344,105 @@ import { DEDUP_JUDGE_PROMPT, parseDedupResult } from "./openai"; import type { DedupResult } from "./openai"; export type { DedupResult } from "./openai"; +// ─── Structured Topic Classifier / Arbitration ─── +// +// Native Bedrock Converse transport for the topic classifier + arbitration. +// Shares prompt strings and the JSON parser with openai.ts so all providers +// stay behaviorally identical modulo the wire format. `endpoint` is required +// here, matching every other Bedrock helper in this file. + +import { + TOPIC_CLASSIFIER_PROMPT, + TOPIC_ARBITRATION_PROMPT, + parseTopicClassifyResult, +} from "./openai"; +import type { TopicClassifyResult } from "./openai"; +export type { TopicClassifyResult } from "./openai"; + +export async function classifyTopicBedrock( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; + const endpoint = cfg.endpoint; + if (!endpoint) { + throw new Error("Bedrock topic-classifier requires 'endpoint'"); + } + + const url = `${endpoint}/model/${model}/converse`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + system: [{ text: TOPIC_CLASSIFIER_PROMPT }], + messages: [{ role: "user", content: [{ text: userContent }] }], + inferenceConfig: { temperature: 0, maxTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Bedrock topic-classifier failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; + const raw = json.output?.message?.content?.[0]?.text?.trim() ?? ""; + log.debug(`Topic classifier raw: "${raw}"`); + return parseTopicClassifyResult(raw, log); +} + +export async function arbitrateTopicSplitBedrock( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; + const endpoint = cfg.endpoint; + if (!endpoint) { + throw new Error("Bedrock topic-arbitration requires 'endpoint'"); + } + + const url = `${endpoint}/model/${model}/converse`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + system: [{ text: TOPIC_ARBITRATION_PROMPT }], + messages: [{ role: "user", content: [{ text: userContent }] }], + inferenceConfig: { temperature: 0, maxTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Bedrock topic-arbitration failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; + const answer = json.output?.message?.content?.[0]?.text?.trim().toUpperCase() ?? ""; + log.debug(`Topic arbitration result: "${answer}"`); + return answer.startsWith("NEW") ? "NEW" : "SAME"; +} + export async function judgeDedupBedrock( newSummary: string, candidates: Array<{ index: number; summary: string; chunkId: string }>, diff --git a/apps/memos-local-openclaw/src/ingest/providers/gemini.ts b/apps/memos-local-openclaw/src/ingest/providers/gemini.ts index 3fafe570d..693245426 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/gemini.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/gemini.ts @@ -336,6 +336,102 @@ import { DEDUP_JUDGE_PROMPT, parseDedupResult } from "./openai"; import type { DedupResult } from "./openai"; export type { DedupResult } from "./openai"; +// ─── Structured Topic Classifier / Arbitration ─── +// +// Native Gemini generateContent transport for the topic classifier + +// arbitration. Shares prompt strings and the JSON parser with openai.ts so +// all providers stay behaviorally identical modulo the wire format. + +import { + TOPIC_CLASSIFIER_PROMPT, + TOPIC_ARBITRATION_PROMPT, + parseTopicClassifyResult, +} from "./openai"; +import type { TopicClassifyResult } from "./openai"; +export type { TopicClassifyResult } from "./openai"; + +export async function classifyTopicGemini( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "gemini-1.5-flash"; + const endpoint = + cfg.endpoint ?? + `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; + + const url = `${endpoint}?key=${cfg.apiKey}`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + systemInstruction: { parts: [{ text: TOPIC_CLASSIFIER_PROMPT }] }, + contents: [{ parts: [{ text: userContent }] }], + generationConfig: { temperature: 0, maxOutputTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Gemini topic-classifier failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; + const raw = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim() ?? ""; + log.debug(`Topic classifier raw: "${raw}"`); + return parseTopicClassifyResult(raw, log); +} + +export async function arbitrateTopicSplitGemini( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "gemini-1.5-flash"; + const endpoint = + cfg.endpoint ?? + `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; + + const url = `${endpoint}?key=${cfg.apiKey}`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + systemInstruction: { parts: [{ text: TOPIC_ARBITRATION_PROMPT }] }, + contents: [{ parts: [{ text: userContent }] }], + generationConfig: { temperature: 0, maxOutputTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Gemini topic-arbitration failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; + const answer = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim().toUpperCase() ?? ""; + log.debug(`Topic arbitration result: "${answer}"`); + return answer.startsWith("NEW") ? "NEW" : "SAME"; +} + export async function judgeDedupGemini( newSummary: string, candidates: Array<{ index: number; summary: string; chunkId: string }>, diff --git a/apps/memos-local-openclaw/src/ingest/providers/index.ts b/apps/memos-local-openclaw/src/ingest/providers/index.ts index c5dd1a85c..972930f9e 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/index.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/index.ts @@ -5,9 +5,9 @@ import { parseJsonOrJson5 } from "../../shared/json5"; import { summarizeOpenAI, summarizeTaskOpenAI, generateTaskTitleOpenAI, judgeNewTopicOpenAI, classifyTopicOpenAI, arbitrateTopicSplitOpenAI, filterRelevantOpenAI, judgeDedupOpenAI, parseFilterResult, parseDedupResult, parseTopicClassifyResult } from "./openai"; import type { FilterResult, DedupResult, TopicClassifyResult } from "./openai"; export type { FilterResult, DedupResult, TopicClassifyResult } from "./openai"; -import { summarizeAnthropic, summarizeTaskAnthropic, generateTaskTitleAnthropic, judgeNewTopicAnthropic, filterRelevantAnthropic, judgeDedupAnthropic } from "./anthropic"; -import { summarizeGemini, summarizeTaskGemini, generateTaskTitleGemini, judgeNewTopicGemini, filterRelevantGemini, judgeDedupGemini } from "./gemini"; -import { summarizeBedrock, summarizeTaskBedrock, generateTaskTitleBedrock, judgeNewTopicBedrock, filterRelevantBedrock, judgeDedupBedrock } from "./bedrock"; +import { summarizeAnthropic, summarizeTaskAnthropic, generateTaskTitleAnthropic, judgeNewTopicAnthropic, filterRelevantAnthropic, judgeDedupAnthropic, classifyTopicAnthropic, arbitrateTopicSplitAnthropic } from "./anthropic"; +import { summarizeGemini, summarizeTaskGemini, generateTaskTitleGemini, judgeNewTopicGemini, filterRelevantGemini, judgeDedupGemini, classifyTopicGemini, arbitrateTopicSplitGemini } from "./gemini"; +import { summarizeBedrock, summarizeTaskBedrock, generateTaskTitleBedrock, judgeNewTopicBedrock, filterRelevantBedrock, judgeDedupBedrock, classifyTopicBedrock, arbitrateTopicSplitBedrock } from "./bedrock"; /** * Resolve a SecretInput (string | SecretRef) to a plain string. @@ -714,9 +714,11 @@ function callTopicClassifier(cfg: SummarizerConfig, taskState: string, newMessag case "voyage": return classifyTopicOpenAI(taskState, newMessage, cfg, log); case "anthropic": + return classifyTopicAnthropic(taskState, newMessage, cfg, log); case "gemini": + return classifyTopicGemini(taskState, newMessage, cfg, log); case "bedrock": - return classifyTopicOpenAI(taskState, newMessage, cfg, log); + return classifyTopicBedrock(taskState, newMessage, cfg, log); default: throw new Error(`Unknown summarizer provider: ${cfg.provider}`); } @@ -737,9 +739,11 @@ function callTopicArbitration(cfg: SummarizerConfig, taskState: string, newMessa case "voyage": return arbitrateTopicSplitOpenAI(taskState, newMessage, cfg, log); case "anthropic": + return arbitrateTopicSplitAnthropic(taskState, newMessage, cfg, log); case "gemini": + return arbitrateTopicSplitGemini(taskState, newMessage, cfg, log); case "bedrock": - return arbitrateTopicSplitOpenAI(taskState, newMessage, cfg, log); + return arbitrateTopicSplitBedrock(taskState, newMessage, cfg, log); default: throw new Error(`Unknown summarizer provider: ${cfg.provider}`); } diff --git a/apps/memos-local-openclaw/src/ingest/providers/openai.ts b/apps/memos-local-openclaw/src/ingest/providers/openai.ts index 60f878e4c..3455a02b5 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/openai.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/openai.ts @@ -269,7 +269,7 @@ export interface TopicClassifyResult { reason: string; // may be empty for compact responses } -const TOPIC_CLASSIFIER_PROMPT = `Classify if NEW MESSAGE continues current task or starts an unrelated one. +export const TOPIC_CLASSIFIER_PROMPT = `Classify if NEW MESSAGE continues current task or starts an unrelated one. Output ONLY JSON: {"d":"S"|"N","c":0.0-1.0} d=S(same) or N(new). c=confidence. Default S. Only N if completely unrelated domain. Sub-questions, tools, methods, details of current topic = S.`; @@ -317,7 +317,7 @@ export async function classifyTopicOpenAI( return parseTopicClassifyResult(raw, log); } -const TOPIC_ARBITRATION_PROMPT = `A classifier flagged this message as possibly new topic (low confidence). Is it truly UNRELATED, or a sub-question/follow-up? +export const TOPIC_ARBITRATION_PROMPT = `A classifier flagged this message as possibly new topic (low confidence). Is it truly UNRELATED, or a sub-question/follow-up? Tools/methods/details of current task = SAME. Shared entity/theme = SAME. Entirely different domain = NEW. Reply one word: NEW or SAME`; diff --git a/apps/memos-local-openclaw/tests/topic-classifier-dispatch.test.ts b/apps/memos-local-openclaw/tests/topic-classifier-dispatch.test.ts new file mode 100644 index 000000000..51dee4819 --- /dev/null +++ b/apps/memos-local-openclaw/tests/topic-classifier-dispatch.test.ts @@ -0,0 +1,270 @@ +/** + * Regression test for issue #1611: + * + * `Summarizer.classifyTopic` and `Summarizer.arbitrateTopicSplit` used to + * dispatch to the OpenAI implementation for EVERY provider — including + * `anthropic`, `gemini`, and `bedrock` — because their case arms in the + * `callTopicClassifier` / `callTopicArbitration` switch statements silently + * fell through to `classifyTopicOpenAI` / `arbitrateTopicSplitOpenAI`. + * + * That broke summarizers configured against Anthropic-only endpoints (e.g. + * Kimi Code's `/coding/v1/messages`), where the OpenAI helper's + * `/chat/completions` URL simply doesn't exist and every call 404'd. + * + * These tests stub `globalThis.fetch` and assert that each provider hits the + * transport-appropriate URL with a body shape the target API actually + * accepts. A regression that routed `anthropic` back through the OpenAI + * transport would either hit `/chat/completions` (wrong URL) or emit an + * `OpenAI topic-classifier failed` error string (wrong label) — both are + * asserted below. + */ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { Summarizer } from "../src/ingest/providers"; +import type { SummarizerConfig, Logger } from "../src/types"; + +const silentLog: Logger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: () => {}, +}; + +interface CapturedRequest { + url: string; + headers: Record; + body: Record; +} + +/** + * Replace global.fetch with a recorder. Each invocation returns a canned + * successful response whose shape matches `provider`. + */ +function installFetchRecorder( + provider: "openai" | "anthropic" | "gemini" | "bedrock", + replyText: string, +): CapturedRequest[] { + const captured: CapturedRequest[] = []; + + const buildResponse = () => { + switch (provider) { + case "openai": + return { choices: [{ message: { content: replyText } }] }; + case "anthropic": + return { content: [{ type: "text", text: replyText }] }; + case "gemini": + return { candidates: [{ content: { parts: [{ text: replyText }] } }] }; + case "bedrock": + return { output: { message: { content: [{ text: replyText }] } } }; + } + }; + + const fakeFetch = vi.fn(async (url: string | URL, init?: RequestInit) => { + const body = init?.body ? JSON.parse(init.body as string) : {}; + const headers: Record = {}; + if (init?.headers) { + const h = init.headers as Record; + for (const [k, v] of Object.entries(h)) headers[k] = String(v); + } + captured.push({ url: String(url), headers, body }); + return new Response(JSON.stringify(buildResponse()), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }); + + vi.stubGlobal("fetch", fakeFetch); + return captured; +} + +function installErrorFetch(status: number, errorBody: string): void { + const fakeFetch = vi.fn(async () => + new Response(errorBody, { status, headers: { "Content-Type": "application/json" } }), + ); + vi.stubGlobal("fetch", fakeFetch); +} + +describe("classifyTopic / arbitrateTopicSplit dispatch by provider (issue #1611)", () => { + beforeEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it("openai_compatible → POSTs /chat/completions with system+user messages", async () => { + const captured = installFetchRecorder("openai", '{"d":"S","c":0.9}'); + const cfg: SummarizerConfig = { + provider: "openai_compatible", + endpoint: "https://api.example.com/v1", + apiKey: "sk-test", + model: "test-model", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.classifyTopic("task state", "new msg"); + expect(result?.decision).toBe("SAME"); + + expect(captured).toHaveLength(1); + expect(captured[0].url).toBe("https://api.example.com/v1/chat/completions"); + expect(captured[0].headers.Authorization).toBe("Bearer sk-test"); + const msgs = captured[0].body.messages as Array<{ role: string; content: string }>; + expect(msgs[0].role).toBe("system"); + expect(msgs[0].content).toContain("Classify if NEW MESSAGE"); + expect(msgs[1].role).toBe("user"); + expect(msgs[1].content).toContain("TASK:\ntask state"); + }); + + it("anthropic → POSTs /v1/messages with x-api-key and system prompt (regression: was hitting /chat/completions)", async () => { + const captured = installFetchRecorder("anthropic", '{"d":"N","c":0.85}'); + const cfg: SummarizerConfig = { + provider: "anthropic", + endpoint: "https://api.kimi.com/coding/v1/messages", + apiKey: "kimi-test", + model: "kimi-for-coding", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.classifyTopic("task state", "new msg"); + expect(result?.decision).toBe("NEW"); + expect(result?.confidence).toBeCloseTo(0.85); + + expect(captured).toHaveLength(1); + // Must NOT contain /chat/completions — that's the pre-fix bug. + expect(captured[0].url).not.toContain("/chat/completions"); + expect(captured[0].url).toBe("https://api.kimi.com/coding/v1/messages"); + expect(captured[0].headers["x-api-key"]).toBe("kimi-test"); + expect(captured[0].headers["anthropic-version"]).toBe("2023-06-01"); + expect(captured[0].body.system).toContain("Classify if NEW MESSAGE"); + expect(captured[0].body.model).toBe("kimi-for-coding"); + const anthropicMessages = captured[0].body.messages as Array<{ role: string; content: string }>; + expect(anthropicMessages[0].role).toBe("user"); + expect(anthropicMessages[0].content).toContain("TASK:\ntask state"); + }); + + it("gemini → POSTs :generateContent with systemInstruction and apiKey query param", async () => { + const captured = installFetchRecorder("gemini", '{"d":"S","c":0.7}'); + const cfg: SummarizerConfig = { + provider: "gemini", + apiKey: "gemini-test", + model: "gemini-1.5-flash", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.classifyTopic("task state", "new msg"); + expect(result?.decision).toBe("SAME"); + + expect(captured).toHaveLength(1); + expect(captured[0].url).toContain(":generateContent?key=gemini-test"); + expect(captured[0].url).toContain("gemini-1.5-flash"); + // Body must use Gemini's structured shape, not OpenAI's. + expect(captured[0].body.messages).toBeUndefined(); + const sysInstr = captured[0].body.systemInstruction as { parts: Array<{ text: string }> }; + expect(sysInstr.parts[0].text).toContain("Classify if NEW MESSAGE"); + const contents = captured[0].body.contents as Array<{ parts: Array<{ text: string }> }>; + expect(contents[0].parts[0].text).toContain("TASK:\ntask state"); + }); + + it("bedrock → POSTs /model//converse with system[] and messages[]", async () => { + const captured = installFetchRecorder("bedrock", '{"d":"N","c":0.6}'); + const cfg: SummarizerConfig = { + provider: "bedrock", + endpoint: "https://bedrock-runtime.us-east-1.amazonaws.com", + apiKey: "aws-test", + model: "anthropic.claude-3-haiku-20240307-v1:0", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.classifyTopic("task state", "new msg"); + expect(result?.decision).toBe("NEW"); + + expect(captured).toHaveLength(1); + expect(captured[0].url).toBe( + "https://bedrock-runtime.us-east-1.amazonaws.com/model/anthropic.claude-3-haiku-20240307-v1:0/converse", + ); + // Body must use Bedrock Converse shape. + expect(captured[0].body.choices).toBeUndefined(); + const system = captured[0].body.system as Array<{ text: string }>; + expect(system[0].text).toContain("Classify if NEW MESSAGE"); + const bedrockMessages = captured[0].body.messages as Array<{ role: string; content: Array<{ text: string }> }>; + expect(bedrockMessages[0].role).toBe("user"); + expect(bedrockMessages[0].content[0].text).toContain("TASK:\ntask state"); + }); + + it("arbitrateTopicSplit for anthropic → POSTs /v1/messages, returns NEW/SAME", async () => { + const captured = installFetchRecorder("anthropic", "NEW"); + const cfg: SummarizerConfig = { + provider: "anthropic", + endpoint: "https://api.kimi.com/coding/v1/messages", + apiKey: "kimi-test", + model: "kimi-for-coding", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.arbitrateTopicSplit("task", "msg"); + expect(result).toBe("NEW"); + expect(captured[0].url).toBe("https://api.kimi.com/coding/v1/messages"); + expect(captured[0].body.system).toContain("A classifier flagged this message"); + }); + + it("arbitrateTopicSplit for gemini → POSTs :generateContent, returns SAME on non-NEW reply", async () => { + const captured = installFetchRecorder("gemini", "same"); + const cfg: SummarizerConfig = { + provider: "gemini", + apiKey: "gemini-test", + model: "gemini-1.5-flash", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.arbitrateTopicSplit("task", "msg"); + expect(result).toBe("SAME"); + expect(captured[0].url).toContain(":generateContent?key=gemini-test"); + }); + + it("arbitrateTopicSplit for bedrock → POSTs /converse, uses system[] shape", async () => { + const captured = installFetchRecorder("bedrock", "NEW\n"); + const cfg: SummarizerConfig = { + provider: "bedrock", + endpoint: "https://bedrock-runtime.us-east-1.amazonaws.com", + apiKey: "aws-test", + model: "anthropic.claude-3-haiku-20240307-v1:0", + }; + const sum = new Summarizer(cfg, silentLog); + + const result = await sum.arbitrateTopicSplit("task", "msg"); + expect(result).toBe("NEW"); + expect(captured[0].url).toContain("/model/anthropic.claude-3-haiku-20240307-v1:0/converse"); + const bsystem = captured[0].body.system as Array<{ text: string }>; + expect(bsystem[0].text).toContain("A classifier flagged this message"); + }); + + it("anthropic 404 surfaces as 'Anthropic topic-classifier failed', not 'OpenAI' — the actual issue #1611 signature", async () => { + installErrorFetch( + 404, + '{"error":{"message":"The requested resource was not found","type":"resource_not_found_error"}}', + ); + const errorMessages: string[] = []; + const captureLog: Logger = { + debug: () => {}, + info: () => {}, + warn: (m) => errorMessages.push(m), + error: (m) => errorMessages.push(m), + }; + const cfg: SummarizerConfig = { + provider: "anthropic", + endpoint: "https://api.kimi.com/coding/v1/messages", + apiKey: "kimi-test", + model: "kimi-for-coding", + }; + const sum = new Summarizer(cfg, captureLog); + + const result = await sum.classifyTopic("state", "msg"); + // No fallback configs, so classifyTopic returns null after logging. + expect(result).toBeNull(); + // The recorded error message must reference the Anthropic transport, not OpenAI. + const combined = errorMessages.join("\n"); + expect(combined).toContain("Anthropic topic-classifier failed"); + expect(combined).not.toContain("OpenAI topic-classifier failed"); + }); +}); diff --git a/packages/memos-core/src/ingest/providers/anthropic.ts b/packages/memos-core/src/ingest/providers/anthropic.ts index e9845d334..4a3bdae3d 100644 --- a/packages/memos-core/src/ingest/providers/anthropic.ts +++ b/packages/memos-core/src/ingest/providers/anthropic.ts @@ -336,6 +336,103 @@ import { DEDUP_JUDGE_PROMPT, parseDedupResult } from "./openai"; import type { DedupResult } from "./openai"; export type { DedupResult } from "./openai"; +// ─── Structured Topic Classifier / Arbitration ─── +// +// Native Anthropic Messages transport for the topic classifier + arbitration. +// The prompt strings (TOPIC_CLASSIFIER_PROMPT / TOPIC_ARBITRATION_PROMPT) and +// the parser (parseTopicClassifyResult) are re-imported from openai.ts so all +// providers stay behaviorally identical modulo the wire format. + +import { + TOPIC_CLASSIFIER_PROMPT, + TOPIC_ARBITRATION_PROMPT, + parseTopicClassifyResult, +} from "./openai"; +import type { TopicClassifyResult } from "./openai"; +export type { TopicClassifyResult } from "./openai"; + +export async function classifyTopicAnthropic( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; + const model = cfg.model ?? "claude-3-haiku-20240307"; + const headers: Record = { + "Content-Type": "application/json", + "x-api-key": cfg.apiKey ?? "", + "anthropic-version": "2023-06-01", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(endpoint, { + method: "POST", + headers, + body: JSON.stringify({ + model, + max_tokens: 60, + temperature: 0, + system: TOPIC_CLASSIFIER_PROMPT, + messages: [{ role: "user", content: userContent }], + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Anthropic topic-classifier failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; + const raw = json.content.find((c) => c.type === "text")?.text?.trim() ?? ""; + log.debug(`Topic classifier raw: "${raw}"`); + return parseTopicClassifyResult(raw, log); +} + +export async function arbitrateTopicSplitAnthropic( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; + const model = cfg.model ?? "claude-3-haiku-20240307"; + const headers: Record = { + "Content-Type": "application/json", + "x-api-key": cfg.apiKey ?? "", + "anthropic-version": "2023-06-01", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(endpoint, { + method: "POST", + headers, + body: JSON.stringify({ + model, + max_tokens: 60, + temperature: 0, + system: TOPIC_ARBITRATION_PROMPT, + messages: [{ role: "user", content: userContent }], + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Anthropic topic-arbitration failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; + const answer = json.content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; + log.debug(`Topic arbitration result: "${answer}"`); + return answer.startsWith("NEW") ? "NEW" : "SAME"; +} + export async function judgeDedupAnthropic( newSummary: string, candidates: Array<{ index: number; summary: string; chunkId: string }>, diff --git a/packages/memos-core/src/ingest/providers/bedrock.ts b/packages/memos-core/src/ingest/providers/bedrock.ts index 1fcd0b359..695ae1f73 100644 --- a/packages/memos-core/src/ingest/providers/bedrock.ts +++ b/packages/memos-core/src/ingest/providers/bedrock.ts @@ -344,6 +344,105 @@ import { DEDUP_JUDGE_PROMPT, parseDedupResult } from "./openai"; import type { DedupResult } from "./openai"; export type { DedupResult } from "./openai"; +// ─── Structured Topic Classifier / Arbitration ─── +// +// Native Bedrock Converse transport for the topic classifier + arbitration. +// Shares prompt strings and the JSON parser with openai.ts so all providers +// stay behaviorally identical modulo the wire format. `endpoint` is required +// here, matching every other Bedrock helper in this file. + +import { + TOPIC_CLASSIFIER_PROMPT, + TOPIC_ARBITRATION_PROMPT, + parseTopicClassifyResult, +} from "./openai"; +import type { TopicClassifyResult } from "./openai"; +export type { TopicClassifyResult } from "./openai"; + +export async function classifyTopicBedrock( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; + const endpoint = cfg.endpoint; + if (!endpoint) { + throw new Error("Bedrock topic-classifier requires 'endpoint'"); + } + + const url = `${endpoint}/model/${model}/converse`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + system: [{ text: TOPIC_CLASSIFIER_PROMPT }], + messages: [{ role: "user", content: [{ text: userContent }] }], + inferenceConfig: { temperature: 0, maxTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Bedrock topic-classifier failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; + const raw = json.output?.message?.content?.[0]?.text?.trim() ?? ""; + log.debug(`Topic classifier raw: "${raw}"`); + return parseTopicClassifyResult(raw, log); +} + +export async function arbitrateTopicSplitBedrock( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; + const endpoint = cfg.endpoint; + if (!endpoint) { + throw new Error("Bedrock topic-arbitration requires 'endpoint'"); + } + + const url = `${endpoint}/model/${model}/converse`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + system: [{ text: TOPIC_ARBITRATION_PROMPT }], + messages: [{ role: "user", content: [{ text: userContent }] }], + inferenceConfig: { temperature: 0, maxTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Bedrock topic-arbitration failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; + const answer = json.output?.message?.content?.[0]?.text?.trim().toUpperCase() ?? ""; + log.debug(`Topic arbitration result: "${answer}"`); + return answer.startsWith("NEW") ? "NEW" : "SAME"; +} + export async function judgeDedupBedrock( newSummary: string, candidates: Array<{ index: number; summary: string; chunkId: string }>, diff --git a/packages/memos-core/src/ingest/providers/gemini.ts b/packages/memos-core/src/ingest/providers/gemini.ts index 3fafe570d..693245426 100644 --- a/packages/memos-core/src/ingest/providers/gemini.ts +++ b/packages/memos-core/src/ingest/providers/gemini.ts @@ -336,6 +336,102 @@ import { DEDUP_JUDGE_PROMPT, parseDedupResult } from "./openai"; import type { DedupResult } from "./openai"; export type { DedupResult } from "./openai"; +// ─── Structured Topic Classifier / Arbitration ─── +// +// Native Gemini generateContent transport for the topic classifier + +// arbitration. Shares prompt strings and the JSON parser with openai.ts so +// all providers stay behaviorally identical modulo the wire format. + +import { + TOPIC_CLASSIFIER_PROMPT, + TOPIC_ARBITRATION_PROMPT, + parseTopicClassifyResult, +} from "./openai"; +import type { TopicClassifyResult } from "./openai"; +export type { TopicClassifyResult } from "./openai"; + +export async function classifyTopicGemini( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "gemini-1.5-flash"; + const endpoint = + cfg.endpoint ?? + `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; + + const url = `${endpoint}?key=${cfg.apiKey}`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + systemInstruction: { parts: [{ text: TOPIC_CLASSIFIER_PROMPT }] }, + contents: [{ parts: [{ text: userContent }] }], + generationConfig: { temperature: 0, maxOutputTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Gemini topic-classifier failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; + const raw = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim() ?? ""; + log.debug(`Topic classifier raw: "${raw}"`); + return parseTopicClassifyResult(raw, log); +} + +export async function arbitrateTopicSplitGemini( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const model = cfg.model ?? "gemini-1.5-flash"; + const endpoint = + cfg.endpoint ?? + `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; + + const url = `${endpoint}?key=${cfg.apiKey}`; + const headers: Record = { + "Content-Type": "application/json", + ...cfg.headers, + }; + + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + + const resp = await fetch(url, { + method: "POST", + headers, + body: JSON.stringify({ + systemInstruction: { parts: [{ text: TOPIC_ARBITRATION_PROMPT }] }, + contents: [{ parts: [{ text: userContent }] }], + generationConfig: { temperature: 0, maxOutputTokens: 60 }, + }), + signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), + }); + + if (!resp.ok) { + const body = await resp.text(); + throw new Error(`Gemini topic-arbitration failed (${resp.status}): ${body}`); + } + + const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; + const answer = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim().toUpperCase() ?? ""; + log.debug(`Topic arbitration result: "${answer}"`); + return answer.startsWith("NEW") ? "NEW" : "SAME"; +} + export async function judgeDedupGemini( newSummary: string, candidates: Array<{ index: number; summary: string; chunkId: string }>, diff --git a/packages/memos-core/src/ingest/providers/index.ts b/packages/memos-core/src/ingest/providers/index.ts index b08818520..005edc23a 100644 --- a/packages/memos-core/src/ingest/providers/index.ts +++ b/packages/memos-core/src/ingest/providers/index.ts @@ -4,9 +4,9 @@ import type { SummarizerConfig, SummaryProvider, Logger, OpenClawAPI } from "../ import { summarizeOpenAI, summarizeTaskOpenAI, generateTaskTitleOpenAI, judgeNewTopicOpenAI, classifyTopicOpenAI, arbitrateTopicSplitOpenAI, filterRelevantOpenAI, judgeDedupOpenAI, parseFilterResult, parseDedupResult, parseTopicClassifyResult } from "./openai"; import type { FilterResult, DedupResult, TopicClassifyResult } from "./openai"; export type { FilterResult, DedupResult, TopicClassifyResult } from "./openai"; -import { summarizeAnthropic, summarizeTaskAnthropic, generateTaskTitleAnthropic, judgeNewTopicAnthropic, filterRelevantAnthropic, judgeDedupAnthropic } from "./anthropic"; -import { summarizeGemini, summarizeTaskGemini, generateTaskTitleGemini, judgeNewTopicGemini, filterRelevantGemini, judgeDedupGemini } from "./gemini"; -import { summarizeBedrock, summarizeTaskBedrock, generateTaskTitleBedrock, judgeNewTopicBedrock, filterRelevantBedrock, judgeDedupBedrock } from "./bedrock"; +import { summarizeAnthropic, summarizeTaskAnthropic, generateTaskTitleAnthropic, judgeNewTopicAnthropic, filterRelevantAnthropic, judgeDedupAnthropic, classifyTopicAnthropic, arbitrateTopicSplitAnthropic } from "./anthropic"; +import { summarizeGemini, summarizeTaskGemini, generateTaskTitleGemini, judgeNewTopicGemini, filterRelevantGemini, judgeDedupGemini, classifyTopicGemini, arbitrateTopicSplitGemini } from "./gemini"; +import { summarizeBedrock, summarizeTaskBedrock, generateTaskTitleBedrock, judgeNewTopicBedrock, filterRelevantBedrock, judgeDedupBedrock, classifyTopicBedrock, arbitrateTopicSplitBedrock } from "./bedrock"; /** * Resolve a SecretInput (string | SecretRef) to a plain string. @@ -713,9 +713,11 @@ function callTopicClassifier(cfg: SummarizerConfig, taskState: string, newMessag case "voyage": return classifyTopicOpenAI(taskState, newMessage, cfg, log); case "anthropic": + return classifyTopicAnthropic(taskState, newMessage, cfg, log); case "gemini": + return classifyTopicGemini(taskState, newMessage, cfg, log); case "bedrock": - return classifyTopicOpenAI(taskState, newMessage, cfg, log); + return classifyTopicBedrock(taskState, newMessage, cfg, log); default: throw new Error(`Unknown summarizer provider: ${cfg.provider}`); } @@ -736,9 +738,11 @@ function callTopicArbitration(cfg: SummarizerConfig, taskState: string, newMessa case "voyage": return arbitrateTopicSplitOpenAI(taskState, newMessage, cfg, log); case "anthropic": + return arbitrateTopicSplitAnthropic(taskState, newMessage, cfg, log); case "gemini": + return arbitrateTopicSplitGemini(taskState, newMessage, cfg, log); case "bedrock": - return arbitrateTopicSplitOpenAI(taskState, newMessage, cfg, log); + return arbitrateTopicSplitBedrock(taskState, newMessage, cfg, log); default: throw new Error(`Unknown summarizer provider: ${cfg.provider}`); } diff --git a/packages/memos-core/src/ingest/providers/openai.ts b/packages/memos-core/src/ingest/providers/openai.ts index 825e2131d..45eeb808f 100644 --- a/packages/memos-core/src/ingest/providers/openai.ts +++ b/packages/memos-core/src/ingest/providers/openai.ts @@ -262,7 +262,7 @@ export interface TopicClassifyResult { reason: string; // may be empty for compact responses } -const TOPIC_CLASSIFIER_PROMPT = `Classify if NEW MESSAGE continues current task or starts an unrelated one. +export const TOPIC_CLASSIFIER_PROMPT = `Classify if NEW MESSAGE continues current task or starts an unrelated one. Output ONLY JSON: {"d":"S"|"N","c":0.0-1.0} d=S(same) or N(new). c=confidence. Default S. Only N if completely unrelated domain. Sub-questions, tools, methods, details of current topic = S.`; @@ -310,7 +310,7 @@ export async function classifyTopicOpenAI( return parseTopicClassifyResult(raw, log); } -const TOPIC_ARBITRATION_PROMPT = `A classifier flagged this message as possibly new topic (low confidence). Is it truly UNRELATED, or a sub-question/follow-up? +export const TOPIC_ARBITRATION_PROMPT = `A classifier flagged this message as possibly new topic (low confidence). Is it truly UNRELATED, or a sub-question/follow-up? Tools/methods/details of current task = SAME. Shared entity/theme = SAME. Entirely different domain = NEW. Reply one word: NEW or SAME`; From 646c28a13e295e88b936c36774d95a4d953bb154 Mon Sep 17 00:00:00 2001 From: MemOS AutoDev Date: Tue, 21 Jul 2026 12:38:35 +0800 Subject: [PATCH 2/3] fix(openclaw): harden topic-classifier/arbitrator providers per OCR review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address code-review findings on the topic-classifier + arbitrator dispatch helpers added in #2133 (issue #1611). Applied to both apps/memos-local-openclaw and packages/memos-core (byte-similar mirrors). Correctness / security fixes: - anthropic: validate cfg.apiKey up front instead of silently sending ""; move cfg.headers spread BEFORE x-api-key + anthropic-version so a misconfigured header cannot silently overwrite the credential - anthropic: guard content parsing — treat missing/non-array json.content as [] to avoid TypeError on unexpected API payloads - anthropic: reduce arbitrateTopicSplit max_tokens from 60 to 10 to match arbitrateTopicSplitOpenAI (single-word NEW/SAME reply) - gemini: validate cfg.apiKey up front; construct URL via new URL() + URLSearchParams so a caller-supplied endpoint that already contains a query string ("?version=v1beta") does not produce malformed "…?version=v1beta?key=…" - bedrock: emit warn log when arbitrateTopicSplit sees empty or unexpected text so silent bias toward SAME becomes visible at normal log levels (production usually suppresses debug) Intentionally not applied (out-of-scope refactors that would touch pre-existing helpers not in this PR): DRY-extract shared low-level callers for anthropic/gemini/bedrock (findings #4/#7/#10/#14) and the DEFAULT_BEDROCK_MODEL constant (finding #11). Existing tests unchanged: tests/topic-classifier-dispatch.test.ts (8/8) and tests/topic-judge-minimax-1315.test.ts (5/5) still pass. The new guards use apiKey values already provided by every test case, so no test edits were required. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/ingest/providers/anthropic.ts | 22 +++++++++++-------- .../src/ingest/providers/bedrock.ts | 5 +++++ .../src/ingest/providers/gemini.ts | 10 +++++++-- .../src/ingest/providers/anthropic.ts | 22 +++++++++++-------- .../src/ingest/providers/bedrock.ts | 5 +++++ .../memos-core/src/ingest/providers/gemini.ts | 10 +++++++-- 6 files changed, 52 insertions(+), 22 deletions(-) diff --git a/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts b/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts index 4a3bdae3d..1b5e1d929 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts @@ -357,13 +357,14 @@ export async function classifyTopicAnthropic( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Anthropic topic-classifier: apiKey is required"); const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; const model = cfg.model ?? "claude-3-haiku-20240307"; const headers: Record = { "Content-Type": "application/json", - "x-api-key": cfg.apiKey ?? "", - "anthropic-version": "2023-06-01", ...cfg.headers, + "x-api-key": cfg.apiKey, + "anthropic-version": "2023-06-01", }; const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; @@ -386,8 +387,9 @@ export async function classifyTopicAnthropic( throw new Error(`Anthropic topic-classifier failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; - const raw = json.content.find((c) => c.type === "text")?.text?.trim() ?? ""; + const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; + const content = Array.isArray(json?.content) ? json.content : []; + const raw = content.find((c) => c.type === "text")?.text?.trim() ?? ""; log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -398,13 +400,14 @@ export async function arbitrateTopicSplitAnthropic( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Anthropic topic-arbitration: apiKey is required"); const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; const model = cfg.model ?? "claude-3-haiku-20240307"; const headers: Record = { "Content-Type": "application/json", - "x-api-key": cfg.apiKey ?? "", - "anthropic-version": "2023-06-01", ...cfg.headers, + "x-api-key": cfg.apiKey, + "anthropic-version": "2023-06-01", }; const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; @@ -414,7 +417,7 @@ export async function arbitrateTopicSplitAnthropic( headers, body: JSON.stringify({ model, - max_tokens: 60, + max_tokens: 10, temperature: 0, system: TOPIC_ARBITRATION_PROMPT, messages: [{ role: "user", content: userContent }], @@ -427,8 +430,9 @@ export async function arbitrateTopicSplitAnthropic( throw new Error(`Anthropic topic-arbitration failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; - const answer = json.content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; + const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; + const content = Array.isArray(json?.content) ? json.content : []; + const answer = content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; log.debug(`Topic arbitration result: "${answer}"`); return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts b/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts index 695ae1f73..082679cd6 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts @@ -440,6 +440,11 @@ export async function arbitrateTopicSplitBedrock( const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; const answer = json.output?.message?.content?.[0]?.text?.trim().toUpperCase() ?? ""; log.debug(`Topic arbitration result: "${answer}"`); + if (!answer) { + log.warn("Bedrock topic-arbitration returned empty text; defaulting to SAME"); + } else if (!answer.startsWith("NEW") && !answer.startsWith("SAME")) { + log.warn(`Bedrock topic-arbitration returned unexpected value "${answer}"; defaulting to SAME`); + } return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/apps/memos-local-openclaw/src/ingest/providers/gemini.ts b/apps/memos-local-openclaw/src/ingest/providers/gemini.ts index 693245426..c40ffeb85 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/gemini.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/gemini.ts @@ -356,12 +356,15 @@ export async function classifyTopicGemini( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Gemini topic-classifier: apiKey is required"); const model = cfg.model ?? "gemini-1.5-flash"; const endpoint = cfg.endpoint ?? `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; - const url = `${endpoint}?key=${cfg.apiKey}`; + const u = new URL(endpoint); + u.searchParams.set("key", cfg.apiKey); + const url = u.toString(); const headers: Record = { "Content-Type": "application/json", ...cfg.headers, @@ -397,12 +400,15 @@ export async function arbitrateTopicSplitGemini( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Gemini topic-arbitration: apiKey is required"); const model = cfg.model ?? "gemini-1.5-flash"; const endpoint = cfg.endpoint ?? `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; - const url = `${endpoint}?key=${cfg.apiKey}`; + const u = new URL(endpoint); + u.searchParams.set("key", cfg.apiKey); + const url = u.toString(); const headers: Record = { "Content-Type": "application/json", ...cfg.headers, diff --git a/packages/memos-core/src/ingest/providers/anthropic.ts b/packages/memos-core/src/ingest/providers/anthropic.ts index 4a3bdae3d..1b5e1d929 100644 --- a/packages/memos-core/src/ingest/providers/anthropic.ts +++ b/packages/memos-core/src/ingest/providers/anthropic.ts @@ -357,13 +357,14 @@ export async function classifyTopicAnthropic( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Anthropic topic-classifier: apiKey is required"); const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; const model = cfg.model ?? "claude-3-haiku-20240307"; const headers: Record = { "Content-Type": "application/json", - "x-api-key": cfg.apiKey ?? "", - "anthropic-version": "2023-06-01", ...cfg.headers, + "x-api-key": cfg.apiKey, + "anthropic-version": "2023-06-01", }; const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; @@ -386,8 +387,9 @@ export async function classifyTopicAnthropic( throw new Error(`Anthropic topic-classifier failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; - const raw = json.content.find((c) => c.type === "text")?.text?.trim() ?? ""; + const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; + const content = Array.isArray(json?.content) ? json.content : []; + const raw = content.find((c) => c.type === "text")?.text?.trim() ?? ""; log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -398,13 +400,14 @@ export async function arbitrateTopicSplitAnthropic( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Anthropic topic-arbitration: apiKey is required"); const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; const model = cfg.model ?? "claude-3-haiku-20240307"; const headers: Record = { "Content-Type": "application/json", - "x-api-key": cfg.apiKey ?? "", - "anthropic-version": "2023-06-01", ...cfg.headers, + "x-api-key": cfg.apiKey, + "anthropic-version": "2023-06-01", }; const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; @@ -414,7 +417,7 @@ export async function arbitrateTopicSplitAnthropic( headers, body: JSON.stringify({ model, - max_tokens: 60, + max_tokens: 10, temperature: 0, system: TOPIC_ARBITRATION_PROMPT, messages: [{ role: "user", content: userContent }], @@ -427,8 +430,9 @@ export async function arbitrateTopicSplitAnthropic( throw new Error(`Anthropic topic-arbitration failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { content: Array<{ type: string; text: string }> }; - const answer = json.content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; + const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; + const content = Array.isArray(json?.content) ? json.content : []; + const answer = content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; log.debug(`Topic arbitration result: "${answer}"`); return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/packages/memos-core/src/ingest/providers/bedrock.ts b/packages/memos-core/src/ingest/providers/bedrock.ts index 695ae1f73..082679cd6 100644 --- a/packages/memos-core/src/ingest/providers/bedrock.ts +++ b/packages/memos-core/src/ingest/providers/bedrock.ts @@ -440,6 +440,11 @@ export async function arbitrateTopicSplitBedrock( const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; const answer = json.output?.message?.content?.[0]?.text?.trim().toUpperCase() ?? ""; log.debug(`Topic arbitration result: "${answer}"`); + if (!answer) { + log.warn("Bedrock topic-arbitration returned empty text; defaulting to SAME"); + } else if (!answer.startsWith("NEW") && !answer.startsWith("SAME")) { + log.warn(`Bedrock topic-arbitration returned unexpected value "${answer}"; defaulting to SAME`); + } return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/packages/memos-core/src/ingest/providers/gemini.ts b/packages/memos-core/src/ingest/providers/gemini.ts index 693245426..c40ffeb85 100644 --- a/packages/memos-core/src/ingest/providers/gemini.ts +++ b/packages/memos-core/src/ingest/providers/gemini.ts @@ -356,12 +356,15 @@ export async function classifyTopicGemini( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Gemini topic-classifier: apiKey is required"); const model = cfg.model ?? "gemini-1.5-flash"; const endpoint = cfg.endpoint ?? `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; - const url = `${endpoint}?key=${cfg.apiKey}`; + const u = new URL(endpoint); + u.searchParams.set("key", cfg.apiKey); + const url = u.toString(); const headers: Record = { "Content-Type": "application/json", ...cfg.headers, @@ -397,12 +400,15 @@ export async function arbitrateTopicSplitGemini( cfg: SummarizerConfig, log: Logger, ): Promise { + if (!cfg.apiKey) throw new Error("Gemini topic-arbitration: apiKey is required"); const model = cfg.model ?? "gemini-1.5-flash"; const endpoint = cfg.endpoint ?? `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; - const url = `${endpoint}?key=${cfg.apiKey}`; + const u = new URL(endpoint); + u.searchParams.set("key", cfg.apiKey); + const url = u.toString(); const headers: Record = { "Content-Type": "application/json", ...cfg.headers, From 9ac07bfb5f620ae1f86fecaa22d2e625c6172987 Mon Sep 17 00:00:00 2001 From: MemOS AutoDev Date: Tue, 21 Jul 2026 13:05:32 +0800 Subject: [PATCH 3/3] fix(openclaw): address remaining OCR round-2 findings on topic-classifier providers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 1 (commit 646c28a1) did not fully converge OCR review. Round 2 targets the 2 repeated findings and 10 newly detected ones without expanding scope beyond the classifier/arbitration functions added by PR #2133. Bedrock (packages/memos-core + apps/memos-local-openclaw): - Fix TRUE MISS from Round 1: `arbitrateTopicSplitBedrock` was still using `maxTokens: 60` while Anthropic's arbitrator was already lowered to 10. Aligned to `maxTokens: 10`. - Introduce `DEFAULT_BEDROCK_TOPIC_MODEL` constant scoped to the two new functions only (`classifyTopicBedrock`, `arbitrateTopicSplitBedrock`). Pre-existing helpers (`judgeDedupBedrock`, `summarizeBedrock`, etc.) keep their inline default to stay out of scope. - Extract `bedrockConverseTopic()` shared transport used by the two new functions to eliminate the DRY duplication OCR flagged. Gemini (packages/memos-core + apps/memos-local-openclaw): - Add empty/unexpected-response `log.warn` branches to `arbitrateTopicSplitGemini`, mirroring the Bedrock arbitrator's defensive handling (Round 1 covered Bedrock only). - Extract `callGeminiTopic()` shared transport used by the two new functions to eliminate the DRY duplication OCR flagged. - Rejected escalating the empty-response case to `throw`: throwing would cascade through `Summarizer.tryChain` and burn tokens on every fallback; warn+default-to-SAME is the safer choice for a boundary detector. Anthropic (packages/memos-core + apps/memos-local-openclaw): - Extract `callAnthropicMessagesTopic()` shared transport used by the two new functions to eliminate the DRY duplication OCR flagged. Header spread order is preserved verbatim: `...cfg.headers` comes BEFORE the credential and `anthropic-version` so user-supplied headers cannot override them (this was the Round 1 security fix — do not undo). Pre-existing helpers in each provider are intentionally untouched to keep the diff minimum-scoped to PR #2133's classifier/arbitrator additions. Verification (executed in apps/memos-local-openclaw): - `./node_modules/.bin/vitest run tests/topic-classifier-dispatch.test.ts \\ tests/topic-judge-minimax-1315.test.ts` → 13/13 passed - `./node_modules/.bin/tsc --noEmit` on the 3 modified provider files → 0 errors - All 6 files (3 apps + 3 packages/memos-core mirrors) are byte-identical --- .../src/ingest/providers/anthropic.ts | 84 +++++++++-------- .../src/ingest/providers/bedrock.ts | 89 +++++++++--------- .../src/ingest/providers/gemini.ts | 90 +++++++++---------- .../src/ingest/providers/anthropic.ts | 84 +++++++++-------- .../src/ingest/providers/bedrock.ts | 89 +++++++++--------- .../memos-core/src/ingest/providers/gemini.ts | 90 +++++++++---------- 6 files changed, 256 insertions(+), 270 deletions(-) diff --git a/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts b/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts index 1b5e1d929..72d23630a 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/anthropic.ts @@ -351,13 +351,19 @@ import { import type { TopicClassifyResult } from "./openai"; export type { TopicClassifyResult } from "./openai"; -export async function classifyTopicAnthropic( - taskState: string, - newMessage: string, +// Shared Anthropic Messages transport for the topic classifier + arbitration. +// Scoped to these two callers only so pre-existing anthropic helpers keep +// their inline implementations (minimum-diff discipline). Header order below +// is deliberate: cfg.headers is spread FIRST so user-supplied headers cannot +// override the credential and required `anthropic-version` that follow. +async function callAnthropicMessagesTopic( + systemPrompt: string, + userContent: string, cfg: SummarizerConfig, - log: Logger, -): Promise { - if (!cfg.apiKey) throw new Error("Anthropic topic-classifier: apiKey is required"); + maxTokens: number, + errorLabel: string, +): Promise { + if (!cfg.apiKey) throw new Error(`Anthropic ${errorLabel}: apiKey is required`); const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; const model = cfg.model ?? "claude-3-haiku-20240307"; const headers: Record = { @@ -367,16 +373,14 @@ export async function classifyTopicAnthropic( "anthropic-version": "2023-06-01", }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - const resp = await fetch(endpoint, { method: "POST", headers, body: JSON.stringify({ model, - max_tokens: 60, + max_tokens: maxTokens, temperature: 0, - system: TOPIC_CLASSIFIER_PROMPT, + system: systemPrompt, messages: [{ role: "user", content: userContent }], }), signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), @@ -384,12 +388,28 @@ export async function classifyTopicAnthropic( if (!resp.ok) { const body = await resp.text(); - throw new Error(`Anthropic topic-classifier failed (${resp.status}): ${body}`); + throw new Error(`Anthropic ${errorLabel} failed (${resp.status}): ${body}`); } const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; const content = Array.isArray(json?.content) ? json.content : []; - const raw = content.find((c) => c.type === "text")?.text?.trim() ?? ""; + return content.find((c) => c.type === "text")?.text?.trim() ?? ""; +} + +export async function classifyTopicAnthropic( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + const raw = await callAnthropicMessagesTopic( + TOPIC_CLASSIFIER_PROMPT, + userContent, + cfg, + 60, + "topic-classifier", + ); log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -400,39 +420,15 @@ export async function arbitrateTopicSplitAnthropic( cfg: SummarizerConfig, log: Logger, ): Promise { - if (!cfg.apiKey) throw new Error("Anthropic topic-arbitration: apiKey is required"); - const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; - const model = cfg.model ?? "claude-3-haiku-20240307"; - const headers: Record = { - "Content-Type": "application/json", - ...cfg.headers, - "x-api-key": cfg.apiKey, - "anthropic-version": "2023-06-01", - }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - - const resp = await fetch(endpoint, { - method: "POST", - headers, - body: JSON.stringify({ - model, - max_tokens: 10, - temperature: 0, - system: TOPIC_ARBITRATION_PROMPT, - messages: [{ role: "user", content: userContent }], - }), - signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), - }); - - if (!resp.ok) { - const body = await resp.text(); - throw new Error(`Anthropic topic-arbitration failed (${resp.status}): ${body}`); - } - - const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; - const content = Array.isArray(json?.content) ? json.content : []; - const answer = content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; + const text = await callAnthropicMessagesTopic( + TOPIC_ARBITRATION_PROMPT, + userContent, + cfg, + 10, + "topic-arbitration", + ); + const answer = text.toUpperCase(); log.debug(`Topic arbitration result: "${answer}"`); return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts b/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts index 082679cd6..a62640324 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/bedrock.ts @@ -359,16 +359,24 @@ import { import type { TopicClassifyResult } from "./openai"; export type { TopicClassifyResult } from "./openai"; -export async function classifyTopicBedrock( - taskState: string, - newMessage: string, +// Default Bedrock model for the topic-classifier / arbitration helpers. +// Scoped to the two functions below to avoid churn in pre-existing helpers. +const DEFAULT_BEDROCK_TOPIC_MODEL = "anthropic.claude-3-haiku-20240307-v1:0"; + +// Shared Converse transport used by the topic classifier and arbitration. +// Only these two callers use it — pre-existing bedrock helpers keep their +// original inline implementations to minimise diff. +async function bedrockConverseTopic( + systemPrompt: string, + userContent: string, cfg: SummarizerConfig, - log: Logger, -): Promise { - const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; + maxTokens: number, + errorLabel: string, +): Promise { + const model = cfg.model ?? DEFAULT_BEDROCK_TOPIC_MODEL; const endpoint = cfg.endpoint; if (!endpoint) { - throw new Error("Bedrock topic-classifier requires 'endpoint'"); + throw new Error(`Bedrock ${errorLabel} requires 'endpoint'`); } const url = `${endpoint}/model/${model}/converse`; @@ -377,26 +385,40 @@ export async function classifyTopicBedrock( ...cfg.headers, }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - const resp = await fetch(url, { method: "POST", headers, body: JSON.stringify({ - system: [{ text: TOPIC_CLASSIFIER_PROMPT }], + system: [{ text: systemPrompt }], messages: [{ role: "user", content: [{ text: userContent }] }], - inferenceConfig: { temperature: 0, maxTokens: 60 }, + inferenceConfig: { temperature: 0, maxTokens }, }), signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), }); if (!resp.ok) { const body = await resp.text(); - throw new Error(`Bedrock topic-classifier failed (${resp.status}): ${body}`); + throw new Error(`Bedrock ${errorLabel} failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; - const raw = json.output?.message?.content?.[0]?.text?.trim() ?? ""; + const json = (await resp.json()) as { output?: { message?: { content?: Array<{ text: string }> } } }; + return json.output?.message?.content?.[0]?.text?.trim() ?? ""; +} + +export async function classifyTopicBedrock( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + const raw = await bedrockConverseTopic( + TOPIC_CLASSIFIER_PROMPT, + userContent, + cfg, + 60, + "topic-classifier", + ); log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -407,38 +429,15 @@ export async function arbitrateTopicSplitBedrock( cfg: SummarizerConfig, log: Logger, ): Promise { - const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; - const endpoint = cfg.endpoint; - if (!endpoint) { - throw new Error("Bedrock topic-arbitration requires 'endpoint'"); - } - - const url = `${endpoint}/model/${model}/converse`; - const headers: Record = { - "Content-Type": "application/json", - ...cfg.headers, - }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - - const resp = await fetch(url, { - method: "POST", - headers, - body: JSON.stringify({ - system: [{ text: TOPIC_ARBITRATION_PROMPT }], - messages: [{ role: "user", content: [{ text: userContent }] }], - inferenceConfig: { temperature: 0, maxTokens: 60 }, - }), - signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), - }); - - if (!resp.ok) { - const body = await resp.text(); - throw new Error(`Bedrock topic-arbitration failed (${resp.status}): ${body}`); - } - - const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; - const answer = json.output?.message?.content?.[0]?.text?.trim().toUpperCase() ?? ""; + const text = await bedrockConverseTopic( + TOPIC_ARBITRATION_PROMPT, + userContent, + cfg, + 10, + "topic-arbitration", + ); + const answer = text.toUpperCase(); log.debug(`Topic arbitration result: "${answer}"`); if (!answer) { log.warn("Bedrock topic-arbitration returned empty text; defaulting to SAME"); diff --git a/apps/memos-local-openclaw/src/ingest/providers/gemini.ts b/apps/memos-local-openclaw/src/ingest/providers/gemini.ts index c40ffeb85..eab5b3d5c 100644 --- a/apps/memos-local-openclaw/src/ingest/providers/gemini.ts +++ b/apps/memos-local-openclaw/src/ingest/providers/gemini.ts @@ -350,13 +350,17 @@ import { import type { TopicClassifyResult } from "./openai"; export type { TopicClassifyResult } from "./openai"; -export async function classifyTopicGemini( - taskState: string, - newMessage: string, +// Shared Gemini generateContent transport for the topic classifier + +// arbitration. Scoped to these two callers only so pre-existing gemini +// helpers keep their inline URL construction (minimum-diff discipline). +async function callGeminiTopic( + systemPrompt: string, + userContent: string, cfg: SummarizerConfig, - log: Logger, -): Promise { - if (!cfg.apiKey) throw new Error("Gemini topic-classifier: apiKey is required"); + maxOutputTokens: number, + errorLabel: string, +): Promise { + if (!cfg.apiKey) throw new Error(`Gemini ${errorLabel}: apiKey is required`); const model = cfg.model ?? "gemini-1.5-flash"; const endpoint = cfg.endpoint ?? @@ -370,26 +374,40 @@ export async function classifyTopicGemini( ...cfg.headers, }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - const resp = await fetch(url, { method: "POST", headers, body: JSON.stringify({ - systemInstruction: { parts: [{ text: TOPIC_CLASSIFIER_PROMPT }] }, + systemInstruction: { parts: [{ text: systemPrompt }] }, contents: [{ parts: [{ text: userContent }] }], - generationConfig: { temperature: 0, maxOutputTokens: 60 }, + generationConfig: { temperature: 0, maxOutputTokens }, }), signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), }); if (!resp.ok) { const body = await resp.text(); - throw new Error(`Gemini topic-classifier failed (${resp.status}): ${body}`); + throw new Error(`Gemini ${errorLabel} failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; - const raw = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim() ?? ""; + const json = (await resp.json()) as { candidates?: Array<{ content?: { parts?: Array<{ text: string }> } }> }; + return json.candidates?.[0]?.content?.parts?.[0]?.text?.trim() ?? ""; +} + +export async function classifyTopicGemini( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + const raw = await callGeminiTopic( + TOPIC_CLASSIFIER_PROMPT, + userContent, + cfg, + 60, + "topic-classifier", + ); log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -400,41 +418,21 @@ export async function arbitrateTopicSplitGemini( cfg: SummarizerConfig, log: Logger, ): Promise { - if (!cfg.apiKey) throw new Error("Gemini topic-arbitration: apiKey is required"); - const model = cfg.model ?? "gemini-1.5-flash"; - const endpoint = - cfg.endpoint ?? - `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; - - const u = new URL(endpoint); - u.searchParams.set("key", cfg.apiKey); - const url = u.toString(); - const headers: Record = { - "Content-Type": "application/json", - ...cfg.headers, - }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - - const resp = await fetch(url, { - method: "POST", - headers, - body: JSON.stringify({ - systemInstruction: { parts: [{ text: TOPIC_ARBITRATION_PROMPT }] }, - contents: [{ parts: [{ text: userContent }] }], - generationConfig: { temperature: 0, maxOutputTokens: 60 }, - }), - signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), - }); - - if (!resp.ok) { - const body = await resp.text(); - throw new Error(`Gemini topic-arbitration failed (${resp.status}): ${body}`); - } - - const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; - const answer = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim().toUpperCase() ?? ""; + const text = await callGeminiTopic( + TOPIC_ARBITRATION_PROMPT, + userContent, + cfg, + 10, + "topic-arbitration", + ); + const answer = text.toUpperCase(); log.debug(`Topic arbitration result: "${answer}"`); + if (!answer) { + log.warn("Gemini topic-arbitration returned empty text; defaulting to SAME"); + } else if (!answer.startsWith("NEW") && !answer.startsWith("SAME")) { + log.warn(`Gemini topic-arbitration returned unexpected value "${answer}"; defaulting to SAME`); + } return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/packages/memos-core/src/ingest/providers/anthropic.ts b/packages/memos-core/src/ingest/providers/anthropic.ts index 1b5e1d929..72d23630a 100644 --- a/packages/memos-core/src/ingest/providers/anthropic.ts +++ b/packages/memos-core/src/ingest/providers/anthropic.ts @@ -351,13 +351,19 @@ import { import type { TopicClassifyResult } from "./openai"; export type { TopicClassifyResult } from "./openai"; -export async function classifyTopicAnthropic( - taskState: string, - newMessage: string, +// Shared Anthropic Messages transport for the topic classifier + arbitration. +// Scoped to these two callers only so pre-existing anthropic helpers keep +// their inline implementations (minimum-diff discipline). Header order below +// is deliberate: cfg.headers is spread FIRST so user-supplied headers cannot +// override the credential and required `anthropic-version` that follow. +async function callAnthropicMessagesTopic( + systemPrompt: string, + userContent: string, cfg: SummarizerConfig, - log: Logger, -): Promise { - if (!cfg.apiKey) throw new Error("Anthropic topic-classifier: apiKey is required"); + maxTokens: number, + errorLabel: string, +): Promise { + if (!cfg.apiKey) throw new Error(`Anthropic ${errorLabel}: apiKey is required`); const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; const model = cfg.model ?? "claude-3-haiku-20240307"; const headers: Record = { @@ -367,16 +373,14 @@ export async function classifyTopicAnthropic( "anthropic-version": "2023-06-01", }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - const resp = await fetch(endpoint, { method: "POST", headers, body: JSON.stringify({ model, - max_tokens: 60, + max_tokens: maxTokens, temperature: 0, - system: TOPIC_CLASSIFIER_PROMPT, + system: systemPrompt, messages: [{ role: "user", content: userContent }], }), signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), @@ -384,12 +388,28 @@ export async function classifyTopicAnthropic( if (!resp.ok) { const body = await resp.text(); - throw new Error(`Anthropic topic-classifier failed (${resp.status}): ${body}`); + throw new Error(`Anthropic ${errorLabel} failed (${resp.status}): ${body}`); } const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; const content = Array.isArray(json?.content) ? json.content : []; - const raw = content.find((c) => c.type === "text")?.text?.trim() ?? ""; + return content.find((c) => c.type === "text")?.text?.trim() ?? ""; +} + +export async function classifyTopicAnthropic( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + const raw = await callAnthropicMessagesTopic( + TOPIC_CLASSIFIER_PROMPT, + userContent, + cfg, + 60, + "topic-classifier", + ); log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -400,39 +420,15 @@ export async function arbitrateTopicSplitAnthropic( cfg: SummarizerConfig, log: Logger, ): Promise { - if (!cfg.apiKey) throw new Error("Anthropic topic-arbitration: apiKey is required"); - const endpoint = cfg.endpoint ?? "https://api.anthropic.com/v1/messages"; - const model = cfg.model ?? "claude-3-haiku-20240307"; - const headers: Record = { - "Content-Type": "application/json", - ...cfg.headers, - "x-api-key": cfg.apiKey, - "anthropic-version": "2023-06-01", - }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - - const resp = await fetch(endpoint, { - method: "POST", - headers, - body: JSON.stringify({ - model, - max_tokens: 10, - temperature: 0, - system: TOPIC_ARBITRATION_PROMPT, - messages: [{ role: "user", content: userContent }], - }), - signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), - }); - - if (!resp.ok) { - const body = await resp.text(); - throw new Error(`Anthropic topic-arbitration failed (${resp.status}): ${body}`); - } - - const json = (await resp.json()) as { content?: Array<{ type: string; text: string }> }; - const content = Array.isArray(json?.content) ? json.content : []; - const answer = content.find((c) => c.type === "text")?.text?.trim().toUpperCase() ?? ""; + const text = await callAnthropicMessagesTopic( + TOPIC_ARBITRATION_PROMPT, + userContent, + cfg, + 10, + "topic-arbitration", + ); + const answer = text.toUpperCase(); log.debug(`Topic arbitration result: "${answer}"`); return answer.startsWith("NEW") ? "NEW" : "SAME"; } diff --git a/packages/memos-core/src/ingest/providers/bedrock.ts b/packages/memos-core/src/ingest/providers/bedrock.ts index 082679cd6..a62640324 100644 --- a/packages/memos-core/src/ingest/providers/bedrock.ts +++ b/packages/memos-core/src/ingest/providers/bedrock.ts @@ -359,16 +359,24 @@ import { import type { TopicClassifyResult } from "./openai"; export type { TopicClassifyResult } from "./openai"; -export async function classifyTopicBedrock( - taskState: string, - newMessage: string, +// Default Bedrock model for the topic-classifier / arbitration helpers. +// Scoped to the two functions below to avoid churn in pre-existing helpers. +const DEFAULT_BEDROCK_TOPIC_MODEL = "anthropic.claude-3-haiku-20240307-v1:0"; + +// Shared Converse transport used by the topic classifier and arbitration. +// Only these two callers use it — pre-existing bedrock helpers keep their +// original inline implementations to minimise diff. +async function bedrockConverseTopic( + systemPrompt: string, + userContent: string, cfg: SummarizerConfig, - log: Logger, -): Promise { - const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; + maxTokens: number, + errorLabel: string, +): Promise { + const model = cfg.model ?? DEFAULT_BEDROCK_TOPIC_MODEL; const endpoint = cfg.endpoint; if (!endpoint) { - throw new Error("Bedrock topic-classifier requires 'endpoint'"); + throw new Error(`Bedrock ${errorLabel} requires 'endpoint'`); } const url = `${endpoint}/model/${model}/converse`; @@ -377,26 +385,40 @@ export async function classifyTopicBedrock( ...cfg.headers, }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - const resp = await fetch(url, { method: "POST", headers, body: JSON.stringify({ - system: [{ text: TOPIC_CLASSIFIER_PROMPT }], + system: [{ text: systemPrompt }], messages: [{ role: "user", content: [{ text: userContent }] }], - inferenceConfig: { temperature: 0, maxTokens: 60 }, + inferenceConfig: { temperature: 0, maxTokens }, }), signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), }); if (!resp.ok) { const body = await resp.text(); - throw new Error(`Bedrock topic-classifier failed (${resp.status}): ${body}`); + throw new Error(`Bedrock ${errorLabel} failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; - const raw = json.output?.message?.content?.[0]?.text?.trim() ?? ""; + const json = (await resp.json()) as { output?: { message?: { content?: Array<{ text: string }> } } }; + return json.output?.message?.content?.[0]?.text?.trim() ?? ""; +} + +export async function classifyTopicBedrock( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + const raw = await bedrockConverseTopic( + TOPIC_CLASSIFIER_PROMPT, + userContent, + cfg, + 60, + "topic-classifier", + ); log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -407,38 +429,15 @@ export async function arbitrateTopicSplitBedrock( cfg: SummarizerConfig, log: Logger, ): Promise { - const model = cfg.model ?? "anthropic.claude-3-haiku-20240307-v1:0"; - const endpoint = cfg.endpoint; - if (!endpoint) { - throw new Error("Bedrock topic-arbitration requires 'endpoint'"); - } - - const url = `${endpoint}/model/${model}/converse`; - const headers: Record = { - "Content-Type": "application/json", - ...cfg.headers, - }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - - const resp = await fetch(url, { - method: "POST", - headers, - body: JSON.stringify({ - system: [{ text: TOPIC_ARBITRATION_PROMPT }], - messages: [{ role: "user", content: [{ text: userContent }] }], - inferenceConfig: { temperature: 0, maxTokens: 60 }, - }), - signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), - }); - - if (!resp.ok) { - const body = await resp.text(); - throw new Error(`Bedrock topic-arbitration failed (${resp.status}): ${body}`); - } - - const json = (await resp.json()) as { output: { message: { content: Array<{ text: string }> } } }; - const answer = json.output?.message?.content?.[0]?.text?.trim().toUpperCase() ?? ""; + const text = await bedrockConverseTopic( + TOPIC_ARBITRATION_PROMPT, + userContent, + cfg, + 10, + "topic-arbitration", + ); + const answer = text.toUpperCase(); log.debug(`Topic arbitration result: "${answer}"`); if (!answer) { log.warn("Bedrock topic-arbitration returned empty text; defaulting to SAME"); diff --git a/packages/memos-core/src/ingest/providers/gemini.ts b/packages/memos-core/src/ingest/providers/gemini.ts index c40ffeb85..eab5b3d5c 100644 --- a/packages/memos-core/src/ingest/providers/gemini.ts +++ b/packages/memos-core/src/ingest/providers/gemini.ts @@ -350,13 +350,17 @@ import { import type { TopicClassifyResult } from "./openai"; export type { TopicClassifyResult } from "./openai"; -export async function classifyTopicGemini( - taskState: string, - newMessage: string, +// Shared Gemini generateContent transport for the topic classifier + +// arbitration. Scoped to these two callers only so pre-existing gemini +// helpers keep their inline URL construction (minimum-diff discipline). +async function callGeminiTopic( + systemPrompt: string, + userContent: string, cfg: SummarizerConfig, - log: Logger, -): Promise { - if (!cfg.apiKey) throw new Error("Gemini topic-classifier: apiKey is required"); + maxOutputTokens: number, + errorLabel: string, +): Promise { + if (!cfg.apiKey) throw new Error(`Gemini ${errorLabel}: apiKey is required`); const model = cfg.model ?? "gemini-1.5-flash"; const endpoint = cfg.endpoint ?? @@ -370,26 +374,40 @@ export async function classifyTopicGemini( ...cfg.headers, }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - const resp = await fetch(url, { method: "POST", headers, body: JSON.stringify({ - systemInstruction: { parts: [{ text: TOPIC_CLASSIFIER_PROMPT }] }, + systemInstruction: { parts: [{ text: systemPrompt }] }, contents: [{ parts: [{ text: userContent }] }], - generationConfig: { temperature: 0, maxOutputTokens: 60 }, + generationConfig: { temperature: 0, maxOutputTokens }, }), signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), }); if (!resp.ok) { const body = await resp.text(); - throw new Error(`Gemini topic-classifier failed (${resp.status}): ${body}`); + throw new Error(`Gemini ${errorLabel} failed (${resp.status}): ${body}`); } - const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; - const raw = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim() ?? ""; + const json = (await resp.json()) as { candidates?: Array<{ content?: { parts?: Array<{ text: string }> } }> }; + return json.candidates?.[0]?.content?.parts?.[0]?.text?.trim() ?? ""; +} + +export async function classifyTopicGemini( + taskState: string, + newMessage: string, + cfg: SummarizerConfig, + log: Logger, +): Promise { + const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; + const raw = await callGeminiTopic( + TOPIC_CLASSIFIER_PROMPT, + userContent, + cfg, + 60, + "topic-classifier", + ); log.debug(`Topic classifier raw: "${raw}"`); return parseTopicClassifyResult(raw, log); } @@ -400,41 +418,21 @@ export async function arbitrateTopicSplitGemini( cfg: SummarizerConfig, log: Logger, ): Promise { - if (!cfg.apiKey) throw new Error("Gemini topic-arbitration: apiKey is required"); - const model = cfg.model ?? "gemini-1.5-flash"; - const endpoint = - cfg.endpoint ?? - `https://generativelanguage.googleapis.com/v1beta/models/${model}:generateContent`; - - const u = new URL(endpoint); - u.searchParams.set("key", cfg.apiKey); - const url = u.toString(); - const headers: Record = { - "Content-Type": "application/json", - ...cfg.headers, - }; - const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`; - - const resp = await fetch(url, { - method: "POST", - headers, - body: JSON.stringify({ - systemInstruction: { parts: [{ text: TOPIC_ARBITRATION_PROMPT }] }, - contents: [{ parts: [{ text: userContent }] }], - generationConfig: { temperature: 0, maxOutputTokens: 60 }, - }), - signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000), - }); - - if (!resp.ok) { - const body = await resp.text(); - throw new Error(`Gemini topic-arbitration failed (${resp.status}): ${body}`); - } - - const json = (await resp.json()) as { candidates: Array<{ content: { parts: Array<{ text: string }> } }> }; - const answer = json.candidates?.[0]?.content?.parts?.[0]?.text?.trim().toUpperCase() ?? ""; + const text = await callGeminiTopic( + TOPIC_ARBITRATION_PROMPT, + userContent, + cfg, + 10, + "topic-arbitration", + ); + const answer = text.toUpperCase(); log.debug(`Topic arbitration result: "${answer}"`); + if (!answer) { + log.warn("Gemini topic-arbitration returned empty text; defaulting to SAME"); + } else if (!answer.startsWith("NEW") && !answer.startsWith("SAME")) { + log.warn(`Gemini topic-arbitration returned unexpected value "${answer}"; defaulting to SAME`); + } return answer.startsWith("NEW") ? "NEW" : "SAME"; }