This commit is contained in:
2026-05-14 19:24:50 +00:00
parent af0628867f
commit a7f218e182
63 changed files with 10539 additions and 0 deletions

31
apps/server/src/auth.ts Normal file
View File

@@ -0,0 +1,31 @@
import type { FastifyInstance, FastifyRequest } from 'fastify';
declare module 'fastify' {
interface FastifyRequest {
user?: string;
}
}
const PUBLIC_PATHS = new Set<string>(['/api/health']);
export function registerAuth(app: FastifyInstance): void {
app.addHook('onRequest', async (req, reply) => {
if (!req.url.startsWith('/api')) return;
if (PUBLIC_PATHS.has(req.routeOptions.url ?? req.url.split('?')[0]!)) return;
const header = req.headers['remote-user'];
const user = Array.isArray(header) ? header[0] : header;
if (!user || user.trim() === '') {
reply.code(401).send({ error: 'unauthenticated' });
return reply;
}
req.user = user.trim();
});
}
export function requireUser(req: FastifyRequest): string {
if (!req.user) {
throw new Error('user not set on request — auth hook must run first');
}
return req.user;
}

28
apps/server/src/config.ts Normal file
View File

@@ -0,0 +1,28 @@
import { z } from 'zod';
const ConfigSchema = z.object({
NODE_ENV: z.enum(['development', 'production', 'test']).default('development'),
PORT: z.coerce.number().int().positive().default(3000),
HOST: z.string().default('0.0.0.0'),
DATABASE_URL: z.string().url(),
LLAMA_SWAP_URL: z.string().url(),
PROJECT_ROOT_WHITELIST: z.string().default('/opt'),
DEFAULT_MODEL: z.string().default('qwen3.6-35b-a3b-mxfp4'),
LOG_LEVEL: z.string().default('info'),
});
export type Config = z.infer<typeof ConfigSchema>;
let cached: Config | null = null;
export function loadConfig(): Config {
if (cached) return cached;
const parsed = ConfigSchema.safeParse(process.env);
if (!parsed.success) {
console.error('Invalid environment configuration:');
console.error(parsed.error.flatten().fieldErrors);
process.exit(1);
}
cached = parsed.data;
return cached;
}

45
apps/server/src/db.ts Normal file
View File

@@ -0,0 +1,45 @@
import postgres from 'postgres';
import { readFile } from 'node:fs/promises';
import { fileURLToPath } from 'node:url';
import { dirname, resolve } from 'node:path';
import type { Config } from './config.js';
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
export type Sql = ReturnType<typeof postgres>;
let sqlInstance: Sql | null = null;
export function getSql(config: Config): Sql {
if (sqlInstance) return sqlInstance;
sqlInstance = postgres(config.DATABASE_URL, {
max: 10,
idle_timeout: 30,
connect_timeout: 10,
onnotice: () => {},
});
return sqlInstance;
}
export async function applySchema(sql: Sql): Promise<void> {
const schemaPath = resolve(__dirname, 'schema.sql');
const ddl = await readFile(schemaPath, 'utf8');
await sql.unsafe(ddl);
}
export async function pingDb(sql: Sql): Promise<boolean> {
try {
await sql`SELECT 1`;
return true;
} catch {
return false;
}
}
export async function closeDb(): Promise<void> {
if (sqlInstance) {
await sqlInstance.end({ timeout: 5 });
sqlInstance = null;
}
}

114
apps/server/src/index.ts Normal file
View File

@@ -0,0 +1,114 @@
import Fastify from 'fastify';
import fastifyStatic from '@fastify/static';
import fastifyWebsocket from '@fastify/websocket';
import { existsSync } from 'node:fs';
import { resolve } from 'node:path';
import { loadConfig } from './config.js';
import { getSql, applySchema, pingDb, closeDb } from './db.js';
import { registerAuth } from './auth.js';
import { registerProjectRoutes } from './routes/projects.js';
import { registerSessionRoutes } from './routes/sessions.js';
import { registerSettingsRoutes } from './routes/settings.js';
import { registerMessageRoutes } from './routes/messages.js';
import { registerWebSocket } from './routes/ws.js';
import { registerModelRoutes } from './routes/models.js';
import { createInferenceRunner } from './services/inference.js';
import { createBroker } from './services/broker.js';
async function main() {
const config = loadConfig();
const app = Fastify({
logger: { level: config.LOG_LEVEL },
});
const sql = getSql(config);
await applySchema(sql);
app.log.info('database schema applied');
await app.register(fastifyWebsocket);
registerAuth(app);
app.get('/api/health', async () => {
const dbOk = await pingDb(sql);
return { status: dbOk ? 'ok' : 'degraded', db: dbOk };
});
registerProjectRoutes(app, sql, config);
registerSessionRoutes(app, sql, config);
registerSettingsRoutes(app, sql);
registerModelRoutes(app, config);
const broker = createBroker();
const inference = createInferenceRunner({
sql,
config,
log: app.log,
publish: (sessionId, frame) => {
broker.publish(sessionId, frame as unknown as Record<string, unknown> & { type: string });
},
});
registerMessageRoutes(app, sql, {
onSend: (sessionId, _userId, assistantId) => {
inference.enqueue(sessionId, assistantId);
},
publishUserMessage: (sessionId, userMessageId, content) => {
broker.publish(sessionId, {
type: 'message_started',
message_id: userMessageId,
role: 'user',
});
broker.publish(sessionId, {
type: 'delta',
message_id: userMessageId,
content,
});
broker.publish(sessionId, {
type: 'message_complete',
message_id: userMessageId,
});
},
});
registerWebSocket(app, sql, broker);
const webDist = process.env.WEB_DIST_PATH ?? resolve(process.cwd(), '../web/dist');
if (existsSync(webDist)) {
await app.register(fastifyStatic, {
root: webDist,
prefix: '/',
wildcard: false,
});
app.setNotFoundHandler((req, reply) => {
if (req.url.startsWith('/api')) {
reply.code(404).send({ error: 'not found' });
return;
}
reply.sendFile('index.html');
});
app.log.info(`serving static frontend from ${webDist}`);
}
const shutdown = async (signal: string) => {
app.log.info(`received ${signal}, shutting down`);
try {
await app.close();
await closeDb();
process.exit(0);
} catch (err) {
app.log.error(err);
process.exit(1);
}
};
process.on('SIGINT', () => void shutdown('SIGINT'));
process.on('SIGTERM', () => void shutdown('SIGTERM'));
await app.listen({ port: config.PORT, host: config.HOST });
app.log.info(`boocode server listening on http://${config.HOST}:${config.PORT}`);
}
main().catch((err) => {
console.error('Fatal startup error:', err);
process.exit(1);
});

