Menu
popagent
publicLatest change da13a7bebe63bf4b2693180d2d4850aabeaa0807 - Add autonomous evolution and self-healing by AkurAI Build
import type {
AgentTask,
Channel,
ChannelMessage,
ChannelSettings,
ChannelSettingsInput,
} from "./api-types";
import { storage } from "./storage";
import { retryableInit } from "./retryable-init";
const CHANNEL_COLUMNS = `id, name, created_at::text AS "createdAt", updated_at::text AS "updatedAt"`;
const SETTINGS_COLUMNS = `s.enabled, c.name AS "channelName", s.dispatch_mode AS "dispatchMode",
s.context_messages AS "contextMessages", s.streaming, s.tool_display AS "toolDisplay",
s.updated_at::text AS "updatedAt"`;
const MESSAGE_COLUMNS = `m.id, m.channel_id AS "channelId", m.workspace_id AS "workspaceId",
m.author_id AS "authorId", m.author_name AS "authorName", m.content,
m.task_id AS "taskId", m.created_at::text AS "createdAt",
t.session_id AS "taskSessionId", t.schedule_id AS "taskScheduleId",
t.source AS "taskSource", t.workspace_id AS "taskWorkspaceId",
t.prompt AS "taskPrompt", t.model AS "taskModel",
t.status AS "taskStatus", t.output AS "taskOutput", t.error AS "taskError",
t.steps_completed AS "taskStepsCompleted", t.progress AS "taskProgress",
t.recovery_count AS "taskRecoveryCount", t.attempt_count AS "taskAttemptCount",
t.max_attempts AS "taskMaxAttempts", t.next_attempt_at::text AS "taskNextAttemptAt",
t.last_error_class AS "taskLastErrorClass", t.dead_lettered_at::text AS "taskDeadLetteredAt",
t.created_at::text AS "taskCreatedAt", t.started_at::text AS "taskStartedAt",
t.completed_at::text AS "taskCompletedAt"`;
const TASK_RETURNING = `id, session_id AS "sessionId", schedule_id AS "scheduleId",
source, workspace_id AS "workspaceId", prompt, model, status, output, error,
steps_completed AS "stepsCompleted", progress, recovery_count AS "recoveryCount",
attempt_count AS "attemptCount", max_attempts AS "maxAttempts",
next_attempt_at::text AS "nextAttemptAt", last_error_class AS "lastErrorClass",
dead_lettered_at::text AS "deadLetteredAt",
created_at::text AS "createdAt", started_at::text AS "startedAt",
completed_at::text AS "completedAt"`;
type MessageRow = Omit<ChannelMessage, "task"> & {
taskSessionId: string | null;
taskScheduleId: string | null;
taskSource: AgentTask["source"] | null;
taskWorkspaceId: string | null;
taskPrompt: string | null;
taskModel: string | null;
taskStatus: AgentTask["status"] | null;
taskOutput: string | null;
taskError: string | null;
taskStepsCompleted: number | null;
taskProgress: string | null;
taskRecoveryCount: number | null;
taskAttemptCount: number | null;
taskMaxAttempts: number | null;
taskNextAttemptAt: string | null;
taskLastErrorClass: AgentTask["lastErrorClass"];
taskDeadLetteredAt: string | null;
taskCreatedAt: string | null;
taskStartedAt: string | null;
taskCompletedAt: string | null;
};
function mapMessage(row: MessageRow): ChannelMessage {
const task = row.taskId && row.taskStatus && row.taskWorkspaceId && row.taskPrompt
&& row.taskModel && row.taskCreatedAt
? {
id: row.taskId,
sessionId: row.taskSessionId,
scheduleId: row.taskScheduleId,
source: row.taskSource ?? "user",
workspaceId: row.taskWorkspaceId,
prompt: row.taskPrompt,
model: row.taskModel,
status: row.taskStatus,
output: row.taskOutput,
error: row.taskError,
stepsCompleted: row.taskStepsCompleted ?? 0,
progress: row.taskProgress,
recoveryCount: row.taskRecoveryCount ?? 0,
attemptCount: row.taskAttemptCount ?? 0,
maxAttempts: row.taskMaxAttempts ?? 3,
nextAttemptAt: row.taskNextAttemptAt,
lastErrorClass: row.taskLastErrorClass,
deadLetteredAt: row.taskDeadLetteredAt,
createdAt: row.taskCreatedAt,
startedAt: row.taskStartedAt,
completedAt: row.taskCompletedAt,
} satisfies AgentTask
: null;
return {
id: row.id,
channelId: row.channelId,
workspaceId: row.workspaceId,
authorId: row.authorId,
authorName: row.authorName,
content: row.content,
taskId: row.taskId,
task,
createdAt: row.createdAt,
};
}
export type ChannelPostInput = {
channelId: string;
workspaceId: string;
content: string;
authorId: string;
authorName: string;
};
export class ChannelStore {
readonly storage = storage;
private readonly initializeOnce = retryableInit(() => this.initialize());
init(): Promise<void> {
return this.initializeOnce();
}
private async initialize() {
await this.storage.init();
const schema = await Bun.file(new URL("./channel-settings.sql", import.meta.url)).text();
await this.storage.db.none(schema);
}
async getSettings(): Promise<ChannelSettings> {
await this.init();
return this.storage.db.one<ChannelSettings>(`
SELECT ${SETTINGS_COLUMNS}
FROM popagent_channel_settings s
CROSS JOIN popagent_channels c
WHERE s.singleton=TRUE AND c.id='general'
`);
}
async updateSettings(input: ChannelSettingsInput): Promise<ChannelSettings> {
await this.init();
return this.storage.db.tx(async (db) => {
await db.none(`
UPDATE popagent_channel_settings
SET enabled=$1, dispatch_mode=$2, context_messages=$3, streaming=$4,
tool_display=$5, updated_at=NOW()
WHERE singleton=TRUE
`, [input.enabled, input.dispatchMode, input.contextMessages, input.streaming, input.toolDisplay]);
await db.none(
"UPDATE popagent_channels SET name=$2,updated_at=NOW() WHERE id=$1",
["general", input.channelName],
);
return db.one<ChannelSettings>(`
SELECT ${SETTINGS_COLUMNS}
FROM popagent_channel_settings s CROSS JOIN popagent_channels c
WHERE s.singleton=TRUE AND c.id='general'
`);
});
}
async listChannels(): Promise<Channel[]> {
await this.init();
return this.storage.db.any<Channel>(`SELECT ${CHANNEL_COLUMNS} FROM popagent_channels ORDER BY created_at, id`);
}
async listMessages(channelId: string, options: { limit?: number; before?: string } = {}): Promise<ChannelMessage[]> {
const limit = options.limit ?? 100;
if (!Number.isInteger(limit) || limit < 1 || limit > 200) {
throw new RangeError("Channel message limit must be between 1 and 200");
}
if (options.before && options.before.length > 256) throw new RangeError("Invalid channel cursor");
await this.init();
const cursor = options.before
? await this.storage.db.oneOrNone<{ createdAt: string; id: string }>(`
SELECT created_at::text AS "createdAt", id FROM popagent_channel_messages
WHERE channel_id=$1 AND id=$2
`, [channelId, options.before])
: undefined;
if (options.before && !cursor) throw new RangeError("Invalid channel cursor");
const rows = await this.storage.db.any<MessageRow>(`
SELECT ${MESSAGE_COLUMNS}
FROM popagent_channel_messages m
LEFT JOIN popagent_agent_tasks t ON t.id=m.task_id
WHERE m.channel_id=$1
AND ($3::timestamptz IS NULL OR (m.created_at,m.id) < ($3::timestamptz,$4))
ORDER BY m.created_at DESC, m.id DESC
LIMIT $2
`, [channelId, limit, cursor?.createdAt ?? null, cursor?.id ?? null]);
return rows.reverse().map(mapMessage);
}
async hasWorkspaceMessages(workspaceId: string): Promise<boolean> {
await this.init();
const row = await this.storage.db.one<{ exists: boolean }>(
"SELECT EXISTS(SELECT 1 FROM popagent_channel_messages WHERE workspace_id=$1) AS exists",
[workspaceId],
);
return row.exists;
}
async findMessageByTask(channelId: string, taskId: string): Promise<ChannelMessage | undefined> {
await this.init();
const row = await this.storage.db.oneOrNone<MessageRow>(`
SELECT ${MESSAGE_COLUMNS}
FROM popagent_channel_messages m
LEFT JOIN popagent_agent_tasks t ON t.id=m.task_id
WHERE m.channel_id=$1 AND m.task_id=$2
`, [channelId, taskId]);
return row ? mapMessage(row) : undefined;
}
async createMessage(input: ChannelPostInput): Promise<ChannelMessage> {
await this.init();
const channel = await this.storage.db.oneOrNone("SELECT id FROM popagent_channels WHERE id=$1", [input.channelId]);
if (!channel) throw new Error("Channel not found");
const row = await this.storage.db.one<Omit<ChannelMessage, "task">>(`
INSERT INTO popagent_channel_messages
(id,channel_id,workspace_id,author_id,author_name,content)
VALUES ($1,$2,$3,$4,$5,$6)
RETURNING id, channel_id AS "channelId", workspace_id AS "workspaceId",
author_id AS "authorId", author_name AS "authorName", content,
task_id AS "taskId", created_at::text AS "createdAt"
`, [crypto.randomUUID(), input.channelId, input.workspaceId, input.authorId, input.authorName, input.content]);
return { ...row, task: null };
}
async createDispatch(input: ChannelPostInput & { model: string; taskPrompt?: string }): Promise<{ message: ChannelMessage; task: AgentTask }> {
await this.init();
return this.storage.db.tx(async (db) => {
const channel = await db.oneOrNone("SELECT id FROM popagent_channels WHERE id=$1 FOR SHARE", [input.channelId]);
if (!channel) throw new Error("Channel not found");
const task = await db.one<AgentTask>(`
INSERT INTO popagent_agent_tasks
(id,session_id,schedule_id,workspace_id,prompt,model,status)
VALUES ($1,NULL,NULL,$2,$3,$4,'queued')
RETURNING ${TASK_RETURNING}
`, [crypto.randomUUID(), input.workspaceId, input.taskPrompt ?? input.content, input.model]);
const row = await db.one<Omit<ChannelMessage, "task">>(`
INSERT INTO popagent_channel_messages
(id,channel_id,workspace_id,author_id,author_name,content,task_id)
VALUES ($1,$2,$3,$4,$5,$6,$7)
RETURNING id, channel_id AS "channelId", workspace_id AS "workspaceId",
author_id AS "authorId", author_name AS "authorName", content,
task_id AS "taskId", created_at::text AS "createdAt"
`, [crypto.randomUUID(), input.channelId, input.workspaceId, input.authorId, input.authorName, input.content, task.id]);
return { message: { ...row, task }, task };
});
}
}
export const channels = new ChannelStore();