Menu
popagent
publicLatest change 7f6c1d0ed24ffc264e50e9966eaf45024245ba71 - feat: add per-task agent communication by AkurAI Build
import type {
AgentTask,
Channel,
ChannelMessage,
ChannelSettings,
ChannelSettingsInput,
} from "./api-types";
import { agentSettings } from "./agent-settings";
import type { TaskAgentCommunicationMessage } from "./task-agent-communication";
import { RESOURCE_ID } from "./sessions";
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.communication_task_id AS "communicationTaskId",
m.communication_to_agent_id AS "communicationToAgentId",
m.communication_to_agent_name AS "communicationToAgentName",
m.reply_to_message_id AS "replyToMessageId", 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.heartbeat_at::text AS "taskHeartbeatAt", 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", heartbeat_at::text AS "heartbeatAt",
created_at::text AS "createdAt", started_at::text AS "startedAt",
completed_at::text AS "completedAt"`;
type MessageRow = Omit<ChannelMessage, "task" | "agentCommunication"> & {
communicationTaskId: string | null;
communicationToAgentId: string | null;
communicationToAgentName: string | null;
replyToMessageId: string | null;
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;
taskHeartbeatAt: 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,
heartbeatAt: row.taskHeartbeatAt,
createdAt: row.taskCreatedAt,
startedAt: row.taskStartedAt,
completedAt: row.taskCompletedAt,
} satisfies AgentTask
: null;
const agentCommunication = row.communicationTaskId && row.communicationToAgentId
&& row.communicationToAgentName
? {
taskId: row.communicationTaskId,
fromAgentId: row.authorId,
toAgentId: row.communicationToAgentId,
toAgentName: row.communicationToAgentName,
replyToMessageId: row.replyToMessageId,
}
: null;
return {
id: row.id,
channelId: row.channelId,
workspaceId: row.workspaceId,
authorId: row.authorId,
authorName: row.authorName,
content: row.content,
taskId: row.taskId,
task,
agentCommunication,
createdAt: row.createdAt,
};
}
export type ChannelPostInput = {
channelId: string;
workspaceId: string;
content: string;
authorId: string;
authorName: string;
};
export function taskUsesChannelTranscript(task: Pick<AgentTask, "source">): boolean {
return task.source === "user";
}
export class ChannelStore {
readonly storage = storage;
private readonly initializeOnce = retryableInit(() => this.initialize());
init(): Promise<void> {
return this.initializeOnce();
}
private async initialize() {
await Promise.all([this.storage.init(), agentSettings.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" | "agentCommunication">>(`
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, agentCommunication: null };
}
async recordAgentCommunication(input: TaskAgentCommunicationMessage): Promise<ChannelMessage> {
if (input.resourceId !== RESOURCE_ID) throw new Error("Task communication resource boundary mismatch");
await this.init();
return this.storage.db.tx(async (db) => {
const task = await db.oneOrNone<{ workspaceId: string; status: AgentTask["status"] }>(`
SELECT workspace_id AS "workspaceId", status
FROM popagent_agent_tasks
WHERE id=$1
FOR SHARE
`, [input.taskId]);
if (!task || task.workspaceId !== input.workspaceId) {
throw new Error("Task communication workspace boundary mismatch");
}
if (task.status !== "running") throw new Error("Task communication task is terminal");
const identities = await db.any<{ id: string; name: string }>(`
SELECT id,name FROM popagent_agents WHERE id IN ($1,$2)
`, [input.from, input.to]);
const sender = identities.find((identity) => identity.id === input.from);
const recipient = identities.find((identity) => identity.id === input.to);
if (!sender || !recipient) throw new Error("Task communication agent identity is unavailable");
if (input.replyTo) {
const replied = await db.oneOrNone<{
taskId: string;
fromAgentId: string;
toAgentId: string;
}>(`
SELECT communication_task_id AS "taskId", author_id AS "fromAgentId",
communication_to_agent_id AS "toAgentId"
FROM popagent_channel_messages
WHERE id=$1
`, [input.replyTo]);
if (
!replied
|| replied.taskId !== input.taskId
|| replied.fromAgentId !== input.to
|| replied.toAgentId !== input.from
) {
throw new Error("Task communication reply boundary mismatch");
}
}
const row = await db.one<Omit<ChannelMessage, "task" | "agentCommunication">>(`
INSERT INTO popagent_channel_messages
(id,channel_id,workspace_id,author_id,author_name,content,
communication_task_id,communication_to_agent_id,
communication_to_agent_name,reply_to_message_id,created_at)
VALUES ($1,'general',$2,$3,$4,$5,$6,$7,$8,$9,$10)
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"
`, [
input.id,
input.workspaceId,
input.from,
sender.name,
input.body,
input.taskId,
input.to,
recipient.name,
input.replyTo ?? null,
input.createdAt,
]);
return {
...row,
task: null,
agentCommunication: {
taskId: input.taskId,
fromAgentId: input.from,
toAgentId: input.to,
toAgentName: recipient.name,
replyToMessageId: input.replyTo ?? null,
},
};
});
}
async announceTaskStart(task: AgentTask): Promise<ChannelMessage | undefined> {
await this.init();
const content = task.source === "build-maintenance"
? "Started reviewed Build maintenance."
: task.source === "self-update"
? "Started autonomous source maintenance."
: `Started task: ${task.prompt.slice(0, 8_000 - "Started task: ".length)}`;
const row = await this.storage.db.oneOrNone<Omit<ChannelMessage, "task" | "agentCommunication">>(`
INSERT INTO popagent_channel_messages
(id,channel_id,workspace_id,author_id,author_name,content,task_id)
VALUES ($1,'general',$2,'orchistrator','Orchistrator',$3,$4)
ON CONFLICT (task_id) DO NOTHING
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(), task.workspaceId, content, task.id]);
return row ? { ...row, task, agentCommunication: null } : undefined;
}
async announceTaskUpdate(task: AgentTask, authorName: string, content: string): Promise<ChannelMessage> {
return this.createMessage({
channelId: "general",
workspaceId: task.workspaceId,
authorId: `task:${task.id}`,
authorName,
content: content.slice(0, 8_000),
});
}
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" | "agentCommunication">>(`
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, agentCommunication: null }, task };
});
}
}
export const channels = new ChannelStore();