View File

@@ -0,0 +1,83 @@
import type { FastifyInstance } from 'fastify';
import { z } from 'zod';
import type { Sql } from '../db.js';
import type { Message, Session } from '../types/api.js';
const SendBody = z.object({
content: z.string().min(1).max(64_000),
});
interface MessageHandlers {
onSend: (sessionId: string, userMessageId: string, assistantMessageId: string) => void;
publishUserMessage: (
sessionId: string,
userMessageId: string,
content: string
) => void;
}
export function registerMessageRoutes(
app: FastifyInstance,
sql: Sql,
handlers: MessageHandlers
): void {
app.get<{ Params: { id: string } }>(
'/api/sessions/:id/messages',
async (req, reply) => {
const session = await sql<Session[]>`SELECT id FROM sessions WHERE id = ${req.params.id}`;
if (session.length === 0) {
reply.code(404);
return { error: 'session not found' };
}
const rows = await sql<Message[]>`
SELECT id, session_id, role, content, tool_calls, tool_results, status, last_seq, created_at
FROM messages
WHERE session_id = ${req.params.id}
ORDER BY created_at ASC, id ASC
`;
return rows;
}
);
app.post<{ Params: { id: string } }>(
'/api/sessions/:id/messages',
async (req, reply) => {
const parsed = SendBody.safeParse(req.body);
if (!parsed.success) {
reply.code(400);
return { error: 'invalid body', details: parsed.error.flatten() };
}
const session = await sql<Session[]>`SELECT id FROM sessions WHERE id = ${req.params.id}`;
if (session.length === 0) {
reply.code(404);
return { error: 'session not found' };
}
const result = await sql.begin(async (tx) => {
const [userMsg] = await tx<{ id: string }[]>`
INSERT INTO messages (session_id, role, content, status, created_at)
VALUES (${req.params.id}, 'user', ${parsed.data.content}, 'complete', clock_timestamp())
RETURNING id
`;
const [assistantMsg] = await tx<{ id: string }[]>`
INSERT INTO messages (session_id, role, content, status, created_at)
VALUES (${req.params.id}, 'assistant', '', 'streaming', clock_timestamp())
RETURNING id
`;
await tx`UPDATE sessions SET updated_at = NOW() WHERE id = ${req.params.id}`;
return { user_message_id: userMsg!.id, assistant_message_id: assistantMsg!.id };
});
handlers.publishUserMessage(
req.params.id,
result.user_message_id,
parsed.data.content
);
handlers.onSend(req.params.id, result.user_message_id, result.assistant_message_id);
reply.code(202);
return result;
}
);
}

View File

@@ -0,0 +1,27 @@
import type { FastifyInstance } from 'fastify';
import type { Config } from '../config.js';
import type { ModelInfo } from '../types/api.js';
interface LlamaSwapModelsResponse {
data?: ModelInfo[];
}
export function registerModelRoutes(app: FastifyInstance, config: Config): void {
app.get('/api/models', async (_req, reply) => {
try {
const res = await fetch(`${config.LLAMA_SWAP_URL}/v1/models`);
if (!res.ok) {
reply.code(502);
return { error: `llama-swap returned ${res.status}` };
}
const parsed = (await res.json()) as LlamaSwapModelsResponse;
return parsed.data ?? [];
} catch (err) {
reply.code(502);
return {
error: 'failed to reach llama-swap',
details: err instanceof Error ? err.message : String(err),
};
}
});
}

View File

@@ -0,0 +1,130 @@
import type { FastifyInstance } from 'fastify';
import { z } from 'zod';
import { realpath, stat, readdir, access } from 'node:fs/promises';
import { basename, resolve, sep } from 'node:path';
import type { Sql } from '../db.js';
import type { Config } from '../config.js';
import type { Project, AvailableProject } from '../types/api.js';
const AddProjectBody = z.object({
path: z.string().min(1),
name: z.string().min(1).optional(),
});
async function isDir(path: string): Promise<boolean> {
try {
const s = await stat(path);
return s.isDirectory();
} catch {
return false;
}
}
async function resolveProjectPath(
raw: string,
whitelist: string
): Promise<{ real: string; name: string } | { error: string }> {
if (!raw.startsWith('/')) return { error: 'path must be absolute' };
let real: string;
try {
real = await realpath(raw);
} catch {
return { error: 'path does not exist' };
}
const whitelistReal = await realpath(whitelist);
if (real !== whitelistReal && !real.startsWith(whitelistReal + sep)) {
return { error: `path must be under ${whitelist}` };
}
if (!(await isDir(real))) return { error: 'path is not a directory' };
return { real, name: basename(real) };
}
export function registerProjectRoutes(
app: FastifyInstance,
sql: Sql,
config: Config
): void {
app.get('/api/projects', async () => {
const rows = await sql<Project[]>`
SELECT id, name, path, added_at, last_session_id
FROM projects
ORDER BY added_at DESC
`;
return rows;
});
app.post('/api/projects', async (req, reply) => {
const parsed = AddProjectBody.safeParse(req.body);
if (!parsed.success) {
reply.code(400);
return { error: 'invalid body', details: parsed.error.flatten() };
}
const resolved = await resolveProjectPath(parsed.data.path, config.PROJECT_ROOT_WHITELIST);
if ('error' in resolved) {
reply.code(400);
return { error: resolved.error };
}
const name = parsed.data.name?.trim() || resolved.name;
try {
const [row] = await sql<Project[]>`
INSERT INTO projects (name, path)
VALUES (${name}, ${resolved.real})
RETURNING id, name, path, added_at, last_session_id
`;
reply.code(201);
return row;
} catch (err) {
if (err instanceof Error && err.message.includes('duplicate key')) {
reply.code(409);
return { error: 'project already exists' };
}
throw err;
}
});
app.delete<{ Params: { id: string } }>('/api/projects/:id', async (req, reply) => {
const id = req.params.id;
const result = await sql`DELETE FROM projects WHERE id = ${id}`;
if (result.count === 0) {
reply.code(404);
return { error: 'not found' };
}
reply.code(204);
return null;
});
app.get('/api/projects/available', async () => {
const whitelist = await realpath(config.PROJECT_ROOT_WHITELIST);
let entries: string[];
try {
entries = await readdir(whitelist);
} catch {
return [] as AvailableProject[];
}
const existing = await sql<{ path: string }[]>`SELECT path FROM projects`;
const existingSet = new Set(existing.map((r) => r.path));
const out: AvailableProject[] = [];
for (const entry of entries) {
const full = resolve(whitelist, entry);
let real: string;
try {
real = await realpath(full);
} catch {
continue;
}
if (real !== whitelist && !real.startsWith(whitelist + sep)) continue;
if (existingSet.has(real)) continue;
if (!(await isDir(real))) continue;
try {
await access(resolve(real, '.git'));
} catch {
continue;
}
out.push({ path: real, name: basename(real) });
}
out.sort((a, b) => a.name.localeCompare(b.name));
return out;
});
}

