1574 lines
71 KiB
TypeScript
1574 lines
71 KiB
TypeScript
import express from "express";
|
|
import { createHash, randomUUID, timingSafeEqual } from "crypto";
|
|
import { createServer } from "http";
|
|
import fs from "fs";
|
|
import os from "os";
|
|
import path from "path";
|
|
import { fileURLToPath } from "url";
|
|
import { WebSocketServer, WebSocket } from "ws";
|
|
import {
|
|
createAgentSession,
|
|
ModelRuntime,
|
|
SessionManager,
|
|
defineTool,
|
|
DefaultResourceLoader,
|
|
SettingsManager,
|
|
type AgentSession,
|
|
type AgentSessionEvent,
|
|
} from "@earendil-works/pi-coding-agent";
|
|
import {
|
|
canReadFile,
|
|
createParsingState,
|
|
markStopped,
|
|
parseDepthIsAllowed,
|
|
recordRead,
|
|
type ParsingLimits,
|
|
} from "./parsing-budget.js";
|
|
import { Type, type Model, type Api } from "@earendil-works/pi-ai";
|
|
import { SessionBroker, type AttachmentHandle, type BrokerRuntime, type BrokerSession } from "./session-broker.js";
|
|
import type {
|
|
ClientMessage,
|
|
ServerMessage,
|
|
ModelInfo,
|
|
SerializedAgentState,
|
|
SessionListItem,
|
|
} from "../shared/protocol.js";
|
|
|
|
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
|
function parseIntEnv(name: string, fallback: number): number {
|
|
const value = Number.parseInt(process.env[name] || "", 10);
|
|
return Number.isFinite(value) ? Math.max(0, value) : fallback;
|
|
}
|
|
|
|
function parseLimitedIntEnv(name: string, fallback: number): number {
|
|
const value = parseIntEnv(name, fallback);
|
|
return value > 0 ? value : fallback;
|
|
}
|
|
|
|
const PORT = parseIntEnv("PORT", 3001);
|
|
const HOST = process.env.HOST || "127.0.0.1";
|
|
const LITELLM_URL = process.env.LITELLM_URL || "";
|
|
const LITELLM_TIMEOUT_MS = parseIntEnv("LITELLM_TIMEOUT_MS", 1500);
|
|
let litellmKey = process.env.LITELLM_KEY || "";
|
|
const APP_TITLE = (process.env.PI_WEBUI_TITLE || "Pi Web UI").trim().slice(0, 120) || "Pi Web UI";
|
|
|
|
const app = express();
|
|
const server = createServer(app);
|
|
const webUiToken = (process.env.PI_WEBUI_TOKEN || "").trim();
|
|
const allowAnonymous = ["1", "true", "yes", "on"].includes((process.env.PI_WEBUI_ALLOW_ANONYMOUS || "").trim().toLowerCase());
|
|
const anonymousLoopbackOnly = HOST === "127.0.0.1" || HOST === "::1" || HOST === "localhost";
|
|
const allowedOrigins = new Set(
|
|
(process.env.PI_WEBUI_ALLOWED_ORIGINS || "")
|
|
.split(",")
|
|
.map((origin) => origin.trim())
|
|
.filter(Boolean),
|
|
);
|
|
|
|
const SECURITY_HEADERS = {
|
|
csp: "default-src 'self'; base-uri 'self'; form-action 'self'; img-src 'self' data: blob:; frame-src 'self' blob:; font-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self' ws: wss:; object-src 'none'; frame-ancestors 'none';",
|
|
hsts: "max-age=31536000; includeSubDomains; preload",
|
|
referrerPolicy: "strict-origin-when-cross-origin",
|
|
frameOptions: "DENY",
|
|
contentTypeOptions: "nosniff",
|
|
};
|
|
|
|
const SKILL_PARSE_LIMITS: ParsingLimits = {
|
|
maxFiles: parseLimitedIntEnv("PI_WEBUI_SKILL_MAX_FILES", 220),
|
|
maxBytesTotal: parseLimitedIntEnv("PI_WEBUI_SKILL_PARSE_BYTE_BUDGET", 3_000_000),
|
|
maxBytesPerFile: parseLimitedIntEnv("PI_WEBUI_SKILL_FILE_MAX_BYTES", 240_000),
|
|
};
|
|
const SKILL_MAX_DEPTH = parseLimitedIntEnv("PI_WEBUI_SKILL_MAX_DEPTH", 8);
|
|
const SKILL_DIRECTORY_ENTRY_LIMIT = parseLimitedIntEnv("PI_WEBUI_SKILL_DIRECTORY_ENTRY_LIMIT", 1000);
|
|
|
|
const SESSION_PARSE_LIMITS: ParsingLimits = {
|
|
maxFiles: parseLimitedIntEnv("PI_WEBUI_SESSION_MAX_FILES", 240),
|
|
maxBytesTotal: parseLimitedIntEnv("PI_WEBUI_SESSION_PARSE_BYTE_BUDGET", 6_000_000),
|
|
maxBytesPerFile: parseLimitedIntEnv("PI_WEBUI_SESSION_FILE_MAX_BYTES", 220_000),
|
|
};
|
|
const SESSION_DIRECTORY_SCAN_LIMIT = parseLimitedIntEnv("PI_WEBUI_SESSION_DIRECTORY_SCAN_LIMIT", 800);
|
|
|
|
function isAllowedWsOrigin(origin: string | undefined, hostHeader: string | undefined): boolean {
|
|
if (!origin) return true;
|
|
if (allowedOrigins.has(origin)) return true;
|
|
try {
|
|
const url = new URL(origin);
|
|
return Boolean(hostHeader) && url.host === hostHeader;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
function isValidToken(candidate: string | undefined): boolean {
|
|
if (!webUiToken && !allowAnonymous) return false;
|
|
if (!webUiToken) return true;
|
|
if (!candidate) return false;
|
|
const expected = Buffer.from(webUiToken);
|
|
const actual = Buffer.from(candidate);
|
|
return expected.length === actual.length && timingSafeEqual(expected, actual);
|
|
}
|
|
|
|
function tokenFromRequestHeader(req: express.Request): string | undefined {
|
|
const header = req.header("x-pi-webui-token") || req.header("authorization") || "";
|
|
if (header.toLowerCase().startsWith("bearer ")) return header.slice(7).trim();
|
|
return header.trim();
|
|
}
|
|
|
|
function tokenFromWebSocketHeaders(headers: Record<string, string | string[] | undefined>): string | undefined {
|
|
const header = headers["x-pi-webui-token"] || headers["X-PI-WEBUI-TOKEN"];
|
|
if (!header) return undefined;
|
|
if (Array.isArray(header)) return header[0]?.trim();
|
|
const value = typeof header === "string" ? header.trim() : "";
|
|
if (value.toLowerCase().startsWith("bearer ")) return value.slice(7).trim();
|
|
return value;
|
|
}
|
|
|
|
function tokenFromWebSocketProtocol(header: string | string[] | undefined): string | undefined {
|
|
const raw = Array.isArray(header) ? header.join(",") : header;
|
|
if (!raw) return undefined;
|
|
for (const item of raw.split(",")) {
|
|
const protocol = item.trim();
|
|
if (!protocol.startsWith("pi-webui-token.")) continue;
|
|
const encoded = protocol.slice("pi-webui-token.".length);
|
|
if (!encoded) continue;
|
|
try {
|
|
return Buffer.from(encoded, "base64url").toString("utf8");
|
|
} catch {
|
|
return undefined;
|
|
}
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
const wss = new WebSocketServer({
|
|
server,
|
|
path: "/api/ws",
|
|
maxPayload: 1024 * 1024,
|
|
verifyClient: (info, done) => {
|
|
const originOk = isAllowedWsOrigin(info.origin, info.req.headers.host);
|
|
const protocolToken = tokenFromWebSocketProtocol(info.req.headers["sec-websocket-protocol"]);
|
|
const headerToken = tokenFromWebSocketHeaders(info.req.headers as Record<string, string | string[] | undefined>);
|
|
const tokenOk = isValidToken(protocolToken || headerToken);
|
|
done(originOk && tokenOk, originOk ? 401 : 403, originOk ? "Unauthorized" : "Forbidden");
|
|
},
|
|
});
|
|
|
|
app.use((_req, res, next) => {
|
|
res.setHeader("Content-Security-Policy", SECURITY_HEADERS.csp);
|
|
res.setHeader("X-Content-Type-Options", SECURITY_HEADERS.contentTypeOptions);
|
|
res.setHeader("Referrer-Policy", SECURITY_HEADERS.referrerPolicy);
|
|
res.setHeader("X-Frame-Options", SECURITY_HEADERS.frameOptions);
|
|
res.setHeader("Strict-Transport-Security", SECURITY_HEADERS.hsts);
|
|
next();
|
|
});
|
|
app.use(express.urlencoded({ extended: false, limit: "64kb" }));
|
|
app.use(express.json({ limit: "256kb" }));
|
|
|
|
wss.on("headers", (headers) => {
|
|
headers.push(`Content-Security-Policy: ${SECURITY_HEADERS.csp}`);
|
|
headers.push(`X-Content-Type-Options: ${SECURITY_HEADERS.contentTypeOptions}`);
|
|
headers.push(`Referrer-Policy: ${SECURITY_HEADERS.referrerPolicy}`);
|
|
headers.push(`X-Frame-Options: ${SECURITY_HEADERS.frameOptions}`);
|
|
headers.push(`Strict-Transport-Security: ${SECURITY_HEADERS.hsts}`);
|
|
});
|
|
|
|
const HOME_DIR = os.homedir();
|
|
const AGENT_DIR = process.env.PI_CODING_AGENT_DIR || path.join(HOME_DIR, ".pi", "agent");
|
|
const AGENT_SETTINGS_PATH = path.join(AGENT_DIR, "settings.json");
|
|
const AGENT_AUTH_PATH = path.join(AGENT_DIR, "auth.json");
|
|
const AGENT_MODELS_PATH = path.join(AGENT_DIR, "models.json");
|
|
const USER_SKILLS_DIR = process.env.PI_WEBUI_USER_SKILLS_DIR || path.join(HOME_DIR, ".pi", "skills");
|
|
const SHARED_SKILLS_DIR = process.env.PI_WEBUI_SHARED_SKILLS_DIR || path.join(AGENT_DIR, "skills");
|
|
const SESSION_LIST_LIMIT = parseIntEnv("SESSION_LIST_LIMIT", 30);
|
|
const SESSION_LIST_MAX_LIMIT = parseIntEnv("SESSION_LIST_MAX_LIMIT", 100);
|
|
const MESSAGE_PAGE_LIMIT = parseIntEnv("MESSAGE_PAGE_LIMIT", 60);
|
|
const MESSAGE_PAGE_MAX_LIMIT = parseIntEnv("MESSAGE_PAGE_MAX_LIMIT", 200);
|
|
const SESSION_LIST_CACHE_MS = parseIntEnv("SESSION_LIST_CACHE_MS", 10000);
|
|
|
|
let modelRuntime: ModelRuntime;
|
|
type ChoiceRequest = {
|
|
id: string;
|
|
prompt: string;
|
|
choices: Array<{ id: string; label: string; description?: string }>;
|
|
allowMultiple: boolean;
|
|
};
|
|
|
|
type RuntimeContext = { sessionManager: SessionManager };
|
|
type WebUiSession = AgentSession & BrokerSession;
|
|
type WebUiRuntime = BrokerRuntime<RuntimeContext, WebUiSession>;
|
|
const clients = new Map<WebSocket, AttachmentHandle<RuntimeContext, WebUiSession>>();
|
|
const attachmentTails = new Map<WebSocket, Promise<void>>();
|
|
const syntheticModels = new Map<string, Model<Api>>();
|
|
let modelConfigStamp = "";
|
|
let modelAuthStamp = "";
|
|
let modelConfigRefreshPromise: Promise<void> | undefined;
|
|
let sessionListCache: { expires: number; files: { filePath: string; mtime: Date }[] } | undefined;
|
|
|
|
function boolEnv(name: string): boolean {
|
|
return ["1", "true", "yes", "on"].includes((process.env[name] || "").trim().toLowerCase());
|
|
}
|
|
|
|
function restrictModelsToScoped(): boolean {
|
|
return boolEnv("PI_WEBUI_RESTRICT_MODELS_TO_SCOPED");
|
|
}
|
|
|
|
function forceModelAliasLabels(): boolean {
|
|
return boolEnv("PI_WEBUI_FORCE_MODEL_ALIAS_LABELS");
|
|
}
|
|
|
|
type SkillSource = "user" | "shared";
|
|
|
|
type SkillListItem = {
|
|
id: string;
|
|
name: string;
|
|
description: string;
|
|
path: string;
|
|
source: SkillSource;
|
|
editable: boolean;
|
|
diagnostics: string[];
|
|
};
|
|
|
|
function isValidSkillName(name: string): boolean {
|
|
if (name.length < 1 || name.length > 64) return false;
|
|
if (name.startsWith("-") || name.endsWith("-") || name.includes("--")) return false;
|
|
for (const ch of name) {
|
|
const ok = (ch >= "a" && ch <= "z") || (ch >= "0" && ch <= "9") || ch === "-";
|
|
if (!ok) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
function normalizeSkillDescription(value: string): string {
|
|
return value.split(" ").filter(Boolean).join(" ").replaceAll("\n", " ").replaceAll("\r", " ").replaceAll("\t", " ").trim();
|
|
}
|
|
|
|
function parseSkillFrontmatter(content: string): { fields: Record<string, string>; diagnostics: string[] } {
|
|
const diagnostics: string[] = [];
|
|
const fields: Record<string, string> = {};
|
|
const lines = content.split(/\r?\n/);
|
|
if (lines[0]?.trim() !== "---") {
|
|
diagnostics.push("missing frontmatter fence");
|
|
return { fields, diagnostics };
|
|
}
|
|
const end = lines.findIndex((line, index) => index > 0 && line.trim() === "---");
|
|
if (end < 0) {
|
|
diagnostics.push("missing closing frontmatter fence");
|
|
return { fields, diagnostics };
|
|
}
|
|
for (let i = 1; i < end; i++) {
|
|
const line = lines[i];
|
|
const colon = line.indexOf(":");
|
|
if (colon <= 0) continue;
|
|
const key = line.slice(0, colon).trim();
|
|
let value = line.slice(colon + 1).trim();
|
|
if ((value.startsWith('"') && value.endsWith('"')) || (value.startsWith("'") && value.endsWith("'"))) value = value.slice(1, -1);
|
|
fields[key] = value;
|
|
}
|
|
if (!fields.name) diagnostics.push("missing name");
|
|
else if (!isValidSkillName(fields.name)) diagnostics.push("invalid name: use lowercase letters, numbers, and single hyphens only");
|
|
if (!fields.description) diagnostics.push("missing description");
|
|
else if (fields.description.length > 1024) diagnostics.push("description exceeds 1024 characters");
|
|
return { fields, diagnostics };
|
|
}
|
|
|
|
async function readSkillFile(filePath: string, source: SkillSource, editable: boolean, state: ReturnType<typeof createParsingState>): Promise<SkillListItem | undefined> {
|
|
const stats = await fs.promises.stat(filePath).catch(() => undefined);
|
|
const sizeBytes = stats?.size ?? 0;
|
|
const decision = canReadFile(sizeBytes, SKILL_PARSE_LIMITS, state);
|
|
if (!decision.canRead) {
|
|
if (decision.stop) markStopped(state, decision.reason!);
|
|
return undefined;
|
|
}
|
|
|
|
const content = await fs.promises.readFile(filePath, "utf8").catch(() => "");
|
|
if (!content) return undefined;
|
|
const { fields, diagnostics } = parseSkillFrontmatter(content);
|
|
if (!fields.description) {
|
|
recordRead(sizeBytes, state);
|
|
return undefined;
|
|
}
|
|
const folderName = path.basename(path.dirname(filePath));
|
|
const name = fields.name || folderName;
|
|
const id = source === "user" && isValidSkillName(name) ? name : createHash("sha256").update(path.resolve(filePath)).digest("hex").slice(0, 16);
|
|
|
|
recordRead(sizeBytes, state);
|
|
return { id, name, description: fields.description || "", path: filePath, source, editable, diagnostics };
|
|
}
|
|
|
|
async function readDirectoryEntriesLimited(dir: string, maxEntries: number): Promise<fs.Dirent[]> {
|
|
const out: fs.Dirent[] = [];
|
|
const handle = await fs.promises.opendir(dir).catch(() => undefined);
|
|
if (!handle) return out;
|
|
try {
|
|
for await (const entry of handle) {
|
|
out.push(entry);
|
|
if (maxEntries > 0 && out.length >= maxEntries) break;
|
|
}
|
|
} finally {
|
|
await handle.close().catch(() => undefined);
|
|
}
|
|
return out;
|
|
}
|
|
|
|
async function collectSkillsFromRoot(root: string, source: SkillSource, editable: boolean): Promise<SkillListItem[]> {
|
|
const out: SkillListItem[] = [];
|
|
const state = createParsingState();
|
|
|
|
const visit = async (dir: string, depth: number, allowRootFiles: boolean): Promise<void> => {
|
|
if (state.stopReason) return;
|
|
if (!parseDepthIsAllowed(depth, SKILL_MAX_DEPTH)) {
|
|
return;
|
|
}
|
|
|
|
const entries = await readDirectoryEntriesLimited(dir, SKILL_DIRECTORY_ENTRY_LIMIT);
|
|
if (entries.some((entry) => entry.isFile() && entry.name === "SKILL.md")) {
|
|
const item = await readSkillFile(path.join(dir, "SKILL.md"), source, editable, state);
|
|
if (item) out.push(item);
|
|
return;
|
|
}
|
|
|
|
for (const entry of entries) {
|
|
if (state.stopReason) break;
|
|
if (entry.name.startsWith(".") || entry.name === "node_modules") continue;
|
|
const fullPath = path.join(dir, entry.name);
|
|
if (entry.isDirectory()) {
|
|
await visit(fullPath, depth + 1, false);
|
|
} else if (allowRootFiles && entry.isFile() && entry.name.endsWith(".md")) {
|
|
const item = await readSkillFile(fullPath, source, editable, state);
|
|
if (item) out.push(item);
|
|
}
|
|
}
|
|
};
|
|
|
|
await visit(root, 1, true);
|
|
return out.sort((a, b) => a.name.localeCompare(b.name));
|
|
}
|
|
|
|
async function listSkills(): Promise<SkillListItem[]> {
|
|
const [userSkills, sharedSkills] = await Promise.all([
|
|
collectSkillsFromRoot(USER_SKILLS_DIR, "user", true),
|
|
collectSkillsFromRoot(SHARED_SKILLS_DIR, "shared", false),
|
|
]);
|
|
const userNames = new Set(userSkills.map((skill) => skill.name));
|
|
return [...userSkills, ...sharedSkills.filter((skill) => !userNames.has(skill.name))];
|
|
}
|
|
|
|
function userSkillPath(name: string): string | undefined {
|
|
if (!isValidSkillName(name)) return undefined;
|
|
const root = path.resolve(USER_SKILLS_DIR);
|
|
const skillFile = path.resolve(root, name, "SKILL.md");
|
|
if (!skillFile.startsWith(`${root}${path.sep}`)) return undefined;
|
|
return skillFile;
|
|
}
|
|
|
|
async function writeSkillAtomic(skillFile: string, content: string) {
|
|
await fs.promises.mkdir(path.dirname(skillFile), { recursive: true, mode: 0o700 });
|
|
const tmp = `${skillFile}.${process.pid}.${Date.now()}.tmp`;
|
|
await fs.promises.writeFile(tmp, content, { mode: 0o600 });
|
|
await fs.promises.rename(tmp, skillFile);
|
|
}
|
|
|
|
async function getSkillContent(source: string, id: string): Promise<{ item: SkillListItem; content: string } | undefined> {
|
|
if (source !== "user" && source !== "shared") return undefined;
|
|
const skills = await listSkills();
|
|
const item = skills.find((skill) => skill.source === source && skill.id === id);
|
|
if (!item) return undefined;
|
|
|
|
const stat = await fs.promises.stat(item.path).catch(() => undefined);
|
|
if (!stat || stat.size > SKILL_PARSE_LIMITS.maxBytesPerFile) return undefined;
|
|
const content = await fs.promises.readFile(item.path, "utf8");
|
|
return { item, content };
|
|
}
|
|
|
|
function tokenFromRequest(req: express.Request): string | undefined {
|
|
return tokenFromRequestHeader(req);
|
|
}
|
|
|
|
function requireEditAuth(req: express.Request, res: express.Response, next: express.NextFunction) {
|
|
if (isValidToken(tokenFromRequest(req))) return next();
|
|
res.status(401).json({ ok: false, error: "unauthorized" });
|
|
}
|
|
|
|
function requireHttpAuth(req: express.Request, res: express.Response, next: express.NextFunction) {
|
|
if (!isAllowedWsOrigin(req.header("Origin"), req.header("Host"))) return res.status(403).type("text/plain").send("Forbidden origin");
|
|
if (isValidToken(tokenFromRequest(req))) return next();
|
|
res.status(401).type("text/plain").send("Unauthorized");
|
|
}
|
|
|
|
function fileStamp(filePath: string): string {
|
|
try {
|
|
const stat = fs.statSync(filePath);
|
|
return `${stat.mtimeMs}:${stat.size}`;
|
|
} catch {
|
|
return "missing";
|
|
}
|
|
}
|
|
|
|
async function refreshModelConfig() {
|
|
const nextAuthStamp = fileStamp(AGENT_AUTH_PATH);
|
|
const nextStamp = [nextAuthStamp, fileStamp(AGENT_MODELS_PATH), fileStamp(AGENT_SETTINGS_PATH)].join("|");
|
|
if (nextStamp === modelConfigStamp) return;
|
|
if (modelAuthStamp && nextAuthStamp !== modelAuthStamp) {
|
|
modelRuntime = await ModelRuntime.create({ authPath: AGENT_AUTH_PATH, modelsPath: AGENT_MODELS_PATH });
|
|
for (const ws of clients.keys()) ws.close(1012, "Model credentials changed");
|
|
await broker?.disposeAll();
|
|
} else {
|
|
await modelRuntime?.refresh({ allowNetwork: false });
|
|
}
|
|
modelAuthStamp = nextAuthStamp;
|
|
modelConfigStamp = nextStamp;
|
|
syntheticModels.clear();
|
|
}
|
|
|
|
function refreshModelConfigIfChanged() {
|
|
if (!modelConfigRefreshPromise) {
|
|
modelConfigRefreshPromise = refreshModelConfig().finally(() => { modelConfigRefreshPromise = undefined; });
|
|
}
|
|
return modelConfigRefreshPromise;
|
|
}
|
|
|
|
function modelToInfo(model: Model<Api>, scoped = false): ModelInfo {
|
|
return {
|
|
provider: model.provider,
|
|
id: model.id,
|
|
name: model.name || model.id,
|
|
scoped,
|
|
forceAliasLabel: forceModelAliasLabels(),
|
|
};
|
|
}
|
|
|
|
function buildModelLookupCandidates(provider: string, modelId: string): Array<{ provider: string; modelId: string }> {
|
|
const normalizedProvider = provider.trim();
|
|
const normalizedModelId = modelId.trim();
|
|
const candidates: Array<{ provider: string; modelId: string }> = [];
|
|
const seen = new Set<string>();
|
|
|
|
const add = (candidateProvider: string, candidateModelId: string) => {
|
|
const p = candidateProvider.trim();
|
|
const id = candidateModelId.trim();
|
|
if (!p || !id) return;
|
|
const key = `${p}/${id}`;
|
|
if (seen.has(key)) return;
|
|
seen.add(key);
|
|
candidates.push({ provider: p, modelId: id });
|
|
};
|
|
|
|
// Exact pair from client message first.
|
|
add(normalizedProvider, normalizedModelId);
|
|
|
|
// Handle merged identifiers like provider=ollama/local and modelId=Qwen...
|
|
if (normalizedProvider.startsWith("ollama/")) {
|
|
const providerSuffix = normalizedProvider.slice("ollama/".length);
|
|
add("ollama", `${providerSuffix}/${normalizedModelId}`);
|
|
}
|
|
|
|
// Handle modelId values like ollama/local/Qwen...
|
|
if (normalizedModelId.startsWith("ollama/")) {
|
|
add("ollama", normalizedModelId.slice("ollama/".length));
|
|
}
|
|
|
|
// Best-effort fallback for Ollama if client sent bare model name.
|
|
if (normalizedProvider === "ollama" && !normalizedModelId.includes("/")) {
|
|
add("ollama", `local/${normalizedModelId}`);
|
|
}
|
|
|
|
return candidates;
|
|
}
|
|
|
|
type MessageCompletionCache = {
|
|
leafId: string | null;
|
|
messageCount: number;
|
|
completedAtByKey: Map<string, number[]>;
|
|
};
|
|
|
|
const messageCompletionCache = new WeakMap<AgentSession, MessageCompletionCache>();
|
|
|
|
function getMessageCompletionTimes(session: AgentSession): Map<string, number[]> {
|
|
const leafId = session.sessionManager.getLeafId();
|
|
const messageCount = session.agent.state.messages.length;
|
|
const cached = messageCompletionCache.get(session);
|
|
if (cached && cached.leafId === leafId && cached.messageCount === messageCount) {
|
|
return new Map([...cached.completedAtByKey].map(([key, values]) => [key, [...values]]));
|
|
}
|
|
|
|
const completedAtByKey = new Map<string, number[]>();
|
|
for (const entry of session.sessionManager.getBranch()) {
|
|
if (entry.type !== "message") continue;
|
|
const timestamp = new Date(entry.timestamp).getTime();
|
|
if (!Number.isFinite(timestamp)) continue;
|
|
const message = entry.message as any;
|
|
const key = `${message.role}:${message.timestamp ?? ""}`;
|
|
const queue = completedAtByKey.get(key) || [];
|
|
queue.push(timestamp);
|
|
completedAtByKey.set(key, queue);
|
|
}
|
|
|
|
messageCompletionCache.set(session, {
|
|
leafId,
|
|
messageCount,
|
|
completedAtByKey: new Map([...completedAtByKey].map(([key, values]) => [key, [...values]])),
|
|
});
|
|
return completedAtByKey;
|
|
}
|
|
|
|
function serializeMessages(session: AgentSession, offset = 0, limit?: number): any[] {
|
|
const source = session.agent.state.messages;
|
|
const messages = limit === undefined ? source.slice(offset) : source.slice(offset, offset + limit);
|
|
const completedAtByKey = getMessageCompletionTimes(session);
|
|
for (let i = 0; i < offset; i++) {
|
|
const message = source[i] as any;
|
|
if (!message) continue;
|
|
const key = `${message.role}:${message.timestamp ?? ""}`;
|
|
completedAtByKey.get(key)?.shift();
|
|
}
|
|
return messages.map((message: any) => {
|
|
const key = `${message.role}:${message.timestamp ?? ""}`;
|
|
const completedAt = completedAtByKey.get(key)?.shift();
|
|
if (message.role !== "assistant" || !completedAt) return message;
|
|
return { ...message, piWebuiCompletedAt: completedAt };
|
|
});
|
|
}
|
|
|
|
function messagePage(session: AgentSession, offset: number, limit = MESSAGE_PAGE_LIMIT) {
|
|
const total = session.agent.state.messages.length;
|
|
const safeLimit = Math.min(Math.max(1, Math.floor(limit)), MESSAGE_PAGE_MAX_LIMIT);
|
|
const safeOffset = Math.min(Math.max(0, Math.floor(offset)), total);
|
|
return {
|
|
messages: serializeMessages(session, safeOffset, safeLimit),
|
|
offset: safeOffset,
|
|
limit: safeLimit,
|
|
total,
|
|
hasMoreBefore: safeOffset > 0,
|
|
};
|
|
}
|
|
|
|
function serializeState(session: AgentSession, overrides: Partial<Pick<SerializedAgentState, "isStreaming">> & { streamingMessage?: any } = {}): SerializedAgentState {
|
|
const state = session.agent.state;
|
|
const hasStreamingMessageOverride = Object.prototype.hasOwnProperty.call(overrides, "streamingMessage");
|
|
const totalMessages = state.messages.length;
|
|
const messagesOffset = Math.max(0, totalMessages - MESSAGE_PAGE_LIMIT);
|
|
const messages = serializeMessages(session, messagesOffset, MESSAGE_PAGE_LIMIT);
|
|
return {
|
|
messages,
|
|
messagesOffset,
|
|
messagesTotal: totalMessages,
|
|
hasMoreMessagesBefore: messagesOffset > 0,
|
|
model: state.model ? modelToInfo(state.model) : undefined,
|
|
thinkingLevel: session.thinkingLevel,
|
|
systemPrompt: state.systemPrompt,
|
|
isStreaming: overrides.isStreaming ?? state.isStreaming,
|
|
streamingMessage: hasStreamingMessageOverride ? overrides.streamingMessage : state.streamingMessage,
|
|
errorMessage: state.errorMessage,
|
|
tools: session.getActiveToolNames(),
|
|
cwd: (session as any)._cwd || (session as any).cwd,
|
|
appTitle: APP_TITLE,
|
|
sessionId: session.sessionId,
|
|
sessionName: session.sessionName,
|
|
sessionPath: session.sessionFile,
|
|
};
|
|
}
|
|
|
|
function send(ws: WebSocket, msg: ServerMessage) {
|
|
if (ws.readyState === WebSocket.OPEN) {
|
|
ws.send(JSON.stringify(msg));
|
|
}
|
|
}
|
|
|
|
function getSessionDir(cwd: string): string {
|
|
const safePath = `--${cwd.replace(/^[/\\]/, "").replace(/[/\\:]/g, "-")}--`;
|
|
return path.join(AGENT_DIR, "sessions", safePath);
|
|
}
|
|
|
|
function extractTextContent(message: any): string {
|
|
const content = message?.content;
|
|
if (typeof content === "string") return content;
|
|
if (!Array.isArray(content)) return "";
|
|
return content
|
|
.filter((part) => part?.type === "text" && typeof part.text === "string")
|
|
.map((part) => part.text)
|
|
.join(" ")
|
|
.trim();
|
|
}
|
|
|
|
function sessionItemFromContent(filePath: string, content: string, mtime: Date): SessionListItem | undefined {
|
|
const lines = content.trim().split("\n").filter(Boolean);
|
|
if (lines.length === 0) return undefined;
|
|
|
|
const entries: any[] = [];
|
|
for (const line of lines) {
|
|
try {
|
|
entries.push(JSON.parse(line));
|
|
} catch {
|
|
// Ignore malformed/incomplete append lines.
|
|
}
|
|
}
|
|
|
|
const header = entries[0];
|
|
if (header?.type !== "session" || typeof header.id !== "string") return undefined;
|
|
|
|
let name: string | undefined;
|
|
let firstMessage = "";
|
|
let messageCount = 0;
|
|
let modifiedTime = new Date(header.timestamp).getTime();
|
|
|
|
for (const entry of entries) {
|
|
if (typeof entry.timestamp === "string") {
|
|
const t = new Date(entry.timestamp).getTime();
|
|
if (!Number.isNaN(t)) modifiedTime = Math.max(modifiedTime || 0, t);
|
|
}
|
|
if (entry.type === "session_info") {
|
|
name = entry.name?.trim() || undefined;
|
|
}
|
|
if (entry.type !== "message") continue;
|
|
messageCount++;
|
|
if (firstMessage || entry.message?.role !== "user") continue;
|
|
firstMessage = extractTextContent(entry.message);
|
|
}
|
|
|
|
const created = new Date(header.timestamp);
|
|
const modified = modifiedTime ? new Date(modifiedTime) : mtime;
|
|
return {
|
|
id: header.id,
|
|
path: filePath,
|
|
name,
|
|
cwd: typeof header.cwd === "string" ? header.cwd : "",
|
|
created: created.toISOString(),
|
|
modified: modified.toISOString(),
|
|
messageCount,
|
|
firstMessage: firstMessage || "(no messages)",
|
|
};
|
|
}
|
|
|
|
async function sessionFiles(): Promise<{ filePath: string; mtime: Date }[]> {
|
|
const now = Date.now();
|
|
if (sessionListCache && sessionListCache.expires > now) return sessionListCache.files;
|
|
|
|
const dir = getSessionDir(HOME_DIR);
|
|
const entries = await readDirectoryEntriesLimited(dir, SESSION_DIRECTORY_SCAN_LIMIT);
|
|
const candidates = entries
|
|
.filter((entry) => entry.isFile() && entry.name.endsWith(".jsonl"))
|
|
// Sort the bounded candidate set before stat calls and pagination.
|
|
.sort((a, b) => b.name.localeCompare(a.name))
|
|
.slice(0, SESSION_DIRECTORY_SCAN_LIMIT);
|
|
const files = (await Promise.all(candidates.map(async (entry) => {
|
|
const filePath = path.join(dir, entry.name);
|
|
const stats = await fs.promises.stat(filePath).catch(() => undefined);
|
|
return stats ? { filePath, mtime: stats.mtime } : undefined;
|
|
})))
|
|
.filter((file): file is { filePath: string; mtime: Date } => Boolean(file))
|
|
.sort((a, b) => b.mtime.getTime() - a.mtime.getTime());
|
|
|
|
sessionListCache = { expires: now + SESSION_LIST_CACHE_MS, files };
|
|
return files;
|
|
}
|
|
|
|
async function readSessionHead(filePath: string, state: ReturnType<typeof createParsingState>): Promise<string | undefined> {
|
|
if (SESSION_PARSE_LIMITS.maxFiles > 0 && state.files >= SESSION_PARSE_LIMITS.maxFiles) {
|
|
markStopped(state, "file-count-limit");
|
|
return undefined;
|
|
}
|
|
const remaining = SESSION_PARSE_LIMITS.maxBytesTotal > 0
|
|
? SESSION_PARSE_LIMITS.maxBytesTotal - state.bytes
|
|
: SESSION_PARSE_LIMITS.maxBytesPerFile;
|
|
const cap = Math.min(SESSION_PARSE_LIMITS.maxBytesPerFile, remaining);
|
|
if (cap <= 0) {
|
|
markStopped(state, "request-byte-limit");
|
|
return undefined;
|
|
}
|
|
const handle = await fs.promises.open(filePath, "r").catch(() => undefined);
|
|
if (!handle) return undefined;
|
|
try {
|
|
const stats = await handle.stat();
|
|
if (!stats.isFile()) return undefined;
|
|
const bytesToRead = Math.min(stats.size, cap);
|
|
const buffer = Buffer.alloc(bytesToRead);
|
|
const { bytesRead } = await handle.read(buffer, 0, bytesToRead, 0);
|
|
recordRead(bytesRead, state);
|
|
return buffer.subarray(0, bytesRead).toString("utf8");
|
|
} finally {
|
|
await handle.close().catch(() => undefined);
|
|
}
|
|
}
|
|
|
|
async function listRecentSessions(options: { offset?: number; limit?: number; query?: string } = {}): Promise<{ items: SessionListItem[]; total: number; hasMore: boolean; offset: number; limit: number; query: string }> {
|
|
const offset = Math.max(0, Math.floor(options.offset || 0));
|
|
const limit = Math.min(Math.max(1, Math.floor(options.limit || SESSION_LIST_LIMIT)), SESSION_LIST_MAX_LIMIT);
|
|
const query = (options.query || "").trim().toLowerCase();
|
|
const files = await sessionFiles();
|
|
const matched: SessionListItem[] = [];
|
|
const state = createParsingState();
|
|
let seenMatches = 0;
|
|
let hasMore = false;
|
|
|
|
for (const { filePath, mtime } of files) {
|
|
if (state.stopReason) {
|
|
hasMore = true;
|
|
break;
|
|
}
|
|
|
|
const content = await readSessionHead(filePath, state);
|
|
if (!content) {
|
|
if (state.stopReason) hasMore = true;
|
|
continue;
|
|
}
|
|
|
|
const item = sessionItemFromContent(filePath, content, mtime);
|
|
if (!item) continue;
|
|
|
|
const haystack = query ? `${item.name || ""}\n${item.firstMessage}\n${content}`.toLowerCase() : "";
|
|
if (query && !haystack.includes(query)) continue;
|
|
if (seenMatches++ < offset) continue;
|
|
if (matched.length >= limit) {
|
|
hasMore = true;
|
|
break;
|
|
}
|
|
matched.push(item);
|
|
}
|
|
|
|
return { items: matched, total: offset + matched.length + (hasMore ? 1 : 0), hasMore, offset, limit, query: options.query || "" };
|
|
}
|
|
|
|
function clearSessionListCache() {
|
|
sessionListCache = undefined;
|
|
}
|
|
|
|
function resolveSessionPath(candidate: string): string | undefined {
|
|
const sessionDir = path.resolve(getSessionDir(HOME_DIR));
|
|
const requestedPath = path.resolve(candidate);
|
|
if (!requestedPath.startsWith(`${sessionDir}${path.sep}`)) return undefined;
|
|
return requestedPath;
|
|
}
|
|
|
|
async function resolveSessionPathById(sessionId: string | undefined): Promise<string | undefined> {
|
|
if (!sessionId || !/^[A-Za-z0-9_-]{6,128}$/.test(sessionId)) return undefined;
|
|
const entries = await readDirectoryEntriesLimited(getSessionDir(HOME_DIR), SESSION_DIRECTORY_SCAN_LIMIT);
|
|
const direct = entries.find((entry) => entry.isFile() && entry.name.endsWith(`_${sessionId}.jsonl`));
|
|
if (direct) {
|
|
const filePath = path.join(getSessionDir(HOME_DIR), direct.name);
|
|
const state = createParsingState();
|
|
const content = await readSessionHead(filePath, state);
|
|
const stats = await fs.promises.stat(filePath).catch(() => undefined);
|
|
if (stats && sessionItemFromContent(filePath, content || "", stats.mtime)?.id === sessionId) return filePath;
|
|
}
|
|
const state = createParsingState();
|
|
for (const { filePath, mtime } of await sessionFiles()) {
|
|
const content = await readSessionHead(filePath, state);
|
|
if (sessionItemFromContent(filePath, content || "", mtime)?.id === sessionId) return filePath;
|
|
if (state.stopReason) break;
|
|
}
|
|
return undefined;
|
|
}
|
|
|
|
async function resolveRequestedSessionPath(request: { sessionPath?: string; sessionId?: string }): Promise<string | undefined> {
|
|
return request.sessionPath ? resolveSessionPath(request.sessionPath) : resolveSessionPathById(request.sessionId);
|
|
}
|
|
|
|
const sessionPathLocks = new Map<string, Promise<void>>();
|
|
|
|
async function withSessionPathLock<T>(filePath: string, fn: () => Promise<T>): Promise<T> {
|
|
const key = path.resolve(filePath);
|
|
const previous = sessionPathLocks.get(key) || Promise.resolve();
|
|
let release!: () => void;
|
|
const gate = new Promise<void>((resolve) => { release = resolve; });
|
|
const tail = previous.then(() => gate);
|
|
sessionPathLocks.set(key, tail);
|
|
await previous;
|
|
try {
|
|
return await fn();
|
|
} finally {
|
|
release();
|
|
if (sessionPathLocks.get(key) === tail) sessionPathLocks.delete(key);
|
|
}
|
|
}
|
|
|
|
async function fetchLiteLLMModels(): Promise<ModelInfo[]> {
|
|
if (!LITELLM_URL) return [];
|
|
|
|
const controller = new AbortController();
|
|
const timeout = setTimeout(() => controller.abort(), LITELLM_TIMEOUT_MS);
|
|
try {
|
|
const headers: Record<string, string> = { "Content-Type": "application/json" };
|
|
if (litellmKey) {
|
|
headers["Authorization"] = `Bearer ${litellmKey}`;
|
|
}
|
|
const res = await fetch(`${LITELLM_URL}/v1/models`, { headers, signal: controller.signal });
|
|
if (!res.ok) {
|
|
console.error(`LiteLLM /v1/models returned ${res.status}`);
|
|
return [];
|
|
}
|
|
const json = await res.json() as { data?: Array<{ id: string; owned_by?: string }> };
|
|
return (json.data || []).map((m) => ({
|
|
provider: "litellm",
|
|
id: m.id.trim(),
|
|
name: m.id.trim(),
|
|
}));
|
|
} catch (err) {
|
|
console.error("Failed to fetch LiteLLM models:", err);
|
|
return [];
|
|
} finally {
|
|
clearTimeout(timeout);
|
|
}
|
|
}
|
|
|
|
function modelKey(provider: string, modelId: string): string {
|
|
return `${provider}/${modelId}`;
|
|
}
|
|
|
|
function parseModelReference(reference: string | undefined): { provider: string; modelId: string } | undefined {
|
|
const trimmed = reference?.trim();
|
|
if (!trimmed) return undefined;
|
|
const thinkingSuffixes = [":off", ":minimal", ":low", ":medium", ":high"];
|
|
const suffix = thinkingSuffixes.find((value) => trimmed.endsWith(value));
|
|
const withoutThinking = suffix ? trimmed.slice(0, -suffix.length) : trimmed;
|
|
const slash = withoutThinking.indexOf("/");
|
|
if (slash <= 0 || slash === withoutThinking.length - 1) return undefined;
|
|
return {
|
|
provider: withoutThinking.slice(0, slash),
|
|
modelId: withoutThinking.slice(slash + 1),
|
|
};
|
|
}
|
|
|
|
function buildSyntheticModel(provider: string, modelId: string): Model<Api> | undefined {
|
|
const key = modelKey(provider, modelId);
|
|
const existing = syntheticModels.get(key);
|
|
if (existing) return existing;
|
|
|
|
const providerModels = modelRuntime.getModels(provider);
|
|
const base = providerModels[0];
|
|
if (!base) return undefined;
|
|
|
|
const model: Model<Api> = {
|
|
...base,
|
|
id: modelId,
|
|
name: modelId,
|
|
};
|
|
syntheticModels.set(key, model);
|
|
return model;
|
|
}
|
|
|
|
function findConfiguredOrSyntheticModel(provider: string, modelId: string): Model<Api> | undefined {
|
|
const model = modelRuntime.getModel(provider, modelId) || syntheticModels.get(modelKey(provider, modelId));
|
|
if (model && modelRuntime.hasConfiguredAuth(provider)) return model;
|
|
const synthetic = buildSyntheticModel(provider, modelId);
|
|
return synthetic && modelRuntime.hasConfiguredAuth(provider) ? synthetic : undefined;
|
|
}
|
|
|
|
function addModelInfo(merged: ModelInfo[], seen: Set<string>, model: Model<Api>, scoped = false) {
|
|
const key = modelKey(model.provider, model.id);
|
|
if (seen.has(key)) {
|
|
if (scoped) {
|
|
const existing = merged.find((m) => modelKey(m.provider, m.id) === key);
|
|
if (existing) existing.scoped = true;
|
|
}
|
|
return;
|
|
}
|
|
seen.add(key);
|
|
merged.push(modelToInfo(model, scoped));
|
|
}
|
|
|
|
function scopedModelReferences(settings: any): string[] {
|
|
return [
|
|
process.env.PI_WEBUI_MODEL,
|
|
process.env.PI_MODEL,
|
|
settings.defaultProvider && settings.defaultModel ? `${settings.defaultProvider}/${settings.defaultModel}` : undefined,
|
|
...(Array.isArray(settings.enabledModels) ? settings.enabledModels : []),
|
|
].filter((reference): reference is string => typeof reference === "string" && reference.trim() !== "");
|
|
}
|
|
|
|
function scopedModelKeys(settings: any): Set<string> {
|
|
const keys = new Set<string>();
|
|
for (const reference of scopedModelReferences(settings)) {
|
|
const parsed = parseModelReference(reference);
|
|
if (parsed) keys.add(modelKey(parsed.provider, parsed.modelId));
|
|
}
|
|
return keys;
|
|
}
|
|
|
|
async function getRegistryModels(): Promise<ModelInfo[]> {
|
|
await refreshModelConfigIfChanged();
|
|
const settings = readAgentSettings();
|
|
const available = await modelRuntime.getAvailable();
|
|
const configured = available.filter((m) => modelRuntime.hasConfiguredAuth(m.provider));
|
|
const source = configured.length > 0 ? configured : available;
|
|
const merged: ModelInfo[] = [];
|
|
const seen = new Set<string>();
|
|
|
|
for (const reference of scopedModelReferences(settings)) {
|
|
const parsed = parseModelReference(reference);
|
|
if (!parsed) continue;
|
|
const model = findConfiguredOrSyntheticModel(parsed.provider, parsed.modelId);
|
|
if (model) addModelInfo(merged, seen, model, true);
|
|
}
|
|
|
|
if (restrictModelsToScoped()) return merged;
|
|
|
|
for (const model of source) {
|
|
addModelInfo(merged, seen, model);
|
|
}
|
|
|
|
return merged;
|
|
}
|
|
|
|
async function getModels(): Promise<ModelInfo[]> {
|
|
const registryModels = await getRegistryModels();
|
|
if (restrictModelsToScoped()) return registryModels;
|
|
const litellmModels = await fetchLiteLLMModels();
|
|
|
|
const seen = new Set(registryModels.map((m) => modelKey(m.provider, m.id)));
|
|
const merged = [...registryModels];
|
|
for (const m of litellmModels) {
|
|
const key = modelKey(m.provider, m.id);
|
|
if (!seen.has(key)) {
|
|
merged.push(m);
|
|
seen.add(key);
|
|
}
|
|
}
|
|
return merged;
|
|
}
|
|
|
|
function findModel(provider: string, modelId: string): Model<Api> | undefined {
|
|
return findConfiguredOrSyntheticModel(provider, modelId);
|
|
}
|
|
|
|
function getModelFromReference(reference: string | undefined): Model<Api> | undefined {
|
|
const parsed = parseModelReference(reference);
|
|
return parsed ? findConfiguredOrSyntheticModel(parsed.provider, parsed.modelId) : undefined;
|
|
}
|
|
|
|
function isModelSelectable(provider: string, modelId: string): boolean {
|
|
if (!restrictModelsToScoped()) return true;
|
|
return scopedModelKeys(readAgentSettings()).has(modelKey(provider, modelId));
|
|
}
|
|
|
|
function defaultThinkingLevelFromSettings(): any {
|
|
const level = readAgentSettings().defaultThinkingLevel;
|
|
return typeof level === "string" && ["off", "minimal", "low", "medium", "high", "xhigh"].includes(level) ? level : "medium";
|
|
}
|
|
|
|
function applyDefaultThinkingLevel(session: AgentSession) {
|
|
if (session.model?.reasoning === true) {
|
|
session.setThinkingLevel(defaultThinkingLevelFromSettings());
|
|
}
|
|
}
|
|
|
|
function restorableSessionModel(sessionManager: SessionManager): Model<Api> | undefined {
|
|
const existingModel = sessionManager.buildSessionContext().model;
|
|
if (!existingModel) return undefined;
|
|
const model = findConfiguredOrSyntheticModel(existingModel.provider, existingModel.modelId);
|
|
if (!model || !isModelSelectable(model.provider, model.id)) return undefined;
|
|
return model;
|
|
}
|
|
|
|
function readAgentSettings(): any {
|
|
try {
|
|
return JSON.parse(fs.readFileSync(AGENT_SETTINGS_PATH, "utf8"));
|
|
} catch (err) {
|
|
console.warn(`Could not read Pi settings from ${AGENT_SETTINGS_PATH}:`, err);
|
|
return {};
|
|
}
|
|
}
|
|
|
|
async function resolveInitialWebUiModel(): Promise<Model<Api> | undefined> {
|
|
const settings = readAgentSettings();
|
|
|
|
// Explicit web UI override without changing Pi's global default.
|
|
const explicit = getModelFromReference(process.env.PI_WEBUI_MODEL || process.env.PI_MODEL);
|
|
if (explicit) return explicit;
|
|
|
|
const defaultProvider = typeof settings.defaultProvider === "string" ? settings.defaultProvider : undefined;
|
|
const defaultModel = typeof settings.defaultModel === "string" ? settings.defaultModel : undefined;
|
|
if (defaultProvider && defaultModel) {
|
|
const configuredDefault = findConfiguredOrSyntheticModel(defaultProvider, defaultModel);
|
|
if (configuredDefault) {
|
|
return configuredDefault;
|
|
}
|
|
console.warn(`Configured Pi default model is not available in this SDK: ${defaultProvider}/${defaultModel}`);
|
|
}
|
|
|
|
// The interactive local Pi agent also scopes/cycles models from enabledModels; use the first
|
|
// configured entry there before falling back to provider defaults like Anthropic Opus.
|
|
if (Array.isArray(settings.enabledModels)) {
|
|
for (const reference of settings.enabledModels) {
|
|
if (typeof reference !== "string") continue;
|
|
const model = getModelFromReference(reference);
|
|
if (model) return model;
|
|
}
|
|
}
|
|
|
|
const available = await modelRuntime.getAvailable();
|
|
return available.find((m) => m.provider === "local" && modelRuntime.hasConfiguredAuth(m.provider))
|
|
|| available.find((m) => m.provider !== "anthropic" && modelRuntime.hasConfiguredAuth(m.provider))
|
|
|| available.find((m) => modelRuntime.hasConfiguredAuth(m.provider));
|
|
}
|
|
|
|
const requestChoiceParams = Type.Object({
|
|
prompt: Type.String({ description: "Short question shown to the user." }),
|
|
choices: Type.Array(Type.Object({
|
|
id: Type.String({ description: "Stable machine-readable choice id." }),
|
|
label: Type.String({ description: "Short button label." }),
|
|
description: Type.Optional(Type.String({ description: "Optional one-line explanation." })),
|
|
}), { description: "2-6 concrete choices." }),
|
|
allowMultiple: Type.Optional(Type.Boolean({ description: "Whether the user may choose multiple options. Default false." })),
|
|
});
|
|
|
|
function createRequestChoiceTool(onChoiceRequest: (request: ChoiceRequest, signal?: AbortSignal) => Promise<string[]>) {
|
|
return defineTool({
|
|
name: "request-choice",
|
|
label: "request-choice",
|
|
description: "Ask the user to choose from concrete options when guessing would be risky.",
|
|
promptSnippet: "request-choice: ask the user to choose one or more concrete options.",
|
|
promptGuidelines: [
|
|
"Use request-choice for discrete user decisions where guessing would be risky. Keep choices few and concrete.",
|
|
"After calling request-choice, stop and wait for the user's choice response instead of continuing to guess.",
|
|
],
|
|
parameters: requestChoiceParams,
|
|
async execute(toolCallId, params, signal) {
|
|
const choices = (params.choices || []).slice(0, 8).map((choice: any, index: number) => ({
|
|
id: String(choice.id || `choice-${index + 1}`).trim().slice(0, 80),
|
|
label: String(choice.label || choice.id || `Choice ${index + 1}`).trim().slice(0, 120),
|
|
description: choice.description ? String(choice.description).trim().slice(0, 300) : undefined,
|
|
})).filter((choice: any) => choice.id && choice.label);
|
|
if (choices.length < 2) {
|
|
const data = { type: "error", error: "request_choice_needs_choices", message: "request-choice requires at least two non-empty choices." };
|
|
return { content: [{ type: "text", text: JSON.stringify(data, null, 2) }], details: data } as any;
|
|
}
|
|
const request: ChoiceRequest = {
|
|
id: toolCallId,
|
|
prompt: String(params.prompt || "Please choose an option.").trim().slice(0, 500),
|
|
choices,
|
|
allowMultiple: Boolean(params.allowMultiple),
|
|
};
|
|
const selected = await onChoiceRequest(request, signal);
|
|
const data = { type: "choice_response", request, selected };
|
|
return { content: [{ type: "text", text: JSON.stringify(data, null, 2) }], details: data } as any;
|
|
},
|
|
});
|
|
}
|
|
|
|
function assertSessionResources(session: AgentSession, resourceLoader: DefaultResourceLoader) {
|
|
const requiredToolNames = ["read", "bash", "edit", "write", "request-choice"];
|
|
const activeToolNames = new Set(session.getActiveToolNames());
|
|
const missingTools = requiredToolNames.filter((name) => !activeToolNames.has(name));
|
|
if (missingTools.length > 0) throw new Error(`Session startup did not activate these tools: ${missingTools.join(", ")}`);
|
|
|
|
const prompt = session.agent.state.systemPrompt;
|
|
const missingSkills = resourceLoader.getSkills().skills
|
|
.filter((skill) => !skill.disableModelInvocation && !prompt.includes(`<name>${skill.name}</name>`))
|
|
.map((skill) => skill.name);
|
|
if (missingSkills.length > 0) throw new Error(`Session startup did not advertise these skills: ${missingSkills.join(", ")}`);
|
|
}
|
|
|
|
async function createWebUiSession(sessionManager: SessionManager, onChoiceRequest: (request: ChoiceRequest, signal?: AbortSignal) => Promise<string[]>) {
|
|
await refreshModelConfigIfChanged();
|
|
const restoredModel = restorableSessionModel(sessionManager);
|
|
const initialModel = restoredModel ? undefined : await resolveInitialWebUiModel();
|
|
const initialThinkingLevel = restoredModel ? undefined : defaultThinkingLevelFromSettings();
|
|
const customTools = [createRequestChoiceTool(onChoiceRequest)];
|
|
const settingsManager = SettingsManager.create(HOME_DIR, AGENT_DIR);
|
|
const resourceLoader = new DefaultResourceLoader({ cwd: HOME_DIR, agentDir: AGENT_DIR, settingsManager, additionalSkillPaths: [USER_SKILLS_DIR, SHARED_SKILLS_DIR] });
|
|
await resourceLoader.reload();
|
|
const result = await createAgentSession({ cwd: HOME_DIR, agentDir: AGENT_DIR, modelRuntime, sessionManager, model: initialModel, thinkingLevel: initialThinkingLevel, customTools, settingsManager, resourceLoader });
|
|
if (!restoredModel) applyDefaultThinkingLevel(result.session);
|
|
assertSessionResources(result.session, resourceLoader);
|
|
return result;
|
|
}
|
|
|
|
function safeSerializeEvent(event: AgentSessionEvent): any {
|
|
try {
|
|
const parsed = JSON.parse(JSON.stringify(event));
|
|
if (parsed?.type === "agent_end" && Array.isArray(parsed.messages)) return { ...parsed, messages: undefined, messagesTruncated: true };
|
|
return parsed;
|
|
} catch { return { type: (event as any).type, _serialized: false }; }
|
|
}
|
|
|
|
function runtimeSnapshot(runtime: WebUiRuntime): SerializedAgentState {
|
|
return serializeState(runtime.session);
|
|
}
|
|
|
|
let broker: SessionBroker<RuntimeContext, WebUiSession>;
|
|
const sessionRuntimes = new WeakMap<object, WebUiRuntime>();
|
|
|
|
function makeRuntimeFactory() {
|
|
return {
|
|
keyForContext: (context: Readonly<RuntimeContext>): string => context.sessionManager.getSessionId(),
|
|
create: async (_key: string, context: Readonly<RuntimeContext>): Promise<WebUiSession> => {
|
|
let sessionForRuntime: WebUiSession;
|
|
const result = await createWebUiSession(context.sessionManager, (request, signal) => {
|
|
const runtime = sessionRuntimes.get(sessionForRuntime as unknown as object);
|
|
if (!runtime) return Promise.reject(new Error("Session runtime is not ready"));
|
|
const pending = runtime.requestChoice({ choiceId: request.id, payload: request }) as Promise<string[]>;
|
|
const reject = () => runtime.rejectChoice(request.id, "Choice cancelled");
|
|
if (signal?.aborted) reject();
|
|
else signal?.addEventListener("abort", reject, { once: true });
|
|
return pending.finally(() => signal?.removeEventListener("abort", reject));
|
|
});
|
|
sessionForRuntime = result.session as WebUiSession;
|
|
return sessionForRuntime;
|
|
},
|
|
};
|
|
}
|
|
|
|
function runtimeForManager(sessionManager: SessionManager): Promise<WebUiRuntime> {
|
|
return broker.getOrCreate({ sessionManager });
|
|
}
|
|
|
|
function makeBroker(): SessionBroker<RuntimeContext, WebUiSession> {
|
|
return new SessionBroker({
|
|
factory: makeRuntimeFactory(),
|
|
serializer: { snapshot: runtimeSnapshot },
|
|
messages: {
|
|
ready: (runtime) => ({ type: "ready", runtimeId: runtime.runtimeId, sessionId: runtime.session.sessionId }),
|
|
stateSync: (_runtime, state) => ({ type: "stateSync", state }),
|
|
event: (event: AgentSessionEvent, runtime) => {
|
|
const safeEvent = safeSerializeEvent(event);
|
|
const messages: unknown[] = [{ type: "agentEvent", event: safeEvent }];
|
|
if (["agent_start", "agent_end", "message_end", "turn_end", "compaction_end"].includes(event.type)) {
|
|
messages.push({ type: "stateSync", state: serializeState(runtime.session, event.type === "agent_end" ? { isStreaming: false, streamingMessage: null } : {}) });
|
|
}
|
|
if (event.type === "message_end" || event.type === "turn_end") clearSessionListCache();
|
|
if (event.type === "message_end") {
|
|
setImmediate(() => {
|
|
listRecentSessions().then((page) => runtime.broadcastMessage({ type: "sessions", sessions: page.items, currentSessionId: runtime.session.sessionId, offset: page.offset, limit: page.limit, total: page.total, hasMore: page.hasMore, query: page.query })).catch((error) => console.error("Failed to refresh sessions after persistence:", error));
|
|
});
|
|
}
|
|
return messages;
|
|
},
|
|
choiceRequest: (request) => ({ type: "agentEvent", event: { type: "choice_request", request: request.payload } }),
|
|
choiceResolved: (requestId, selected) => ({ type: "agentEvent", event: { type: "choice_resolved", requestId, selected } }),
|
|
promptAccepted: (requestId) => ({ type: "promptAccepted", requestId }),
|
|
promptRejected: (requestId, message) => ({ type: "promptRejected", requestId, message: message || "Prompt was rejected" }),
|
|
},
|
|
idleTtlMs: parseLimitedIntEnv("PI_WEBUI_DETACHED_IDLE_TTL_MS", 15 * 60_000),
|
|
activeMaxAgeMs: parseLimitedIntEnv("PI_WEBUI_ACTIVE_MAX_AGE_MS", 24 * 60 * 60_000),
|
|
maxRuntimes: parseLimitedIntEnv("PI_WEBUI_MAX_RUNTIMES", 20),
|
|
onDelete: async (_key, runtime) => {
|
|
const sessionPath = runtime.session.sessionFile;
|
|
if (sessionPath) await fs.promises.unlink(sessionPath).catch((error: any) => { if (error?.code !== "ENOENT") throw error; });
|
|
clearSessionListCache();
|
|
},
|
|
onError: (error) => console.error("Session broker error:", error),
|
|
});
|
|
}
|
|
|
|
function isClientMessage(value: any): value is ClientMessage {
|
|
if (!value || typeof value.type !== "string") return false;
|
|
switch (value.type) {
|
|
case "prompt":
|
|
return typeof value.text === "string" && value.text.length <= 100_000 && typeof value.requestId === "string" && value.requestId.length >= 1 && value.requestId.length <= 200;
|
|
case "steer":
|
|
case "followUp":
|
|
return typeof value.text === "string" && value.text.length <= 100_000;
|
|
case "setModel":
|
|
return typeof value.provider === "string" && typeof value.modelId === "string";
|
|
case "setThinkingLevel":
|
|
return typeof value.level === "string";
|
|
case "loadSession":
|
|
case "deleteSession":
|
|
return typeof value.sessionPath === "string" || typeof value.sessionId === "string";
|
|
case "renameSession":
|
|
return (typeof value.sessionPath === "string" || typeof value.sessionId === "string") && typeof value.name === "string" && value.name.length <= 200;
|
|
case "choiceResponse":
|
|
return typeof value.requestId === "string" && value.requestId.length <= 200 && Array.isArray(value.selected) && value.selected.length <= 8 && value.selected.every((item: any) => typeof item === "string" && item.length <= 80);
|
|
case "getMessages":
|
|
return typeof value.offset === "number";
|
|
case "abort":
|
|
case "getModels":
|
|
case "getState":
|
|
case "newSession":
|
|
case "getSessions":
|
|
return true;
|
|
default:
|
|
return false;
|
|
}
|
|
}
|
|
|
|
async function handleClientMessage(ws: WebSocket, msg: ClientMessage) {
|
|
const reply = (message: ServerMessage) => {
|
|
const current = clients.get(ws);
|
|
if (current && current.runtime.isAttachmentValid(current)) send(ws, message);
|
|
};
|
|
try {
|
|
const handle = clients.get(ws);
|
|
if (!handle || !handle.runtime.isAttachmentValid(handle)) {
|
|
send(ws, { type: "error", message: "Session is not ready yet" });
|
|
return;
|
|
}
|
|
const runtime = handle.runtime;
|
|
const session = runtime.session;
|
|
const assertAttachment = (attachment: AttachmentHandle<RuntimeContext, WebUiSession> = handle): void => {
|
|
if (!attachment.runtime.isAttachmentValid(attachment)) throw new Error("Session attachment is no longer valid");
|
|
};
|
|
switch (msg.type) {
|
|
case "prompt": {
|
|
const fingerprint = createHash("sha256").update(JSON.stringify({ text: msg.text })).digest("hex");
|
|
await runtime.prompt(handle, { requestId: msg.requestId, text: msg.text }, fingerprint);
|
|
break;
|
|
}
|
|
case "steer": await runtime.steer(handle, msg.text); break;
|
|
case "followUp": await runtime.followUp(handle, msg.text); break;
|
|
case "choiceResponse": {
|
|
const request = runtime.getChoice(msg.requestId)?.payload as ChoiceRequest | undefined;
|
|
if (!request) { reply({ type: "error", message: "Choice request is no longer active" }); break; }
|
|
const selected = msg.selected.filter((id) => request.choices.some((choice: any) => choice.id === id));
|
|
if (!request.allowMultiple && selected.length > 1) selected.splice(1);
|
|
if (!selected.length) { reply({ type: "error", message: "Please choose one of the listed options" }); break; }
|
|
await runtime.resolveChoice(handle, msg.requestId, selected);
|
|
break;
|
|
}
|
|
case "abort": await runtime.abort(handle); reply({ type: "stateSync", state: serializeState(session) }); break;
|
|
case "getModels": {
|
|
const models = await getModels();
|
|
reply({ type: "models", models, current: session.model ? modelToInfo(session.model) : undefined, thinkingLevel: session.thinkingLevel });
|
|
break;
|
|
}
|
|
case "setModel": {
|
|
await refreshModelConfigIfChanged();
|
|
let model: Model<Api> | undefined;
|
|
for (const candidate of buildModelLookupCandidates(msg.provider, msg.modelId)) { model = findModel(candidate.provider, candidate.modelId); if (model) break; }
|
|
if (!model) throw new Error(`Model not found: ${msg.provider}/${msg.modelId}`);
|
|
if (!isModelSelectable(model.provider, model.id)) throw new Error(`Model is not enabled for this Web UI: ${model.provider}/${model.id}`);
|
|
await runtime.invoke(handle, async () => { await session.setModel(model!); applyDefaultThinkingLevel(session); });
|
|
reply({ type: "modelChanged", model: modelToInfo(model), thinkingLevel: session.thinkingLevel });
|
|
break;
|
|
}
|
|
case "setThinkingLevel":
|
|
await runtime.invoke(handle, async () => { session.setThinkingLevel(msg.level as any); });
|
|
reply({ type: "modelChanged", model: session.model ? modelToInfo(session.model) : { provider: "", id: "", name: "" }, thinkingLevel: session.thinkingLevel });
|
|
break;
|
|
case "getState": reply({ type: "stateSync", state: serializeState(session) }); break;
|
|
case "getMessages": {
|
|
const page = messagePage(session, msg.offset, msg.limit);
|
|
reply({ type: "messagePage", messages: page.messages, offset: page.offset, limit: page.limit, total: page.total, hasMoreBefore: page.hasMoreBefore });
|
|
break;
|
|
}
|
|
case "getSessions": {
|
|
const page = await listRecentSessions({ offset: msg.offset, limit: msg.limit, query: msg.query });
|
|
reply({ type: "sessions", sessions: page.items, currentSessionId: session.sessionId, offset: page.offset, limit: page.limit, total: page.total, hasMore: page.hasMore, query: page.query });
|
|
break;
|
|
}
|
|
case "renameSession": {
|
|
const requestedPath = await resolveRequestedSessionPath(msg);
|
|
if (!requestedPath) throw new Error("Session not found or outside the configured session directory");
|
|
const name = msg.name.trim().replace(/[\r\n\t]+/g, " ").replace(/\s+/g, " ").slice(0, 120);
|
|
await withSessionPathLock(requestedPath, async () => {
|
|
if (!fs.existsSync(requestedPath)) throw new Error("Session not found or outside the configured session directory");
|
|
if (session.sessionFile && path.resolve(session.sessionFile) === path.resolve(requestedPath)) {
|
|
await runtime.invoke(handle, async () => { (session as any).setSessionName(name); });
|
|
reply({ type: "stateSync", state: serializeState(session) });
|
|
} else {
|
|
const item = sessionItemFromContent(requestedPath, await fs.promises.readFile(requestedPath, "utf8"), new Date());
|
|
const target = item ? broker.get(item.id) : undefined;
|
|
if (target) await target.invokeOrdered(async () => { (target.session as any).setSessionName(name); });
|
|
else SessionManager.open(requestedPath, getSessionDir(HOME_DIR), HOME_DIR).appendSessionInfo(name);
|
|
}
|
|
clearSessionListCache();
|
|
});
|
|
const page = await listRecentSessions();
|
|
reply({ type: "sessions", sessions: page.items, currentSessionId: session.sessionId, offset: page.offset, limit: page.limit, total: page.total, hasMore: page.hasMore, query: page.query });
|
|
break;
|
|
}
|
|
case "newSession": {
|
|
const next = await runtimeForManager(SessionManager.create(HOME_DIR, getSessionDir(HOME_DIR)));
|
|
const nextHandle = next.attach({ isOpen: () => ws.readyState === WebSocket.OPEN, send: (message) => ws.send(JSON.stringify(message)) });
|
|
sessionRuntimes.set(next.session as unknown as object, next);
|
|
const current = clients.get(ws);
|
|
if (!current) throw new Error("Session attachment is no longer valid");
|
|
current.detach();
|
|
clients.set(ws, nextHandle);
|
|
send(ws, { type: "sessionChanged", sessionId: next.session.sessionId });
|
|
console.log(`New session created: ${next.session.sessionId}`);
|
|
break;
|
|
}
|
|
case "loadSession": {
|
|
assertAttachment(handle);
|
|
let next = typeof msg.sessionId === "string" ? broker.get(msg.sessionId) : undefined;
|
|
const requestedPath = msg.sessionPath ? resolveSessionPath(msg.sessionPath) : undefined;
|
|
const attachNext = (target: WebUiRuntime): void => {
|
|
assertAttachment(handle);
|
|
const nextHandle = target.attach({ isOpen: () => ws.readyState === WebSocket.OPEN, send: (message) => ws.send(JSON.stringify(message)) });
|
|
sessionRuntimes.set(target.session as unknown as object, target);
|
|
const current = clients.get(ws);
|
|
if (current) current.detach();
|
|
clients.set(ws, nextHandle);
|
|
send(ws, { type: "sessionChanged", sessionId: target.session.sessionId });
|
|
};
|
|
if (!next && requestedPath) {
|
|
await withSessionPathLock(requestedPath, async () => {
|
|
assertAttachment(handle);
|
|
if (!fs.existsSync(requestedPath)) throw new Error("Session not found or outside the configured session directory");
|
|
const content = await fs.promises.readFile(requestedPath, "utf8");
|
|
const item = sessionItemFromContent(requestedPath, content, new Date());
|
|
if (!item) throw new Error("Session file is invalid");
|
|
next = broker.get(item.id) || await runtimeForManager(SessionManager.open(requestedPath, getSessionDir(HOME_DIR), HOME_DIR));
|
|
attachNext(next);
|
|
});
|
|
} else if (!next) {
|
|
const resolvedPath = await resolveSessionPathById(msg.sessionId);
|
|
assertAttachment(handle);
|
|
if (!resolvedPath) throw new Error("Session not found or outside the configured session directory");
|
|
await withSessionPathLock(resolvedPath, async () => {
|
|
assertAttachment(handle);
|
|
if (!fs.existsSync(resolvedPath)) throw new Error("Session not found or outside the configured session directory");
|
|
const content = await fs.promises.readFile(resolvedPath, "utf8");
|
|
const item = sessionItemFromContent(resolvedPath, content, new Date());
|
|
if (!item) throw new Error("Session file is invalid");
|
|
next = broker.get(item.id) || await runtimeForManager(SessionManager.open(resolvedPath, getSessionDir(HOME_DIR), HOME_DIR));
|
|
attachNext(next);
|
|
});
|
|
} else {
|
|
attachNext(next);
|
|
}
|
|
break;
|
|
}
|
|
case "deleteSession": {
|
|
const requestedPath = await resolveRequestedSessionPath(msg);
|
|
if (!requestedPath) throw new Error("Session not found or outside the configured session directory");
|
|
await withSessionPathLock(requestedPath, async () => {
|
|
assertAttachment(handle);
|
|
if (!fs.existsSync(requestedPath)) throw new Error("Session not found or outside the configured session directory");
|
|
const item = sessionItemFromContent(requestedPath, await fs.promises.readFile(requestedPath, "utf8"), new Date());
|
|
const target = session.sessionFile && path.resolve(session.sessionFile) === path.resolve(requestedPath) ? runtime : undefined;
|
|
const targetRuntime = target || (item ? broker.get(item.id) : undefined);
|
|
if (targetRuntime) {
|
|
if (targetRuntime.isActive || targetRuntime.attachmentCount > (target ? 1 : 0)) throw new Error("Cannot delete a conversation that is active or attached elsewhere");
|
|
await broker.delete(targetRuntime.key);
|
|
if (target) {
|
|
const replacement = await runtimeForManager(SessionManager.create(HOME_DIR, getSessionDir(HOME_DIR)));
|
|
const nextHandle = replacement.attach({ isOpen: () => ws.readyState === WebSocket.OPEN, send: (message) => ws.send(JSON.stringify(message)) });
|
|
sessionRuntimes.set(replacement.session as unknown as object, replacement);
|
|
clients.set(ws, nextHandle);
|
|
send(ws, { type: "sessionChanged", sessionId: replacement.session.sessionId });
|
|
}
|
|
} else {
|
|
await fs.promises.unlink(requestedPath);
|
|
}
|
|
clearSessionListCache();
|
|
});
|
|
const current = clients.get(ws)?.runtime.session || session;
|
|
const page = await listRecentSessions();
|
|
reply({ type: "sessions", sessions: page.items, currentSessionId: current.sessionId, offset: page.offset, limit: page.limit, total: page.total, hasMore: page.hasMore, query: page.query });
|
|
break;
|
|
}
|
|
}
|
|
} catch (err: any) { console.error(`Error handling ${msg.type}:`, err); reply({ type: "error", message: err.message || String(err) }); }
|
|
}
|
|
|
|
app.get("/api/skills", requireEditAuth, async (_req, res) => {
|
|
const skills = await listSkills();
|
|
res.json({
|
|
ok: true,
|
|
userRoot: USER_SKILLS_DIR,
|
|
sharedRoot: SHARED_SKILLS_DIR,
|
|
skills: skills.map((skill) => ({ ...skill, path: skill.editable ? skill.path : undefined })),
|
|
});
|
|
});
|
|
|
|
app.get("/api/skills/:source/:id", requireEditAuth, async (req, res) => {
|
|
const result = await getSkillContent(String(req.params.source || ""), String(req.params.id || ""));
|
|
if (!result) return res.status(404).json({ ok: false, error: "not_found" });
|
|
const publicItem = result.item.editable ? result.item : { ...result.item, path: undefined };
|
|
res.json({ ok: true, skill: publicItem, content: result.content });
|
|
});
|
|
|
|
app.post("/api/skills", requireEditAuth, async (req, res) => {
|
|
const name = String(req.body?.name || "").trim().toLowerCase();
|
|
const description = normalizeSkillDescription(String(req.body?.description || ""));
|
|
if (!isValidSkillName(name)) return res.status(400).json({ ok: false, error: "invalid_name", message: "Use lowercase letters, numbers, and single hyphens only." });
|
|
if (!description || description.length > 1024) return res.status(400).json({ ok: false, error: "invalid_description", message: "Description is required and must be at most 1024 characters." });
|
|
const skillFile = userSkillPath(name);
|
|
if (!skillFile) return res.status(400).json({ ok: false, error: "invalid_name" });
|
|
if (fs.existsSync(skillFile)) return res.status(409).json({ ok: false, error: "already_exists" });
|
|
const escapedDescription = description.replaceAll('"', "'");
|
|
const content = `---\nname: ${name}\ndescription: "${escapedDescription}"\n---\n\n# ${name}\n\n## When to use\n\n${description}\n\n## Instructions\n\n- Describe the workflow this skill should guide.\n- Add any setup, commands, references, or safety constraints the agent should follow.\n`;
|
|
await writeSkillAtomic(skillFile, content);
|
|
const item = await readSkillFile(skillFile, "user", true, createParsingState());
|
|
res.json({ ok: true, skill: item, content, note: "Start a new session or reload the current one for Pi to advertise new/changed skills." });
|
|
});
|
|
|
|
app.put("/api/skills/user/:id", requireEditAuth, async (req, res) => {
|
|
const id = String(req.params.id || "").trim();
|
|
const content = String(req.body?.content || "");
|
|
if (!isValidSkillName(id)) return res.status(400).json({ ok: false, error: "invalid_name" });
|
|
if (Buffer.byteLength(content, "utf8") > 100_000) return res.status(413).json({ ok: false, error: "content_too_large", message: "Skill content must be at most 100 KB." });
|
|
const parsed = parseSkillFrontmatter(content);
|
|
if (parsed.diagnostics.length > 0) return res.status(400).json({ ok: false, error: "invalid_skill", diagnostics: parsed.diagnostics });
|
|
if (parsed.fields.name !== id) return res.status(400).json({ ok: false, error: "name_mismatch", message: "The frontmatter name must match the skill being edited. Create a new skill to rename." });
|
|
const skillFile = userSkillPath(id);
|
|
if (!skillFile || !fs.existsSync(skillFile)) return res.status(404).json({ ok: false, error: "not_found" });
|
|
await writeSkillAtomic(skillFile, content);
|
|
const item = await readSkillFile(skillFile, "user", true, createParsingState());
|
|
res.json({ ok: true, skill: item, note: "Start a new session or reload the current one for Pi to advertise new/changed skills." });
|
|
});
|
|
|
|
app.delete("/api/skills/user/:id", requireEditAuth, async (req, res) => {
|
|
const id = String(req.params.id || "").trim();
|
|
const skillFile = userSkillPath(id);
|
|
if (!skillFile || !fs.existsSync(skillFile)) return res.status(404).json({ ok: false, error: "not_found" });
|
|
await fs.promises.rm(path.dirname(skillFile), { recursive: true, force: true });
|
|
res.json({ ok: true, note: "Deleted. Start a new session or reload the current one to remove it from advertised skills." });
|
|
});
|
|
|
|
// Serve static client files in production after API routes.
|
|
const clientDist = path.resolve(__dirname, "../dist");
|
|
const TEXT_PREVIEW_EXTENSIONS = new Set([
|
|
".txt", ".md", ".markdown", ".json", ".jsonl", ".csv", ".tsv", ".log", ".xml", ".html", ".htm", ".css", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", ".py", ".php", ".sh", ".bash", ".zsh", ".yaml", ".yml", ".toml", ".ini", ".conf", ".sql",
|
|
]);
|
|
|
|
async function resolveHomeFilePath(requested: string): Promise<string | undefined> {
|
|
const trimmed = requested.trim();
|
|
if (!trimmed) return undefined;
|
|
const home = await fs.promises.realpath(os.homedir()).catch(() => undefined);
|
|
const requestedPath = path.resolve(path.isAbsolute(trimmed) ? trimmed : path.join(os.homedir(), trimmed));
|
|
const resolved = await fs.promises.realpath(requestedPath).catch(() => undefined);
|
|
if (!home || !resolved || (resolved !== home && !resolved.startsWith(`${home}${path.sep}`))) return undefined;
|
|
return resolved;
|
|
}
|
|
|
|
function canPreviewTextFile(filePath: string, size: number): boolean {
|
|
return size <= 2 * 1024 * 1024 && TEXT_PREVIEW_EXTENSIONS.has(path.extname(filePath).toLowerCase());
|
|
}
|
|
|
|
app.get("/api/files", requireHttpAuth, async (req, res) => {
|
|
const requested = typeof req.query.path === "string" ? req.query.path : "";
|
|
const filePath = await resolveHomeFilePath(requested);
|
|
if (!filePath) return res.status(403).type("text/plain").send("File is outside the workspace");
|
|
const stats = await fs.promises.stat(filePath).catch(() => undefined);
|
|
if (!stats?.isFile()) return res.status(404).type("text/plain").send("File not found");
|
|
if (req.query.download === "1") return res.download(filePath, path.basename(filePath));
|
|
if (req.query.inline === "1") {
|
|
const inlineTypes: Record<string, string> = { ".pdf": "application/pdf", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp" };
|
|
const type = inlineTypes[path.extname(filePath).toLowerCase()];
|
|
if (!type) return res.status(415).type("text/plain").send("File type is not previewable");
|
|
res.type(type);
|
|
return res.sendFile(filePath);
|
|
}
|
|
if (req.query.raw === "1") {
|
|
if (!canPreviewTextFile(filePath, stats.size)) return res.status(415).type("text/plain").send("File type is not previewable as text");
|
|
res.type("text/plain");
|
|
return res.sendFile(filePath);
|
|
}
|
|
return res.status(400).type("text/plain").send("Use the file preview page");
|
|
});
|
|
|
|
app.use(express.static(clientDist));
|
|
app.get("/{*path}", (_req, res) => {
|
|
res.sendFile(path.join(clientDist, "client", "index.html"));
|
|
});
|
|
|
|
async function main() {
|
|
console.log("Initializing Pi agent session...");
|
|
|
|
if (!webUiToken && (!allowAnonymous || !anonymousLoopbackOnly)) {
|
|
throw new Error("PI_WEBUI_TOKEN is required. Anonymous mode is allowed only when PI_WEBUI_ALLOW_ANONYMOUS=true and HOST is loopback.");
|
|
}
|
|
|
|
modelRuntime = await ModelRuntime.create({ authPath: AGENT_AUTH_PATH, modelsPath: AGENT_MODELS_PATH });
|
|
modelAuthStamp = fileStamp(AGENT_AUTH_PATH);
|
|
modelConfigStamp = [modelAuthStamp, fileStamp(AGENT_MODELS_PATH), fileStamp(AGENT_SETTINGS_PATH)].join("|");
|
|
setInterval(() => refreshModelConfigIfChanged().catch((err) => console.error("Model configuration refresh failed:", err)), 2000).unref();
|
|
|
|
if (!litellmKey) {
|
|
const auth = await modelRuntime.getAuth("ollama");
|
|
if (auth?.auth.apiKey) litellmKey = auth.auth.apiKey;
|
|
}
|
|
|
|
broker = makeBroker();
|
|
const sweepTimer = setInterval(() => broker.sweep().catch((error) => console.error("Session broker sweep failed:", error)), 60_000);
|
|
sweepTimer.unref();
|
|
let shuttingDown = false;
|
|
const shutdown = async () => {
|
|
if (shuttingDown) return;
|
|
shuttingDown = true;
|
|
clearInterval(sweepTimer);
|
|
for (const ws of wss.clients) ws.close(1001, "Server shutting down");
|
|
const deadline = new Promise<void>((resolve) => setTimeout(resolve, 30_000).unref());
|
|
await Promise.race([broker.shutdown(30_000), deadline]);
|
|
await Promise.race([
|
|
new Promise<void>((resolve) => server.close(() => resolve())),
|
|
deadline,
|
|
]);
|
|
for (const ws of wss.clients) ws.terminate();
|
|
};
|
|
process.once("SIGTERM", () => void shutdown().finally(() => process.exit(0)));
|
|
process.once("SIGINT", () => void shutdown().finally(() => process.exit(0)));
|
|
|
|
wss.on("connection", async (ws, request) => {
|
|
let closed = false;
|
|
const sink = { isOpen: () => !closed && ws.readyState === WebSocket.OPEN, send: (message: unknown) => ws.send(JSON.stringify(message)) };
|
|
let handle: AttachmentHandle<RuntimeContext, WebUiSession> | undefined;
|
|
const detach = () => {
|
|
closed = true;
|
|
const current = clients.get(ws);
|
|
if (current) current.detach();
|
|
clients.delete(ws);
|
|
attachmentTails.delete(ws);
|
|
console.log(`Client disconnected (${clients.size} total)`);
|
|
};
|
|
ws.once("close", detach);
|
|
try {
|
|
const url = new URL(request.url || "/api/ws", `http://${request.headers.host || "localhost"}`);
|
|
const requestedId = url.searchParams.get("session") || url.searchParams.get("sessionId") || undefined;
|
|
const liveRuntime = !url.searchParams.get("sessionPath") && requestedId ? broker.get(requestedId) : undefined;
|
|
let runtime: WebUiRuntime;
|
|
if (liveRuntime) {
|
|
runtime = liveRuntime;
|
|
} else {
|
|
const requestedPath = url.searchParams.get("sessionPath")
|
|
? resolveSessionPath(url.searchParams.get("sessionPath") || "")
|
|
: await resolveSessionPathById(requestedId);
|
|
if (requestedId && !requestedPath) throw new Error("Session not found or outside the configured session directory");
|
|
if (requestedPath) {
|
|
runtime = await withSessionPathLock(requestedPath, async () => {
|
|
if (!fs.existsSync(requestedPath)) throw new Error("Session not found or outside the configured session directory");
|
|
const content = await fs.promises.readFile(requestedPath, "utf8");
|
|
const item = sessionItemFromContent(requestedPath, content, new Date());
|
|
if (!item) throw new Error("Session file is invalid");
|
|
const context: RuntimeContext = { sessionManager: SessionManager.open(requestedPath, getSessionDir(HOME_DIR), HOME_DIR) };
|
|
return runtimeForManager(context.sessionManager);
|
|
});
|
|
} else {
|
|
const context: RuntimeContext = { sessionManager: SessionManager.create(HOME_DIR, getSessionDir(HOME_DIR)) };
|
|
runtime = await runtimeForManager(context.sessionManager);
|
|
}
|
|
}
|
|
sessionRuntimes.set(runtime.session as unknown as object, runtime);
|
|
handle = runtime.attach(sink);
|
|
clients.set(ws, handle);
|
|
console.log(`Client session ${runtime.session.sessionId}: ${runtime.session.model?.provider}/${runtime.session.model?.id}, thinking ${runtime.session.thinkingLevel}, tools ${runtime.session.getActiveToolNames().join(", ")}`);
|
|
} catch (err: any) {
|
|
console.error("Failed to attach client session:", err);
|
|
if (!closed) { send(ws, { type: "error", message: err.message || String(err) }); ws.close(); }
|
|
return;
|
|
}
|
|
ws.on("message", (data) => {
|
|
try {
|
|
const parsed = JSON.parse(data.toString());
|
|
if (!isClientMessage(parsed)) { send(ws, { type: "error", message: "Invalid client message" }); return; }
|
|
if (["newSession", "loadSession", "deleteSession", "renameSession"].includes(parsed.type)) {
|
|
const previous = attachmentTails.get(ws) || Promise.resolve();
|
|
const current = previous.catch(() => {}).then(() => handleClientMessage(ws, parsed));
|
|
attachmentTails.set(ws, current.catch(() => {}));
|
|
} else {
|
|
void handleClientMessage(ws, parsed);
|
|
}
|
|
} catch (err) { console.error("Invalid message:", err); send(ws, { type: "error", message: "Invalid JSON message" }); }
|
|
});
|
|
});
|
|
|
|
server.listen(PORT, HOST, () => {
|
|
console.log(`Pi WebUI server listening on http://${HOST}:${PORT}`);
|
|
|
|
if (webUiToken) {
|
|
console.log("PI_WEBUI_TOKEN is required and enabled. Open the UI with ?token=<token> once; the client stores it in localStorage and then removes it from the URL.");
|
|
}
|
|
|
|
if (!webUiToken) {
|
|
console.log("PI_WEBUI_ALLOW_ANONYMOUS=true is set on loopback; tokenless local development access is enabled.");
|
|
}
|
|
});
|
|
}
|
|
|
|
main().catch((err) => {
|
|
console.error("Fatal:", err);
|
|
process.exit(1);
|
|
});
|