Menu
popagent
publicLatest change 7cecf6a89b6f39dae8c5678f5b369014199aeb3c - Add self-hosted Mastra observability by AkurAI Build
import { handleChatStream } from "@mastra/ai-sdk";
import {
createUIMessageStream,
createUIMessageStreamResponse,
type UIMessage,
type UIMessageStreamWriter,
} from "ai";
import { z } from "zod";
import index from "./ui/index.html";
import {
AGENT_TOOL_NAMES,
DEFAULT_WORKSPACE_ID,
MAX_AGENT_INSTRUCTIONS_CHARACTERS,
type AgentActivity,
type AgentRuntimeSettingsInput,
type BrowserSettingsInput,
type AgentSettingsInput,
type AgentSkillInput,
type ChatMessage,
type MemorySettingsInput,
} from "./api-types";
import { agentActivity, AgentExecutionRuntime, createAgentTaskExecutor, createTraceId } from "./agent-execution";
import { agentWorkspaces, type AgentWorkspaceStore } from "./agent-workspaces";
import { agent, mastra } from "./agent";
import { agentRuntimeSettings, type AgentRuntimeSettingsStore } from "./agent-runtime-settings";
import { agentSettings, type AgentSettingsStore } from "./agent-settings";
import { agentSkills, type AgentSkillStore } from "./agent-skills";
import { browserRuntime, browserSettings, type BrowserRuntime, type BrowserSettingsStore } from "./browser-settings";
import { latestUserText, replaceLatestUserText } from "./chat-messages";
import { createChannelRoutes } from "./channel-routes";
import { channels } from "./channels";
import { hookAudit } from "./hook-audit";
import { createTaskLifecycleHooks } from "./hook-lifecycle";
import { createHookEvent, HookBlockedError, loadHookRuntime, type HookRuntime } from "./hooks";
import { longTermMemory, MAX_FACT_CHARACTERS, type LongTermMemoryStore } from "./long-term-memory";
import { memorySettings, type MemorySettingsStore } from "./memory-settings";
import { isKnownModel, listModelCatalog } from "./models";
import {
getObservabilityOverview,
getObservabilityTrace,
listObservabilityLogs,
listObservabilityTraces,
observabilityHealth,
parseObservabilityQuery,
} from "./observability-store";
import { RESOURCE_ID, SessionStore } from "./sessions";
import { AgentTaskRuntime, AgentTaskStore } from "./tasks";
import { requireApiKey } from "./auth";
import { startMaintenance } from "./maintenance";
import { appLogger, observability } from "./observability";
import { storage } from "./storage";
const observabilityFeedbackSchema = z.object({
traceId: z.string().regex(/^[a-f0-9]{32}$/),
value: z.union([z.literal(1), z.literal(-1)]),
comment: z.string().trim().max(2_000).optional(),
}).strict();
const skillInputSchema: z.ZodType<AgentSkillInput> = z.object({
name: z.string().regex(/^[a-z0-9]+(?:-[a-z0-9]+)*$/).max(64),
description: z.string().min(1).max(1024),
instructions: z.string().min(1),
references: z.record(z.string(), z.string()),
enabled: z.boolean(),
userInvocable: z.boolean(),
});
const browserHostSchema = z.string().trim().min(1).max(255).regex(/^(?:\*\.)?[a-z0-9.-]+(?::\d{1,5})?$/i);
const browserSettingsSchema: z.ZodType<BrowserSettingsInput> = z.object({
enabled: z.boolean(),
scope: z.enum(["thread", "shared"]),
viewportWidth: z.number().int().min(320).max(3840),
viewportHeight: z.number().int().min(240).max(2160),
timeoutMs: z.number().int().min(1_000).max(120_000),
maxSessions: z.number().int().min(1).max(16),
idleTimeoutMs: z.number().int().min(60_000).max(86_400_000),
screencastEnabled: z.boolean(),
screenshotsEnabled: z.boolean(),
multiTabEnabled: z.boolean(),
formsEnabled: z.boolean(),
dialogsEnabled: z.boolean(),
dragEnabled: z.boolean(),
evaluateEnabled: z.boolean(),
recordingEnabled: z.boolean(),
recordingRetentionDays: z.number().int().min(1).max(90),
recordingMaxFiles: z.number().int().min(1).max(100),
allowHosts: z.array(browserHostSchema).max(100),
denyHosts: z.array(browserHostSchema).max(100),
}).strict();
const memorySettingsSchema: z.ZodType<MemorySettingsInput> = z.object({
autoCompact: z.boolean(),
observationTokens: z.number().int().min(4_000).max(200_000),
reflectionTokens: z.number().int().min(4_000).max(400_000),
recentMessagePercent: z.number().int().min(5).max(75),
bufferOnIdle: z.boolean(),
});
const agentSettingsInputSchema: z.ZodType<AgentSettingsInput> = z.object({
name: z.string().trim().min(1).max(120),
description: z.string().trim().min(1).max(1024),
instructions: z.string().min(1).max(MAX_AGENT_INSTRUCTIONS_CHARACTERS),
workspaceAccess: z.enum(["read-write", "read-only", "none"]),
browserAccess: z.enum(["interactive", "read-only", "none"]),
delegationEnabled: z.boolean(),
tools: z.array(z.enum(AGENT_TOOL_NAMES)).max(AGENT_TOOL_NAMES.length)
.refine((tools) => new Set(tools).size === tools.length, "Tools must be unique"),
}).strict();
const agentRuntimeSettingsSchema: z.ZodType<AgentRuntimeSettingsInput> = z.object({
defaultModel: z.string().trim().min(1).max(200),
supervisorMaxSteps: z.number().int().min(1).max(128),
specialistMaxSteps: z.number().int().min(1).max(128),
toolConcurrency: z.number().int().min(1).max(16),
delegationContextMessages: z.number().int().min(1).max(100),
delegationResultCharacters: z.number().int().min(1_000).max(100_000),
maxProcessorRetries: z.number().int().min(0).max(10),
finalResponseFeedback: z.string().trim().min(1).max(4_000),
delegationFailureFeedback: z.string().trim().min(1).max(4_000),
delegationResultTruncationMarker: z.string().min(1).max(500),
taskConcurrency: z.number().int().min(1).max(16),
taskPollIntervalMs: z.number().int().min(250).max(300_000),
taskTimeoutMs: z.number().int().min(1_000).max(86_400_000),
taskStaleAfterMs: z.number().int().min(60_000).max(604_800_000),
}).strict();
const memoryKeySchema = z.string().trim().min(1).max(128);
const memoryContentSchema = z.string().trim().min(1).max(MAX_FACT_CHARACTERS);
const memoryCreateSchema = z.object({
content: memoryContentSchema,
key: memoryKeySchema.optional(),
importance: z.number().min(0).max(1).default(0.8),
sessionId: z.string().trim().min(1).max(256).default("memory-manager"),
}).strict();
const memoryUpdateSchema = z.object({
content: memoryContentSchema,
key: memoryKeySchema.optional(),
importance: z.number().min(0).max(1),
}).strict();
const workspaceInputSchema = z.object({
name: z.string().trim().min(1).max(120),
repositoryPath: z.string().trim().min(1).max(512),
}).strict();
const sessionPostSchema = z.object({
model: z.string().min(1).max(200).optional(),
workspaceId: z.string().min(1).max(256).default(DEFAULT_WORKSPACE_ID),
}).strict();
const messageSchema = z.object({
id: z.string().optional(),
role: z.enum(["user", "assistant", "system"]),
parts: z.array(z.unknown()),
}).passthrough();
const sessionPutSchema = z.object({
model: z.string().min(1).max(200),
messages: z.array(messageSchema).max(2000),
revision: z.number().int().min(0),
});
const chatPostSchema = z.object({
id: z.string().min(1).max(256).optional(),
model: z.string().min(1).max(200).optional(),
workspaceId: z.string().min(1).max(256).optional(),
messages: z.array(messageSchema).min(1).max(2000),
});
const taskPostSchema = z.object({
prompt: z.string().trim().min(1).max(8000),
model: z.string().min(1).max(200).optional(),
sessionId: z.string().min(1).max(256).optional(),
workspaceId: z.string().min(1).max(256).default(DEFAULT_WORKSPACE_ID),
});
const scheduleSchema = z.object({
name: z.string().trim().min(1).max(120),
prompt: z.string().trim().min(1).max(8000),
model: z.string().min(1).max(200).optional(),
sessionId: z.string().min(1).max(256).optional(),
workspaceId: z.string().min(1).max(256).default(DEFAULT_WORKSPACE_ID),
cron: z.string().trim().min(1).max(120),
enabled: z.boolean().optional(),
});
const MAX_BODY_BYTES = 2 * 1024 * 1024;
async function readJson(req: Request): Promise<unknown> {
const text = await req.text();
if (text.length > MAX_BODY_BYTES) throw new RangeError("body too large");
return JSON.parse(text);
}
function guardApiRoutes<T extends Record<string, unknown>>(routes: T): T {
const guarded = { ...routes } as T;
for (const [path, route] of Object.entries(routes)) {
if (!path.startsWith("/api/")) continue;
if (typeof route === "function") {
(guarded as Record<string, unknown>)[path] = async (req: Request) => requireApiKey(req) ?? route(req);
} else if (route && typeof route === "object") {
const methods = { ...(route as Record<string, unknown>) };
for (const [method, handler] of Object.entries(methods)) {
if (typeof handler === "function") methods[method] = async (req: Request) => requireApiKey(req) ?? handler(req);
}
(guarded as Record<string, unknown>)[path] = methods;
}
}
return guarded;
}
export const sessions = new SessionStore();
const hooks = loadHookRuntime((run) => hookAudit.record(run));
const taskStore = new AgentTaskStore();
const agentExecution = new AgentExecutionRuntime(
agent,
hooks,
longTermMemory,
browserRuntime,
memorySettings,
agentRuntimeSettings,
);
const taskRuntime = new AgentTaskRuntime(
taskStore,
createAgentTaskExecutor(agent, agentExecution),
createTaskLifecycleHooks(hooks),
);
type AgentActivityMessage = UIMessage<unknown, { agentActivity: AgentActivity }>;
export function startServer(
port = 3000,
sessionStore = sessions,
hookRuntime: HookRuntime = hooks,
memoryStore: LongTermMemoryStore = longTermMemory,
agentStore: AgentSettingsStore = agentSettings,
browserStore: BrowserSettingsStore = browserSettings,
activeBrowser: BrowserRuntime = browserRuntime,
skillStore: AgentSkillStore = agentSkills,
tasks: AgentTaskStore = taskStore,
taskRunner: AgentTaskRuntime = taskRuntime,
memoryConfigStore: MemorySettingsStore = memorySettings,
workspaceStore: AgentWorkspaceStore = agentWorkspaces,
runtimeConfigStore: Pick<AgentRuntimeSettingsStore, "get" | "update"> = agentRuntimeSettings,
) {
const startedSessions = new Set<string>();
const executionRuntime = new AgentExecutionRuntime(
agent,
hookRuntime,
memoryStore,
activeBrowser,
memoryConfigStore,
runtimeConfigStore,
);
type ApiRequest = Request & { params: { id: string; skillId: string } };
type ApiHandler = (req: ApiRequest) => Response | Promise<Response>;
type RouteDef = typeof index | ApiHandler | Partial<Record<"GET" | "POST" | "PUT" | "PATCH" | "DELETE", ApiHandler>>;
const routes: Record<string, RouteDef> = {
"/": index,
...createChannelRoutes({
channelStore: channels,
workspaceStore,
runtimeSettings: runtimeConfigStore,
taskStore: tasks,
taskRunner,
}),
"/api/models": async () => {
const runtimeSettings = await runtimeConfigStore.get();
return Response.json({ ...(await listModelCatalog()), defaultModel: runtimeSettings.defaultModel });
},
"/api/workspaces": {
GET: async () => Response.json({ workspaces: await workspaceStore.list() }),
POST: async (req) => {
const input = workspaceInputSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid workspace", issues: input.error.issues }, { status: 400 });
}
try {
return Response.json(await workspaceStore.create(input.data), { status: 201 });
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
},
"/api/workspaces/:id": {
PATCH: async (req) => {
const input = workspaceInputSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid workspace", issues: input.error.issues }, { status: 400 });
}
try {
const workspace = await workspaceStore.update(req.params.id, input.data);
return workspace
? Response.json(workspace)
: Response.json({ error: "Workspace not found" }, { status: 404 });
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
DELETE: async (req) => {
if (req.params.id === DEFAULT_WORKSPACE_ID) {
return Response.json({ error: "The default workspace cannot be deleted" }, { status: 409 });
}
if (!(await workspaceStore.get(req.params.id))) {
return Response.json({ error: "Workspace not found" }, { status: 404 });
}
const [active, archived, workspaceTasks, schedules] = await Promise.all([
sessionStore.list(false, req.params.id),
sessionStore.list(true, req.params.id),
tasks.listTasks(req.params.id),
tasks.listSchedules(req.params.id),
]);
if (active.length || archived.length || workspaceTasks.length || schedules.length) {
return Response.json({ error: "Workspace still has sessions, tasks, or schedules" }, { status: 409 });
}
await workspaceStore.delete(req.params.id);
return new Response(null, { status: 204 });
},
},
"/api/agents": {
GET: async () => Response.json({ agents: await agentStore.list() }),
},
"/api/agents/:id": {
PATCH: async (req) => {
const input = agentSettingsInputSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid agent settings", issues: input.error.issues }, { status: 400 });
}
try {
const settings = await agentStore.update(req.params.id, input.data);
return settings
? Response.json(settings)
: Response.json({ error: "Agent not found" }, { status: 404 });
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
},
"/api/settings/browser": {
GET: async () => Response.json(await browserStore.get()),
PATCH: async (req) => {
const input = browserSettingsSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid browser settings", issues: input.error.issues }, { status: 400 });
}
const settings = await browserStore.update(input.data);
await activeBrowser.apply(agent, settings);
return Response.json(settings);
},
},
"/api/settings/memory": {
GET: async () => Response.json(await memoryConfigStore.get()),
PATCH: async (req) => {
const input = memorySettingsSchema.safeParse(await req.json().catch(() => ({})));
if (!input.success) {
return Response.json({ error: "Invalid memory settings", issues: input.error.issues }, { status: 400 });
}
return Response.json(await memoryConfigStore.update(input.data));
},
},
"/api/settings/agent-runtime": {
GET: async () => Response.json(await runtimeConfigStore.get()),
PATCH: async (req) => {
const input = agentRuntimeSettingsSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid agent runtime settings", issues: input.error.issues }, { status: 400 });
}
if (!(await isKnownModel(input.data.defaultModel))) {
return Response.json({ error: `Unknown model: ${input.data.defaultModel}` }, { status: 400 });
}
try {
const settings = await runtimeConfigStore.update(input.data);
taskRunner.configure(settings);
return Response.json(settings);
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
},
"/api/agents/:id/skills": {
GET: async (req) => Response.json({ skills: await skillStore.list(req.params.id) }),
POST: async (req) => {
const input = skillInputSchema.safeParse(await req.json().catch(() => ({})));
if (!input.success) return Response.json({ error: "Invalid skill", issues: input.error.issues }, { status: 400 });
return Response.json(await skillStore.create(req.params.id, input.data), { status: 201 });
},
},
"/api/agents/:id/skills/:skillId": {
PATCH: async (req) => {
const input = skillInputSchema.safeParse(await req.json().catch(() => ({})));
if (!input.success) return Response.json({ error: "Invalid skill", issues: input.error.issues }, { status: 400 });
const skill = await skillStore.update(req.params.id, req.params.skillId, input.data);
return skill ? Response.json(skill) : Response.json({ error: "Skill not found" }, { status: 404 });
},
DELETE: async (req) => await skillStore.delete(req.params.id, req.params.skillId)
? new Response(null, { status: 204 })
: Response.json({ error: "Skill not found" }, { status: 404 }),
},
"/api/sessions": {
GET: async (req) => {
const params = new URL(req.url).searchParams;
const archived = params.get("archived") === "true";
const workspaceId = params.get("workspaceId") || undefined;
return Response.json({ sessions: await sessionStore.list(archived, workspaceId) });
},
POST: async (req) => {
const input = sessionPostSchema.safeParse(await readJson(req).catch(() => ({})));
if (!input.success) {
return Response.json({ error: "Invalid session", issues: input.error.issues }, { status: 400 });
}
const model = input.data.model ?? (await runtimeConfigStore.get()).defaultModel;
if (!(await isKnownModel(model))) {
return Response.json({ error: `Unknown model: ${model}` }, { status: 400 });
}
if (!(await workspaceStore.get(input.data.workspaceId))) {
return Response.json({ error: "Workspace not found" }, { status: 404 });
}
const session = await sessionStore.create(model, input.data.workspaceId);
try {
await hookRuntime.dispatch(createHookEvent("SessionStart", {
sessionId: session.id,
model,
detail: {
source: "create",
session: { title: session.title, model, workspaceId: session.workspaceId },
},
}));
startedSessions.add(session.id);
} catch (error) {
appLogger().warn("hook.session_start.failed", { error: messageOf(error) });
}
return Response.json(session, { status: 201 });
},
},
"/api/sessions/:id": {
GET: async (req) => {
const session = await sessionStore.get(req.params.id);
return session
? Response.json(session)
: Response.json({ error: "Session not found" }, { status: 404 });
},
PUT: async (req) => {
const input = sessionPutSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid session", issues: input.error.issues }, { status: 400 });
}
if (!(await isKnownModel(input.data.model))) {
return Response.json({ error: `Unknown model: ${input.data.model}` }, { status: 400 });
}
const session = await sessionStore.save(req.params.id, {
model: input.data.model,
messages: input.data.messages as ChatMessage[],
expectedRevision: input.data.revision,
});
if (session === "conflict") {
const current = await sessionStore.get(req.params.id);
return Response.json({ error: "revision conflict", session: current }, { status: 409 });
}
return session
? Response.json(session)
: Response.json({ error: "Session not found" }, { status: 404 });
},
PATCH: async (req) => {
const body = (await req.json().catch(() => ({}))) as {
title?: string;
archived?: boolean;
};
if (typeof body.title === "string") {
const title = body.title.trim();
if (!title) return Response.json({ error: "title is required" }, { status: 400 });
const session = await sessionStore.rename(req.params.id, title);
return session
? Response.json(session)
: Response.json({ error: "Session not found" }, { status: 404 });
}
if (typeof body.archived === "boolean") {
const session = await sessionStore.setArchived(req.params.id, body.archived);
return session
? Response.json(session)
: Response.json({ error: "Session not found" }, { status: 404 });
}
return Response.json({ error: "title or archived is required" }, { status: 400 });
},
DELETE: async (req) => {
const session = await sessionStore.get(req.params.id);
if (!session) return Response.json({ error: "Session not found" }, { status: 404 });
try {
await hookRuntime.dispatch(createHookEvent("SessionEnd", {
sessionId: session.id,
model: session.model,
detail: { source: "delete" },
}));
} catch (error) {
appLogger().warn("hook.session_end.failed", { error: messageOf(error) });
}
await hookAudit.deleteForSession(session.id);
await sessionStore.delete(session.id);
startedSessions.delete(session.id);
return new Response(null, { status: 204 });
},
},
"/api/sessions/:id/hooks": {
GET: async (req) => Response.json({ runs: await hookAudit.listForSession(req.params.id) }),
},
"/api/memories": {
GET: async (req) => {
const params = new URL(req.url).searchParams;
const kind = params.get("kind");
if (kind !== null && kind !== "fact" && kind !== "episode") {
return Response.json({ error: "kind must be fact or episode" }, { status: 400 });
}
return Response.json({
memories: await memoryStore.list({
resourceId: RESOURCE_ID,
query: params.get("query") ?? undefined,
kind: kind ?? undefined,
}),
});
},
POST: async (req) => {
const input = memoryCreateSchema.safeParse(await req.json().catch(() => ({})));
if (!input.success) {
return Response.json({ error: "Invalid memory", issues: input.error.issues }, { status: 400 });
}
await memoryStore.rememberFact({
resourceId: RESOURCE_ID,
sessionId: input.data.sessionId,
key: input.data.key,
content: input.data.content,
importance: input.data.importance,
});
const memories = await memoryStore.list({
resourceId: RESOURCE_ID,
query: input.data.content,
kind: "fact",
limit: 1,
});
return Response.json(memories[0], { status: 201 });
},
},
"/api/memories/:id": {
PATCH: async (req) => {
const input = memoryUpdateSchema.safeParse(await req.json().catch(() => ({})));
if (!input.success) {
return Response.json({ error: "Invalid memory", issues: input.error.issues }, { status: 400 });
}
const memory = await memoryStore.updateFact({
id: req.params.id,
resourceId: RESOURCE_ID,
key: input.data.key,
content: input.data.content,
importance: input.data.importance,
});
return memory
? Response.json(memory)
: Response.json({ error: "Editable fact not found" }, { status: 404 });
},
DELETE: async (req) => {
const deleted = await memoryStore.delete(req.params.id, RESOURCE_ID);
return deleted
? new Response(null, { status: 204 })
: Response.json({ error: "Memory not found" }, { status: 404 });
},
},
"/api/tasks": {
GET: async (req) => {
const workspaceId = new URL(req.url).searchParams.get("workspaceId") || undefined;
return Response.json({ tasks: await tasks.listTasks(workspaceId) });
},
POST: async (req) => {
const input = taskPostSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) return Response.json({ error: "Invalid task", issues: input.error.issues }, { status: 400 });
const model = input.data.model ?? (await runtimeConfigStore.get()).defaultModel;
if (!(await isKnownModel(model))) return Response.json({ error: `Unknown model: ${model}` }, { status: 400 });
if (!(await workspaceStore.get(input.data.workspaceId))) return Response.json({ error: "Workspace not found" }, { status: 404 });
const task = await tasks.createTask({
sessionId: input.data.sessionId,
workspaceId: input.data.workspaceId,
prompt: input.data.prompt,
model,
});
taskRunner.enqueue(task);
return Response.json(task, { status: 201 });
},
},
"/api/tasks/:id/cancel": {
POST: async (req) => await taskRunner.cancel(req.params.id)
? new Response(null, { status: 204 })
: Response.json({ error: "Task cannot be cancelled" }, { status: 409 }),
},
"/api/schedules": {
GET: async (req) => {
const workspaceId = new URL(req.url).searchParams.get("workspaceId") || undefined;
return Response.json({ schedules: await tasks.listSchedules(workspaceId) });
},
POST: async (req) => {
const input = scheduleSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) return Response.json({ error: "Invalid schedule", issues: input.error.issues }, { status: 400 });
const model = input.data.model ?? (await runtimeConfigStore.get()).defaultModel;
if (!(await isKnownModel(model))) return Response.json({ error: `Unknown model: ${model}` }, { status: 400 });
if (!(await workspaceStore.get(input.data.workspaceId))) return Response.json({ error: "Workspace not found" }, { status: 404 });
try {
return Response.json(await tasks.createSchedule({
sessionId: input.data.sessionId,
workspaceId: input.data.workspaceId,
name: input.data.name,
prompt: input.data.prompt,
model,
cron: input.data.cron,
enabled: input.data.enabled ?? true,
}), { status: 201 });
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
},
"/api/schedules/:id": {
PATCH: async (req) => {
const input = scheduleSchema.extend({ enabled: z.boolean() }).safeParse(await readJson(req).catch(() => undefined));
if (!input.success) return Response.json({ error: "Invalid schedule", issues: input.error.issues }, { status: 400 });
const model = input.data.model ?? (await runtimeConfigStore.get()).defaultModel;
if (!(await isKnownModel(model))) return Response.json({ error: `Unknown model: ${model}` }, { status: 400 });
if (!(await workspaceStore.get(input.data.workspaceId))) return Response.json({ error: "Workspace not found" }, { status: 404 });
try {
const schedule = await tasks.updateSchedule(req.params.id, {
workspaceId: input.data.workspaceId,
name: input.data.name,
prompt: input.data.prompt,
model,
cron: input.data.cron,
enabled: input.data.enabled,
});
return schedule ? Response.json(schedule) : Response.json({ error: "Schedule not found" }, { status: 404 });
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
DELETE: async (req) => await tasks.deleteSchedule(req.params.id)
? new Response(null, { status: 204 })
: Response.json({ error: "Schedule not found" }, { status: 404 }),
},
"/api/observability/overview": async (req) => {
try {
return Response.json(await getObservabilityOverview(parseObservabilityQuery(req.url)));
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
"/api/observability/traces": async (req) => {
try {
return Response.json(await listObservabilityTraces(parseObservabilityQuery(req.url)));
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
"/api/observability/traces/:id": async (req) => {
if (!/^[a-f0-9]{32}$/.test(req.params.id)) {
return Response.json({ error: "Invalid trace ID" }, { status: 400 });
}
const trace = await getObservabilityTrace(req.params.id);
return trace ? Response.json(trace) : Response.json({ error: "Trace not found" }, { status: 404 });
},
"/api/observability/logs": async (req) => {
try {
return Response.json(await listObservabilityLogs(parseObservabilityQuery(req.url)));
} catch (error) {
return Response.json({ error: messageOf(error) }, { status: 400 });
}
},
"/api/observability/health": async () => {
const health = await observabilityHealth();
return Response.json(health, { status: health.status === "ok" ? 200 : 503 });
},
"/api/observability/feedback": {
POST: async (req) => {
const input = observabilityFeedbackSchema.safeParse(await readJson(req).catch(() => undefined));
if (!input.success) {
return Response.json({ error: "Invalid feedback", issues: input.error.issues }, { status: 400 });
}
if (!await getObservabilityTrace(input.data.traceId)) {
return Response.json({ error: "Trace not found" }, { status: 404 });
}
await observability.addFeedback({
traceId: input.data.traceId,
feedback: {
feedbackSource: "user",
feedbackType: "thumbs",
value: input.data.value,
comment: input.data.comment,
},
});
return Response.json({ recorded: true });
},
},
"/api/chat": {
POST: async (req) => {
try {
const parsed = chatPostSchema.safeParse(await readJson(req).catch(() => undefined));
if (!parsed.success) return Response.json({ error: "Invalid chat request", issues: parsed.error.issues }, { status: 400 });
const body = parsed.data;
const model = body.model ?? (await runtimeConfigStore.get()).defaultModel;
if (!(await isKnownModel(model))) return Response.json({ error: `Unknown model: ${model}` }, { status: 400 });
const sessionId = body.id ?? crypto.randomUUID();
const turnId = crypto.randomUUID();
const traceId = createTraceId();
const messages = body.messages as ChatMessage[];
const userText = latestUserText(messages);
if (!userText) return Response.json({ error: "A user message is required" }, { status: 400 });
const existing = body.id ? await sessionStore.get(body.id) : undefined;
const workspaceId = existing?.workspaceId ?? body.workspaceId ?? DEFAULT_WORKSPACE_ID;
if (!(await workspaceStore.get(workspaceId))) {
return Response.json({ error: "Workspace not found" }, { status: 404 });
}
if (!startedSessions.has(sessionId)) {
await hookRuntime.dispatch(createHookEvent("SessionStart", {
sessionId,
model,
detail: { source: existing ? "resume" : "create", workspaceId },
}));
startedSessions.add(sessionId);
}
let activityWriter: UIMessageStreamWriter<AgentActivityMessage> | undefined;
const pendingActivity: Array<Parameters<UIMessageStreamWriter<AgentActivityMessage>["write"]>[0]> = [];
const lifecycle = { sessionId, turnId, model };
const prepared = await executionRuntime.prepare({
...lifecycle,
traceId,
workspaceId,
executionSource: "chat",
prompt: userText,
onIterationComplete: async (iteration) => {
if (iteration.agentId === "popagent") return;
const chunk = {
type: "data-agentActivity" as const,
id: iteration.runId,
data: agentActivity(turnId, iteration),
};
if (activityWriter) activityWriter.write(chunk);
else pendingActivity.push(chunk);
},
});
if (prepared.promptWasReplaced) replaceLatestUserText(messages, prepared.prompt);
const agentStream = await handleChatStream({
mastra,
agentId: "popagent",
version: "v6",
params: {
id: sessionId,
messages: [messages.at(-1)!],
memory: { thread: sessionId, resource: RESOURCE_ID },
} as Parameters<typeof handleChatStream>[0]["params"],
defaultOptions: prepared.options as Parameters<typeof handleChatStream>[0]["defaultOptions"],
onError: (error) => {
void hookRuntime.dispatch(createHookEvent("StopFailure", {
...lifecycle,
detail: { error: { message: messageOf(error) } },
})).catch(() => undefined);
return "The agent could not complete this turn.";
},
messageMetadata: ({ part }) =>
part.type === "finish" && "totalUsage" in part
? { custom: { usage: part.totalUsage, turnId, traceId } }
: undefined,
});
const stream = createUIMessageStream<AgentActivityMessage>({
execute: ({ writer }) => {
activityWriter = writer;
for (const chunk of pendingActivity) writer.write(chunk);
pendingActivity.length = 0;
writer.merge(agentStream as unknown as Parameters<typeof writer.merge>[0]);
},
});
return createUIMessageStreamResponse({
stream: stream as unknown as Parameters<typeof createUIMessageStreamResponse>[0]["stream"],
});
} catch (error) {
if (error instanceof HookBlockedError) return Response.json({ error: error.reason }, { status: 403 });
appLogger().error("chat.turn.failed", { error });
return Response.json({ error: "Unable to start the agent turn" }, { status: 500 });
}
},
},
"/api/*": () => Response.json({ error: "Not found" }, { status: 404 }),
"/*": index,
};
return Bun.serve({
port,
routes: guardApiRoutes(routes) as Bun.Serve.Routes<undefined, string>,
});
}
const messageOf = (error: unknown) => error instanceof Error ? error.message : String(error);
if (import.meta.main) {
const server = startServer(Number(process.env.PORT ?? 3000));
await taskRuntime.start();
const maintenance = startMaintenance();
appLogger().info("server.started", { url: String(server.url) });
let stopping = false;
const shutdown = async (signal: string) => {
if (stopping) return;
stopping = true;
appLogger().info("server.stopping", { signal });
maintenance.stop();
server.stop(false);
await taskRuntime.stop();
await mastra.observability.flush();
await mastra.observability.shutdown();
await storage.close();
process.exit(0);
};
process.once("SIGINT", () => void shutdown("SIGINT"));
process.once("SIGTERM", () => void shutdown("SIGTERM"));
}