View File

@@ -0,0 +1,138 @@
import type { FastifyInstance } from 'fastify';
import { z } from 'zod';
import type { Sql } from '../db.js';
import type { Config } from '../config.js';
import type { Session } from '../types/api.js';
import { getSetting } from './settings.js';
const CreateBody = z.object({
name: z.string().min(1).max(200).optional(),
model: z.string().min(1).max(200).optional(),
system_prompt: z.string().max(8000).optional(),
});
const PatchBody = z.object({
name: z.string().min(1).max(200).optional(),
model: z.string().min(1).max(200).optional(),
system_prompt: z.string().max(8000).optional(),
});
async function resolveDefaultModel(sql: Sql, config: Config): Promise<string> {
const fromDb = await getSetting<string>(sql, 'default_model');
if (typeof fromDb === 'string' && fromDb.length > 0) return fromDb;
return config.DEFAULT_MODEL;
}
export function registerSessionRoutes(
app: FastifyInstance,
sql: Sql,
config: Config
): void {
app.get<{ Params: { id: string } }>(
'/api/projects/:id/sessions',
async (req, reply) => {
const project = await sql`SELECT id FROM projects WHERE id = ${req.params.id}`;
if (project.length === 0) {
reply.code(404);
return { error: 'project not found' };
}
const rows = await sql<Session[]>`
SELECT id, project_id, name, model, system_prompt, created_at, updated_at
FROM sessions
WHERE project_id = ${req.params.id}
ORDER BY updated_at DESC
`;
return rows;
}
);
app.post<{ Params: { id: string } }>(
'/api/projects/:id/sessions',
async (req, reply) => {
const parsed = CreateBody.safeParse(req.body ?? {});
if (!parsed.success) {
reply.code(400);
return { error: 'invalid body', details: parsed.error.flatten() };
}
const project = await sql`SELECT id FROM projects WHERE id = ${req.params.id}`;
if (project.length === 0) {
reply.code(404);
return { error: 'project not found' };
}
let model = parsed.data.model;
if (!model) {
const lastUsed = await sql<{ model: string }[]>`
SELECT model FROM sessions
WHERE project_id = ${req.params.id}
ORDER BY created_at DESC
LIMIT 1
`;
model = lastUsed[0]?.model ?? (await resolveDefaultModel(sql, config));
}
const name = parsed.data.name ?? 'New session';
const systemPrompt = parsed.data.system_prompt ?? '';
const [row] = await sql<Session[]>`
INSERT INTO sessions (project_id, name, model, system_prompt)
VALUES (${req.params.id}, ${name}, ${model}, ${systemPrompt})
RETURNING id, project_id, name, model, system_prompt, created_at, updated_at
`;
reply.code(201);
return row;
}
);
app.get<{ Params: { id: string } }>('/api/sessions/:id', async (req, reply) => {
const rows = await sql<Session[]>`
SELECT id, project_id, name, model, system_prompt, created_at, updated_at
FROM sessions WHERE id = ${req.params.id}
`;
if (rows.length === 0) {
reply.code(404);
return { error: 'session not found' };
}
return rows[0];
});
app.patch<{ Params: { id: string } }>(
'/api/sessions/:id',
async (req, reply) => {
const parsed = PatchBody.safeParse(req.body);
if (!parsed.success) {
reply.code(400);
return { error: 'invalid body', details: parsed.error.flatten() };
}
const { name, model, system_prompt } = parsed.data;
const rows = await sql<Session[]>`
UPDATE sessions
SET
name = COALESCE(${name ?? null}, name),
model = COALESCE(${model ?? null}, model),
system_prompt = COALESCE(${system_prompt ?? null}, system_prompt),
updated_at = NOW()
WHERE id = ${req.params.id}
RETURNING id, project_id, name, model, system_prompt, created_at, updated_at
`;
if (rows.length === 0) {
reply.code(404);
return { error: 'session not found' };
}
return rows[0];
}
);
app.delete<{ Params: { id: string } }>(
'/api/sessions/:id',
async (req, reply) => {
const result = await sql`DELETE FROM sessions WHERE id = ${req.params.id}`;
if (result.count === 0) {
reply.code(404);
return { error: 'not found' };
}
reply.code(204);
return null;
}
);
}

View File

