Fix #1611: classifyTopic fails when summarizer uses Anthropic endpoint (Kimi Code)

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";
+
+// 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,
+  maxTokens: number,
+  errorLabel: string,
+): Promise<string> {
+  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<string, string> = {
+    "Content-Type": "application/json",
+    ...cfg.headers,
+    "x-api-key": cfg.apiKey,
+    "anthropic-version": "2023-06-01",
+  };
+
+  const resp = await fetch(endpoint, {
+    method: "POST",
+    headers,
+    body: JSON.stringify({
+      model,
+      max_tokens: maxTokens,
+      temperature: 0,
+      system: systemPrompt,
+      messages: [{ role: "user", content: userContent }],
+    }),
+    signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000),
+  });
+
+  if (!resp.ok) {
+    const body = await resp.text();
+    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 : [];
+  return content.find((c) => c.type === "text")?.text?.trim() ?? "";
+}
+
+export async function classifyTopicAnthropic(
+  taskState: string,
+  newMessage: string,
+  cfg: SummarizerConfig,
+  log: Logger,
+): Promise<TopicClassifyResult> {
+  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);
+}
+
+export async function arbitrateTopicSplitAnthropic(
+  taskState: string,
+  newMessage: string,
+  cfg: SummarizerConfig,
+  log: Logger,
+): Promise<string> {
+  const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`;
+  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";
+}
+
 export async function judgeDedupAnthropic(
   newSummary: string,
   candidates: Array<{ index: number; summary: string; chunkId: string }>,

apps/memos-local-openclaw/src/ingest/providers/bedrock.ts

@@ -344,6 +344,109 @@ 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";
+
+// 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,
+  maxTokens: number,
+  errorLabel: string,
+): Promise<string> {
+  const model = cfg.model ?? DEFAULT_BEDROCK_TOPIC_MODEL;
+  const endpoint = cfg.endpoint;
+  if (!endpoint) {
+    throw new Error(`Bedrock ${errorLabel} requires 'endpoint'`);
+  }
+
+  const url = `${endpoint}/model/${model}/converse`;
+  const headers: Record<string, string> = {
+    "Content-Type": "application/json",
+    ...cfg.headers,
+  };
+
+  const resp = await fetch(url, {
+    method: "POST",
+    headers,
+    body: JSON.stringify({
+      system: [{ text: systemPrompt }],
+      messages: [{ role: "user", content: [{ text: userContent }] }],
+      inferenceConfig: { temperature: 0, maxTokens },
+    }),
+    signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000),
+  });
+
+  if (!resp.ok) {
+    const body = await resp.text();
+    throw new Error(`Bedrock ${errorLabel} failed (${resp.status}): ${body}`);
+  }
+
+  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<TopicClassifyResult> {
+  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);
+}
+
+export async function arbitrateTopicSplitBedrock(
+  taskState: string,
+  newMessage: string,
+  cfg: SummarizerConfig,
+  log: Logger,
+): Promise<string> {
+  const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`;
+  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");
+  } 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";
+}
+
 export async function judgeDedupBedrock(
   newSummary: string,
   candidates: Array<{ index: number; summary: string; chunkId: string }>,

apps/memos-local-openclaw/src/ingest/providers/gemini.ts

@@ -336,6 +336,106 @@ 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";
+
+// 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,
+  maxOutputTokens: number,
+  errorLabel: string,
+): Promise<string> {
+  if (!cfg.apiKey) throw new Error(`Gemini ${errorLabel}: 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<string, string> = {
+    "Content-Type": "application/json",
+    ...cfg.headers,
+  };
+
+  const resp = await fetch(url, {
+    method: "POST",
+    headers,
+    body: JSON.stringify({
+      systemInstruction: { parts: [{ text: systemPrompt }] },
+      contents: [{ parts: [{ text: userContent }] }],
+      generationConfig: { temperature: 0, maxOutputTokens },
+    }),
+    signal: AbortSignal.timeout(cfg.timeoutMs ?? 15_000),
+  });
+
+  if (!resp.ok) {
+    const body = await resp.text();
+    throw new Error(`Gemini ${errorLabel} failed (${resp.status}): ${body}`);
+  }
+
+  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<TopicClassifyResult> {
+  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);
+}
+
+export async function arbitrateTopicSplitGemini(
+  taskState: string,
+  newMessage: string,
+  cfg: SummarizerConfig,
+  log: Logger,
+): Promise<string> {
+  const userContent = `TASK:\n${taskState}\n\nMSG:\n${newMessage}`;
+  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";
+}
+
 export async function judgeDedupGemini(
   newSummary: string,
   candidates: Array<{ index: number; summary: string; chunkId: string }>,

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}`);
   }

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`;