diff --git a/.changeset/memory-cloudflare-embeddings.md b/.changeset/memory-cloudflare-embeddings.md new file mode 100644 index 0000000000..b981bc177e --- /dev/null +++ b/.changeset/memory-cloudflare-embeddings.md @@ -0,0 +1,5 @@ +--- +'@roomote/web': minor +--- + +Memory embeddings auto-wire to Cloudflare Workers AI when its token and account are configured, so semantic recall no longer needs a separate OpenAI or OpenRouter key. diff --git a/.env.production.example b/.env.production.example index c2a6ab4f5b..0608337bba 100644 --- a/.env.production.example +++ b/.env.production.example @@ -176,6 +176,16 @@ DEFAULT_COMPUTE_PROVIDER=docker # R_BRAIN_INFERENCE_UPSTREAM_API_KEY= # R_BRAIN_EMBEDDING_MODEL=BAAI/bge-m3 # R_BRAIN_EMBEDDING_DIMENSIONS=1024 +# Cloudflare Workers AI embeddings (no OpenAI/OpenRouter key required). +# Auto-wires only when those keys are unset. Leave +# R_BRAIN_EMBEDDINGS_UPSTREAM_URL unset so Roomote uses the Workers AI /ai +# base (it appends /v1/embeddings). Do not paste the embeddings path into +# R_BRAIN_EMBEDDINGS_UPSTREAM_URL. +# The embedding pair is create-time: set it BEFORE Memory's first boot. +# CLOUDFLARE_WORKERS_AI_API_TOKEN= +# CLOUDFLARE_WORKERS_AI_ACCOUNT_ID= +# R_BRAIN_EMBEDDING_MODEL=@cf/baai/bge-m3 +# R_BRAIN_EMBEDDING_DIMENSIONS=1024 # Self-run judgment model for routing and triage decisions. Any server that # answers the typed decisions request at /v1/decisions works; the bundled one # is COMPOSE_PROFILES=judgment with R_JUDGMENT_UPSTREAM_URL=http://judgment:8080 diff --git a/apps/api/src/handlers/brain-inference/__tests__/brain-inference.test.ts b/apps/api/src/handlers/brain-inference/__tests__/brain-inference.test.ts index c65db8568b..b4f696df9f 100644 --- a/apps/api/src/handlers/brain-inference/__tests__/brain-inference.test.ts +++ b/apps/api/src/handlers/brain-inference/__tests__/brain-inference.test.ts @@ -6,12 +6,14 @@ const { mockGetBrainGatewayToken, mockResolveBrainInferenceProvider, mockMapBrainModelName, + mockResolveBrainEmbeddingsUpstream, mockGenerateTrackedNonTaskText, mockEnv, } = vi.hoisted(() => ({ mockGetBrainGatewayToken: vi.fn(), mockResolveBrainInferenceProvider: vi.fn(), mockMapBrainModelName: vi.fn(), + mockResolveBrainEmbeddingsUpstream: vi.fn(), mockGenerateTrackedNonTaskText: vi.fn(), mockEnv: {} as Record, })); @@ -22,6 +24,7 @@ vi.mock('@roomote/sdk/server', () => ({ getBrainGatewayToken: mockGetBrainGatewayToken, resolveBrainInferenceProvider: mockResolveBrainInferenceProvider, mapBrainModelName: mockMapBrainModelName, + resolveBrainEmbeddingsUpstream: mockResolveBrainEmbeddingsUpstream, })); vi.mock('@roomote/cloud-agents/server/non-task-provider-usage', () => ({ @@ -70,6 +73,17 @@ beforeEach(() => { } mockGetBrainGatewayToken.mockReturnValue(GATEWAY_TOKEN); mockResolveBrainInferenceProvider.mockResolvedValue(OPENROUTER); + mockResolveBrainEmbeddingsUpstream.mockImplementation(async () => { + const configured = mockEnv.R_BRAIN_EMBEDDINGS_UPSTREAM_URL?.trim(); + if (!configured) { + return null; + } + return { + baseUrl: configured.replace(/\/$/, ''), + apiKey: mockEnv.R_BRAIN_INFERENCE_UPSTREAM_API_KEY?.trim() || undefined, + source: 'configured' as const, + }; + }); mockMapBrainModelName.mockImplementation((requested: string) => requested === 'text-embedding-3-small' ? 'openai/text-embedding-3-small' @@ -294,6 +308,78 @@ describe('local inference upstreams', () => { expect(response.status).toBe(401); expect(fetchMock).not.toHaveBeenCalled(); }); + + it('passes the requested embedding model to the local upstream resolver', async () => { + mockResolveBrainEmbeddingsUpstream.mockResolvedValue(null); + const fetchMock = vi + .fn() + .mockResolvedValue(new Response('{}', { status: 200 })); + vi.stubGlobal('fetch', fetchMock); + + await post('/v1/embeddings', { + token: GATEWAY_TOKEN, + body: { model: '@cf/baai/bge-m3', input: 'hello' }, + }); + + expect(mockResolveBrainEmbeddingsUpstream).toHaveBeenCalledWith( + '@cf/baai/bge-m3', + ); + }); + + it('routes embeddings to Cloudflare Workers AI without an OpenAI key', async () => { + mockResolveBrainInferenceProvider.mockResolvedValue(null); + mockResolveBrainEmbeddingsUpstream.mockResolvedValue({ + baseUrl: `https://api.cloudflare.com/client/v4/accounts/${'a1b2c3d4e5f6789012345678abcdef90'}/ai`, + apiKey: 'cfat_test', + source: 'cloudflare-workers-ai', + }); + const fetchMock = vi.fn().mockResolvedValue( + new Response(JSON.stringify({ data: [] }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }), + ); + vi.stubGlobal('fetch', fetchMock); + + const response = await post('/v1/embeddings', { + token: GATEWAY_TOKEN, + body: { model: '@cf/baai/bge-m3', input: 'hello' }, + }); + + expect(response.status).toBe(200); + const [url, init] = fetchMock.mock.calls[0]!; + expect(url).toBe( + 'https://api.cloudflare.com/client/v4/accounts/a1b2c3d4e5f6789012345678abcdef90/ai/v1/embeddings', + ); + const headers = new Headers((init as RequestInit).headers); + expect(headers.get('authorization')).toBe('Bearer cfat_test'); + expect(JSON.parse((init as RequestInit).body as string)).toMatchObject({ + model: '@cf/baai/bge-m3', + }); + expect(mockMapBrainModelName).not.toHaveBeenCalled(); + expect(mockResolveBrainInferenceProvider).not.toHaveBeenCalled(); + }); + + it('keeps chat on the provider when Cloudflare embeddings are wired', async () => { + mockResolveBrainEmbeddingsUpstream.mockResolvedValue({ + baseUrl: + 'https://api.cloudflare.com/client/v4/accounts/a1b2c3d4e5f6789012345678abcdef90/ai', + apiKey: 'cfat_test', + source: 'cloudflare-workers-ai', + }); + const fetchMock = vi + .fn() + .mockResolvedValue(new Response('{}', { status: 200 })); + vi.stubGlobal('fetch', fetchMock); + + await post('/v1/chat/completions', { + token: GATEWAY_TOKEN, + body: { model: 'gpt-5.6-luna', messages: [] }, + }); + + expect(mockResolveBrainInferenceProvider).toHaveBeenCalled(); + expect(String(fetchMock.mock.calls[0]![0])).toContain('openrouter.ai'); + }); }); describe('helper-model synthesis', () => { diff --git a/apps/api/src/handlers/brain-inference/auth.ts b/apps/api/src/handlers/brain-inference/auth.ts new file mode 100644 index 0000000000..9e666525f6 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/auth.ts @@ -0,0 +1,24 @@ +import { timingSafeEqual } from 'node:crypto'; + +export function presentedToken(header: string | undefined): string | null { + if (!header) { + return null; + } + + const match = header.match(/^Bearer\s+(.+)$/i); + + return match?.[1]?.trim() || null; +} + +/** Constant-time compare that does not leak length through early return. */ +export function tokenMatches(presented: string, expected: string): boolean { + const a = Buffer.from(presented); + const b = Buffer.from(expected); + + if (a.length !== b.length) { + timingSafeEqual(a, a); + return false; + } + + return timingSafeEqual(a, b); +} diff --git a/apps/api/src/handlers/brain-inference/constants.ts b/apps/api/src/handlers/brain-inference/constants.ts new file mode 100644 index 0000000000..946a18cfba --- /dev/null +++ b/apps/api/src/handlers/brain-inference/constants.ts @@ -0,0 +1,51 @@ +export const LOG_PREFIX = '[Brain Inference]'; + +/** + * Sentinel chat-model id the Brain requests in gateway mode. It is not a + * real provider model: the gateway answers it through the deployment's + * helper model. An operator's `R_BRAIN_MODEL` still wins. + */ +export const BRAIN_HELPER_MODEL_ID = 'roomote/helper'; + +export const HELPER_SYNTHESIS_TIMEOUT_MS = 120_000; +export const HELPER_SYNTHESIS_DEFAULT_MAX_OUTPUT_TOKENS = 2048; + +/** + * Embeddings for recall and chat for sourced synthesis. Narrower than the + * task-sandbox gateway because this credential is a static deployment secret. + */ +export const BRAIN_ALLOWED_PATHS = new Set([ + '/v1/embeddings', + '/v1/chat/completions', + '/v1/responses', +]); + +/** Never forwarded upstream: the Brain's gateway token is not a provider key. */ +export const REQUEST_HEADER_DENYLIST = new Set([ + 'authorization', + 'x-api-key', + 'cookie', + 'host', + 'connection', + 'content-length', + 'keep-alive', + 'proxy-authenticate', + 'proxy-authorization', + 'te', + 'trailer', + 'transfer-encoding', + 'upgrade', + 'expect', + 'forwarded', + 'x-forwarded-for', + 'x-forwarded-host', + 'x-forwarded-proto', + 'x-real-ip', +]); + +export const RESPONSE_HEADER_DENYLIST = new Set([ + 'content-encoding', + 'content-length', + 'transfer-encoding', + 'connection', +]); diff --git a/apps/api/src/handlers/brain-inference/copy-headers.ts b/apps/api/src/handlers/brain-inference/copy-headers.ts new file mode 100644 index 0000000000..9933d2ee88 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/copy-headers.ts @@ -0,0 +1,14 @@ +export function copyAllowedHeaders( + source: Headers, + denylist: Set, +): Headers { + const headers = new Headers(); + + for (const [name, value] of source.entries()) { + if (!denylist.has(name.toLowerCase())) { + headers.set(name, value); + } + } + + return headers; +} diff --git a/apps/api/src/handlers/brain-inference/forward-local.ts b/apps/api/src/handlers/brain-inference/forward-local.ts new file mode 100644 index 0000000000..c0cb9449e4 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/forward-local.ts @@ -0,0 +1,92 @@ +import type { Context } from 'hono'; +import { formatSingleLineLog } from '@roomote/types'; +import { resolveBrainEmbeddingsUpstream } from '@roomote/sdk/server'; + +import type { Variables } from '../../types'; +import { + LOG_PREFIX, + REQUEST_HEADER_DENYLIST, + RESPONSE_HEADER_DENYLIST, +} from './constants'; +import { copyAllowedHeaders } from './copy-headers'; + +type GatewayContext = Context<{ Variables: Variables }>; + +function embeddingModelFromBody(body: string): string | undefined { + try { + const parsed: unknown = JSON.parse(body); + + if ( + parsed && + typeof parsed === 'object' && + 'model' in parsed && + typeof parsed.model === 'string' + ) { + return parsed.model.trim() || undefined; + } + } catch { + // Forward the body unchanged; a missing model only affects routing. + } + + return undefined; +} + +export async function forwardLocalEmbeddings( + c: GatewayContext, + upstreamPath: string, + requestUrl: URL, +): Promise { + if (upstreamPath !== '/v1/embeddings') { + return null; + } + + const body = await c.req.text(); + const localUpstream = await resolveBrainEmbeddingsUpstream( + embeddingModelFromBody(body), + ); + + if (!localUpstream) { + return null; + } + + const headers = copyAllowedHeaders( + c.req.raw.headers, + REQUEST_HEADER_DENYLIST, + ); + + if (localUpstream.apiKey) { + headers.set('authorization', `Bearer ${localUpstream.apiKey}`); + } + + let upstream: Response; + + try { + upstream = await fetch( + `${localUpstream.baseUrl}${upstreamPath}${requestUrl.search}`, + { method: 'POST', headers, body }, + ); + } catch (error) { + console.warn( + formatSingleLineLog(`${LOG_PREFIX} Local upstream request failed`, { + upstreamPath, + error: error instanceof Error ? error.message : String(error), + }), + ); + + return c.json({ error: 'Brain inference upstream is unreachable' }, 502); + } + + if (!upstream.ok) { + console.warn( + formatSingleLineLog(`${LOG_PREFIX} Local upstream returned an error`, { + upstreamPath, + status: upstream.status, + }), + ); + } + + return new Response(upstream.body, { + status: upstream.status, + headers: copyAllowedHeaders(upstream.headers, RESPONSE_HEADER_DENYLIST), + }); +} diff --git a/apps/api/src/handlers/brain-inference/forward-provider.ts b/apps/api/src/handlers/brain-inference/forward-provider.ts new file mode 100644 index 0000000000..8a51c6a7b0 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/forward-provider.ts @@ -0,0 +1,89 @@ +import type { Context } from 'hono'; +import { + formatSingleLineLog, + getInferenceGatewayProvider, +} from '@roomote/types'; +import { resolveBrainInferenceProvider } from '@roomote/sdk/server'; + +import type { Variables } from '../../types'; +import { LOG_PREFIX, RESPONSE_HEADER_DENYLIST } from './constants'; +import { copyAllowedHeaders } from './copy-headers'; +import { providerHeaders } from './provider-headers'; +import { rewriteBody } from './rewrite-body'; + +type GatewayContext = Context<{ Variables: Variables }>; + +export async function forwardProviderInference( + c: GatewayContext, + upstreamPath: string, + requestUrl: URL, + helperOverrideBody: string | undefined, +): Promise { + const resolved = await resolveBrainInferenceProvider(); + + if (!resolved) { + return c.json( + { + error: + 'No model provider is configured for this deployment. Add a provider key in Settings to enable the Brain.', + }, + 503, + ); + } + + const provider = getInferenceGatewayProvider(resolved.providerId); + + if (!provider?.authHeader) { + console.error( + formatSingleLineLog(`${LOG_PREFIX} Provider is not proxyable`, { + providerId: resolved.providerId, + }), + ); + + return c.json({ error: 'Brain inference provider is unavailable' }, 500); + } + + const headers = providerHeaders(c.req.raw.headers, resolved, { + authHeader: provider.authHeader, + }); + const body = await rewriteBody( + helperOverrideBody ?? (await c.req.text()), + resolved, + ); + const startedAt = Date.now(); + + let upstream: Response; + + try { + upstream = await fetch( + `${provider.upstreamBaseUrl}${upstreamPath}${requestUrl.search}`, + { method: 'POST', headers, body }, + ); + } catch (error) { + console.warn( + formatSingleLineLog(`${LOG_PREFIX} Upstream request failed`, { + providerId: resolved.providerId, + upstreamPath, + error: error instanceof Error ? error.message : String(error), + }), + ); + + return c.json({ error: 'Brain inference upstream is unreachable' }, 502); + } + + if (!upstream.ok) { + console.warn( + formatSingleLineLog(`${LOG_PREFIX} Upstream returned an error`, { + providerId: resolved.providerId, + upstreamPath, + status: upstream.status, + durationMs: Date.now() - startedAt, + }), + ); + } + + return new Response(upstream.body, { + status: upstream.status, + headers: copyAllowedHeaders(upstream.headers, RESPONSE_HEADER_DENYLIST), + }); +} diff --git a/apps/api/src/handlers/brain-inference/helper-answer-json.ts b/apps/api/src/handlers/brain-inference/helper-answer-json.ts new file mode 100644 index 0000000000..d418a89e24 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/helper-answer-json.ts @@ -0,0 +1,47 @@ +import { randomUUID } from 'node:crypto'; + +import type { Context } from 'hono'; + +import type { Variables } from '../../types'; +import { BRAIN_HELPER_MODEL_ID } from './constants'; + +type GatewayContext = Context<{ Variables: Variables }>; + +export function helperChatJson(c: GatewayContext, text: string): Response { + return c.json({ + id: `brain-helper-${randomUUID()}`, + object: 'chat.completion', + created: Math.floor(Date.now() / 1000), + model: BRAIN_HELPER_MODEL_ID, + choices: [ + { + index: 0, + message: { role: 'assistant', content: text }, + finish_reason: 'stop', + }, + ], + usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, + }); +} + +export function helperResponsesJson(c: GatewayContext, text: string): Response { + return c.json({ + id: `resp_${randomUUID()}`, + object: 'response', + created_at: Math.floor(Date.now() / 1000), + status: 'completed', + model: BRAIN_HELPER_MODEL_ID, + error: null, + incomplete_details: null, + output: [ + { + id: `msg_${randomUUID()}`, + type: 'message', + status: 'completed', + role: 'assistant', + content: [{ type: 'output_text', text, annotations: [] }], + }, + ], + usage: { input_tokens: 0, output_tokens: 0, total_tokens: 0 }, + }); +} diff --git a/apps/api/src/handlers/brain-inference/helper-answer.ts b/apps/api/src/handlers/brain-inference/helper-answer.ts new file mode 100644 index 0000000000..ef303592e7 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/helper-answer.ts @@ -0,0 +1,82 @@ +import type { Context } from 'hono'; +import { + generateTrackedNonTaskText, + NON_TASK_INFERENCE_SURFACES, +} from '@roomote/cloud-agents/server/non-task-provider-usage'; +import { formatSingleLineLog } from '@roomote/types'; + +import type { Variables } from '../../types'; +import { + HELPER_SYNTHESIS_DEFAULT_MAX_OUTPUT_TOKENS, + HELPER_SYNTHESIS_TIMEOUT_MS, + LOG_PREFIX, +} from './constants'; +import { helperChatJson, helperResponsesJson } from './helper-answer-json'; +import { jsonResponseInstruction, toHelperPromptParts } from './helper-prompt'; +import { responsesHelperBody } from './helper-responses'; + +type GatewayContext = Context<{ Variables: Variables }>; + +export async function answerWithHelperModel( + c: GatewayContext, + body: Record, + isResponses: boolean, +): Promise { + if (body.stream === true) { + return c.json( + { + error: + 'The Brain helper model does not support streaming. Retry without stream.', + }, + 400, + ); + } + + const helperBody = isResponses ? responsesHelperBody(body) : body; + if (!helperBody) { + return c.json( + { + error: + 'The Brain helper model supports text-only Responses input without tools or previous responses.', + }, + 400, + ); + } + + const { system, prompt } = toHelperPromptParts(helperBody.messages); + const jsonInstruction = jsonResponseInstruction(helperBody.response_format); + const systemWithFormat = [system, jsonInstruction] + .filter((part): part is string => Boolean(part)) + .join('\n\n'); + + try { + const text = await generateTrackedNonTaskText({ + surface: NON_TASK_INFERENCE_SURFACES.brainSynthesis, + modelRole: 'small', + system: systemWithFormat || undefined, + prompt, + maxOutputTokens: helperMaxTokens(helperBody.max_tokens), + timeoutMs: HELPER_SYNTHESIS_TIMEOUT_MS, + }); + + return isResponses ? helperResponsesJson(c, text) : helperChatJson(c, text); + } catch (error) { + const detail = ( + error instanceof Error ? error.message : String(error) + ).replace(/\s+/g, ' '); + + console.warn( + formatSingleLineLog(`${LOG_PREFIX} Helper synthesis failed`, { + error: detail, + }), + ); + + return c.json({ error: 'Brain helper-model synthesis failed' }, 502); + } +} + +function helperMaxTokens(value: unknown): number { + return typeof value === 'number' && Number.isFinite(value) && value > 0 + ? value + : HELPER_SYNTHESIS_DEFAULT_MAX_OUTPUT_TOKENS; +} diff --git a/apps/api/src/handlers/brain-inference/helper-prompt.ts b/apps/api/src/handlers/brain-inference/helper-prompt.ts new file mode 100644 index 0000000000..6c03af14eb --- /dev/null +++ b/apps/api/src/handlers/brain-inference/helper-prompt.ts @@ -0,0 +1,79 @@ +function messageContentText(content: unknown): string { + if (typeof content === 'string') { + return content; + } + + if (Array.isArray(content)) { + return content + .flatMap((part) => { + const record = + part && typeof part === 'object' + ? (part as Record) + : undefined; + + return typeof record?.text === 'string' ? [record.text] : []; + }) + .join('\n'); + } + + return ''; +} + +/** + * Convert an OpenAI chat request into the system/prompt pair + * generateTrackedNonTaskText speaks. + */ +export function toHelperPromptParts(messages: unknown): { + system: string; + prompt: string; +} { + const systemParts: string[] = []; + const promptParts: string[] = []; + + for (const message of Array.isArray(messages) ? messages : []) { + const record = + message && typeof message === 'object' + ? (message as Record) + : undefined; + const role = typeof record?.role === 'string' ? record.role : 'user'; + const text = messageContentText(record?.content); + + if (!text.trim()) { + continue; + } + + if (role === 'system') { + systemParts.push(text); + } else if (role === 'user') { + promptParts.push(text); + } else { + promptParts.push( + `${role.charAt(0).toUpperCase()}${role.slice(1)}: ${text}`, + ); + } + } + + return { + system: systemParts.join('\n\n'), + prompt: promptParts.join('\n\n'), + }; +} + +export function jsonResponseInstruction( + responseFormat: unknown, +): string | null { + const record = + responseFormat && typeof responseFormat === 'object' + ? (responseFormat as Record) + : undefined; + + if (record?.type === 'json_object') { + return 'Respond with only valid JSON. No prose, no code fences.'; + } + + if (record?.type === 'json_schema') { + return `Respond with only valid JSON that conforms to this JSON Schema. No prose, no code fences.\n${JSON.stringify(record.json_schema ?? {})}`; + } + + return null; +} diff --git a/apps/api/src/handlers/brain-inference/helper-responses.ts b/apps/api/src/handlers/brain-inference/helper-responses.ts new file mode 100644 index 0000000000..30b6c85199 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/helper-responses.ts @@ -0,0 +1,77 @@ +/** Translate the native OpenAI Responses text contract into the helper input. */ +export function responsesHelperBody( + body: Record, +): Record | null { + if ( + (Array.isArray(body.tools) && body.tools.length > 0) || + body.previous_response_id + ) { + return null; + } + + if (typeof body.input !== 'string' && !Array.isArray(body.input)) { + return null; + } + + const input = + typeof body.input === 'string' + ? [{ role: 'user', content: body.input }] + : Array.isArray(body.input) + ? body.input + : []; + + if (!input.every(isHelperResponseMessage)) { + return null; + } + + const text = + body.text && typeof body.text === 'object' + ? (body.text as Record) + : undefined; + const format = + text?.format && typeof text.format === 'object' + ? (text.format as Record) + : undefined; + + return { + ...body, + messages: [ + ...(typeof body.instructions === 'string' + ? [{ role: 'system', content: body.instructions }] + : []), + ...input.map((message) => + message && typeof message === 'object' && message.role === 'developer' + ? { ...message, role: 'system' } + : message, + ), + ], + max_tokens: body.max_output_tokens, + response_format: + format?.type === 'json_schema' + ? { type: 'json_schema', json_schema: format } + : format, + }; +} + +function isHelperResponseMessage(message: unknown): boolean { + if (!message || typeof message !== 'object') return false; + const record = message as Record; + if ( + !['system', 'developer', 'user', 'assistant'].includes(String(record.role)) + ) { + return false; + } + if (record.type !== undefined && record.type !== 'message') return false; + if (typeof record.content === 'string') return true; + return ( + Array.isArray(record.content) && + record.content.every((part: unknown) => { + if (!part || typeof part !== 'object') return false; + const value = part as Record; + return ( + ['input_text', 'output_text'].includes(String(value.type)) && + typeof value.text === 'string' + ); + }) + ); +} diff --git a/apps/api/src/handlers/brain-inference/index.ts b/apps/api/src/handlers/brain-inference/index.ts index ed5f828c23..02588346ef 100644 --- a/apps/api/src/handlers/brain-inference/index.ts +++ b/apps/api/src/handlers/brain-inference/index.ts @@ -1,342 +1,27 @@ -import { randomUUID, timingSafeEqual } from 'node:crypto'; - import { Hono } from 'hono'; -import { - generateTrackedNonTaskText, - NON_TASK_INFERENCE_SURFACES, -} from '@roomote/cloud-agents/server/non-task-provider-usage'; -import { Env } from '@roomote/env'; - -import { - formatSingleLineLog, - getInferenceGatewayProvider, -} from '@roomote/types'; -import { - getBrainGatewayToken, - mapBrainModelName, - resolveBrainInferenceProvider, - type ResolvedBrainInference, -} from '@roomote/sdk/server'; +import { formatSingleLineLog } from '@roomote/types'; +import { getBrainGatewayToken } from '@roomote/sdk/server'; import type { Variables } from '../../types'; +import { presentedToken, tokenMatches } from './auth'; +import { + BRAIN_ALLOWED_PATHS, + BRAIN_HELPER_MODEL_ID, + LOG_PREFIX, +} from './constants'; +import { forwardLocalEmbeddings } from './forward-local'; +import { forwardProviderInference } from './forward-provider'; +import { maybeAnswerChatWithHelper } from './maybe-helper'; -const LOG_PREFIX = '[Brain Inference]'; - -/** - * Sentinel chat-model id the Brain requests in gateway mode. It is not a real - * provider model: the gateway answers it itself through the deployment's - * helper ("small") model, which is what lets a Brain synthesize without any - * Brain-specific provider key. An operator's `R_BRAIN_MODEL` still wins — the - * sentinel is only the default the gbrain entrypoint configures. - */ -export const BRAIN_HELPER_MODEL_ID = 'roomote/helper'; - -/** How long a helper-model synthesis call may run before failing the request. */ -const HELPER_SYNTHESIS_TIMEOUT_MS = 120_000; - -/** - * gbrain caps its own synthesis output; when it does not say, stay modest — - * the helper model is a summarizer, not a long-form writer. - */ -const HELPER_SYNTHESIS_DEFAULT_MAX_OUTPUT_TOKENS = 2048; - -/** - * The Brain's whole inference surface: embeddings for recall and chat for - * sourced synthesis and query expansion. Deliberately narrower than the - * task-sandbox gateway's allowlist, because this credential is a static - * deployment secret rather than a short-lived run token. Reranking is not - * part of the Brain: retrieval is hybrid RRF, and the reranker is disabled - * per-brain by the gbrain entrypoint. - */ -const BRAIN_ALLOWED_PATHS = new Set([ - '/v1/embeddings', - '/v1/chat/completions', - '/v1/responses', -]); - -/** Never forwarded upstream: the Brain's gateway token is not a provider key. */ -const REQUEST_HEADER_DENYLIST = new Set([ - 'authorization', - 'x-api-key', - 'cookie', - 'host', - 'connection', - 'content-length', - 'keep-alive', - 'proxy-authenticate', - 'proxy-authorization', - 'te', - 'trailer', - 'transfer-encoding', - 'upgrade', - 'expect', - 'forwarded', - 'x-forwarded-for', - 'x-forwarded-host', - 'x-forwarded-proto', - 'x-real-ip', -]); - -const RESPONSE_HEADER_DENYLIST = new Set([ - 'content-encoding', - 'content-length', - 'transfer-encoding', - 'connection', -]); - -function presentedToken(header: string | undefined): string | null { - if (!header) { - return null; - } - - const match = header.match(/^Bearer\s+(.+)$/i); - - return match?.[1]?.trim() || null; -} - -/** Constant-time compare that does not leak length through early return. */ -function tokenMatches(presented: string, expected: string): boolean { - const a = Buffer.from(presented); - const b = Buffer.from(expected); - - if (a.length !== b.length) { - // Still burn a comparison so a wrong-length guess is not measurably - // faster than a right-length one. - timingSafeEqual(a, a); - return false; - } - - return timingSafeEqual(a, b); -} - -/** - * Rewrite the requested model into the resolved provider's naming. The Brain - * always speaks OpenAI model names because it is configured as an - * OpenAI-compatible client; the deployment may be routing through OpenRouter, - * which namespaces the same models by vendor. - */ -async function rewriteBody( - rawBody: string, - resolved: ResolvedBrainInference, -): Promise { - if (!rawBody) { - return rawBody; - } - - try { - const parsed = JSON.parse(rawBody) as Record; - - if (typeof parsed.model !== 'string') { - return rawBody; - } - - return JSON.stringify({ - ...parsed, - model: mapBrainModelName(parsed.model, resolved), - }); - } catch { - // Not JSON we understand; forward untouched rather than failing the call. - return rawBody; - } -} - -/** - * A self-run inference upstream for one gateway path. Embeddings are the - * Brain's bulk data path (memory text in, vectors out), so they are the one - * a deployment may want on its own hardware; chat synthesis stays with the - * configured model provider. Model names pass through unrewritten — the - * upstream owns its own model registry, and every Brain is locked to its - * embedding model at creation, so the name must mean exactly one thing - * forever. - */ -function resolveLocalUpstream( - upstreamPath: string, -): { baseUrl: string; apiKey?: string } | null { - const baseUrl = - upstreamPath === '/v1/embeddings' - ? Env.R_BRAIN_EMBEDDINGS_UPSTREAM_URL - : undefined; - - if (!baseUrl?.trim()) { - return null; - } - - return { - baseUrl: baseUrl.trim().replace(/\/$/, ''), - apiKey: Env.R_BRAIN_INFERENCE_UPSTREAM_API_KEY?.trim() || undefined, - }; -} - -/** - * Flatten OpenAI-style message content to plain text. Array content keeps its - * text parts (joined) and drops the rest; the helper path is text-only. - */ -function messageContentText(content: unknown): string { - if (typeof content === 'string') { - return content; - } - - if (Array.isArray(content)) { - return content - .flatMap((part) => { - const record = - part && typeof part === 'object' - ? (part as Record) - : undefined; - - return typeof record?.text === 'string' ? [record.text] : []; - }) - .join('\n'); - } - - return ''; -} - -/** - * Convert an OpenAI chat request into the system/prompt pair - * generateTrackedNonTaskText speaks. System messages concatenate into the - * system string; everything else concatenates in order into the prompt, with - * non-user roles labeled so multi-turn context stays attributable. - */ -function toHelperPromptParts(messages: unknown): { - system: string; - prompt: string; -} { - const systemParts: string[] = []; - const promptParts: string[] = []; - - for (const message of Array.isArray(messages) ? messages : []) { - const record = - message && typeof message === 'object' - ? (message as Record) - : undefined; - const role = typeof record?.role === 'string' ? record.role : 'user'; - const text = messageContentText(record?.content); - - if (!text.trim()) { - continue; - } - - if (role === 'system') { - systemParts.push(text); - } else if (role === 'user') { - promptParts.push(text); - } else { - promptParts.push( - `${role.charAt(0).toUpperCase()}${role.slice(1)}: ${text}`, - ); - } - } - - return { - system: systemParts.join('\n\n'), - prompt: promptParts.join('\n\n'), - }; -} - -/** - * gbrain relies on `response_format` for its structured synthesis calls, but - * the helper path runs through a plain-text prompt; translate the contract - * into a strict instruction instead of dropping it silently. - */ -function jsonResponseInstruction(responseFormat: unknown): string | null { - const record = - responseFormat && typeof responseFormat === 'object' - ? (responseFormat as Record) - : undefined; - - if (record?.type === 'json_object') { - return 'Respond with only valid JSON. No prose, no code fences.'; - } - - if (record?.type === 'json_schema') { - return `Respond with only valid JSON that conforms to this JSON Schema. No prose, no code fences.\n${JSON.stringify(record.json_schema ?? {})}`; - } - - return null; -} - -/** Translate the native OpenAI Responses text contract into the helper input. */ -function responsesHelperBody( - body: Record, -): Record | null { - // The helper is stateless and text-only; do not silently drop tool/image - // input or pretend a previous response has been loaded. - if ( - (Array.isArray(body.tools) && body.tools.length > 0) || - body.previous_response_id - ) - return null; - if (typeof body.input !== 'string' && !Array.isArray(body.input)) return null; - const input = - typeof body.input === 'string' - ? [{ role: 'user', content: body.input }] - : Array.isArray(body.input) - ? body.input - : []; - if ( - !input.every((message) => { - if (!message || typeof message !== 'object') return false; - if (!['system', 'developer', 'user', 'assistant'].includes(message.role)) - return false; - if (message.type !== undefined && message.type !== 'message') - return false; - if (typeof message.content === 'string') return true; - return ( - Array.isArray(message.content) && - message.content.every((part: unknown) => { - if (!part || typeof part !== 'object') return false; - const value = part as Record; - return ( - ['input_text', 'output_text'].includes(String(value.type)) && - typeof value.text === 'string' - ); - }) - ); - }) - ) - return null; - const text = - body.text && typeof body.text === 'object' - ? (body.text as Record) - : undefined; - const format = - text?.format && typeof text.format === 'object' - ? (text.format as Record) - : undefined; - return { - ...body, - messages: [ - ...(typeof body.instructions === 'string' - ? [{ role: 'system', content: body.instructions }] - : []), - ...input.map((message) => - message && typeof message === 'object' && message.role === 'developer' - ? { ...message, role: 'system' } - : message, - ), - ], - max_tokens: body.max_output_tokens, - response_format: - format?.type === 'json_schema' - ? { type: 'json_schema', json_schema: format } - : format, - }; -} +export { BRAIN_HELPER_MODEL_ID }; /** * Inference gateway for this deployment's Brain. * * The Brain container holds no provider credential. It is pointed at this * route with a shared gateway token, and Roomote injects whichever provider - * key an admin configured, resolved per request. That is what makes the - * Brain's credential a Settings value rather than a container environment - * variable: rotating or switching it takes effect on the next call, with no - * redeploy and no key ever present in the Brain. - * - * The route authenticates its own caller (policy class `webhook`): the token - * is a deployment secret shared with exactly one sibling service. + * key an admin configured, resolved per request. */ export const brainInference = new Hono<{ Variables: Variables }>(); @@ -344,9 +29,6 @@ brainInference.post('/*', async (c) => { const expectedToken = getBrainGatewayToken(); if (!expectedToken) { - // No token configured means no Brain is wired to this deployment. Closed - // rather than open: an unauthenticated request must never reach a - // provider key. return c.json({ error: 'The Brain inference gateway is not enabled' }, 404); } @@ -372,301 +54,22 @@ brainInference.post('/*', async (c) => { return c.json({ error: 'Path is not allowed through this gateway' }, 403); } - // The helper-model sentinel is answered here, before provider resolution, - // because it exists precisely for deployments with no Brain provider key: - // synthesis rides the deployment's helper model instead. An operator's - // R_BRAIN_MODEL still wins — the sentinel is rewritten to it and forwarded - // through the ordinary provider path below. - let helperOverrideBody: string | undefined; - - const isResponses = upstreamPath === '/v1/responses'; - if (upstreamPath === '/v1/chat/completions' || isResponses) { - let parsedBody: Record | undefined; - - try { - const candidate = JSON.parse(await c.req.text()) as unknown; - - parsedBody = - candidate && typeof candidate === 'object' && !Array.isArray(candidate) - ? (candidate as Record) - : undefined; - } catch { - // Not JSON we understand; the provider path forwards it untouched. - } - - const answerWithHelperModel = async (body: Record) => { - if (body.stream === true) { - // gbrain's gateway chat is non-streaming by design; refuse rather - // than pretend an SSE stream that would never come. - return c.json( - { - error: - 'The Brain helper model does not support streaming. Retry without stream.', - }, - 400, - ); - } - - const helperBody = isResponses ? responsesHelperBody(body) : body; - if (!helperBody) { - return c.json( - { - error: - 'The Brain helper model supports text-only Responses input without tools or previous responses.', - }, - 400, - ); - } - const { system, prompt } = toHelperPromptParts(helperBody.messages); - const jsonInstruction = jsonResponseInstruction( - helperBody.response_format, - ); - const systemWithFormat = [system, jsonInstruction] - .filter((part): part is string => Boolean(part)) - .join('\n\n'); - - try { - const text = await generateTrackedNonTaskText({ - surface: NON_TASK_INFERENCE_SURFACES.brainSynthesis, - modelRole: 'small', - system: systemWithFormat || undefined, - prompt, - maxOutputTokens: - typeof helperBody.max_tokens === 'number' && - Number.isFinite(helperBody.max_tokens) && - helperBody.max_tokens > 0 - ? helperBody.max_tokens - : HELPER_SYNTHESIS_DEFAULT_MAX_OUTPUT_TOKENS, - timeoutMs: HELPER_SYNTHESIS_TIMEOUT_MS, - }); - - if (isResponses) { - return c.json({ - id: `resp_${randomUUID()}`, - object: 'response', - created_at: Math.floor(Date.now() / 1000), - status: 'completed', - model: BRAIN_HELPER_MODEL_ID, - error: null, - incomplete_details: null, - output: [ - { - id: `msg_${randomUUID()}`, - type: 'message', - status: 'completed', - role: 'assistant', - content: [{ type: 'output_text', text, annotations: [] }], - }, - ], - // Actual usage is recorded by the tracked helper call above. - usage: { input_tokens: 0, output_tokens: 0, total_tokens: 0 }, - }); - } - return c.json({ - id: `brain-helper-${randomUUID()}`, - object: 'chat.completion', - created: Math.floor(Date.now() / 1000), - model: BRAIN_HELPER_MODEL_ID, - choices: [ - { - index: 0, - message: { role: 'assistant', content: text }, - finish_reason: 'stop', - }, - ], - // Advisory only: gbrain logs usage but never bills from it, and - // the real usage is already recorded by the tracked call above. - usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, - }); - } catch (error) { - const detail = ( - error instanceof Error ? error.message : String(error) - ).replace(/\s+/g, ' '); - - console.warn( - formatSingleLineLog(`${LOG_PREFIX} Helper synthesis failed`, { - error: detail, - }), - ); - - return c.json({ error: 'Brain helper-model synthesis failed' }, 502); - } - }; - - if (parsedBody?.model === BRAIN_HELPER_MODEL_ID) { - const overrideModel = Env.R_BRAIN_MODEL?.trim(); - - if (overrideModel) { - helperOverrideBody = JSON.stringify({ - ...parsedBody, - model: overrideModel, - }); - } else { - return answerWithHelperModel(parsedBody); - } - } else if (parsedBody && !(await resolveBrainInferenceProvider())) { - // No Brain provider key configured. gbrain's expansion and chat send - // concrete model ids, not the sentinel, so without this they would - // 503 on the provider path below — instead every chat request rides - // the deployment's helper model, exactly like the sentinel. A - // Brain-specific key, when configured, still takes this path's place - // so an operator can bill Memory inference separately. - return answerWithHelperModel(parsedBody); - } - } - - const localUpstream = resolveLocalUpstream(upstreamPath); - - if (localUpstream) { - const headers = new Headers(); - - // Same denylist as the provider path: the gateway token in - // `authorization` must never reach any upstream. - for (const [name, value] of c.req.raw.headers.entries()) { - if (!REQUEST_HEADER_DENYLIST.has(name.toLowerCase())) { - headers.set(name, value); - } - } - - if (localUpstream.apiKey) { - headers.set('authorization', `Bearer ${localUpstream.apiKey}`); - } - - let upstream: Response; + const helperResult = await maybeAnswerChatWithHelper(c, upstreamPath); - try { - upstream = await fetch( - `${localUpstream.baseUrl}${upstreamPath}${requestUrl.search}`, - { method: 'POST', headers, body: await c.req.text() }, - ); - } catch (error) { - console.warn( - formatSingleLineLog(`${LOG_PREFIX} Local upstream request failed`, { - upstreamPath, - error: error instanceof Error ? error.message : String(error), - }), - ); - - return c.json({ error: 'Brain inference upstream is unreachable' }, 502); - } - - if (!upstream.ok) { - console.warn( - formatSingleLineLog(`${LOG_PREFIX} Local upstream returned an error`, { - upstreamPath, - status: upstream.status, - }), - ); - } - - const responseHeaders = new Headers(); - - for (const [name, value] of upstream.headers.entries()) { - if (!RESPONSE_HEADER_DENYLIST.has(name.toLowerCase())) { - responseHeaders.set(name, value); - } - } - - return new Response(upstream.body, { - status: upstream.status, - headers: responseHeaders, - }); + if (helperResult instanceof Response) { + return helperResult; } - const resolved = await resolveBrainInferenceProvider(); + const local = await forwardLocalEmbeddings(c, upstreamPath, requestUrl); - if (!resolved) { - // A deployed but unconfigured Brain lands here. 503 rather than 500: the - // deployment is healthy, it just has no model provider configured yet. - return c.json( - { - error: - 'No model provider is configured for this deployment. Add a provider key in Settings to enable the Brain.', - }, - 503, - ); + if (local) { + return local; } - const provider = getInferenceGatewayProvider(resolved.providerId); - - if (!provider?.authHeader) { - console.error( - formatSingleLineLog(`${LOG_PREFIX} Provider is not proxyable`, { - providerId: resolved.providerId, - }), - ); - - return c.json({ error: 'Brain inference provider is unavailable' }, 500); - } - - const headers = new Headers(); - - for (const [name, value] of c.req.raw.headers.entries()) { - if (!REQUEST_HEADER_DENYLIST.has(name.toLowerCase())) { - headers.set(name, value); - } - } - - headers.set( - provider.authHeader.name, - provider.authHeader.scheme === 'bearer' - ? `Bearer ${resolved.apiKey}` - : resolved.apiKey, + return forwardProviderInference( + c, + upstreamPath, + requestUrl, + helperResult?.helperOverrideBody, ); - - const body = await rewriteBody( - helperOverrideBody ?? (await c.req.text()), - resolved, - ); - const startedAt = Date.now(); - - let upstream: Response; - - try { - upstream = await fetch( - // Carry the query string: none of today's allowed paths use one, but - // dropping it silently is the kind of thing that only surfaces as a - // confusing upstream error much later (Azure's ?api-version=). - `${provider.upstreamBaseUrl}${upstreamPath}${requestUrl.search}`, - { - method: 'POST', - headers, - body, - }, - ); - } catch (error) { - console.warn( - formatSingleLineLog(`${LOG_PREFIX} Upstream request failed`, { - providerId: resolved.providerId, - upstreamPath, - error: error instanceof Error ? error.message : String(error), - }), - ); - - return c.json({ error: 'Brain inference upstream is unreachable' }, 502); - } - - if (!upstream.ok) { - console.warn( - formatSingleLineLog(`${LOG_PREFIX} Upstream returned an error`, { - providerId: resolved.providerId, - upstreamPath, - status: upstream.status, - durationMs: Date.now() - startedAt, - }), - ); - } - - const responseHeaders = new Headers(); - - for (const [name, value] of upstream.headers.entries()) { - if (!RESPONSE_HEADER_DENYLIST.has(name.toLowerCase())) { - responseHeaders.set(name, value); - } - } - - return new Response(upstream.body, { - status: upstream.status, - headers: responseHeaders, - }); }); diff --git a/apps/api/src/handlers/brain-inference/maybe-helper.ts b/apps/api/src/handlers/brain-inference/maybe-helper.ts new file mode 100644 index 0000000000..7a6d991548 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/maybe-helper.ts @@ -0,0 +1,52 @@ +import type { Context } from 'hono'; +import { Env } from '@roomote/env'; +import { resolveBrainInferenceProvider } from '@roomote/sdk/server'; + +import type { Variables } from '../../types'; +import { BRAIN_HELPER_MODEL_ID } from './constants'; +import { answerWithHelperModel } from './helper-answer'; + +type GatewayContext = Context<{ Variables: Variables }>; + +export async function maybeAnswerChatWithHelper( + c: GatewayContext, + upstreamPath: string, +): Promise { + const isResponses = upstreamPath === '/v1/responses'; + if (upstreamPath !== '/v1/chat/completions' && !isResponses) { + return null; + } + + let parsedBody: Record | undefined; + + try { + const candidate = JSON.parse(await c.req.text()) as unknown; + parsedBody = + candidate && typeof candidate === 'object' && !Array.isArray(candidate) + ? (candidate as Record) + : undefined; + } catch { + // Not JSON we understand; the provider path forwards it untouched. + } + + if (parsedBody?.model === BRAIN_HELPER_MODEL_ID) { + const overrideModel = Env.R_BRAIN_MODEL?.trim(); + + if (overrideModel) { + return { + helperOverrideBody: JSON.stringify({ + ...parsedBody, + model: overrideModel, + }), + }; + } + + return answerWithHelperModel(c, parsedBody, isResponses); + } + + if (parsedBody && !(await resolveBrainInferenceProvider())) { + return answerWithHelperModel(c, parsedBody, isResponses); + } + + return {}; +} diff --git a/apps/api/src/handlers/brain-inference/provider-headers.ts b/apps/api/src/handlers/brain-inference/provider-headers.ts new file mode 100644 index 0000000000..75012ab95d --- /dev/null +++ b/apps/api/src/handlers/brain-inference/provider-headers.ts @@ -0,0 +1,19 @@ +import type { ResolvedBrainInference } from '@roomote/sdk/server'; + +import { REQUEST_HEADER_DENYLIST } from './constants'; +import { copyAllowedHeaders } from './copy-headers'; + +export function providerHeaders( + requestHeaders: Headers, + resolved: ResolvedBrainInference, + provider: { authHeader: { name: string; scheme?: string } }, +): Headers { + const headers = copyAllowedHeaders(requestHeaders, REQUEST_HEADER_DENYLIST); + headers.set( + provider.authHeader.name, + provider.authHeader.scheme === 'bearer' + ? `Bearer ${resolved.apiKey}` + : resolved.apiKey, + ); + return headers; +} diff --git a/apps/api/src/handlers/brain-inference/rewrite-body.ts b/apps/api/src/handlers/brain-inference/rewrite-body.ts new file mode 100644 index 0000000000..8f31d0b314 --- /dev/null +++ b/apps/api/src/handlers/brain-inference/rewrite-body.ts @@ -0,0 +1,33 @@ +import { + mapBrainModelName, + type ResolvedBrainInference, +} from '@roomote/sdk/server'; + +/** + * Rewrite the requested model into the resolved provider's naming. The Brain + * always speaks OpenAI model names; the deployment may route through + * OpenRouter, which namespaces the same models by vendor. + */ +export async function rewriteBody( + rawBody: string, + resolved: ResolvedBrainInference, +): Promise { + if (!rawBody) { + return rawBody; + } + + try { + const parsed = JSON.parse(rawBody) as Record; + + if (typeof parsed.model !== 'string') { + return rawBody; + } + + return JSON.stringify({ + ...parsed, + model: mapBrainModelName(parsed.model, resolved), + }); + } catch { + return rawBody; + } +} diff --git a/apps/docs/environment-variables.mdx b/apps/docs/environment-variables.mdx index e53e4c83b3..fce5b3488d 100644 --- a/apps/docs/environment-variables.mdx +++ b/apps/docs/environment-variables.mdx @@ -253,8 +253,10 @@ manifest changes. | `R_BRAIN_MODEL` | Optional | Memory synthesis model in the configured provider's naming. Changes apply immediately. | | `R_BRAIN_EMBEDDING_MODEL` | Before first Memory boot | Embedding model id that sizes Memory's vector storage. Changing it later requires re-embedding. | | `R_BRAIN_EMBEDDING_DIMENSIONS` | Before first Memory boot | Output width for `R_BRAIN_EMBEDDING_MODEL`; it must match the served model. | -| `R_BRAIN_EMBEDDINGS_UPSTREAM_URL` | Optional | OpenAI-compatible embeddings endpoint used instead of the Memory provider. | +| `R_BRAIN_EMBEDDINGS_UPSTREAM_URL` | Optional | OpenAI-compatible embeddings endpoint used instead of the Memory provider. Wins over Cloudflare Workers AI auto-wire. | | `R_BRAIN_INFERENCE_UPSTREAM_API_KEY` | Optional | Bearer key for the self-run embeddings upstream; omit it for a trusted private-network service. | +| `CLOUDFLARE_WORKERS_AI_API_TOKEN` | Optional | Workers AI token used for Memory embeddings when no embeddings upstream URL and no OpenAI/OpenRouter Memory key is set. | +| `CLOUDFLARE_WORKERS_AI_ACCOUNT_ID` | Optional | 32-hex Cloudflare account id paired with the Workers AI token. | ### Sandbox providers diff --git a/apps/docs/memory.mdx b/apps/docs/memory.mdx index 969b3cc33b..2b80eaf556 100644 --- a/apps/docs/memory.mdx +++ b/apps/docs/memory.mdx @@ -140,6 +140,25 @@ with no redeploy. OpenRouter and OpenAI both support memory embedding calls. +### Cloudflare Workers AI + +Deployments that already have a Workers AI token and no OpenAI or OpenRouter +Memory key can use it for Memory embeddings. Set +`CLOUDFLARE_WORKERS_AI_API_TOKEN` and `CLOUDFLARE_WORKERS_AI_ACCOUNT_ID`. +Roomote forwards embeddings to Workers AI's OpenAI-compatible endpoint. An +existing OpenAI-model Memory keeps its provider path; set +`R_BRAIN_EMBEDDINGS_UPSTREAM_URL` to the `/ai` base to opt in. Set the +embedding model and dimensions before Memory's first boot: + +```sh +R_BRAIN_EMBEDDING_MODEL=@cf/baai/bge-m3 +R_BRAIN_EMBEDDING_DIMENSIONS=1024 +``` + +Leave `R_BRAIN_EMBEDDINGS_UPSTREAM_URL` unset unless you want to override +that auto-wired endpoint. An explicit upstream URL must be the Workers AI +`/ai` base; Roomote appends `/v1/embeddings`. + ### Run embeddings locally Self-hosted Compose deployments can keep embeddings on their own hardware @@ -212,10 +231,10 @@ pages written over the last 30 days and the memories written most recently. source. Selecting a new memory opens it in the browser and updates the page URL without adding browser history. -**Status** reports whether recall is semantic or keyword-only and which provider -is serving its embeddings and synthesis. Memory that is running but has no -provider key is called out as needing attention rather than shown as healthy: -without one it can only match keywords, so recall would look real while missing +**Status** reports whether recall is semantic or keyword-only and which path is +serving embeddings and synthesis. Memory that is running but has no embeddings +path is called out as needing attention rather than shown as healthy: without +embeddings it can only match keywords, so recall would look real while missing everything semantic. **Sources** shows connected sources with their current collection state, when @@ -312,10 +331,11 @@ it takes effect on the next request with nothing to restart. ## If you turned memory on later -Memory that first boots without a provider key is created with semantic +Memory that first boots without an embeddings path is created with semantic recall switched off, because the embedding model sizes its vector storage at -creation time. Adding a key later is still fine: memory notices on its next -start, enables semantic recall, and embeds whatever it already holds. It logs +creation time. Adding a provider key, Cloudflare Workers AI, or an embeddings +upstream later is still fine: memory notices on its next start, enables +semantic recall, and embeds whatever it already holds. It logs `semantic recall enabled` when it does. That repair runs once and is safe, since memory in this state has no diff --git a/apps/web/src/trpc/commands/brain/brain-corpus-types.ts b/apps/web/src/trpc/commands/brain/brain-corpus-types.ts new file mode 100644 index 0000000000..bd3211c984 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-corpus-types.ts @@ -0,0 +1,24 @@ +import type { BrainNamespaceBucketId } from '@roomote/types'; + +export type BrainNamespaceSummary = { + id: BrainNamespaceBucketId; + label: string; + pages: number; +}; + +export type BrainCorpusSummary = { + reachable: boolean; + /** Pages in the exhaustive cached corpus listing. */ + listedPages: number; + /** + * The corpus's exact page count from gbrain's admin census, or null when + * the admin API did not answer. + */ + totalPages: number | null; + namespaces: BrainNamespaceSummary[]; + /** + * Pages written per UTC day over the trailing window, oldest first, + * zero-filled. Computed from the exhaustive corpus listing. + */ + activityByDay: Array<{ date: string; pages: number }>; +}; diff --git a/apps/web/src/trpc/commands/brain/brain-credentials.ts b/apps/web/src/trpc/commands/brain/brain-credentials.ts new file mode 100644 index 0000000000..81c79c965c --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-credentials.ts @@ -0,0 +1,23 @@ +import { and, db, eq, isNull, mcpConnections } from '@roomote/db/server'; +import { BRAIN_MCP_ID, isMcpConnectionGbrainConfig } from '@roomote/types'; +import { Env } from '@/lib/server/env'; + +/** + * Whether Roomote holds usable Brain credentials, read without minting or + * provisioning anything. Deliberately not `resolveBrainConnection`: the + * corpus read already takes that path once per request. + */ +export async function hasBrainCredentials(): Promise { + if (Env.R_GBRAIN_AGENT_TOKEN) { + return true; + } + + const stored = await db.query.mcpConnections.findFirst({ + where: and( + eq(mcpConnections.mcpId, BRAIN_MCP_ID), + isNull(mcpConnections.userId), + ), + }); + + return isMcpConnectionGbrainConfig(stored?.authConfig); +} diff --git a/apps/web/src/trpc/commands/brain/brain-settings-status.test.ts b/apps/web/src/trpc/commands/brain/brain-settings-status.test.ts new file mode 100644 index 0000000000..83b2e2abe7 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-settings-status.test.ts @@ -0,0 +1,38 @@ +import { resolveBrainSettingsStatus } from './brain-settings-status'; + +const ready = { + configured: true, + url: 'http://gbrain:8931', + embeddingsAvailable: true, + corpusReachable: true, + provisioned: true, +}; + +describe('resolveBrainSettingsStatus', () => { + it('is incomplete without embeddings, even when Memory is on', () => { + const result = resolveBrainSettingsStatus({ + ...ready, + embeddingsAvailable: false, + corpusReachable: false, + }); + + expect(result.status).toBe('incomplete'); + expect(result.statusDetail).toMatch(/no embeddings provider/); + }); + + it('does not warn about embeddings when Cloudflare or another embedder is ready', () => { + const result = resolveBrainSettingsStatus({ + ...ready, + embeddingsAvailable: true, + corpusReachable: true, + }); + + expect(result).toEqual({ status: 'connected', statusDetail: null }); + }); + + it('is not_configured when Memory is off', () => { + expect( + resolveBrainSettingsStatus({ ...ready, configured: false }).status, + ).toBe('not_configured'); + }); +}); diff --git a/apps/web/src/trpc/commands/brain/brain-settings-status.ts b/apps/web/src/trpc/commands/brain/brain-settings-status.ts new file mode 100644 index 0000000000..281d674861 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-settings-status.ts @@ -0,0 +1,72 @@ +/** + * - `connected`: a Brain is configured, provisioned, and answering. + * - `unreachable`: configured, but the corpus did not answer. Ingestion is + * holding its checkpoints; this is an outage, not data loss. + * - `incomplete`: half-configured. A Brain without embeddings still answers + * keyword queries, which looks real while missing everything semantic, so + * it is reported as a fault rather than as a working Brain. + * - `not_configured`: this deployment has no Brain. + */ +export type BrainStatus = + | 'connected' + | 'unreachable' + | 'incomplete' + | 'not_configured'; + +type BrainSettingsStatusInput = { + configured: boolean; + url: string | null; + embeddingsAvailable: boolean; + corpusReachable: boolean; + provisioned: boolean; +}; + +const EMBEDDINGS_DETAIL = + 'Memory has no embeddings provider, so it can only match keywords. Configure a provider key, Cloudflare Workers AI, or an embeddings upstream to enable semantic recall.'; + +/** + * Settings status for Memory. Semantic recall needs embeddings from a + * provider key, a configured upstream, or Cloudflare Workers AI. + */ +export function resolveBrainSettingsStatus(input: BrainSettingsStatusInput): { + status: BrainStatus; + statusDetail: string | null; +} { + if (!input.configured) { + return { + status: 'not_configured', + statusDetail: + 'Memory is turned off for this deployment. Enable it to give agents shared memory.', + }; + } + + if (!input.url) { + return { + status: 'incomplete', + statusDetail: + 'A Memory provider key is set, but no Memory service URL is configured, so there is nowhere to store memories.', + }; + } + + if (!input.embeddingsAvailable) { + return { status: 'incomplete', statusDetail: EMBEDDINGS_DETAIL }; + } + + if (input.corpusReachable) { + return { status: 'connected', statusDetail: null }; + } + + if (!input.provisioned) { + return { + status: 'incomplete', + statusDetail: + 'Roomote has not been able to provision its Memory credentials yet. This resolves on its own once Memory has started.', + }; + } + + return { + status: 'unreachable', + statusDetail: + 'Memory did not answer. Ingestion holds its position while it is down, so nothing is lost.', + }; +} diff --git a/apps/web/src/trpc/commands/brain/brain-settings-types.ts b/apps/web/src/trpc/commands/brain/brain-settings-types.ts new file mode 100644 index 0000000000..18d781d7be --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-settings-types.ts @@ -0,0 +1,50 @@ +import type { BrainCorpusSummary } from './brain-corpus-types'; +import type { BrainSourceSummary } from './brain-source-types'; +import type { BrainStatus } from './brain-settings-status'; + +export type BrainSettings = { + status: BrainStatus; + /** Why the status is not `connected`, in one sentence, or null. */ + statusDetail: string | null; + /** Effective Brain on/off state, as the Settings toggle should render it. */ + enabled: boolean; + /** + * True when `enabled` comes from the legacy activation signal (an explicit + * R_BRAIN_* provider key) rather than an explicit Settings choice. + */ + enabledFromLegacyKey: boolean; + url: string | null; + inferenceProvider: 'openrouter' | 'openai' | 'helper' | null; + /** + * Whether the serving key is the Brain's own (`R_BRAIN_*`) or the + * deployment's general provider key. + */ + keySource: 'brain' | 'deployment' | null; + /** + * Recall health. `semantic`/`keyword-only` are measured from gbrain's own + * embedding counts; `unknown` means the admin census did not answer. + */ + recall: { + mode: 'semantic' | 'keyword-only' | 'unknown'; + embeddedCount: number | null; + chunkCount: number | null; + }; + /** + * Whether an embedding path exists at all (a dedicated embedder upstream + * or an embeddings-capable provider key). + */ + embeddingsAvailable: boolean; + corpus: BrainCorpusSummary; + sources: BrainSourceSummary[]; + taskMemories: { + byStatus: Record< + 'pending' | 'processing' | 'done' | 'skipped' | 'failed', + number + >; + total: number; + lastProcessedAt: Date | null; + lastError: string | null; + historicalCompletedRunsWithoutEvent: number; + recentCompletedRunsWithoutEvent: number; + }; +}; diff --git a/apps/web/src/trpc/commands/brain/brain-source-types.ts b/apps/web/src/trpc/commands/brain/brain-source-types.ts new file mode 100644 index 0000000000..c67e0bfc74 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-source-types.ts @@ -0,0 +1,23 @@ +import type { BrainSourceId } from '@roomote/types'; + +export type BrainSourceStatus = + | 'ingesting' + | 'backfilling' + | 'idle' + | 'not_connected'; + +export type BrainSourceSummary = { + id: BrainSourceId; + label: string; + description: string; + namespaceLabel: string; + status: BrainSourceStatus; + /** Newest checkpoint across the source's partitions. */ + lastSyncedAt: Date | null; + /** Live partitions (channels, repositories, workspaces) being read. */ + streams: number; + /** Deep-replay progress while `backfilling`, or null when unknowable. */ + backfillProgress: { read: number; total: number } | null; + /** Upstream objects this source is tracking (pages it owns and refreshes). */ + trackedItems: number; +}; diff --git a/apps/web/src/trpc/commands/brain/brain-task-memories.ts b/apps/web/src/trpc/commands/brain/brain-task-memories.ts new file mode 100644 index 0000000000..565decd113 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/brain-task-memories.ts @@ -0,0 +1,28 @@ +import { + backfillBrainMemoryEvents, + db, + requeueFailedBrainMemoryEvents, +} from '@roomote/db/server'; +import type { UserAuthSuccess } from '@/types'; + +import { assertAdmin } from '../setup/shared'; + +/** + * Enqueue a memory for every completed run that does not have one. + * Idempotent by the outbox's unique(runId). + */ +export async function backfillBrainTaskMemoriesCommand( + auth: UserAuthSuccess, +): Promise<{ queued: number }> { + assertAdmin(auth); + + return { queued: await backfillBrainMemoryEvents(db) }; +} + +export async function retryFailedBrainTaskMemoriesCommand( + auth: UserAuthSuccess, +): Promise<{ requeued: number }> { + assertAdmin(auth); + + return { requeued: await requeueFailedBrainMemoryEvents(db) }; +} diff --git a/apps/web/src/trpc/commands/brain/get-brain-page.ts b/apps/web/src/trpc/commands/brain/get-brain-page.ts new file mode 100644 index 0000000000..12a4c4bb66 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/get-brain-page.ts @@ -0,0 +1,49 @@ +import { readBrainPage } from '@roomote/sdk/server'; +import type { UserAuthSuccess } from '@/types'; + +import { assertAdmin } from '../setup/shared'; + +/** + * How much of a page body the preview ships. A Brain page can be an entire + * day of channel history; the dialog is a reading pane, not an export. + */ +const PAGE_CONTENT_PREVIEW_LIMIT = 32_000; + +type BrainPageContent = { + slug: string; + title: string; + updatedAt: Date | null; + /** + * The page body as the Brain stores it. Untrusted content distilled from + * tasks and integrations: the client renders it as plain text. + */ + content: string | null; + /** The body exceeded the preview bound and was cut. */ + contentTruncated: boolean; +}; + +export async function getBrainPageCommand( + auth: UserAuthSuccess, + input: { slug: string }, +): Promise { + assertAdmin(auth); + + const page = await readBrainPage(input.slug); + + if (!page) { + return null; + } + + const contentTruncated = + (page.content?.length ?? 0) > PAGE_CONTENT_PREVIEW_LIMIT; + + return { + slug: page.slug, + title: page.title ?? page.slug, + updatedAt: page.updatedAt, + content: contentTruncated + ? page.content!.slice(0, PAGE_CONTENT_PREVIEW_LIMIT) + : page.content, + contentTruncated, + }; +} diff --git a/apps/web/src/trpc/commands/brain/get-brain-settings.ts b/apps/web/src/trpc/commands/brain/get-brain-settings.ts new file mode 100644 index 0000000000..c4adabac16 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/get-brain-settings.ts @@ -0,0 +1,78 @@ +import { + resolveBrainEnabledState, + resolveModelProviderEnvValue, +} from '@roomote/db/server'; +import type { UserAuthSuccess } from '@/types'; +import { Env } from '@/lib/server/env'; + +import { assertAdmin } from '../setup/shared'; +import { hasBrainCredentials } from './brain-credentials'; +import type { BrainSettings } from './brain-settings-types'; +import { resolveBrainSettingsStatus } from './brain-settings-status'; +import { loadBrainSettings } from './load-brain-settings'; +import { summarizeSources } from './summarize-sources'; + +export async function getBrainSettingsCommand( + auth: UserAuthSuccess, +): Promise { + assertAdmin(auth); + + const enabledState = await resolveBrainEnabledState(); + const configured = enabledState.enabled; + const url = Env.R_GBRAIN_URL ?? null; + const loaded = await loadBrainSettings(configured); + const provisioned = + configured && !loaded.corpus.reachable ? await hasBrainCredentials() : true; + const keySource: BrainSettings['keySource'] = !loaded.inference + ? null + : ( + await resolveModelProviderEnvValue([ + loaded.inference.providerId === 'openrouter' + ? 'R_BRAIN_OPENROUTER_API_KEY' + : 'R_BRAIN_OPENAI_API_KEY', + ]) + )?.trim() + ? 'brain' + : 'deployment'; + const memoryTotal = Object.values(loaded.memories.byStatus).reduce( + (total, value) => total + value, + 0, + ); + const { status, statusDetail } = resolveBrainSettingsStatus({ + configured, + url, + embeddingsAvailable: loaded.embeddingsAvailable, + corpusReachable: loaded.corpus.reachable, + provisioned, + }); + + return { + status, + statusDetail, + enabled: enabledState.enabled, + enabledFromLegacyKey: enabledState.enabled && enabledState.fromLegacyKey, + url, + inferenceProvider: loaded.inference?.providerId ?? 'helper', + embeddingsAvailable: loaded.embeddingsAvailable, + keySource, + recall: loaded.recall, + corpus: loaded.corpus, + sources: summarizeSources({ + syncStates: loaded.syncStates, + itemCounts: loaded.itemCounts, + requirements: loaded.requirements, + taskMemoriesActive: memoryTotal > 0, + taskMemoriesLastProcessedAt: loaded.memories.lastProcessedAt, + }), + taskMemories: { + byStatus: loaded.memories.byStatus, + total: memoryTotal, + lastProcessedAt: loaded.memories.lastProcessedAt, + lastError: loaded.memories.lastError, + historicalCompletedRunsWithoutEvent: + loaded.memories.historicalCompletedRunsWithoutEvent, + recentCompletedRunsWithoutEvent: + loaded.memories.recentCompletedRunsWithoutEvent, + }, + }; +} diff --git a/apps/web/src/trpc/commands/brain/index.ts b/apps/web/src/trpc/commands/brain/index.ts index 9d03704bfa..ecf9ff17f2 100644 --- a/apps/web/src/trpc/commands/brain/index.ts +++ b/apps/web/src/trpc/commands/brain/index.ts @@ -1,721 +1,22 @@ -import { - and, - backfillBrainMemoryEvents, - countBrainCollectorItemsByCollector, - db, - eq, - getBrainMemoryEventSummary, - isNull, - resolveBrainEnabledState, - resolveModelProviderEnvValue, - listBrainSyncStates, - mcpConnections, - requeueFailedBrainMemoryEvents, - setBrainEnabled, -} from '@roomote/db/server'; -import { - isBrainEmbeddingAvailable, - readBrainCorpus, - readBrainPage, - readBrainStats, - requestBrainBackfill, - resolveBrainSourceRequirements, - resolveBrainInferenceProvider, - type BrainCorpusSnapshot, -} from '@roomote/sdk/server'; -import { - BRAIN_MCP_ID, - BRAIN_SOURCES, - BRAIN_OTHER_NAMESPACE_ID, - brainNamespaceLabel, - resolveBrainNamespaceId, - isMcpConnectionGbrainConfig, - parseBrainBackfillCompletedCount, - resolveBrainSourceIdForCollector, - resolveBrainSourceIdForCurrentCollector, - type BrainNamespaceBucketId, - type BrainSourceId, - type BrainSourceRequirement, -} from '@roomote/types'; - -import type { UserAuthSuccess } from '@/types'; -import { Env } from '@/lib/server/env'; - -import { assertAdmin } from '../setup/shared'; - -const EMPTY_MEMORY_SUMMARY: Awaited< - ReturnType -> = { - byStatus: { pending: 0, processing: 0, done: 0, skipped: 0, failed: 0 }, - lastProcessedAt: null, - lastError: null, - historicalCompletedRunsWithoutEvent: 0, - recentCompletedRunsWithoutEvent: 0, -}; - -/** - * - `connected`: a Brain is configured, provisioned, and answering. - * - `unreachable`: configured, but the corpus did not answer. Ingestion is - * holding its checkpoints; this is an outage, not data loss. - * - `incomplete`: half-configured. A Brain container without an inference - * provider still answers keyword queries, which is worse than absent - * because recall looks real while missing everything semantic — so it is - * reported as a fault rather than as a working Brain. - * - `not_configured`: this deployment has no Brain. - */ -export type BrainStatus = - | 'connected' - | 'unreachable' - | 'incomplete' - | 'not_configured'; - -export type BrainSourceStatus = - | 'ingesting' - | 'backfilling' - | 'idle' - | 'not_connected'; - -export type BrainSourceSummary = { - id: BrainSourceId; - label: string; - description: string; - namespaceLabel: string; - status: BrainSourceStatus; - /** Newest checkpoint across the source's partitions. */ - lastSyncedAt: Date | null; - /** Live partitions (channels, repositories, workspaces) being read. */ - streams: number; - /** Deep-replay progress while `backfilling`, or null when unknowable. */ - backfillProgress: { read: number; total: number } | null; - /** Upstream objects this source is tracking (pages it owns and refreshes). */ - trackedItems: number; -}; - -export type BrainNamespaceSummary = { - id: BrainNamespaceBucketId; - label: string; - pages: number; -}; - -export type BrainCorpusSummary = { - reachable: boolean; - /** Pages in the exhaustive cached corpus listing. */ - listedPages: number; - /** - * The corpus's exact page count from gbrain's admin census, or null when - * the admin API did not answer. - */ - totalPages: number | null; - namespaces: BrainNamespaceSummary[]; - /** - * Pages written per UTC day over the trailing window, oldest first, - * zero-filled. Computed from the exhaustive corpus listing. - */ - activityByDay: Array<{ date: string; pages: number }>; -}; - -export type BrainSettings = { - status: BrainStatus; - /** Why the status is not `connected`, in one sentence, or null. */ - statusDetail: string | null; - /** Effective Brain on/off state, as the Settings toggle should render it. */ - enabled: boolean; - /** - * True when `enabled` comes from the legacy activation signal (an explicit - * R_BRAIN_* provider key) rather than an explicit Settings choice. - */ - enabledFromLegacyKey: boolean; - url: string | null; - inferenceProvider: 'openrouter' | 'openai' | 'helper' | null; - /** - * Whether the serving key is the Brain's own (`R_BRAIN_*`) or the - * deployment's general provider key. The provider preference order can - * legitimately pick a general key even when a brain-specific one exists - * for the other provider, and the page must not claim otherwise. - */ - keySource: 'brain' | 'deployment' | null; - /** - * Recall health. `semantic`/`keyword-only` are measured from gbrain's own - * embedding counts; `unknown` means the admin census did not answer and - * the UI falls back to inferring from provider presence. - */ - recall: { - mode: 'semantic' | 'keyword-only' | 'unknown'; - embeddedCount: number | null; - chunkCount: number | null; - }; - /** - * Whether an embedding path exists at all (a dedicated embedder upstream - * or an embeddings-capable provider key). What the `unknown` recall mode - * falls back to: the helper model answers synthesis but cannot embed, so - * inference presence alone must not imply semantic recall. - */ - embeddingsAvailable: boolean; - corpus: BrainCorpusSummary; - sources: BrainSourceSummary[]; - taskMemories: { - byStatus: Record< - 'pending' | 'processing' | 'done' | 'skipped' | 'failed', - number - >; - total: number; - lastProcessedAt: Date | null; - lastError: string | null; - historicalCompletedRunsWithoutEvent: number; - recentCompletedRunsWithoutEvent: number; - }; -}; - -/** Days of page-writing activity summarized for the ingestion chart. */ -const ACTIVITY_WINDOW_DAYS = 30; - -function buildActivityByDay( - pages: Array<{ updatedAt: Date | null }>, -): Array<{ date: string; pages: number }> { - const counts = new Map(); - const today = new Date(); - - for (let offset = ACTIVITY_WINDOW_DAYS - 1; offset >= 0; offset--) { - const day = new Date(today); - day.setUTCDate(day.getUTCDate() - offset); - counts.set(day.toISOString().slice(0, 10), 0); - } - - for (const page of pages) { - if (!page.updatedAt) { - continue; - } - - const key = page.updatedAt.toISOString().slice(0, 10); - - if (counts.has(key)) { - counts.set(key, counts.get(key)! + 1); - } - } - - return [...counts.entries()].map(([date, count]) => ({ - date, - pages: count, - })); -} - -function summarizeCorpus( - snapshot: Awaited>, -): BrainCorpusSummary { - if (!snapshot) { - return { - reachable: false, - listedPages: 0, - totalPages: null, - namespaces: [], - activityByDay: [], - }; - } - - const counts = new Map(); - - for (const page of snapshot.pages) { - const namespaceId = resolveBrainNamespaceId(page.slug); - counts.set(namespaceId, (counts.get(namespaceId) ?? 0) + 1); - } - - const namespaces = [...counts.entries()] - .map(([id, pages]) => ({ id, label: brainNamespaceLabel(id), pages })) - // Largest first, with the catch-all bucket last however big it is: it is - // the least informative slice and should never lead the chart. - .sort((left, right) => { - if (left.id === BRAIN_OTHER_NAMESPACE_ID) return 1; - if (right.id === BRAIN_OTHER_NAMESPACE_ID) return -1; - return right.pages - left.pages || left.label.localeCompare(right.label); - }); - - return { - reachable: true, - listedPages: snapshot.pages.length, - totalPages: null, - namespaces, - activityByDay: buildActivityByDay(snapshot.pages), - }; -} - -/** Exported for unit tests; the command below is its only runtime caller. */ -export function summarizeSources(input: { - syncStates: Awaited>; - itemCounts: Awaited>; - requirements: Record; - taskMemoriesActive: boolean; - taskMemoriesLastProcessedAt: Date | null; -}): BrainSourceSummary[] { - const statesBySource = new Map< - BrainSourceId, - Array<(typeof input.syncStates)[number]> - >(); - - for (const state of input.syncStates) { - // Current-version rows only: rows a version bump superseded, and - // auxiliary rows sharing a source's leading segment (inventories, - // censuses), would otherwise inflate stream counts and report their own - // completion as the source's. - const sourceId = resolveBrainSourceIdForCurrentCollector(state.collectorId); - - if (!sourceId) { - continue; - } - - statesBySource.set(sourceId, [ - ...(statesBySource.get(sourceId) ?? []), - state, - ]); - } - - const itemsBySource = new Map(); - - for (const row of input.itemCounts) { - const sourceId = resolveBrainSourceIdForCollector(row.collectorId); - - if (!sourceId) { - continue; - } - - itemsBySource.set(sourceId, (itemsBySource.get(sourceId) ?? 0) + row.items); - } - - return BRAIN_SOURCES.map((source) => { - const states = statesBySource.get(source.id) ?? []; - const connected = source.requires - ? input.requirements[source.requires] - : true; - const lastSyncedAt = states.reduce( - (latest, state) => - state.watermark && (!latest || state.watermark > latest) - ? state.watermark - : latest, - // The outbox records when a memory actually landed, which is fresher - // than the backfill checkpoint its sync-state row carries. - source.id === 'task-memories' ? input.taskMemoriesLastProcessedAt : null, - ); - const parents = states.filter((state) => - source.collectorIds.includes(state.collectorId), - ); - const children = states.filter( - (state) => !source.collectorIds.includes(state.collectorId), - ); - // A deep replay in progress is a parent row holding only a cursor. - // A row that also carries a watermark is a live stream whose cursor is a - // rolling checkpoint or mode-state (pull-request facts, member sweeps, - // the Notion incremental scan): steady ingestion, not history reading. - const backfilling = parents.some( - (state) => - state.backfillCursor && !state.backfillCompletedAt && !state.watermark, - ); - const streams = children.length > 0 ? children.length : states.length; - const backfillProgress = ((): { read: number; total: number } | null => { - if (!backfilling) { - return null; - } - - for (const parent of parents) { - if (!parent.backfillCursor || parent.backfillCompletedAt) { - continue; - } - - // Fan-out walks record their completed partitions in the cursor; - // counting rows with a completion timestamp instead would show 0 - // forever, because partition rows never carry one. - const read = parseBrainBackfillCompletedCount(parent.backfillCursor); - - if (read !== null && children.length > 0) { - return { - read: Math.min(read, children.length), - total: children.length, - }; - } - } - - const read = states.filter((state) => state.backfillCompletedAt).length; - - return streams > 0 - ? { read: Math.min(read, streams), total: streams } - : null; - })(); - const active = - source.id === 'task-memories' - ? input.taskMemoriesActive || states.length > 0 - : states.length > 0; - - return { - id: source.id, - label: source.label, - description: source.description, - namespaceLabel: brainNamespaceLabel(source.namespaceId), - status: !connected - ? ('not_connected' as const) - : backfilling - ? ('backfilling' as const) - : active - ? ('ingesting' as const) - : ('idle' as const), - lastSyncedAt, - streams, - backfillProgress, - trackedItems: itemsBySource.get(source.id) ?? 0, - }; - }); -} - -/** - * Whether Roomote holds usable Brain credentials, read without minting or - * provisioning anything. Deliberately not `resolveBrainConnection` (which - * registers OAuth clients headlessly on first use): the corpus read already - * takes that path once per request, and this check runs strictly after it, - * so it reads whatever state that attempt left behind rather than racing it. - */ -async function hasBrainCredentials(): Promise { - if (Env.R_GBRAIN_AGENT_TOKEN) { - return true; - } - - const stored = await db.query.mcpConnections.findFirst({ - where: and( - eq(mcpConnections.mcpId, BRAIN_MCP_ID), - isNull(mcpConnections.userId), - ), - }); - - return isMcpConnectionGbrainConfig(stored?.authConfig); -} - -export async function getBrainSettingsCommand( - auth: UserAuthSuccess, -): Promise { - assertAdmin(auth); - - // Activation is the Settings toggle, falling back to the legacy explicit - // brain-specific provider key. The gateway token and R_GBRAIN_URL exist as - // plumbing on deployments that never opted in, so neither can mean "the - // Brain is on". - const enabledState = await resolveBrainEnabledState(); - const configured = enabledState.enabled; - const url = Env.R_GBRAIN_URL ?? null; - - // The rollups only describe a Brain that exists; on an unconfigured - // deployment they would scan real tables (the missing-memory check - // anti-joins task runs) just to be thrown away by the status card. - const [ - inference, - corpusSnapshot, - stats, - syncStates, - itemCounts, - memories, - requirements, - embeddingsAvailable, - ] = await Promise.all([ - resolveBrainInferenceProvider(), - configured ? readBrainCorpus() : null, - configured ? readBrainStats() : null, - configured ? listBrainSyncStates(db) : [], - configured ? countBrainCollectorItemsByCollector(db) : [], - configured ? getBrainMemoryEventSummary(db) : EMPTY_MEMORY_SUMMARY, - resolveBrainSourceRequirements(), - isBrainEmbeddingAvailable(), - ]); - - const corpus = summarizeCorpus(corpusSnapshot); - - corpus.totalPages = stats?.pageCount ?? null; - - // Measured, not inferred: gbrain's own census says whether chunks are - // actually embedded. Chunks with zero embeddings is the keyword-only - // silent failure this page exists to catch. - const recall: BrainSettings['recall'] = - stats && (stats.chunkCount ?? 0) > 0 - ? { - mode: (stats.embeddedCount ?? 0) > 0 ? 'semantic' : 'keyword-only', - embeddedCount: stats.embeddedCount, - chunkCount: stats.chunkCount, - } - : { mode: 'unknown', embeddedCount: null, chunkCount: null }; - - // Read only after (and only when) the corpus failed to answer: it decides - // between "credentials not provisioned yet" and "service down", and the - // corpus read above is the call that may have just provisioned them. - const provisioned = - configured && !corpus.reachable ? await hasBrainCredentials() : true; - - const keySource: BrainSettings['keySource'] = !inference - ? null - : ( - await resolveModelProviderEnvValue([ - inference.providerId === 'openrouter' - ? 'R_BRAIN_OPENROUTER_API_KEY' - : 'R_BRAIN_OPENAI_API_KEY', - ]) - )?.trim() - ? 'brain' - : 'deployment'; - const memoryTotal = Object.values(memories.byStatus).reduce( - (total, value) => total + value, - 0, - ); - - const { status, statusDetail } = ((): { - status: BrainStatus; - statusDetail: string | null; - } => { - if (!configured) { - return { - status: 'not_configured', - statusDetail: - 'Memory is turned off for this deployment. Enable it to give agents shared memory.', - }; - } - - if (!url) { - return { - status: 'incomplete', - statusDetail: - 'A Memory provider key is set, but no Memory service URL is configured, so there is nowhere to store memories.', - }; - } - - // Synthesis rides the deployment's helper model in gateway mode, but - // semantic recall still needs embeddings: a provider key or a configured - // embeddings upstream. - if (!inference && !Env.R_BRAIN_EMBEDDINGS_UPSTREAM_URL) { - return { - status: 'incomplete', - statusDetail: - 'Memory has no embeddings provider, so it can only match keywords. Configure a provider key or an embeddings upstream to enable semantic recall.', - }; - } - - if (corpus.reachable) { - return { status: 'connected', statusDetail: null }; - } - - if (!provisioned) { - return { - status: 'incomplete', - statusDetail: - 'Roomote has not been able to provision its Memory credentials yet. This resolves on its own once Memory has started.', - }; - } - - return { - status: 'unreachable', - statusDetail: - 'Memory did not answer. Ingestion holds its position while it is down, so nothing is lost.', - }; - })(); - - return { - status, - statusDetail, - enabled: enabledState.enabled, - enabledFromLegacyKey: enabledState.enabled && enabledState.fromLegacyKey, - url, - // Chat and expansion always have somewhere to go: without a - // Brain-specific provider key, the gateway answers them with the - // deployment's helper model. - inferenceProvider: inference?.providerId ?? 'helper', - embeddingsAvailable, - keySource, - recall, - corpus, - sources: summarizeSources({ - syncStates, - itemCounts, - requirements, - taskMemoriesActive: memoryTotal > 0, - taskMemoriesLastProcessedAt: memories.lastProcessedAt, - }), - taskMemories: { - byStatus: memories.byStatus, - total: memoryTotal, - lastProcessedAt: memories.lastProcessedAt, - lastError: memories.lastError, - historicalCompletedRunsWithoutEvent: - memories.historicalCompletedRunsWithoutEvent, - recentCompletedRunsWithoutEvent: memories.recentCompletedRunsWithoutEvent, - }, - }; -} - -/** - * Turn the Brain on or off for the deployment. Writes the explicit Settings - * choice, which from then on wins over the legacy R_BRAIN_* key fallback in - * both directions. - */ -export async function setMemoryEnabledCommand( - auth: UserAuthSuccess, - input: { enabled: boolean }, -): Promise<{ enabled: boolean }> { - assertAdmin(auth); - - await setBrainEnabled(input.enabled); - - if (input.enabled) { - // Start the initial backfill now rather than waiting out the 15-minute - // collector and PR-sync schedules: a freshly enabled Memory should show - // content landing within moments, not ticks from now. - void requestBrainBackfill('memory-enabled'); - } - - return { enabled: input.enabled }; -} - -/** - * Enqueue a memory for every completed run that does not have one. Idempotent - * by the outbox's unique(runId), so running it twice costs nothing; the - * drainer distills the backlog newest-first. - */ -export async function backfillBrainTaskMemoriesCommand( - auth: UserAuthSuccess, -): Promise<{ queued: number }> { - assertAdmin(auth); - - return { queued: await backfillBrainMemoryEvents(db) }; -} - -export async function retryFailedBrainTaskMemoriesCommand( - auth: UserAuthSuccess, -): Promise<{ requeued: number }> { - assertAdmin(auth); - - return { requeued: await requeueFailedBrainMemoryEvents(db) }; -} - -export type BrainPageListing = { - reachable: boolean; - total: number; - nextOffset: number | null; - pages: Array<{ - slug: string; - title: string; - namespaceId: BrainNamespaceBucketId; - namespaceLabel: string; - updatedAt: Date | null; - }>; -}; - -function toListedPage(page: { - slug: string; - title: string | null; - updatedAt: Date | null; -}): BrainPageListing['pages'][number] { - const namespaceId = resolveBrainNamespaceId(page.slug); - - return { - slug: page.slug, - title: page.title ?? page.slug, - namespaceId, - namespaceLabel: brainNamespaceLabel(namespaceId), - updatedAt: page.updatedAt, - }; -} - -/** Exported for focused command tests; the command below owns auth and I/O. */ -export function paginateBrainCorpus( - snapshot: BrainCorpusSnapshot, - input: { - search?: string; - namespaceId?: string; - offset: number; - limit: number; - }, -): Omit { - const needle = input.search?.trim().toLowerCase() ?? ''; - const filtered = snapshot.pages.filter((page) => { - const namespaceId = resolveBrainNamespaceId(page.slug); - - return ( - (!input.namespaceId || namespaceId === input.namespaceId) && - (!needle || - page.slug.toLowerCase().includes(needle) || - page.title?.toLowerCase().includes(needle)) - ); - }); - const pages = filtered - .slice(input.offset, input.offset + input.limit) - .map(toListedPage); - - return { - total: filtered.length, - nextOffset: - input.offset + pages.length < filtered.length - ? input.offset + pages.length - : null, - pages, - }; -} - -/** - * One server-filtered page from the exhaustive cached corpus. The browser only - * renders this bounded result, so searching a large Brain does not ship or - * repeatedly filter the full listing on every keystroke. - */ -export async function listBrainPagesCommand( - auth: UserAuthSuccess, - input: { - search?: string; - namespaceId?: string; - offset: number; - limit: number; - }, -): Promise { - assertAdmin(auth); - - const snapshot = await readBrainCorpus(); - - if (!snapshot) { - return { reachable: false, total: 0, nextOffset: null, pages: [] }; - } - - return { reachable: true, ...paginateBrainCorpus(snapshot, input) }; -} - -/** - * How much of a page body the preview ships. A Brain page can be an entire - * day of channel history; the dialog is a reading pane, not an export. - */ -const PAGE_CONTENT_PREVIEW_LIMIT = 32_000; - -type BrainPageContent = { - slug: string; - title: string; - updatedAt: Date | null; - /** - * The page body as the Brain stores it. Untrusted content distilled from - * tasks and integrations: the client renders it as plain text. - */ - content: string | null; - /** The body exceeded the preview bound and was cut. */ - contentTruncated: boolean; -}; - -export async function getBrainPageCommand( - auth: UserAuthSuccess, - input: { slug: string }, -): Promise { - assertAdmin(auth); - - const page = await readBrainPage(input.slug); - - if (!page) { - return null; - } - - const contentTruncated = - (page.content?.length ?? 0) > PAGE_CONTENT_PREVIEW_LIMIT; - - return { - slug: page.slug, - title: page.title ?? page.slug, - updatedAt: page.updatedAt, - content: contentTruncated - ? page.content!.slice(0, PAGE_CONTENT_PREVIEW_LIMIT) - : page.content, - contentTruncated, - }; -} +export type { BrainStatus } from './brain-settings-status'; +export type { + BrainSourceStatus, + BrainSourceSummary, +} from './brain-source-types'; +export type { + BrainCorpusSummary, + BrainNamespaceSummary, +} from './brain-corpus-types'; +export type { BrainSettings } from './brain-settings-types'; +export type { BrainPageListing } from './paginate-brain-corpus'; + +export { summarizeSources } from './summarize-sources'; +export { getBrainSettingsCommand } from './get-brain-settings'; +export { setMemoryEnabledCommand } from './set-memory-enabled'; +export { + backfillBrainTaskMemoriesCommand, + retryFailedBrainTaskMemoriesCommand, +} from './brain-task-memories'; +export { paginateBrainCorpus } from './paginate-brain-corpus'; +export { listBrainPagesCommand } from './list-brain-pages'; +export { getBrainPageCommand } from './get-brain-page'; diff --git a/apps/web/src/trpc/commands/brain/list-brain-pages.ts b/apps/web/src/trpc/commands/brain/list-brain-pages.ts new file mode 100644 index 0000000000..e4f29ba255 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/list-brain-pages.ts @@ -0,0 +1,33 @@ +import { readBrainCorpus } from '@roomote/sdk/server'; +import type { UserAuthSuccess } from '@/types'; + +import { assertAdmin } from '../setup/shared'; +import { + paginateBrainCorpus, + type BrainPageListing, +} from './paginate-brain-corpus'; + +/** + * One server-filtered page from the exhaustive cached corpus. The browser + * only renders this bounded result, so searching a large Brain does not + * ship the full listing on every keystroke. + */ +export async function listBrainPagesCommand( + auth: UserAuthSuccess, + input: { + search?: string; + namespaceId?: string; + offset: number; + limit: number; + }, +): Promise { + assertAdmin(auth); + + const snapshot = await readBrainCorpus(); + + if (!snapshot) { + return { reachable: false, total: 0, nextOffset: null, pages: [] }; + } + + return { reachable: true, ...paginateBrainCorpus(snapshot, input) }; +} diff --git a/apps/web/src/trpc/commands/brain/load-brain-settings.ts b/apps/web/src/trpc/commands/brain/load-brain-settings.ts new file mode 100644 index 0000000000..3ba1ffb25f --- /dev/null +++ b/apps/web/src/trpc/commands/brain/load-brain-settings.ts @@ -0,0 +1,71 @@ +import { + countBrainCollectorItemsByCollector, + db, + getBrainMemoryEventSummary, + listBrainSyncStates, +} from '@roomote/db/server'; +import { + isBrainEmbeddingAvailable, + readBrainCorpus, + readBrainStats, + resolveBrainSourceRequirements, + resolveBrainInferenceProvider, +} from '@roomote/sdk/server'; + +import type { BrainSettings } from './brain-settings-types'; +import { summarizeCorpus } from './summarize-corpus'; + +const EMPTY_MEMORY_SUMMARY: Awaited< + ReturnType +> = { + byStatus: { pending: 0, processing: 0, done: 0, skipped: 0, failed: 0 }, + lastProcessedAt: null, + lastError: null, + historicalCompletedRunsWithoutEvent: 0, + recentCompletedRunsWithoutEvent: 0, +}; + +export async function loadBrainSettings(configured: boolean) { + const [ + inference, + corpusSnapshot, + stats, + syncStates, + itemCounts, + memories, + requirements, + embeddingsAvailable, + ] = await Promise.all([ + resolveBrainInferenceProvider(), + configured ? readBrainCorpus() : null, + configured ? readBrainStats() : null, + configured ? listBrainSyncStates(db) : [], + configured ? countBrainCollectorItemsByCollector(db) : [], + configured ? getBrainMemoryEventSummary(db) : EMPTY_MEMORY_SUMMARY, + resolveBrainSourceRequirements(), + isBrainEmbeddingAvailable(), + ]); + + const corpus = summarizeCorpus(corpusSnapshot); + corpus.totalPages = stats?.pageCount ?? null; + + const recall: BrainSettings['recall'] = + stats && (stats.chunkCount ?? 0) > 0 + ? { + mode: (stats.embeddedCount ?? 0) > 0 ? 'semantic' : 'keyword-only', + embeddedCount: stats.embeddedCount, + chunkCount: stats.chunkCount, + } + : { mode: 'unknown', embeddedCount: null, chunkCount: null }; + + return { + inference, + corpus, + syncStates, + itemCounts, + memories, + requirements, + embeddingsAvailable, + recall, + }; +} diff --git a/apps/web/src/trpc/commands/brain/paginate-brain-corpus.ts b/apps/web/src/trpc/commands/brain/paginate-brain-corpus.ts new file mode 100644 index 0000000000..b42062735b --- /dev/null +++ b/apps/web/src/trpc/commands/brain/paginate-brain-corpus.ts @@ -0,0 +1,70 @@ +import type { BrainCorpusSnapshot } from '@roomote/sdk/server'; +import { + brainNamespaceLabel, + resolveBrainNamespaceId, + type BrainNamespaceBucketId, +} from '@roomote/types'; + +export type BrainPageListing = { + reachable: boolean; + total: number; + nextOffset: number | null; + pages: Array<{ + slug: string; + title: string; + namespaceId: BrainNamespaceBucketId; + namespaceLabel: string; + updatedAt: Date | null; + }>; +}; + +function toListedPage(page: { + slug: string; + title: string | null; + updatedAt: Date | null; +}): BrainPageListing['pages'][number] { + const namespaceId = resolveBrainNamespaceId(page.slug); + + return { + slug: page.slug, + title: page.title ?? page.slug, + namespaceId, + namespaceLabel: brainNamespaceLabel(namespaceId), + updatedAt: page.updatedAt, + }; +} + +/** Exported for focused command tests; the command below owns auth and I/O. */ +export function paginateBrainCorpus( + snapshot: BrainCorpusSnapshot, + input: { + search?: string; + namespaceId?: string; + offset: number; + limit: number; + }, +): Omit { + const needle = input.search?.trim().toLowerCase() ?? ''; + const filtered = snapshot.pages.filter((page) => { + const namespaceId = resolveBrainNamespaceId(page.slug); + + return ( + (!input.namespaceId || namespaceId === input.namespaceId) && + (!needle || + page.slug.toLowerCase().includes(needle) || + page.title?.toLowerCase().includes(needle)) + ); + }); + const pages = filtered + .slice(input.offset, input.offset + input.limit) + .map(toListedPage); + + return { + total: filtered.length, + nextOffset: + input.offset + pages.length < filtered.length + ? input.offset + pages.length + : null, + pages, + }; +} diff --git a/apps/web/src/trpc/commands/brain/set-memory-enabled.ts b/apps/web/src/trpc/commands/brain/set-memory-enabled.ts new file mode 100644 index 0000000000..7a1baeff85 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/set-memory-enabled.ts @@ -0,0 +1,27 @@ +import { setBrainEnabled } from '@roomote/db/server'; +import { requestBrainBackfill } from '@roomote/sdk/server'; +import type { UserAuthSuccess } from '@/types'; + +import { assertAdmin } from '../setup/shared'; + +/** + * Turn the Brain on or off for the deployment. Writes the explicit Settings + * choice, which from then on wins over the legacy R_BRAIN_* key fallback. + */ +export async function setMemoryEnabledCommand( + auth: UserAuthSuccess, + input: { enabled: boolean }, +): Promise<{ enabled: boolean }> { + assertAdmin(auth); + + await setBrainEnabled(input.enabled); + + if (input.enabled) { + // Start the initial backfill now rather than waiting out the 15-minute + // collector schedules: a freshly enabled Memory should show content + // landing within moments. + void requestBrainBackfill('memory-enabled'); + } + + return { enabled: input.enabled }; +} diff --git a/apps/web/src/trpc/commands/brain/source-backfill-progress.ts b/apps/web/src/trpc/commands/brain/source-backfill-progress.ts new file mode 100644 index 0000000000..3398e33255 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/source-backfill-progress.ts @@ -0,0 +1,42 @@ +import { parseBrainBackfillCompletedCount } from '@roomote/types'; +import type { listBrainSyncStates } from '@roomote/db/server'; + +type SyncState = Awaited>[number]; + +/** + * A deep replay in progress is a parent row holding only a cursor. A row + * that also carries a watermark is a live stream whose cursor is a rolling + * checkpoint: steady ingestion, not history reading. + */ +export function sourceBackfillProgress( + parents: SyncState[], + children: SyncState[], + states: SyncState[], + streams: number, + backfilling: boolean, +): { read: number; total: number } | null { + if (!backfilling) { + return null; + } + + for (const parent of parents) { + if (!parent.backfillCursor || parent.backfillCompletedAt) { + continue; + } + + // Fan-out walks record completed partitions in the cursor; counting + // rows with a completion timestamp would show 0 forever. + const read = parseBrainBackfillCompletedCount(parent.backfillCursor); + + if (read !== null && children.length > 0) { + return { + read: Math.min(read, children.length), + total: children.length, + }; + } + } + + const read = states.filter((state) => state.backfillCompletedAt).length; + + return streams > 0 ? { read: Math.min(read, streams), total: streams } : null; +} diff --git a/apps/web/src/trpc/commands/brain/source-sync-groups.ts b/apps/web/src/trpc/commands/brain/source-sync-groups.ts new file mode 100644 index 0000000000..df2962ba67 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/source-sync-groups.ts @@ -0,0 +1,52 @@ +import { + resolveBrainSourceIdForCollector, + resolveBrainSourceIdForCurrentCollector, + type BrainSourceId, +} from '@roomote/types'; +import type { + countBrainCollectorItemsByCollector, + listBrainSyncStates, +} from '@roomote/db/server'; + +type SyncState = Awaited>[number]; + +export function groupSyncStatesBySource( + syncStates: Awaited>, +): Map { + const statesBySource = new Map(); + + for (const state of syncStates) { + // Current-version rows only: superseded and auxiliary rows would + // otherwise inflate stream counts. + const sourceId = resolveBrainSourceIdForCurrentCollector(state.collectorId); + + if (!sourceId) { + continue; + } + + statesBySource.set(sourceId, [ + ...(statesBySource.get(sourceId) ?? []), + state, + ]); + } + + return statesBySource; +} + +export function groupItemCountsBySource( + itemCounts: Awaited>, +): Map { + const itemsBySource = new Map(); + + for (const row of itemCounts) { + const sourceId = resolveBrainSourceIdForCollector(row.collectorId); + + if (!sourceId) { + continue; + } + + itemsBySource.set(sourceId, (itemsBySource.get(sourceId) ?? 0) + row.items); + } + + return itemsBySource; +} diff --git a/apps/web/src/trpc/commands/brain/summarize-corpus.ts b/apps/web/src/trpc/commands/brain/summarize-corpus.ts new file mode 100644 index 0000000000..fb65b31313 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/summarize-corpus.ts @@ -0,0 +1,79 @@ +import { + BRAIN_OTHER_NAMESPACE_ID, + brainNamespaceLabel, + resolveBrainNamespaceId, + type BrainNamespaceBucketId, +} from '@roomote/types'; +import type { readBrainCorpus } from '@roomote/sdk/server'; + +import type { BrainCorpusSummary } from './brain-corpus-types'; + +/** Days of page-writing activity summarized for the ingestion chart. */ +const ACTIVITY_WINDOW_DAYS = 30; + +function buildActivityByDay( + pages: Array<{ updatedAt: Date | null }>, +): Array<{ date: string; pages: number }> { + const counts = new Map(); + const today = new Date(); + + for (let offset = ACTIVITY_WINDOW_DAYS - 1; offset >= 0; offset--) { + const day = new Date(today); + day.setUTCDate(day.getUTCDate() - offset); + counts.set(day.toISOString().slice(0, 10), 0); + } + + for (const page of pages) { + if (!page.updatedAt) { + continue; + } + + const key = page.updatedAt.toISOString().slice(0, 10); + + if (counts.has(key)) { + counts.set(key, counts.get(key)! + 1); + } + } + + return [...counts.entries()].map(([date, count]) => ({ + date, + pages: count, + })); +} + +export function summarizeCorpus( + snapshot: Awaited>, +): BrainCorpusSummary { + if (!snapshot) { + return { + reachable: false, + listedPages: 0, + totalPages: null, + namespaces: [], + activityByDay: [], + }; + } + + const counts = new Map(); + + for (const page of snapshot.pages) { + const namespaceId = resolveBrainNamespaceId(page.slug); + counts.set(namespaceId, (counts.get(namespaceId) ?? 0) + 1); + } + + const namespaces = [...counts.entries()] + .map(([id, pages]) => ({ id, label: brainNamespaceLabel(id), pages })) + .sort((left, right) => { + if (left.id === BRAIN_OTHER_NAMESPACE_ID) return 1; + if (right.id === BRAIN_OTHER_NAMESPACE_ID) return -1; + return right.pages - left.pages || left.label.localeCompare(right.label); + }); + + return { + reachable: true, + listedPages: snapshot.pages.length, + totalPages: null, + namespaces, + activityByDay: buildActivityByDay(snapshot.pages), + }; +} diff --git a/apps/web/src/trpc/commands/brain/summarize-sources.ts b/apps/web/src/trpc/commands/brain/summarize-sources.ts new file mode 100644 index 0000000000..8c352a6fe7 --- /dev/null +++ b/apps/web/src/trpc/commands/brain/summarize-sources.ts @@ -0,0 +1,81 @@ +import { + countBrainCollectorItemsByCollector, + listBrainSyncStates, +} from '@roomote/db/server'; +import { + BRAIN_SOURCES, + brainNamespaceLabel, + type BrainSourceRequirement, +} from '@roomote/types'; + +import type { BrainSourceSummary } from './brain-source-types'; +import { sourceBackfillProgress } from './source-backfill-progress'; +import { + groupItemCountsBySource, + groupSyncStatesBySource, +} from './source-sync-groups'; + +/** Exported for unit tests; the settings command is its only runtime caller. */ +export function summarizeSources(input: { + syncStates: Awaited>; + itemCounts: Awaited>; + requirements: Record; + taskMemoriesActive: boolean; + taskMemoriesLastProcessedAt: Date | null; +}): BrainSourceSummary[] { + const statesBySource = groupSyncStatesBySource(input.syncStates); + const itemsBySource = groupItemCountsBySource(input.itemCounts); + + return BRAIN_SOURCES.map((source) => { + const states = statesBySource.get(source.id) ?? []; + const connected = source.requires + ? input.requirements[source.requires] + : true; + const lastSyncedAt = states.reduce( + (latest, state) => + state.watermark && (!latest || state.watermark > latest) + ? state.watermark + : latest, + source.id === 'task-memories' ? input.taskMemoriesLastProcessedAt : null, + ); + const parents = states.filter((state) => + source.collectorIds.includes(state.collectorId), + ); + const children = states.filter( + (state) => !source.collectorIds.includes(state.collectorId), + ); + const backfilling = parents.some( + (state) => + state.backfillCursor && !state.backfillCompletedAt && !state.watermark, + ); + const streams = children.length > 0 ? children.length : states.length; + const active = + source.id === 'task-memories' + ? input.taskMemoriesActive || states.length > 0 + : states.length > 0; + + return { + id: source.id, + label: source.label, + description: source.description, + namespaceLabel: brainNamespaceLabel(source.namespaceId), + status: !connected + ? ('not_connected' as const) + : backfilling + ? ('backfilling' as const) + : active + ? ('ingesting' as const) + : ('idle' as const), + lastSyncedAt, + streams, + backfillProgress: sourceBackfillProgress( + parents, + children, + states, + streams, + backfilling, + ), + trackedItems: itemsBySource.get(source.id) ?? 0, + }; + }); +} diff --git a/deploy/compose/docker-compose.prod.yml b/deploy/compose/docker-compose.prod.yml index cb06b03bf0..e727a228b7 100644 --- a/deploy/compose/docker-compose.prod.yml +++ b/deploy/compose/docker-compose.prod.yml @@ -104,6 +104,12 @@ x-roomote-inference-env: &roomote-inference-env OPENCODE_GO_API_KEY: ${OPENCODE_GO_API_KEY:-} BASETEN_API_KEY: ${BASETEN_API_KEY:-} TOGETHER_API_KEY: ${TOGETHER_API_KEY:-} + # Memory embeddings auto-wire to Workers AI when these are set, + # R_BRAIN_EMBEDDINGS_UPSTREAM_URL is empty, and no OpenAI/OpenRouter + # Memory key is set. Set R_BRAIN_EMBEDDING_MODEL and + # R_BRAIN_EMBEDDING_DIMENSIONS to @cf/baai/bge-m3 / 1024 before first boot. + CLOUDFLARE_WORKERS_AI_API_TOKEN: ${CLOUDFLARE_WORKERS_AI_API_TOKEN:-} + CLOUDFLARE_WORKERS_AI_ACCOUNT_ID: ${CLOUDFLARE_WORKERS_AI_ACCOUNT_ID:-} x-roomote-web-env: &roomote-web-env <<: *roomote-inference-env diff --git a/docker-compose.production.yml b/docker-compose.production.yml index 1a9f12cb30..3dd3907eb9 100644 --- a/docker-compose.production.yml +++ b/docker-compose.production.yml @@ -60,6 +60,8 @@ x-roomote-production-env: &roomote-production-env OPENCODE_GO_API_KEY: ${OPENCODE_GO_API_KEY:-} BASETEN_API_KEY: ${BASETEN_API_KEY:-} TOGETHER_API_KEY: ${TOGETHER_API_KEY:-} + CLOUDFLARE_WORKERS_AI_API_TOKEN: ${CLOUDFLARE_WORKERS_AI_API_TOKEN:-} + CLOUDFLARE_WORKERS_AI_ACCOUNT_ID: ${CLOUDFLARE_WORKERS_AI_ACCOUNT_ID:-} GITHUB_TOKEN: ${GITHUB_TOKEN:-} R_GITHUB_APP_SLUG: ${R_GITHUB_APP_SLUG:?R_GITHUB_APP_SLUG is required} R_GITHUB_ADDITIONAL_APP_SLUGS: ${R_GITHUB_ADDITIONAL_APP_SLUGS:-} diff --git a/docker-compose.self-host.yml b/docker-compose.self-host.yml index 4fa82fe21d..f4ff590b88 100644 --- a/docker-compose.self-host.yml +++ b/docker-compose.self-host.yml @@ -84,6 +84,8 @@ x-roomote-env: &roomote-env OPENCODE_GO_API_KEY: ${OPENCODE_GO_API_KEY:-} BASETEN_API_KEY: ${BASETEN_API_KEY:-} TOGETHER_API_KEY: ${TOGETHER_API_KEY:-} + CLOUDFLARE_WORKERS_AI_API_TOKEN: ${CLOUDFLARE_WORKERS_AI_API_TOKEN:-} + CLOUDFLARE_WORKERS_AI_ACCOUNT_ID: ${CLOUDFLARE_WORKERS_AI_ACCOUNT_ID:-} GITHUB_TOKEN: ${GITHUB_TOKEN:-} R_GITHUB_APP_ID: ${R_GITHUB_APP_ID:-} R_GITHUB_APP_PRIVATE_KEY: ${R_GITHUB_APP_PRIVATE_KEY:-} diff --git a/packages/sdk/src/server/lib/__tests__/brain-inference.test.ts b/packages/sdk/src/server/lib/__tests__/brain-inference.test.ts index ae7c3f35b7..85ccf27466 100644 --- a/packages/sdk/src/server/lib/__tests__/brain-inference.test.ts +++ b/packages/sdk/src/server/lib/__tests__/brain-inference.test.ts @@ -7,14 +7,19 @@ vi.mock('@roomote/db/server', () => ({ const mockEnv: { R_BRAIN_MODEL?: string; R_BRAIN_EMBEDDING_MODEL?: string; + R_BRAIN_EMBEDDING_DIMENSIONS?: number; R_BRAIN_EMBEDDINGS_UPSTREAM_URL?: string; + R_BRAIN_INFERENCE_UPSTREAM_API_KEY?: string; } = {}; vi.mock('@roomote/env', () => ({ Env: mockEnv })); const { mapBrainModelName, + describeBrainModels, isBrainEmbeddingAvailable, + resolveBrainEmbeddingsUpstream, + resetBrainEmbeddingsUpstreamCache, resetBrainInferenceProviderCache, } = await import('../brain-inference'); const { resolveModelProviderEnvValue } = await import('@roomote/db/server'); @@ -24,14 +29,54 @@ const mockResolveModelProviderEnvValue = vi.mocked( const OPENROUTER = { providerId: 'openrouter', apiKey: 'sk-or' } as const; const OPENAI = { providerId: 'openai', apiKey: 'sk' } as const; +const CF_ACCOUNT_ID = 'a1b2c3d4e5f6789012345678abcdef90'; + +function envVarNames(names: string | readonly string[]): readonly string[] { + return typeof names === 'string' ? [names] : names; +} + +function mockCloudflareWorkersAi(): void { + mockResolveModelProviderEnvValue.mockImplementation(async (names) => { + const list = envVarNames(names); + if (list.includes('CLOUDFLARE_WORKERS_AI_API_TOKEN')) { + return 'cfat_test'; + } + if (list.includes('CLOUDFLARE_WORKERS_AI_ACCOUNT_ID')) { + return CF_ACCOUNT_ID; + } + return undefined; + }); +} + +function mockCloudflareAndOpenRouter(): void { + mockResolveModelProviderEnvValue.mockImplementation(async (names) => { + const list = envVarNames(names); + if (list.includes('CLOUDFLARE_WORKERS_AI_API_TOKEN')) { + return 'cfat_test'; + } + if (list.includes('CLOUDFLARE_WORKERS_AI_ACCOUNT_ID')) { + return CF_ACCOUNT_ID; + } + if ( + list.includes('R_BRAIN_OPENROUTER_API_KEY') || + list.includes('OPENROUTER_API_KEY') + ) { + return 'sk-or'; + } + return undefined; + }); +} beforeEach(() => { delete mockEnv.R_BRAIN_MODEL; delete mockEnv.R_BRAIN_EMBEDDING_MODEL; + delete mockEnv.R_BRAIN_EMBEDDING_DIMENSIONS; delete mockEnv.R_BRAIN_EMBEDDINGS_UPSTREAM_URL; + delete mockEnv.R_BRAIN_INFERENCE_UPSTREAM_API_KEY; mockResolveModelProviderEnvValue.mockReset(); mockResolveModelProviderEnvValue.mockResolvedValue(undefined); resetBrainInferenceProviderCache(); + resetBrainEmbeddingsUpstreamCache(); }); describe('mapBrainModelName', () => { @@ -104,4 +149,135 @@ describe('isBrainEmbeddingAvailable', () => { mockEnv.R_BRAIN_EMBEDDINGS_UPSTREAM_URL = ' '; await expect(isBrainEmbeddingAvailable()).resolves.toBe(false); }); + + it('is ready when Cloudflare Workers AI can embed, with no OpenAI key', async () => { + mockCloudflareWorkersAi(); + await expect(isBrainEmbeddingAvailable()).resolves.toBe(true); + }); + + it('is not ready with a Cloudflare token and no account id', async () => { + mockResolveModelProviderEnvValue.mockImplementation(async (names) => { + const list = envVarNames(names); + if (list.includes('CLOUDFLARE_WORKERS_AI_API_TOKEN')) { + return 'cfat_test'; + } + return undefined; + }); + await expect(isBrainEmbeddingAvailable()).resolves.toBe(false); + }); + + it('is not ready with a Cloudflare account id and no token', async () => { + mockResolveModelProviderEnvValue.mockImplementation(async (names) => { + const list = envVarNames(names); + if (list.includes('CLOUDFLARE_WORKERS_AI_ACCOUNT_ID')) { + return CF_ACCOUNT_ID; + } + return undefined; + }); + await expect(isBrainEmbeddingAvailable()).resolves.toBe(false); + }); + + it('is not ready with an account id that is not 32 hex characters', async () => { + mockResolveModelProviderEnvValue.mockImplementation(async (names) => { + const list = envVarNames(names); + if (list.includes('CLOUDFLARE_WORKERS_AI_API_TOKEN')) { + return 'cfat_test'; + } + if (list.includes('CLOUDFLARE_WORKERS_AI_ACCOUNT_ID')) { + return 'not-a-cloudflare-account-id'; + } + return undefined; + }); + await expect(isBrainEmbeddingAvailable()).resolves.toBe(false); + }); + + it('is not ready with a 32-character account id that is not hex', async () => { + mockResolveModelProviderEnvValue.mockImplementation(async (names) => { + const list = envVarNames(names); + if (list.includes('CLOUDFLARE_WORKERS_AI_API_TOKEN')) { + return 'cfat_test'; + } + if (list.includes('CLOUDFLARE_WORKERS_AI_ACCOUNT_ID')) { + return 'zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz'; + } + return undefined; + }); + await expect(isBrainEmbeddingAvailable()).resolves.toBe(false); + }); +}); + +describe('resolveBrainEmbeddingsUpstream', () => { + it('returns the configured URL and key, without reading Cloudflare', async () => { + mockEnv.R_BRAIN_EMBEDDINGS_UPSTREAM_URL = 'https://embedder.example/v1/'; + mockEnv.R_BRAIN_INFERENCE_UPSTREAM_API_KEY = 'local-key'; + mockCloudflareWorkersAi(); + + await expect(resolveBrainEmbeddingsUpstream()).resolves.toEqual({ + baseUrl: 'https://embedder.example/v1', + apiKey: 'local-key', + source: 'configured', + }); + expect(mockResolveModelProviderEnvValue).not.toHaveBeenCalled(); + }); + + it('wires Cloudflare Workers AI as an OpenAI-compatible /ai base', async () => { + mockCloudflareWorkersAi(); + + await expect(resolveBrainEmbeddingsUpstream()).resolves.toEqual({ + baseUrl: `https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/ai`, + apiKey: 'cfat_test', + source: 'cloudflare-workers-ai', + }); + }); + + it('does not auto-wire Cloudflare when an OpenAI or OpenRouter key already serves embeddings', async () => { + mockCloudflareAndOpenRouter(); + + await expect(resolveBrainEmbeddingsUpstream()).resolves.toBeNull(); + await expect( + resolveBrainEmbeddingsUpstream('text-embedding-3-small'), + ).resolves.toBeNull(); + await expect(isBrainEmbeddingAvailable()).resolves.toBe(true); + }); + + it('keeps Cloudflare-model corpora on Cloudflare when a provider key is added', async () => { + mockCloudflareAndOpenRouter(); + + await expect(resolveBrainEmbeddingsUpstream()).resolves.toBeNull(); + await expect( + resolveBrainEmbeddingsUpstream('@cf/baai/bge-m3'), + ).resolves.toEqual({ + baseUrl: `https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/ai`, + apiKey: 'cfat_test', + source: 'cloudflare-workers-ai', + }); + }); + + it('caches a Cloudflare miss so an unconfigured tick does not re-read settings', async () => { + await expect(resolveBrainEmbeddingsUpstream()).resolves.toBeNull(); + const afterFirst = mockResolveModelProviderEnvValue.mock.calls.length; + expect(afterFirst).toBeGreaterThan(0); + await expect(resolveBrainEmbeddingsUpstream()).resolves.toBeNull(); + expect(mockResolveModelProviderEnvValue).toHaveBeenCalledTimes(afterFirst); + }); + + it('caches a Cloudflare hit across embeddings requests', async () => { + mockCloudflareWorkersAi(); + await resolveBrainEmbeddingsUpstream(); + const afterFirst = mockResolveModelProviderEnvValue.mock.calls.length; + expect(afterFirst).toBeGreaterThan(0); + await resolveBrainEmbeddingsUpstream(); + expect(mockResolveModelProviderEnvValue).toHaveBeenCalledTimes(afterFirst); + }); +}); + +describe('describeBrainModels', () => { + it('reports 1024 dimensions for bge-m3', () => { + mockEnv.R_BRAIN_EMBEDDING_MODEL = '@cf/baai/bge-m3'; + + expect(describeBrainModels('openai')).toMatchObject({ + embeddingModel: '@cf/baai/bge-m3', + embeddingDimensions: 1024, + }); + }); }); diff --git a/packages/sdk/src/server/lib/brain-embeddings-cloudflare.ts b/packages/sdk/src/server/lib/brain-embeddings-cloudflare.ts new file mode 100644 index 0000000000..c0845bdf5e --- /dev/null +++ b/packages/sdk/src/server/lib/brain-embeddings-cloudflare.ts @@ -0,0 +1,38 @@ +import { resolveModelProviderEnvValue } from '@roomote/db/server'; + +/** Workers AI account ids are 32 hex chars. */ +const CLOUDFLARE_WORKERS_AI_ACCOUNT_ID_RE = /^[a-f0-9]{32}$/i; + +export const CLOUDFLARE_WORKERS_AI_EMBEDDING_MODEL = '@cf/baai/bge-m3'; +export const CLOUDFLARE_WORKERS_AI_EMBEDDING_DIMENSIONS = 1024; + +type CloudflareWorkersAiEmbeddings = { + baseUrl: string; + apiKey: string; +}; + +export function isCloudflareWorkersAiEmbeddingModel(model: string): boolean { + return model.startsWith('@cf/'); +} + +export async function readCloudflareWorkersAiEmbeddings(): Promise { + const apiKey = ( + await resolveModelProviderEnvValue(['CLOUDFLARE_WORKERS_AI_API_TOKEN']) + )?.trim(); + const accountId = ( + await resolveModelProviderEnvValue(['CLOUDFLARE_WORKERS_AI_ACCOUNT_ID']) + )?.trim(); + + if ( + !apiKey || + !accountId || + !CLOUDFLARE_WORKERS_AI_ACCOUNT_ID_RE.test(accountId) + ) { + return null; + } + + return { + baseUrl: `https://api.cloudflare.com/client/v4/accounts/${accountId}/ai`, + apiKey, + }; +} diff --git a/packages/sdk/src/server/lib/brain-embeddings-upstream-read.ts b/packages/sdk/src/server/lib/brain-embeddings-upstream-read.ts new file mode 100644 index 0000000000..dc94a2d737 --- /dev/null +++ b/packages/sdk/src/server/lib/brain-embeddings-upstream-read.ts @@ -0,0 +1,43 @@ +import { Env } from '@roomote/env'; + +import { + isCloudflareWorkersAiEmbeddingModel, + readCloudflareWorkersAiEmbeddings, +} from './brain-embeddings-cloudflare'; +import { resolveBrainInferenceProvider } from './brain-inference-provider'; + +export type BrainEmbeddingsUpstream = { + baseUrl: string; + apiKey?: string; + source: 'configured' | 'cloudflare-workers-ai'; +}; + +export async function readBrainEmbeddingsUpstream( + requestedModel?: string, +): Promise { + const configured = Env.R_BRAIN_EMBEDDINGS_UPSTREAM_URL?.trim(); + + if (configured) { + return { + baseUrl: configured.replace(/\/$/, ''), + apiKey: Env.R_BRAIN_INFERENCE_UPSTREAM_API_KEY?.trim() || undefined, + source: 'configured', + }; + } + + const cloudflare = await readCloudflareWorkersAiEmbeddings(); + + if (!cloudflare) { + return null; + } + + const useCloudflare = requestedModel + ? isCloudflareWorkersAiEmbeddingModel(requestedModel) + : !(await resolveBrainInferenceProvider()); + + if (!useCloudflare) { + return null; + } + + return { ...cloudflare, source: 'cloudflare-workers-ai' }; +} diff --git a/packages/sdk/src/server/lib/brain-embeddings-upstream.ts b/packages/sdk/src/server/lib/brain-embeddings-upstream.ts new file mode 100644 index 0000000000..01815de4cf --- /dev/null +++ b/packages/sdk/src/server/lib/brain-embeddings-upstream.ts @@ -0,0 +1,70 @@ +import { + type BrainEmbeddingsUpstream, + readBrainEmbeddingsUpstream, +} from './brain-embeddings-upstream-read'; +import { resolveBrainInferenceProvider } from './brain-inference-provider'; + +export { + CLOUDFLARE_WORKERS_AI_EMBEDDING_DIMENSIONS, + CLOUDFLARE_WORKERS_AI_EMBEDDING_MODEL, +} from './brain-embeddings-cloudflare'; +export type { BrainEmbeddingsUpstream } from './brain-embeddings-upstream-read'; + +const UPSTREAM_CACHE_TTL_MS = 30_000; + +let upstreamCache: { + values: Map; + expiresAtMs: number; +} | null = null; + +/** Drop the cached upstream so the next call re-reads settings. */ +export function resetBrainEmbeddingsUpstreamCache(): void { + upstreamCache = null; +} + +/** + * Where Memory embeddings are forwarded. `R_BRAIN_EMBEDDINGS_UPSTREAM_URL` + * always wins. A `@cf/` request stays on Workers AI even after a chat/task + * provider key is added. With no requested model, Cloudflare auto-wires only + * when no OpenAI or OpenRouter Memory key is set. Base `.../ai`. + */ +export async function resolveBrainEmbeddingsUpstream( + requestedModel?: string, +): Promise { + const cacheKey = requestedModel ?? ''; + const cached = upstreamCache; + + if ( + cached && + cached.expiresAtMs > Date.now() && + cached.values.has(cacheKey) + ) { + return cached.values.get(cacheKey)!; + } + + const value = await readBrainEmbeddingsUpstream(requestedModel); + + if (!upstreamCache || upstreamCache.expiresAtMs <= Date.now()) { + upstreamCache = { + values: new Map(), + expiresAtMs: Date.now() + UPSTREAM_CACHE_TTL_MS, + }; + } + + upstreamCache.values.set(cacheKey, value); + return value; +} + +/** + * Whether this deployment can embed Brain pages. An explicit upstream + * URL wins; Cloudflare Workers AI auto-wires only with no OpenAI or + * OpenRouter key; otherwise that provider key is required. Synthesis is + * not part of this gate. + */ +export async function isBrainEmbeddingAvailable(): Promise { + if (await resolveBrainEmbeddingsUpstream()) { + return true; + } + + return Boolean(await resolveBrainInferenceProvider()); +} diff --git a/packages/sdk/src/server/lib/brain-gateway-token.ts b/packages/sdk/src/server/lib/brain-gateway-token.ts new file mode 100644 index 0000000000..be27fa34e3 --- /dev/null +++ b/packages/sdk/src/server/lib/brain-gateway-token.ts @@ -0,0 +1,32 @@ +import { readFileSync } from 'node:fs'; + +import { Env } from '@roomote/env'; + +/** + * The shared secret the Brain presents to /api/brain/inference. Absent + * means the route is closed: a deployment without this configured has no + * Brain calling it, and an unauthenticated request must never fall through + * to a provider key. + */ +export function getBrainGatewayToken(): string | null { + const direct = Env.R_BRAIN_GATEWAY_TOKEN?.trim(); + + if (direct) { + return direct; + } + + // Compose brings the Brain up without anyone choosing a value: the + // container writes one to its volume on first boot and the app services + // mount that read-only. Absent simply means no Brain is wired up yet. + if (Env.R_BRAIN_GATEWAY_TOKEN_FILE) { + try { + return ( + readFileSync(Env.R_BRAIN_GATEWAY_TOKEN_FILE, 'utf8').trim() || null + ); + } catch { + return null; + } + } + + return null; +} diff --git a/packages/sdk/src/server/lib/brain-inference-model-names.ts b/packages/sdk/src/server/lib/brain-inference-model-names.ts new file mode 100644 index 0000000000..08317b4b95 --- /dev/null +++ b/packages/sdk/src/server/lib/brain-inference-model-names.ts @@ -0,0 +1,24 @@ +/** + * Same model under each provider's naming, so a request can be translated + * without changing which model actually runs. `kind` matters: a chat model + * is free to be overridden per deployment, an embedding model is not, + * because its output width is baked into the Brain's vector column at + * creation. + */ +export const MODEL_TRANSLATIONS: ReadonlyArray<{ + kind: 'embedding' | 'chat'; + openai: string; + openrouter: string; +}> = [ + { + kind: 'embedding', + openai: 'text-embedding-3-small', + openrouter: 'openai/text-embedding-3-small', + }, + { + kind: 'embedding', + openai: 'text-embedding-3-large', + openrouter: 'openai/text-embedding-3-large', + }, + { kind: 'chat', openai: 'gpt-5.6-luna', openrouter: 'openai/gpt-5.6-luna' }, +]; diff --git a/packages/sdk/src/server/lib/brain-inference-model-summary.ts b/packages/sdk/src/server/lib/brain-inference-model-summary.ts new file mode 100644 index 0000000000..53677746e4 --- /dev/null +++ b/packages/sdk/src/server/lib/brain-inference-model-summary.ts @@ -0,0 +1,60 @@ +import { Env } from '@roomote/env'; + +import { MODEL_TRANSLATIONS } from './brain-inference-model-names'; +import type { BrainInferenceProviderId } from './brain-inference-provider'; + +export type BrainModelSummary = { + /** In the serving provider's own naming, as the gateway will forward it. */ + synthesisModel: string; + synthesisSource: 'default' | 'override'; + embeddingModel: string; + /** Known for the stock embedding models; null for an operator's own. */ + embeddingDimensions: number | null; +}; + +function knownEmbeddingDimensions(model: string): number | null { + if (model.includes('text-embedding-3-large')) { + return 3072; + } + + if (model.includes('text-embedding-3-small')) { + return 1536; + } + + // Cloudflare Workers AI `@cf/baai/bge-m3` and Infinity `BAAI/bge-m3`. + if (model.includes('bge-m3')) { + return 1024; + } + + return null; +} + +/** + * Models the Brain is effectively running, for display. Synthesis follows + * `R_BRAIN_MODEL` or the translation table; embeddings follow + * `R_BRAIN_EMBEDDING_MODEL` (create-time). + */ +export function describeBrainModels( + providerId: BrainInferenceProviderId, +): BrainModelSummary { + const override = Env.R_BRAIN_MODEL?.trim(); + const chat = MODEL_TRANSLATIONS.find((entry) => entry.kind === 'chat')!; + const stockEmbedding = MODEL_TRANSLATIONS.find( + (entry) => entry.kind === 'embedding', + )!; + const embeddingModel = + Env.R_BRAIN_EMBEDDING_MODEL?.trim() || + (providerId === 'openrouter' + ? stockEmbedding.openrouter + : stockEmbedding.openai); + + return { + synthesisModel: + override || (providerId === 'openrouter' ? chat.openrouter : chat.openai), + synthesisSource: override ? 'override' : 'default', + embeddingModel, + embeddingDimensions: + Env.R_BRAIN_EMBEDDING_DIMENSIONS ?? + knownEmbeddingDimensions(embeddingModel), + }; +} diff --git a/packages/sdk/src/server/lib/brain-inference-models.ts b/packages/sdk/src/server/lib/brain-inference-models.ts new file mode 100644 index 0000000000..6d149ee5ce --- /dev/null +++ b/packages/sdk/src/server/lib/brain-inference-models.ts @@ -0,0 +1,39 @@ +import { Env } from '@roomote/env'; + +import { MODEL_TRANSLATIONS } from './brain-inference-model-names'; +import type { ResolvedBrainInference } from './brain-inference-provider'; + +/** + * Rewrite the model name the Brain asked for into the one this deployment + * should actually call. + * + * The Brain always speaks OpenAI's names because it is an OpenAI-compatible + * client. For the synthesis model only, `R_BRAIN_MODEL` wins outright. An + * unrecognized model with no override passes through untouched. + */ +export function mapBrainModelName( + requested: string, + resolved: ResolvedBrainInference, +): string { + const normalized = requested.trim(); + const row = MODEL_TRANSLATIONS.find( + (entry) => entry.openai === normalized || entry.openrouter === normalized, + ); + + if (!row) { + return normalized; + } + + // Deliberately no embedding override here. Which embedding model runs is + // fixed when the Brain is created, because its output width sizes a vector + // column that cannot be resized in place. + if (row.kind === 'chat') { + const override = Env.R_BRAIN_MODEL?.trim(); + + if (override) { + return override; + } + } + + return resolved.providerId === 'openrouter' ? row.openrouter : row.openai; +} diff --git a/packages/sdk/src/server/lib/brain-inference-provider.ts b/packages/sdk/src/server/lib/brain-inference-provider.ts new file mode 100644 index 0000000000..c96c464bea --- /dev/null +++ b/packages/sdk/src/server/lib/brain-inference-provider.ts @@ -0,0 +1,81 @@ +/** + * Which provider the Brain's own inference runs through. + * + * The Brain holds no provider credential. Roomote resolves the key per + * request from Settings first, then the deployment environment. + */ + +import { resolveModelProviderEnvValue } from '@roomote/db/server'; + +export type BrainInferenceProviderId = 'openrouter' | 'openai'; + +export type ResolvedBrainInference = { + providerId: BrainInferenceProviderId; + apiKey: string; +}; + +/** + * OpenRouter first mirrors the Brain container's default, so a deployment + * with both keys behaves the same through the gateway or a direct call. + */ +const BRAIN_PROVIDER_PREFERENCE: readonly BrainInferenceProviderId[] = [ + 'openrouter', + 'openai', +]; + +/** Brain-specific keys win so Memory can be billed apart from task inference. */ +const BRAIN_PROVIDER_ENV_VAR_NAMES: Record< + BrainInferenceProviderId, + readonly string[] +> = { + openrouter: ['R_BRAIN_OPENROUTER_API_KEY', 'OPENROUTER_API_KEY'], + openai: ['R_BRAIN_OPENAI_API_KEY', 'OPENAI_API_KEY'], +}; + +/** + * Resolution reads persisted settings. A 30s cache keeps a backfill from + * issuing a database round trip per page; rotating a key still takes effect + * promptly. + */ +const PROVIDER_CACHE_TTL_MS = 30_000; + +let providerCache: { + value: ResolvedBrainInference | null; + expiresAtMs: number; +} | null = null; + +/** Drop the cached provider, so the next call re-reads settings. */ +export function resetBrainInferenceProviderCache(): void { + providerCache = null; +} + +export async function resolveBrainInferenceProvider(): Promise { + const cached = providerCache; + + if (cached && cached.expiresAtMs > Date.now()) { + return cached.value; + } + + const resolved = await readBrainInferenceProvider(); + + providerCache = { + value: resolved, + expiresAtMs: Date.now() + PROVIDER_CACHE_TTL_MS, + }; + + return resolved; +} + +async function readBrainInferenceProvider(): Promise { + for (const providerId of BRAIN_PROVIDER_PREFERENCE) { + const apiKey = await resolveModelProviderEnvValue( + BRAIN_PROVIDER_ENV_VAR_NAMES[providerId], + ); + + if (apiKey?.trim()) { + return { providerId, apiKey: apiKey.trim() }; + } + } + + return null; +} diff --git a/packages/sdk/src/server/lib/brain-inference.ts b/packages/sdk/src/server/lib/brain-inference.ts index 7ccbe96a6f..75b9e72131 100644 --- a/packages/sdk/src/server/lib/brain-inference.ts +++ b/packages/sdk/src/server/lib/brain-inference.ts @@ -1,277 +1,30 @@ /** * Which provider the Brain's own inference runs through, and under which - * model names. - * - * The Brain container holds no provider credential. It is configured with a - * base URL pointing back at this deployment plus a shared gateway token, and - * every embedding and synthesis call arrives at /api/brain/inference for - * Roomote to forward. The real key is resolved here, per request, from - * wherever an admin configured it: Settings first (encrypted in the database), - * then the deployment's environment. Rotating it takes effect on the next - * call, with no restart and no key ever reaching the Brain. - */ - -import { readFileSync } from 'node:fs'; - -import { resolveModelProviderEnvValue } from '@roomote/db/server'; -import { Env } from '@roomote/env'; - -/** - * Same model under each provider's naming, so a request can be translated - * without changing which model actually runs. `kind` matters: a chat model is - * free to be overridden per deployment, an embedding model is not, because - * its output width is baked into the Brain's vector column at creation. - */ -const MODEL_TRANSLATIONS: ReadonlyArray<{ - kind: 'embedding' | 'chat'; - openai: string; - openrouter: string; -}> = [ - { - kind: 'embedding', - openai: 'text-embedding-3-small', - openrouter: 'openai/text-embedding-3-small', - }, - { - kind: 'embedding', - openai: 'text-embedding-3-large', - openrouter: 'openai/text-embedding-3-large', - }, - { kind: 'chat', openai: 'gpt-5.6-luna', openrouter: 'openai/gpt-5.6-luna' }, -]; - -export type BrainInferenceProviderId = 'openrouter' | 'openai'; - -/** - * Preference order when more than one provider is configured. OpenRouter - * first mirrors the Brain container's own default, so a deployment that has - * both keys behaves the same whether it routes through the gateway or (on the - * env-var fallback path) talks to a provider directly. - */ -const BRAIN_PROVIDER_PREFERENCE: readonly BrainInferenceProviderId[] = [ - 'openrouter', - 'openai', -]; - -/** - * Brain-specific keys win over the deployment's general model-provider keys, - * so an operator can bill the Brain separately from task inference without - * the two settings fighting. - */ -const BRAIN_PROVIDER_ENV_VAR_NAMES: Record< - BrainInferenceProviderId, - readonly string[] -> = { - openrouter: ['R_BRAIN_OPENROUTER_API_KEY', 'OPENROUTER_API_KEY'], - openai: ['R_BRAIN_OPENAI_API_KEY', 'OPENAI_API_KEY'], -}; - -/** - * Resolution reads persisted deployment settings, so without this every - * embedding call is a database round trip. A backfill embeds a page at a - * time and would issue hundreds in a burst against the same database the - * drainer is working. Short enough that rotating a key in Settings still - * takes effect promptly; misses are cheap either way. - */ -const PROVIDER_CACHE_TTL_MS = 30_000; - -let providerCache: { - value: ResolvedBrainInference | null; - expiresAtMs: number; -} | null = null; - -/** Drop the cached provider, so the next call re-reads settings. */ -export function resetBrainInferenceProviderCache(): void { - providerCache = null; -} - -export type ResolvedBrainInference = { - providerId: BrainInferenceProviderId; - apiKey: string; -}; - -/** - * Resolve the provider the Brain's inference should use right now, or null - * when this deployment has no usable credential. Null is a normal state: it - * means the Brain is deployed but nobody has configured a model provider yet. - */ -export async function resolveBrainInferenceProvider(): Promise { - const cached = providerCache; - - if (cached && cached.expiresAtMs > Date.now()) { - return cached.value; - } - - const resolved = await readBrainInferenceProvider(); - - providerCache = { - value: resolved, - expiresAtMs: Date.now() + PROVIDER_CACHE_TTL_MS, - }; - - return resolved; -} - -/** - * Whether this deployment has a way to embed Brain pages at all — the - * readiness bar for running collectors and draining the memory outbox. - * - * Provider-agnostic by design. A dedicated embedder is the normal path: - * `R_BRAIN_EMBEDDINGS_UPSTREAM_URL` points embeddings at a self-run model - * (the shared Modal endpoint for every hosting-managed tenant; the local - * Infinity service for the self-host compose local-inference profile), so no - * model-provider key is involved in embedding at all. Only when no embedder - * is configured does this fall back to requiring an embeddings-capable - * provider key (the direct self-host path, where OpenAI/OpenRouter does the - * embedding). - * - * Synthesis is deliberately NOT part of this gate: it rides the deployment's - * helper model through the inference gateway and therefore works with whatever - * provider runs the deployment's tasks — Anthropic, OpenAI, OpenRouter, or a - * minted trial key alike. Gating on an OpenAI/OpenRouter key would have locked - * every Anthropic-only or trial tenant out of Memory even though both halves - * of its inference are actually available. - */ -export async function isBrainEmbeddingAvailable(): Promise { - if (Env.R_BRAIN_EMBEDDINGS_UPSTREAM_URL?.trim()) { - return true; - } - - return Boolean(await resolveBrainInferenceProvider()); -} - -async function readBrainInferenceProvider(): Promise { - for (const providerId of BRAIN_PROVIDER_PREFERENCE) { - const apiKey = await resolveModelProviderEnvValue( - BRAIN_PROVIDER_ENV_VAR_NAMES[providerId], - ); - - if (apiKey?.trim()) { - return { providerId, apiKey: apiKey.trim() }; - } - } - - return null; -} - -/** - * Rewrite the model name the Brain asked for into the one this deployment - * should actually call. - * - * Two things happen here. The Brain always speaks OpenAI's names because it - * is configured as an OpenAI-compatible client, so its request is translated - * onto whichever provider is serving it. And for the synthesis model only, an - * operator override (`R_BRAIN_MODEL`, written in that provider's own naming) - * wins outright, which is what makes the synthesis model a deployment setting - * rather than a property of the Brain container. - * - * An unrecognized model with no override passes through untouched rather than - * being rejected: an operator who set GBRAIN_MODEL on the container directly - * has already taken ownership of matching it to their provider. - */ -export function mapBrainModelName( - requested: string, - resolved: ResolvedBrainInference, -): string { - const normalized = requested.trim(); - const row = MODEL_TRANSLATIONS.find( - (entry) => entry.openai === normalized || entry.openrouter === normalized, - ); - - if (!row) { - return normalized; - } - - // Deliberately no embedding override here. Which embedding model runs is - // fixed when the Brain is created, because its output width sizes a vector - // column that cannot be resized in place; substituting a different one at - // request time would send vectors of the wrong width into that column. The - // Brain is told its embedding model at init instead, and this only - // translates the name it asks for into the serving provider's spelling. - if (row.kind === 'chat') { - const override = Env.R_BRAIN_MODEL?.trim(); - - if (override) { - return override; - } - } - - return resolved.providerId === 'openrouter' ? row.openrouter : row.openai; -} - -export type BrainModelSummary = { - /** In the serving provider's own naming, as the gateway will forward it. */ - synthesisModel: string; - synthesisSource: 'default' | 'override'; - embeddingModel: string; - /** Known for the stock embedding models; null for an operator's own. */ - embeddingDimensions: number | null; -}; - -/** - * The models the Brain is effectively running, for display. Mirrors - * `mapBrainModelName`: the synthesis model is whatever the override or the - * translation table produces at forward time, while the embedding model is - * whatever the container was told at init (`R_BRAIN_EMBEDDING_MODEL`, else - * the stock default), which is why the Settings page presents it as fixed. - */ -export function describeBrainModels( - providerId: BrainInferenceProviderId, -): BrainModelSummary { - // Truthiness on the trimmed value, exactly like `mapBrainModelName`: a - // whitespace-only override falls through to the default in both places. - const override = Env.R_BRAIN_MODEL?.trim(); - const chat = MODEL_TRANSLATIONS.find((entry) => entry.kind === 'chat')!; - const stockEmbedding = MODEL_TRANSLATIONS.find( - (entry) => entry.kind === 'embedding', - )!; - const embeddingModel = - Env.R_BRAIN_EMBEDDING_MODEL?.trim() || - (providerId === 'openrouter' - ? stockEmbedding.openrouter - : stockEmbedding.openai); - const embeddingDimensions = - Env.R_BRAIN_EMBEDDING_DIMENSIONS ?? - (embeddingModel.includes('text-embedding-3-large') - ? 3072 - : embeddingModel.includes('text-embedding-3-small') - ? 1536 - : null); - - return { - synthesisModel: - override || (providerId === 'openrouter' ? chat.openrouter : chat.openai), - synthesisSource: override ? 'override' : 'default', - embeddingModel, - embeddingDimensions, - }; -} - -/** - * The shared secret the Brain presents to /api/brain/inference. Absent means - * the route is closed: a deployment without this configured has no Brain - * calling it, and an unauthenticated request must never fall through to a - * provider key. - */ -export function getBrainGatewayToken(): string | null { - const direct = Env.R_BRAIN_GATEWAY_TOKEN?.trim(); - - if (direct) { - return direct; - } - - // Compose brings the Brain up without anyone choosing a value: the - // container writes one to its volume on first boot and the app services - // mount that read-only. Absent simply means no Brain is wired up yet. - if (Env.R_BRAIN_GATEWAY_TOKEN_FILE) { - try { - return ( - readFileSync(Env.R_BRAIN_GATEWAY_TOKEN_FILE, 'utf8').trim() || null - ); - } catch { - return null; - } - } - - return null; -} + * model names. Split by responsibility in the sibling modules; this file + * is the public surface `export * from './lib/brain-inference'` re-exports. + */ + +export { + type BrainInferenceProviderId, + type ResolvedBrainInference, + resetBrainInferenceProviderCache, + resolveBrainInferenceProvider, +} from './brain-inference-provider'; + +export { + CLOUDFLARE_WORKERS_AI_EMBEDDING_DIMENSIONS, + CLOUDFLARE_WORKERS_AI_EMBEDDING_MODEL, + type BrainEmbeddingsUpstream, + isBrainEmbeddingAvailable, + resetBrainEmbeddingsUpstreamCache, + resolveBrainEmbeddingsUpstream, +} from './brain-embeddings-upstream'; + +export { mapBrainModelName } from './brain-inference-models'; + +export { + type BrainModelSummary, + describeBrainModels, +} from './brain-inference-model-summary'; + +export { getBrainGatewayToken } from './brain-gateway-token';