@@ -0,0 +1,49 @@
import type { FastifyInstance } from 'fastify';
import { z } from 'zod';
import type { Sql } from '../db.js';
export async function getSetting<T = unknown>(
sql: Sql,
key: string
): Promise<T | null> {
const rows = await sql<{ value: T }[]>`SELECT value FROM settings WHERE key = ${key}`;
return rows[0]?.value ?? null;
}
export async function setSetting(
sql: Sql,
key: string,
value: unknown
): Promise<void> {
await sql`
INSERT INTO settings (key, value)
VALUES (${key}, ${sql.json(value as never)})
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
`;
}
const PatchBody = z.record(z.string(), z.unknown());
export function registerSettingsRoutes(app: FastifyInstance, sql: Sql): void {
app.get('/api/settings', async () => {
const rows = await sql<{ key: string; value: unknown }[]>`SELECT key, value FROM settings`;
const out: Record<string, unknown> = {};
for (const r of rows) out[r.key] = r.value;
return out;
});
app.patch('/api/settings', async (req, reply) => {
const parsed = PatchBody.safeParse(req.body);
if (!parsed.success) {
reply.code(400);
return { error: 'invalid body', details: parsed.error.flatten() };
}
for (const [k, v] of Object.entries(parsed.data)) {
await setSetting(sql, k, v);
}
const rows = await sql<{ key: string; value: unknown }[]>`SELECT key, value FROM settings`;
const out: Record<string, unknown> = {};
for (const r of rows) out[r.key] = r.value;
return out;
});
}

View File

@@ -0,0 +1,45 @@
import type { FastifyInstance } from 'fastify';
import type { Sql } from '../db.js';
import type { Broker } from '../services/broker.js';
import type { Message } from '../types/api.js';
export function registerWebSocket(
app: FastifyInstance,
sql: Sql,
broker: Broker
): void {
app.get<{ Params: { id: string } }>(
'/api/ws/sessions/:id',
{ websocket: true },
async (socket, req) => {
const sessionId = req.params.id;
const session = await sql<{ id: string }[]>`SELECT id FROM sessions WHERE id = ${sessionId}`;
if (session.length === 0) {
socket.send(JSON.stringify({ type: 'error', error: 'session not found' }));
socket.close(1008, 'session not found');
return;
}
const messages = await sql<Message[]>`
SELECT id, session_id, role, content, tool_calls, tool_results, status, last_seq, created_at
FROM messages
WHERE session_id = ${sessionId}
ORDER BY created_at ASC, id ASC
`;
socket.send(JSON.stringify({ type: 'snapshot', messages }));
const unsubscribe = broker.subscribe(sessionId, (frame) => {
if (socket.readyState !== socket.OPEN) return;
try {
socket.send(JSON.stringify(frame));
} catch (err) {
app.log.warn({ err, sessionId }, 'ws send failed');
}
});
socket.on('close', () => unsubscribe());
socket.on('error', () => unsubscribe());
}
);
}

View File

@@ -0,0 +1,40 @@
CREATE TABLE IF NOT EXISTS projects (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name TEXT NOT NULL,
path TEXT NOT NULL UNIQUE,
added_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
last_session_id UUID
);
CREATE TABLE IF NOT EXISTS sessions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
name TEXT NOT NULL,
model TEXT NOT NULL,
system_prompt TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_sessions_project ON sessions(project_id, updated_at DESC);
CREATE TABLE IF NOT EXISTS messages (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
session_id UUID NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
role TEXT NOT NULL CHECK (role IN ('user', 'assistant', 'tool')),
content TEXT NOT NULL DEFAULT '',
tool_calls JSONB,
tool_results JSONB,
status TEXT NOT NULL DEFAULT 'complete' CHECK (status IN ('streaming', 'complete', 'failed')),
last_seq INT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, created_at);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value JSONB NOT NULL
);
INSERT INTO settings (key, value) VALUES ('default_model', '"qwen3.6-35b-a3b-mxfp4"') ON CONFLICT (key) DO NOTHING;

View File

@@ -0,0 +1,38 @@
export type Frame = Record<string, unknown> & { type: string };
export type Listener = (frame: Frame) => void;
export interface Broker {
publish(sessionId: string, frame: Frame): void;
subscribe(sessionId: string, listener: Listener): () => void;
}
export function createBroker(): Broker {
const topics = new Map<string, Set<Listener>>();
return {
publish(sessionId, frame) {
const set = topics.get(sessionId);
if (!set) return;
for (const listener of set) {
try {
listener(frame);
} catch {
// ignore listener errors so one bad subscriber doesn't break the rest
}
}
},
subscribe(sessionId, listener) {
let set = topics.get(sessionId);
if (!set) {
set = new Set();
topics.set(sessionId, set);
}
set.add(listener);
return () => {
const s = topics.get(sessionId);
if (!s) return;
s.delete(listener);
if (s.size === 0) topics.delete(sessionId);
};
},
};
}

View File

@@ -0,0 +1,471 @@
import type { FastifyBaseLogger } from 'fastify';
import type { Sql } from '../db.js';
import type { Config } from '../config.js';
import type { Message, Project, Session, ToolCall } from '../types/api.js';
import { ALL_TOOLS, TOOLS_BY_NAME, toolJsonSchemas } from './tools.js';
import { PathScopeError, resolveProjectRoot } from './path_guard.js';
const BASE_SYSTEM_PROMPT = (projectPath: string) =>
`You are BooCode Chat, a code investigation assistant. The user is working on a project located at ${projectPath}. Use the file-read tools (view_file, list_dir, grep, find_files) to investigate code when needed. Be concise. Cite file paths and line numbers when discussing code. Do not hallucinate file contents — read the file first. Tool results may be truncated; if so, narrow your query rather than guessing.`;
const DB_FLUSH_INTERVAL_MS = 500;
const MAX_TOOL_LOOP_DEPTH = 5;
export interface InferenceFrame {
type: 'message_started' | 'delta' | 'tool_call' | 'tool_result' | 'message_complete' | 'error';
message_id?: string;
tool_message_id?: string;
tool_call_id?: string;
role?: 'assistant' | 'tool' | 'user';
content?: string;
tool_call?: ToolCall;
output?: unknown;
truncated?: boolean;
error?: string;
}
export type FramePublisher = (sessionId: string, frame: InferenceFrame) => void;
interface OpenAiMessage {
role: 'system' | 'user' | 'assistant' | 'tool';
content: string | null;
tool_calls?: Array<{
id: string;
type: 'function';
function: { name: string; arguments: string };
}>;
tool_call_id?: string;
}
interface ChatCompletionDelta {
role?: string;
content?: string | null;
tool_calls?: Array<{
index: number;
id?: string;
type?: 'function';
function?: { name?: string; arguments?: string };
}>;
}
interface ChatCompletionChunk {
choices: Array<{
delta: ChatCompletionDelta;
finish_reason: string | null;
}>;
}
interface InferenceContext {
sql: Sql;
config: Config;
log: FastifyBaseLogger;
publish: FramePublisher;
}
export function buildMessagesPayload(
session: Session,
project: Project,
history: Message[]
): OpenAiMessage[] {
const out: OpenAiMessage[] = [];
let systemPrompt = BASE_SYSTEM_PROMPT(project.path);
if (session.system_prompt && session.system_prompt.trim().length > 0) {
systemPrompt += '\n\n' + session.system_prompt.trim();
}
out.push({ role: 'system', content: systemPrompt });
for (const m of history) {
if (m.role === 'assistant' && m.status === 'streaming') continue;
if (m.role === 'tool') {
const tr = m.tool_results;
if (!tr) continue;
const outputText = tr.error
? `error: ${tr.error}`
: typeof tr.output === 'string'
? tr.output
: JSON.stringify(tr.output);
out.push({
role: 'tool',
content: outputText,
tool_call_id: tr.tool_call_id,
});
continue;
}
if (m.role === 'assistant') {
const msg: OpenAiMessage = {
role: 'assistant',
content: m.content && m.content.length > 0 ? m.content : null,
};
if (m.tool_calls && m.tool_calls.length > 0) {
msg.tool_calls = m.tool_calls.map((tc) => ({
id: tc.id,
type: 'function' as const,
function: { name: tc.name, arguments: JSON.stringify(tc.args) },
}));
}
out.push(msg);
continue;
}
out.push({ role: 'user', content: m.content });
}
return out;
}
async function loadContext(
sql: Sql,
sessionId: string
): Promise<{ session: Session; project: Project; history: Message[] } | null> {
const sessionRows = await sql<Session[]>`
SELECT id, project_id, name, model, system_prompt, created_at, updated_at
FROM sessions WHERE id = ${sessionId}
`;
if (sessionRows.length === 0) return null;
const session = sessionRows[0]!;
const projectRows = await sql<Project[]>`
SELECT id, name, path, added_at, last_session_id
FROM projects WHERE id = ${session.project_id}
`;
if (projectRows.length === 0) return null;
const project = projectRows[0]!;
const history = await sql<Message[]>`
SELECT id, session_id, role, content, tool_calls, tool_results, status, last_seq, created_at
FROM messages
WHERE session_id = ${sessionId}
ORDER BY created_at ASC, id ASC
`;
return { session, project, history };
}
async function* sseLines(stream: ReadableStream<Uint8Array>): AsyncGenerator<string> {
const reader = stream.getReader();
const decoder = new TextDecoder('utf-8');
let buffer = '';
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let idx;
while ((idx = buffer.indexOf('\n')) >= 0) {
const line = buffer.slice(0, idx).replace(/\r$/, '');
buffer = buffer.slice(idx + 1);
if (line.length === 0) continue;
yield line;
}
}
if (buffer.length > 0) yield buffer;
} finally {
reader.releaseLock();
}
}
async function streamCompletion(
ctx: InferenceContext,
model: string,
messages: OpenAiMessage[],
includeTools: boolean,
onDelta: (content: string) => void
): Promise<{ finishReason: string | null; content: string; toolCalls: ToolCall[] }> {
const body: Record<string, unknown> = { model, messages, stream: true };
if (includeTools) {
body['tools'] = toolJsonSchemas();
body['tool_choice'] = 'auto';
}
const res = await fetch(`${ctx.config.LLAMA_SWAP_URL}/v1/chat/completions`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body),
});
if (!res.ok || !res.body) {
const text = await res.text().catch(() => '');
throw new Error(`llama-swap returned ${res.status}: ${text.slice(0, 200)}`);
}
let content = '';
let finishReason: string | null = null;
const toolCallsBuffer = new Map<number, { id: string; name: string; argsText: string }>();
for await (const line of sseLines(res.body)) {
if (!line.startsWith('data:')) continue;
const payload = line.slice(5).trim();
if (payload === '[DONE]') break;
let parsed: ChatCompletionChunk;
try {
parsed = JSON.parse(payload);
} catch {
continue;
}
const choice = parsed.choices?.[0];
if (!choice) continue;
const delta = choice.delta ?? {};
if (typeof delta.content === 'string' && delta.content.length > 0) {
content += delta.content;
onDelta(delta.content);
}
if (Array.isArray(delta.tool_calls)) {
for (const tc of delta.tool_calls) {
const idx = tc.index;
const existing = toolCallsBuffer.get(idx) ?? { id: '', name: '', argsText: '' };
if (tc.id) existing.id = tc.id;
if (tc.function?.name) existing.name = tc.function.name;
if (typeof tc.function?.arguments === 'string') existing.argsText += tc.function.arguments;
toolCallsBuffer.set(idx, existing);
}
}
if (choice.finish_reason) finishReason = choice.finish_reason;
}
const toolCalls: ToolCall[] = [];
for (const [, t] of [...toolCallsBuffer.entries()].sort(([a], [b]) => a - b)) {
let args: Record<string, unknown> = {};
if (t.argsText.length > 0) {
try {
args = JSON.parse(t.argsText);
} catch {
args = { _raw: t.argsText };
}
}
toolCalls.push({ id: t.id || `call_${toolCalls.length}`, name: t.name, args });
}
return { finishReason, content, toolCalls };
}
async function executeToolCall(
projectRoot: string,
toolCall: ToolCall
): Promise<{ output: unknown; truncated: boolean; error?: string }> {
const tool = TOOLS_BY_NAME[toolCall.name];
if (!tool) {
return { output: null, truncated: false, error: `unknown tool: ${toolCall.name}` };
}
const parsed = tool.inputSchema.safeParse(toolCall.args);
if (!parsed.success) {
return {
output: null,
truncated: false,
error: `invalid input: ${JSON.stringify(parsed.error.flatten())}`,
};
}
try {
const output = await tool.execute(parsed.data, projectRoot);
const truncated =
typeof output === 'object' && output !== null && 'truncated' in output
? Boolean((output as { truncated: unknown }).truncated)
: false;
return { output, truncated };
} catch (err) {
if (err instanceof PathScopeError) {
return { output: null, truncated: false, error: err.message };
}
return {
output: null,
truncated: false,
error: err instanceof Error ? err.message : String(err),
};
}
}
async function runAssistantTurn(
ctx: InferenceContext,
sessionId: string,
assistantMessageId: string,
depth: number
): Promise<void> {
if (depth > MAX_TOOL_LOOP_DEPTH) {
await ctx.sql`
UPDATE messages
SET status = 'failed', content = ${'tool loop depth exceeded'}
WHERE id = ${assistantMessageId}
`;
ctx.publish(sessionId, {
type: 'error',
message_id: assistantMessageId,
error: 'tool loop depth exceeded',
});
return;
}
const loaded = await loadContext(ctx.sql, sessionId);
if (!loaded) {
ctx.log.warn({ sessionId }, 'inference: session or project missing');
return;
}
const { session, project, history } = loaded;
const projectRoot = await resolveProjectRoot(project.path);
const messages = buildMessagesPayload(session, project, history);
ctx.publish(sessionId, {
type: 'message_started',
message_id: assistantMessageId,
role: 'assistant',
});
let accumulated = '';
let pendingFlushTimer: NodeJS.Timeout | null = null;
let flushPromise: Promise<unknown> = Promise.resolve();
const flushNow = () => {
if (pendingFlushTimer) {
clearTimeout(pendingFlushTimer);
pendingFlushTimer = null;
}
const snapshot = accumulated;
flushPromise = flushPromise.then(() =>
ctx.sql`UPDATE messages SET content = ${snapshot} WHERE id = ${assistantMessageId}`
);
};
const scheduleFlush = () => {
if (pendingFlushTimer) return;
pendingFlushTimer = setTimeout(() => {
pendingFlushTimer = null;
flushNow();
}, DB_FLUSH_INTERVAL_MS);
};
let content = '';
let finishReason: string | null = null;
let toolCalls: ToolCall[] = [];
try {
const result = await streamCompletion(
ctx,
session.model,
messages,
true,
(delta) => {
accumulated += delta;
ctx.publish(sessionId, {
type: 'delta',
message_id: assistantMessageId,
content: delta,
});
ctx.log.debug({ sessionId, delta }, 'inference delta');
scheduleFlush();
}
);
content = result.content;
finishReason = result.finishReason;
toolCalls = result.toolCalls;
} catch (err) {
if (pendingFlushTimer) {
clearTimeout(pendingFlushTimer);
pendingFlushTimer = null;
}
const errMsg = err instanceof Error ? err.message : String(err);
await ctx.sql`
UPDATE messages
SET status = 'failed', content = ${accumulated}
WHERE id = ${assistantMessageId}
`;
ctx.publish(sessionId, {
type: 'error',
message_id: assistantMessageId,
error: errMsg,
});
ctx.log.error({ err, sessionId, assistantMessageId }, 'inference failed');
return;
}
if (pendingFlushTimer) {
clearTimeout(pendingFlushTimer);
pendingFlushTimer = null;
}
await flushPromise;
if (toolCalls.length > 0) {
await ctx.sql`
UPDATE messages
SET content = ${content}, status = 'complete',
tool_calls = ${ctx.sql.json(toolCalls as never)}
WHERE id = ${assistantMessageId}
`;
for (const tc of toolCalls) {
ctx.publish(sessionId, {
type: 'tool_call',
message_id: assistantMessageId,
tool_call: tc,
});
}
ctx.publish(sessionId, {
type: 'message_complete',
message_id: assistantMessageId,
});
await Promise.all(
toolCalls.map(async (tc) => {
const [toolRow] = await ctx.sql<{ id: string }[]>`
INSERT INTO messages (session_id, role, content, status, created_at)
VALUES (${sessionId}, 'tool', '', 'complete', clock_timestamp())
RETURNING id
`;
const toolMessageId = toolRow!.id;
const result = await executeToolCall(projectRoot, tc);
const stored = {
tool_call_id: tc.id,
output: result.output,
truncated: result.truncated,
...(result.error ? { error: result.error } : {}),
};
await ctx.sql`
UPDATE messages
SET tool_results = ${ctx.sql.json(stored as never)}
WHERE id = ${toolMessageId}
`;
ctx.publish(sessionId, {
type: 'tool_result',
tool_message_id: toolMessageId,
tool_call_id: tc.id,
output: result.output,
truncated: result.truncated,
...(result.error ? { error: result.error } : {}),
});
})
);
const [nextAssistant] = await ctx.sql<{ id: string }[]>`
INSERT INTO messages (session_id, role, content, status, created_at)
VALUES (${sessionId}, 'assistant', '', 'streaming', clock_timestamp())
RETURNING id
`;
await runAssistantTurn(ctx, sessionId, nextAssistant!.id, depth + 1);
return;
}
await ctx.sql`
UPDATE messages
SET content = ${content}, status = 'complete'
WHERE id = ${assistantMessageId}
`;
ctx.publish(sessionId, {
type: 'message_complete',
message_id: assistantMessageId,
});
ctx.log.info({ sessionId, assistantMessageId, finishReason, chars: content.length }, 'inference complete');
}
export async function runInference(
ctx: InferenceContext,
sessionId: string,
assistantMessageId: string
): Promise<void> {
return runAssistantTurn(ctx, sessionId, assistantMessageId, 0);
}
export function createInferenceRunner(ctx: InferenceContext) {
return {
enqueue(sessionId: string, assistantMessageId: string) {
void runInference(ctx, sessionId, assistantMessageId).catch((err) => {
ctx.log.error({ err }, 'unhandled inference error');
});
},
};
}
// Reference to keep ALL_TOOLS imported for type checks if needed
export const _toolNames = ALL_TOOLS.map((t) => t.name);

View File

@@ -0,0 +1,39 @@
import { realpath } from 'node:fs/promises';
import { isAbsolute, resolve, sep } from 'node:path';
export class PathScopeError extends Error {
constructor(message: string) {
super(message);
this.name = 'PathScopeError';
}
}
export async function resolveProjectRoot(projectPath: string): Promise<string> {
try {
return await realpath(projectPath);
} catch {
throw new PathScopeError(`project path does not exist: ${projectPath}`);
}
}
export async function pathGuard(
projectRoot: string,
requested: string
): Promise<string> {
if (typeof requested !== 'string' || requested.length === 0) {
throw new PathScopeError('path is required');
}
const candidate = isAbsolute(requested) ? requested : resolve(projectRoot, requested);
let real: string;
try {
real = await realpath(candidate);
} catch {
throw new PathScopeError(`path does not exist: ${requested}`);
}
if (real !== projectRoot && !real.startsWith(projectRoot + sep)) {
throw new PathScopeError(
`path escapes project root: ${requested} -> ${real}`
);
}
return real;
}

View File

@@ -0,0 +1,371 @@
import { readFile, readdir, stat } from 'node:fs/promises';
import { resolve, basename, relative } from 'node:path';
import { spawn } from 'node:child_process';
import { z } from 'zod';
import { pathGuard, PathScopeError } from './path_guard.js';
const MAX_FILE_BYTES = 5 * 1024 * 1024;
const DEFAULT_VIEW_LINES = 200;
const MAX_GREP_RESULTS = 200;
const DEFAULT_GREP_RESULTS = 100;
const MAX_FIND_RESULTS = 200;
const DEFAULT_FIND_RESULTS = 100;
const MAX_DIR_ENTRIES = 500;
export interface ToolJsonSchema {
type: 'function';
function: {
name: string;
description: string;
parameters: Record<string, unknown>;
};
}
export interface ToolDef<TInput> {
name: string;
description: string;
inputSchema: z.ZodType<TInput>;
jsonSchema: ToolJsonSchema;
execute(input: TInput, projectRoot: string): Promise<unknown>;
}
const ViewFileInput = z.object({
path: z.string().min(1),
start_line: z.number().int().positive().optional(),
end_line: z.number().int().positive().optional(),
});
type ViewFileInputT = z.infer<typeof ViewFileInput>;
export const viewFile: ToolDef<ViewFileInputT> = {
name: 'view_file',
description:
"Read a file under the project. Returns first 200 lines by default, or a slice via start_line/end_line (1-indexed, inclusive). Files larger than 5MB are refused. Output is truncated if longer than the slice; the response indicates truncation.",
inputSchema: ViewFileInput,
jsonSchema: {
type: 'function',
function: {
name: 'view_file',
description:
"Read a file under the project. Returns first 200 lines by default, or a slice via start_line/end_line (1-indexed, inclusive). Files larger than 5MB are refused.",
parameters: {
type: 'object',
properties: {
path: { type: 'string', description: 'absolute or project-relative path' },
start_line: { type: 'integer', description: 'first line (1-indexed)' },
end_line: { type: 'integer', description: 'last line (1-indexed, inclusive)' },
},
required: ['path'],
additionalProperties: false,
},
},
},
async execute(input, projectRoot) {
const real = await pathGuard(projectRoot, input.path);
const s = await stat(real);
if (!s.isFile()) {
throw new PathScopeError(`not a file: ${input.path}`);
}
if (s.size > MAX_FILE_BYTES) {
throw new Error(`file too large (${s.size} bytes, max ${MAX_FILE_BYTES})`);
}
const raw = await readFile(real, 'utf8');
const lines = raw.split('\n');
const total = lines.length;
let start = input.start_line ?? 1;
let end = input.end_line ?? Math.min(total, start + DEFAULT_VIEW_LINES - 1);
if (input.start_line == null && input.end_line == null) {
end = Math.min(total, DEFAULT_VIEW_LINES);
}
if (start < 1) start = 1;
if (end > total) end = total;
if (end < start) end = start;
const slice = lines.slice(start - 1, end);
const content = slice.join('\n');
const truncated = total > end || start > 1;
return {
path: relative(projectRoot, real) || basename(real),
content,
total_lines: total,
returned_lines: [start, end],
truncated,
};
},
};
const ListDirInput = z.object({
path: z.string().min(1),
show_hidden: z.boolean().optional(),
});
type ListDirInputT = z.infer<typeof ListDirInput>;
export const listDir: ToolDef<ListDirInputT> = {
name: 'list_dir',
description: 'List entries in a directory (up to 500). Hidden files excluded unless show_hidden=true.',
inputSchema: ListDirInput,
jsonSchema: {
type: 'function',
function: {
name: 'list_dir',
description:
'List entries in a directory (up to 500). Hidden files (dot-prefixed) excluded unless show_hidden=true.',
parameters: {
type: 'object',
properties: {
path: { type: 'string' },
show_hidden: { type: 'boolean' },
},
required: ['path'],
additionalProperties: false,
},
},
},
async execute(input, projectRoot) {
const real = await pathGuard(projectRoot, input.path);
const s = await stat(real);
if (!s.isDirectory()) {
throw new PathScopeError(`not a directory: ${input.path}`);
}
const entries = await readdir(real, { withFileTypes: true });
const filtered = input.show_hidden
? entries
: entries.filter((e) => !e.name.startsWith('.'));
const total = filtered.length;
const slice = filtered.slice(0, MAX_DIR_ENTRIES);
const out = await Promise.all(
slice.map(async (e) => {
const child = resolve(real, e.name);
let size: number | undefined;
if (e.isFile()) {
try {
const cs = await stat(child);
size = cs.size;
} catch {
/* ignore */
}
}
return {
name: e.name,
type: e.isDirectory() ? ('dir' as const) : ('file' as const),
...(size != null ? { size } : {}),
};
})
);
return {
path: relative(projectRoot, real) || '.',
entries: out,
total,
truncated: total > MAX_DIR_ENTRIES,
};
},
};
const GrepInput = z.object({
pattern: z.string().min(1),
path: z.string().optional(),
case_sensitive: z.boolean().optional(),
max_results: z.number().int().positive().optional(),
hidden: z.boolean().optional(),
});
type GrepInputT = z.infer<typeof GrepInput>;
interface RipgrepMatch {
type: string;
data?: {
path?: { text?: string };
line_number?: number;
lines?: { text?: string };
};
}
export const grep: ToolDef<GrepInputT> = {
name: 'grep',
description:
'Search file contents with ripgrep. Default path is project root. Max 100 results (200 cap).',
inputSchema: GrepInput,
jsonSchema: {
type: 'function',
function: {
name: 'grep',
description:
'Search file contents with ripgrep. Returns up to 100 matches (cap 200). Set hidden=true to include dot-prefixed files.',
parameters: {
type: 'object',
properties: {
pattern: { type: 'string' },
path: { type: 'string' },
case_sensitive: { type: 'boolean' },
max_results: { type: 'integer' },
hidden: { type: 'boolean' },
},
required: ['pattern'],
additionalProperties: false,
},
},
},
async execute(input, projectRoot) {
const target = await pathGuard(projectRoot, input.path ?? projectRoot);
const limit = Math.min(
Math.max(input.max_results ?? DEFAULT_GREP_RESULTS, 1),
MAX_GREP_RESULTS
);
const args = [
'--json',
'--max-count',
String(limit),
'--max-columns',
'300',
];
if (!input.case_sensitive) args.push('--ignore-case');
if (input.hidden) args.push('--hidden');
args.push('--', input.pattern, target);
return await new Promise((resolveP, rejectP) => {
const child = spawn('rg', args, { cwd: projectRoot });
const matches: Array<{ path: string; line: number; content: string }> = [];
let buf = '';
let stderr = '';
child.stdout.setEncoding('utf8');
child.stderr.setEncoding('utf8');
child.stdout.on('data', (chunk: string) => {
buf += chunk;
let idx;
while ((idx = buf.indexOf('\n')) >= 0) {
const line = buf.slice(0, idx);
buf = buf.slice(idx + 1);
if (!line) continue;
if (matches.length >= limit) continue;
try {
const parsed = JSON.parse(line) as RipgrepMatch;
if (parsed.type !== 'match' || !parsed.data) continue;
const path = parsed.data.path?.text ?? '';
const lineNumber = parsed.data.line_number ?? 0;
const content = parsed.data.lines?.text ?? '';
matches.push({
path: relative(projectRoot, path) || path,
line: lineNumber,
content: content.replace(/\n$/, ''),
});
} catch {
/* ignore non-json */
}
}
if (matches.length >= limit) {
child.kill();
}
});
child.stderr.on('data', (chunk: string) => {
stderr += chunk;
});
child.on('error', (err) => rejectP(err));
child.on('close', (code) => {
// rg exits 1 when no matches, 2 on real error
if (code === 2 && matches.length === 0) {
rejectP(new Error(`ripgrep failed: ${stderr.slice(0, 300)}`));
return;
}
resolveP({
matches,
total: matches.length,
truncated: matches.length >= limit,
});
});
});
},
};
const FindFilesInput = z.object({
pattern: z.string().min(1),
path: z.string().optional(),
max_results: z.number().int().positive().optional(),
});
type FindFilesInputT = z.infer<typeof FindFilesInput>;
export const findFiles: ToolDef<FindFilesInputT> = {
name: 'find_files',
description: 'Glob for filenames. Default path is project root. Max 100 results (200 cap).',
inputSchema: FindFilesInput,
jsonSchema: {
type: 'function',
function: {
name: 'find_files',
description:
'Glob for filenames under a directory. Default path is project root. Max 100 results (cap 200). Pattern uses standard glob (e.g. "**/*.ts").',
parameters: {
type: 'object',
properties: {
pattern: { type: 'string' },
path: { type: 'string' },
max_results: { type: 'integer' },
},
required: ['pattern'],
additionalProperties: false,
},
},
},
async execute(input, projectRoot) {
const target = await pathGuard(projectRoot, input.path ?? projectRoot);
const limit = Math.min(
Math.max(input.max_results ?? DEFAULT_FIND_RESULTS, 1),
MAX_FIND_RESULTS
);
return await new Promise((resolveP, rejectP) => {
const args = ['--files', '--glob', input.pattern, target];
const child = spawn('rg', args, { cwd: projectRoot });
const paths: string[] = [];
let total = 0;
let buf = '';
let stderr = '';
child.stdout.setEncoding('utf8');
child.stderr.setEncoding('utf8');
child.stdout.on('data', (chunk: string) => {
buf += chunk;
let idx;
while ((idx = buf.indexOf('\n')) >= 0) {
const line = buf.slice(0, idx);
buf = buf.slice(idx + 1);
if (!line) continue;
total++;
if (paths.length < limit) {
paths.push(relative(projectRoot, line) || line);
}
}
});
child.stderr.on('data', (chunk: string) => {
stderr += chunk;
});
child.on('error', (err) => rejectP(err));
child.on('close', (code) => {
if (code === 2) {
rejectP(new Error(`ripgrep failed: ${stderr.slice(0, 300)}`));
return;
}
if (buf.length > 0) {
total++;
if (paths.length < limit) {
paths.push(relative(projectRoot, buf) || buf);
}
}
resolveP({
paths,
total,
truncated: total > paths.length,
});
});
});
},
};
export const ALL_TOOLS: ReadonlyArray<ToolDef<unknown>> = [
viewFile as ToolDef<unknown>,
listDir as ToolDef<unknown>,
grep as ToolDef<unknown>,
findFiles as ToolDef<unknown>,
];
export const TOOLS_BY_NAME: Record<string, ToolDef<unknown>> = Object.fromEntries(
ALL_TOOLS.map((t) => [t.name, t])
);
export function toolJsonSchemas(): ToolJsonSchema[] {
return ALL_TOOLS.map((t) => t.jsonSchema);
}

View File

@@ -0,0 +1,55 @@
export interface Project {
id: string;
name: string;
path: string;
added_at: string;
last_session_id: string | null;
}
export interface AvailableProject {
path: string;
name: string;
}
export interface Session {
id: string;
project_id: string;
name: string;
model: string;
system_prompt: string;
created_at: string;
updated_at: string;
}
export type MessageRole = 'user' | 'assistant' | 'tool';
export type MessageStatus = 'streaming' | 'complete' | 'failed';
export interface ToolCall {
id: string;
name: string;
args: Record<string, unknown>;
}
export interface ToolResult {
tool_call_id: string;
output: unknown;
truncated: boolean;
error?: string;
}
export interface Message {
id: string;
session_id: string;
role: MessageRole;
content: string;
tool_calls: ToolCall[] | null;
tool_results: ToolResult | null;
status: MessageStatus;
last_seq: number;
created_at: string;
}
export interface ModelInfo {
id: string;
[key: string]: unknown;
}