Menu
popagent
publicLatest change b61deab8cc219052959fa9b2e7525335a79519f8 - Isolate tests and remove terminal tasks by AkurAI Build
import { Cron } from "croner";
import { agentRuntimeSettings, type AgentRuntimeSettingsStore } from "./agent-runtime-settings";
import {
DEFAULT_WORKSPACE_ID,
type AgentRuntimeSettings,
type AgentSchedule,
type AgentTask,
type AgentTaskErrorClass,
type AgentTaskProgress,
} from "./api-types";
import { storage } from "./storage";
import { appLogger } from "./observability";
import { retryableInit } from "./retryable-init";
const HOST_TIMEZONE = Intl.DateTimeFormat().resolvedOptions().timeZone;
const TASK_COLUMNS = `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"`;
const SCHEDULE_COLUMNS = `id, session_id AS "sessionId", workspace_id AS "workspaceId",
source, name, prompt, model, max_attempts AS "maxAttempts", cron, timezone, enabled,
next_run_at::text AS "nextRunAt", last_run_at::text AS "lastRunAt",
created_at::text AS "createdAt", updated_at::text AS "updatedAt"`;
const RETRY_BASE_DELAY_MS = 1_000;
const RETRY_MAX_DELAY_MS = 5 * 60_000;
const TRANSIENT_ERROR_CLASSES: Partial<Record<AgentTaskErrorClass, true>> = {
timeout: true,
provider: true,
"rate-limit": true,
browser: true,
database: true,
infrastructure: true,
};
const TRANSIENT_DATABASE_CODES: Record<string, true> = {
"40001": true,
"40P01": true,
"53300": true,
"57P01": true,
"57P02": true,
"57P03": true,
};
const TRANSIENT_INFRASTRUCTURE_CODES: Record<string, true> = {
ECONNABORTED: true,
ECONNREFUSED: true,
ECONNRESET: true,
EHOSTUNREACH: true,
EPIPE: true,
ETIMEDOUT: true,
ENETDOWN: true,
ENETRESET: true,
ENETUNREACH: true,
EAI_AGAIN: true,
UND_ERR_CONNECT_TIMEOUT: true,
UND_ERR_HEADERS_TIMEOUT: true,
UND_ERR_SOCKET: true,
};
function errorChain(error: unknown): Array<Record<string, unknown>> {
const chain: Array<Record<string, unknown>> = [];
let current = error;
while (current && typeof current === "object" && chain.length < 4) {
const record = current as Record<string, unknown>;
chain.push(record);
current = record.cause;
}
return chain;
}
function classifyFailure(error: unknown): AgentTaskErrorClass {
const chain = errorChain(error);
const names = chain.map((item) => String(item.name ?? "")).join(" ").toLowerCase();
const messages = chain.map((item) => String(item.message ?? "")).join(" ").toLowerCase();
const codes = chain.map((item) => String(item.code ?? "").toUpperCase());
const statuses = chain
.map((item) => Number(item.statusCode ?? item.status))
.filter(Number.isFinite);
if (names.includes("hookblockederror")) return "hook-denial";
if (names.includes("aborterror") || /\bcancell?ed\b|\bcancellation\b/.test(messages)) return "cancelled";
if (
names.includes("zoderror")
|| names.includes("validationerror")
|| names.includes("rangeerror")
|| names.includes("syntaxerror")
) return "validation";
if (
statuses.some((status) => status === 401 || status === 403)
|| /\bpermission denied\b|\baccess denied\b|\bforbidden\b|\bunauthori[sz]ed\b|\bnot authori[sz]ed\b|\bcontainment\b|\bread-only\b/.test(messages)
) return "permission";
if (
/\bnot configured\b|\bconfiguration (?:is )?(?:missing|required)\b|\bmissing configuration\b|\bnot enabled\b|\brequires .* configuration\b|\b(?:workspace|memory|secret|agent) context (?:is )?(?:missing|required|unavailable)\b/.test(messages)
) return "configuration";
if (
statuses.includes(429)
|| codes.some((code) => code === "RATE_LIMIT" || code === "RATE_LIMITED")
|| /\brate[- ]?limit(?:ed|ing)?\b|\btoo many requests\b|\bquota temporarily exhausted\b/.test(messages)
) return "rate-limit";
if (
codes.some((code) => code.startsWith("08") || TRANSIENT_DATABASE_CODES[code])
) return "database";
if (codes.some((code) => TRANSIENT_INFRASTRUCTURE_CODES[code])) return "infrastructure";
if (
chain.some((item) => item.isRetryable === true)
|| (
/\bapi(?:call)?error\b|\bprovidererror\b|\bretryerror\b/.test(names)
&& statuses.some((status) => [408, 425, 500, 502, 503, 504].includes(status))
)
|| /(?:provider|model|llm|gateway).*(?:temporar|unavailable|overload|capacity|exhaust|upstream|connection)|(?:temporar|unavailable|overload|capacity|exhaust|upstream).*(?:provider|model|llm|gateway)/.test(messages)
) return "provider";
if (
/(?:browser|chromium|playwright|page|context).*(?:disconnect|closed|crash|temporar|unavailable)|(?:disconnect|crash).*(?:browser|chromium|playwright)/.test(messages)
) return "browser";
if (
/(?:database|postgres|sql).*(?:temporar|unavailable|connection|deadlock|serializ)|(?:deadlock|serializ).*(?:database|postgres|sql)/.test(messages)
) return "database";
if (
names.includes("timeouterror")
|| /\b(?:request|operation|background task) (?:timed out|timeout)\b|\btimeout (?:after|while|waiting)\b|\btimed out (?:after|while|waiting)\b/.test(messages)
) return "timeout";
if (
statuses.some((status) => [408, 425, 502, 503, 504].includes(status))
|| /\bservice unavailable\b|\bconnection (?:reset|refused|closed)\b|\bsocket hang up\b|\bnetwork unreachable\b|\btemporary failure\b|\bbad gateway\b|\bgateway timeout\b|\bupstream connect\b/.test(messages)
) return "infrastructure";
if (/\binvalid\b|\bmalformed\b|\bvalidation\b/.test(messages)) return "validation";
return "unknown";
}
function retryAt(attemptNumber: number): string {
const exponential = RETRY_BASE_DELAY_MS * 2 ** Math.min(Math.max(0, attemptNumber - 1), 20);
const jittered = exponential * (0.75 + Math.random() * 0.5);
return new Date(Date.now() + Math.min(RETRY_MAX_DELAY_MS, Math.round(jittered))).toISOString();
}
export class AgentTaskStore {
readonly storage = storage;
private readonly initializeOnce = retryableInit(() => this.initialize());
init(): Promise<void> {
return this.initializeOnce();
}
private async initialize() {
await this.storage.init();
await this.storage.db.none(`
CREATE TABLE IF NOT EXISTS popagent_agent_schedules (
id TEXT PRIMARY KEY,
session_id TEXT,
workspace_id TEXT NOT NULL DEFAULT '${DEFAULT_WORKSPACE_ID}',
source TEXT NOT NULL DEFAULT 'user' CHECK (source IN ('user','self-update')),
name TEXT NOT NULL,
prompt TEXT NOT NULL,
model TEXT NOT NULL,
max_attempts INTEGER NOT NULL DEFAULT 3 CHECK (max_attempts >= 1),
cron TEXT NOT NULL,
timezone TEXT NOT NULL DEFAULT 'UTC',
enabled BOOLEAN NOT NULL DEFAULT TRUE,
next_run_at TIMESTAMPTZ,
last_run_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS popagent_agent_tasks (
id TEXT PRIMARY KEY,
session_id TEXT,
workspace_id TEXT NOT NULL DEFAULT '${DEFAULT_WORKSPACE_ID}',
schedule_id TEXT REFERENCES popagent_agent_schedules(id) ON DELETE SET NULL,
source TEXT NOT NULL DEFAULT 'user' CHECK (source IN ('user','self-update')),
prompt TEXT NOT NULL,
model TEXT NOT NULL,
status TEXT NOT NULL,
output TEXT,
error TEXT,
steps_completed INTEGER NOT NULL DEFAULT 0,
progress TEXT,
recovery_count INTEGER NOT NULL DEFAULT 0,
attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
max_attempts INTEGER NOT NULL DEFAULT 3 CHECK (max_attempts >= 1),
next_attempt_at TIMESTAMPTZ,
last_error_class TEXT,
dead_lettered_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ
);
CREATE TABLE IF NOT EXISTS popagent_data_migrations (
id TEXT PRIMARY KEY,
applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
`);
await this.storage.db.tx(async (db) => {
const migrationId = "2026-08-14-agent-task-retry-v1";
await db.one("SELECT pg_advisory_xact_lock(hashtext($1))", [migrationId]);
if (await db.oneOrNone("SELECT id FROM popagent_data_migrations WHERE id=$1", [migrationId])) return;
await db.none(`
ALTER TABLE popagent_agent_schedules
ADD COLUMN IF NOT EXISTS workspace_id TEXT NOT NULL DEFAULT '${DEFAULT_WORKSPACE_ID}';
ALTER TABLE popagent_agent_schedules
ADD COLUMN IF NOT EXISTS max_attempts INTEGER NOT NULL DEFAULT 3 CHECK (max_attempts >= 1);
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS workspace_id TEXT NOT NULL DEFAULT '${DEFAULT_WORKSPACE_ID}';
CREATE INDEX IF NOT EXISTS popagent_agent_tasks_workspace_created_idx
ON popagent_agent_tasks (workspace_id, created_at DESC);
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS steps_completed INTEGER NOT NULL DEFAULT 0;
ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS progress TEXT;
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS recovery_count INTEGER NOT NULL DEFAULT 0;
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0);
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS max_attempts INTEGER NOT NULL DEFAULT 3 CHECK (max_attempts >= 1);
ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS next_attempt_at TIMESTAMPTZ;
ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS last_error_class TEXT;
ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS dead_lettered_at TIMESTAMPTZ;
CREATE INDEX IF NOT EXISTS popagent_agent_tasks_queued_retry_idx
ON popagent_agent_tasks (next_attempt_at, created_at)
WHERE status='queued';
ALTER TABLE popagent_agent_tasks DROP COLUMN IF EXISTS agent_id;
ALTER TABLE popagent_agent_schedules DROP COLUMN IF EXISTS agent_id;
`);
const schedules = await db.any<{ id: string; cron: string; enabled: boolean }>(
"SELECT id, cron, enabled FROM popagent_agent_schedules",
);
for (const schedule of schedules) {
const next = schedule.enabled ? new Cron(schedule.cron, { timezone: HOST_TIMEZONE }).nextRun() : null;
await db.none(
"UPDATE popagent_agent_schedules SET timezone=$2,next_run_at=$3 WHERE id=$1",
[schedule.id, HOST_TIMEZONE, next],
);
}
await db.none("INSERT INTO popagent_data_migrations (id) VALUES ($1)", [migrationId]);
});
await this.storage.db.tx(async (db) => {
const migrationId = "2026-08-14-task-source-v1";
await db.one("SELECT pg_advisory_xact_lock(hashtext($1))", [migrationId]);
if (await db.oneOrNone("SELECT id FROM popagent_data_migrations WHERE id=$1", [migrationId])) return;
await db.none(`
ALTER TABLE popagent_agent_schedules
ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'user'
CHECK (source IN ('user','self-update'));
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'user'
CHECK (source IN ('user','self-update'));
UPDATE popagent_agent_schedules SET source='user' WHERE source IS NULL;
UPDATE popagent_agent_tasks SET source='user' WHERE source IS NULL;
ALTER TABLE popagent_agent_schedules ALTER COLUMN source SET DEFAULT 'user';
ALTER TABLE popagent_agent_schedules ALTER COLUMN source SET NOT NULL;
ALTER TABLE popagent_agent_tasks ALTER COLUMN source SET DEFAULT 'user';
ALTER TABLE popagent_agent_tasks ALTER COLUMN source SET NOT NULL;
DO $migration$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conrelid='popagent_agent_schedules'::regclass
AND conname='popagent_agent_schedules_source_check'
) THEN
ALTER TABLE popagent_agent_schedules
ADD CONSTRAINT popagent_agent_schedules_source_check
CHECK (source IN ('user','self-update'));
END IF;
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conrelid='popagent_agent_tasks'::regclass
AND conname='popagent_agent_tasks_source_check'
) THEN
ALTER TABLE popagent_agent_tasks
ADD CONSTRAINT popagent_agent_tasks_source_check
CHECK (source IN ('user','self-update'));
END IF;
END
$migration$;
CREATE UNIQUE INDEX IF NOT EXISTS popagent_agent_schedules_self_update_workspace_idx
ON popagent_agent_schedules (workspace_id)
WHERE source='self-update';
`);
await db.none("INSERT INTO popagent_data_migrations (id) VALUES ($1)", [migrationId]);
});
}
async listTasks(workspaceId?: string): Promise<AgentTask[]> {
await this.init();
return this.storage.db.any<AgentTask>(
`SELECT ${TASK_COLUMNS} FROM popagent_agent_tasks
WHERE ($1::text IS NULL OR workspace_id=$1)
ORDER BY created_at DESC LIMIT 100`,
[workspaceId ?? null],
);
}
async deleteTerminalTask(id: string): Promise<"deleted" | "active" | "missing"> {
await this.init();
return this.storage.db.tx(async (db) => {
const deleted = await db.oneOrNone<{ id: string }>(
`DELETE FROM popagent_agent_tasks
WHERE id=$1 AND status IN ('completed','failed','cancelled','dead-letter')
RETURNING id`,
[id],
);
if (deleted) return "deleted";
return await db.oneOrNone("SELECT id FROM popagent_agent_tasks WHERE id=$1", [id])
? "active"
: "missing";
});
}
async listQueuedTasks(): Promise<AgentTask[]> {
await this.init();
return this.storage.db.any<AgentTask>(
`SELECT ${TASK_COLUMNS} FROM popagent_agent_tasks
WHERE status='queued'
AND attempt_count < max_attempts
AND (next_attempt_at IS NULL OR next_attempt_at <= NOW())
ORDER BY COALESCE(next_attempt_at, created_at), created_at`,
);
}
async recoverInterruptedTasks(staleAfterMs: number): Promise<number> {
if (!Number.isFinite(staleAfterMs) || staleAfterMs <= 0) {
throw new RangeError("Task stale threshold must be positive");
}
await this.init();
return this.storage.db.tx(async (db) => {
await db.none(
`UPDATE popagent_agent_tasks
SET status='failed',error='stale after restart',last_error_class='restart-stale',
progress='Failed without retry (stale after restart)',next_attempt_at=NULL,
completed_at=NOW(),dead_lettered_at=NULL
WHERE status='running'
AND started_at < NOW() - ($1::double precision * INTERVAL '1 millisecond')`,
[staleAfterMs],
);
await db.none(
`UPDATE popagent_agent_tasks
SET status='dead-letter',
error=COALESCE(error,'retry limit exhausted before restart'),
last_error_class=COALESCE(last_error_class,'infrastructure'),
progress=format(
'Dead-lettered after %s of %s attempts (%s)',
attempt_count,
max_attempts,
COALESCE(last_error_class,'infrastructure')
),
next_attempt_at=NULL,completed_at=NOW(),dead_lettered_at=NOW()
WHERE status='queued' AND attempt_count >= max_attempts`,
);
await db.none(
`UPDATE popagent_agent_tasks
SET status='dead-letter',error='interrupted by restart after final attempt',
last_error_class='infrastructure',
progress=format(
'Dead-lettered after %s of %s attempts (infrastructure)',
attempt_count,
max_attempts
),
next_attempt_at=NULL,completed_at=NOW(),dead_lettered_at=NOW()
WHERE status='running' AND attempt_count >= max_attempts`,
);
const recovered = await db.any<{ id: string }>(
`UPDATE popagent_agent_tasks
SET status='queued',started_at=NULL,steps_completed=0,
error='interrupted by restart',last_error_class='infrastructure',
next_attempt_at=NOW(),completed_at=NULL,dead_lettered_at=NULL,
progress=format(
'Retry scheduled after infrastructure interruption (attempt %s of %s)',
attempt_count,
max_attempts
),
recovery_count=recovery_count+1
WHERE status='running' AND attempt_count < max_attempts
RETURNING id`,
);
await db.none(
`UPDATE popagent_agent_tasks
SET status='cancelled',progress='Cancelled',last_error_class='cancelled',
next_attempt_at=NULL,completed_at=NOW()
WHERE status='cancelling'`,
);
return recovered.length;
});
}
async createTask(input: {
sessionId?: string;
workspaceId?: string;
scheduleId?: string;
prompt: string;
model: string;
maxAttempts?: number;
}): Promise<AgentTask> {
const maxAttempts = input.maxAttempts ?? 3;
if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
throw new RangeError("Task max attempts must be a positive integer");
}
await this.init();
return this.storage.db.one<AgentTask>(
`INSERT INTO popagent_agent_tasks
(id, session_id, schedule_id, workspace_id, source, prompt, model, status, max_attempts)
SELECT $1,$2,$3,$4,'user',$5,$6,'queued',$7
WHERE $3::text IS NULL OR EXISTS (
SELECT 1 FROM popagent_agent_schedules WHERE id=$3 AND source='user'
)
RETURNING ${TASK_COLUMNS}`,
[
crypto.randomUUID(),
input.sessionId ?? null,
input.scheduleId ?? null,
input.workspaceId ?? DEFAULT_WORKSPACE_ID,
input.prompt,
input.model,
maxAttempts,
],
);
}
async claim(id: string): Promise<boolean> {
await this.init();
return Boolean(await this.storage.db.oneOrNone(
`UPDATE popagent_agent_tasks
SET status='running',started_at=NOW(),attempt_count=attempt_count+1,
next_attempt_at=NULL,steps_completed=0,
progress=format('Starting attempt %s of %s',attempt_count+1,max_attempts)
WHERE id=$1 AND status='queued' AND attempt_count < max_attempts
AND (next_attempt_at IS NULL OR next_attempt_at <= NOW())
RETURNING id`,
[id],
));
}
async updateProgress(id: string, update: AgentTaskProgress) {
await this.init();
await this.storage.db.none(
`UPDATE popagent_agent_tasks
SET steps_completed=GREATEST(steps_completed,$2),progress=$3
WHERE id=$1 AND status='running'`,
[id, update.stepsCompleted, update.progress],
);
}
async complete(id: string, output: string): Promise<"completed" | "cancelled" | undefined> {
await this.init();
const result = await this.storage.db.oneOrNone<{ status: "completed" | "cancelled" }>(
`UPDATE popagent_agent_tasks
SET status=CASE WHEN status='cancelling' THEN 'cancelled' ELSE 'completed' END,
output=$2,error=NULL,
last_error_class=CASE WHEN status='cancelling' THEN 'cancelled' ELSE NULL END,
next_attempt_at=NULL,dead_lettered_at=NULL,
progress=CASE WHEN status='cancelling' THEN 'Cancelled' ELSE 'Completed' END,
completed_at=NOW()
WHERE id=$1 AND status IN ('running','cancelling')
RETURNING status`,
[id, output],
);
return result?.status;
}
async fail(
id: string,
error: string,
errorClass: AgentTaskErrorClass = "unknown",
nextAttemptAt?: string,
): Promise<"queued" | "failed" | "dead-letter" | undefined> {
await this.init();
const eligibleRetryAt = TRANSIENT_ERROR_CLASSES[errorClass] ? nextAttemptAt : undefined;
const result = await this.storage.db.oneOrNone<{
status: "queued" | "failed" | "dead-letter";
}>(
`UPDATE popagent_agent_tasks
SET status=CASE
WHEN source='self-update' AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN 'queued'
WHEN source='self-update' AND $4::timestamptz IS NOT NULL THEN 'dead-letter'
ELSE 'failed'
END,
error=$2::text,last_error_class=$3::text,
next_attempt_at=CASE
WHEN source='self-update' AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN $4
ELSE NULL
END,
started_at=CASE
WHEN source='self-update' AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN NULL
ELSE started_at
END,
progress=CASE
WHEN source='self-update' AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts
THEN format('Retry scheduled after %s failure (attempt %s of %s)',$3::text,attempt_count,max_attempts)
WHEN source='self-update' AND $4::timestamptz IS NOT NULL
THEN format('Dead-lettered after %s of %s attempts (%s)',attempt_count,max_attempts,$3::text)
ELSE format('Failed without retry (%s)',$3::text)
END,
completed_at=CASE
WHEN source='self-update' AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN NULL
ELSE NOW()
END,
dead_lettered_at=CASE
WHEN source='self-update' AND $4::timestamptz IS NOT NULL AND attempt_count >= max_attempts THEN NOW()
ELSE NULL
END
WHERE id=$1 AND status='running'
RETURNING status`,
[id, error, errorClass, eligibleRetryAt ?? null],
);
return result?.status;
}
async cancel(id: string): Promise<"cancelling" | "cancelled" | undefined> {
await this.init();
const result = await this.storage.db.oneOrNone<{ status: "cancelling" | "cancelled" }>(
`UPDATE popagent_agent_tasks
SET status=CASE status WHEN 'queued' THEN 'cancelled' ELSE 'cancelling' END,
progress=CASE status WHEN 'queued' THEN 'Cancelled' ELSE 'Cancelling' END,
last_error_class='cancelled',next_attempt_at=NULL,dead_lettered_at=NULL,
completed_at=CASE status WHEN 'queued' THEN NOW() ELSE NULL END
WHERE id=$1 AND status IN ('queued','running')
RETURNING status`,
[id],
);
return result?.status;
}
async finishCancellation(id: string): Promise<void> {
await this.init();
await this.storage.db.none(
`UPDATE popagent_agent_tasks
SET status='cancelled',progress='Cancelled',last_error_class='cancelled',
next_attempt_at=NULL,dead_lettered_at=NULL,completed_at=NOW()
WHERE id=$1 AND status='cancelling'`,
[id],
);
}
async listSchedules(workspaceId?: string): Promise<AgentSchedule[]> {
await this.init();
return this.storage.db.any<AgentSchedule>(
`SELECT ${SCHEDULE_COLUMNS} FROM popagent_agent_schedules
WHERE ($1::text IS NULL OR workspace_id=$1)
ORDER BY name`,
[workspaceId ?? null],
);
}
async createSchedule(input: {
sessionId?: string;
workspaceId?: string;
name: string;
prompt: string;
model: string;
maxAttempts?: number;
cron: string;
enabled: boolean;
}): Promise<AgentSchedule> {
const maxAttempts = input.maxAttempts ?? 3;
if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
throw new RangeError("Schedule max attempts must be a positive integer");
}
await this.init();
const next = input.enabled ? new Cron(input.cron, { timezone: HOST_TIMEZONE }).nextRun() : null;
return this.storage.db.one<AgentSchedule>(
`INSERT INTO popagent_agent_schedules
(id,session_id,workspace_id,source,name,prompt,model,max_attempts,cron,timezone,enabled,next_run_at)
VALUES ($1,$2,$3,'user',$4,$5,$6,$7,$8,$9,$10,$11) RETURNING ${SCHEDULE_COLUMNS}`,
[
crypto.randomUUID(),
input.sessionId ?? null,
input.workspaceId ?? DEFAULT_WORKSPACE_ID,
input.name,
input.prompt,
input.model,
maxAttempts,
input.cron,
HOST_TIMEZONE,
input.enabled,
next,
],
);
}
async updateSchedule(
id: string,
input: {
workspaceId?: string;
name: string;
prompt: string;
model: string;
maxAttempts?: number;
cron: string;
enabled: boolean;
},
): Promise<AgentSchedule | undefined> {
if (input.maxAttempts !== undefined && (!Number.isInteger(input.maxAttempts) || input.maxAttempts < 1)) {
throw new RangeError("Schedule max attempts must be a positive integer");
}
await this.init();
const next = input.enabled ? new Cron(input.cron, { timezone: HOST_TIMEZONE }).nextRun() : null;
return (await this.storage.db.oneOrNone<AgentSchedule>(
`UPDATE popagent_agent_schedules
SET workspace_id=COALESCE($2,workspace_id),name=$3,prompt=$4,model=$5,
max_attempts=COALESCE($6,max_attempts),cron=$7,timezone=$8,
enabled=$9,next_run_at=$10,updated_at=NOW()
WHERE id=$1 AND source='user' RETURNING ${SCHEDULE_COLUMNS}`,
[
id,
input.workspaceId ?? null,
input.name,
input.prompt,
input.model,
input.maxAttempts ?? null,
input.cron,
HOST_TIMEZONE,
input.enabled,
next,
],
)) ?? undefined;
}
async deleteSchedule(id: string): Promise<boolean> {
await this.init();
return Boolean(await this.storage.db.oneOrNone(
"DELETE FROM popagent_agent_schedules WHERE id=$1 AND source='user' RETURNING id",
[id],
));
}
async upsertSelfUpdateSchedule(input: {
workspaceId: string;
name: string;
prompt: string;
model: string;
maxAttempts: number;
cron: string;
enabled: boolean;
}): Promise<AgentSchedule> {
if (!Number.isInteger(input.maxAttempts) || input.maxAttempts < 1) {
throw new RangeError("Schedule max attempts must be a positive integer");
}
await this.init();
const next = input.enabled ? new Cron(input.cron, { timezone: HOST_TIMEZONE }).nextRun() : null;
return this.storage.db.one<AgentSchedule>(
`INSERT INTO popagent_agent_schedules
(id,session_id,workspace_id,source,name,prompt,model,max_attempts,cron,timezone,enabled,next_run_at)
VALUES ($1,NULL,$2,'self-update',$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (workspace_id) WHERE source='self-update'
DO UPDATE SET name=EXCLUDED.name,prompt=EXCLUDED.prompt,model=EXCLUDED.model,
max_attempts=EXCLUDED.max_attempts,cron=EXCLUDED.cron,timezone=EXCLUDED.timezone,
enabled=EXCLUDED.enabled,next_run_at=EXCLUDED.next_run_at,updated_at=NOW()
RETURNING ${SCHEDULE_COLUMNS}`,
[
crypto.randomUUID(),
input.workspaceId,
input.name,
input.prompt,
input.model,
input.maxAttempts,
input.cron,
HOST_TIMEZONE,
input.enabled,
next,
],
);
}
async disableSelfUpdateSchedules(exceptWorkspaceIds: string[] = []): Promise<void> {
await this.init();
await this.storage.db.none(
`UPDATE popagent_agent_schedules
SET enabled=false,next_run_at=NULL,updated_at=NOW()
WHERE source='self-update' AND enabled
AND NOT (workspace_id=ANY($1::text[]))`,
[exceptWorkspaceIds],
);
}
async materializeDueTasks(): Promise<AgentTask[]> {
await this.init();
return this.storage.db.tx(async (db) => {
const schedules = await db.any<AgentSchedule>(
`SELECT ${SCHEDULE_COLUMNS} FROM popagent_agent_schedules
WHERE source='user' AND enabled AND next_run_at <= NOW()
ORDER BY next_run_at, id
FOR UPDATE SKIP LOCKED`,
);
const autonomyTable = await db.one<{ present: boolean }>(
"SELECT to_regclass('popagent_autonomy_settings') IS NOT NULL AS present",
);
if (autonomyTable.present) {
schedules.push(...await db.any<AgentSchedule>(
`SELECT ${SCHEDULE_COLUMNS} FROM popagent_agent_schedules AS schedules
WHERE schedules.source='self-update'
AND schedules.enabled
AND schedules.next_run_at <= NOW()
AND EXISTS (
SELECT 1 FROM popagent_autonomy_settings
WHERE singleton AND enabled AND self_update_enabled
FOR KEY SHARE
)
ORDER BY schedules.next_run_at, schedules.id
FOR UPDATE OF schedules SKIP LOCKED`,
));
}
const tasks: AgentTask[] = [];
for (const schedule of schedules) {
const next = new Cron(schedule.cron, { timezone: HOST_TIMEZONE }).nextRun();
const task = await db.one<AgentTask>(
`INSERT INTO popagent_agent_tasks
(id, session_id, schedule_id, workspace_id, source, prompt, model, status, max_attempts)
VALUES ($1,$2,$3,$4,$5,$6,$7,'queued',$8) RETURNING ${TASK_COLUMNS}`,
[
crypto.randomUUID(),
schedule.sessionId,
schedule.id,
schedule.workspaceId,
schedule.source,
schedule.prompt,
schedule.model,
schedule.maxAttempts,
],
);
await db.none(
"UPDATE popagent_agent_schedules SET last_run_at=NOW(),next_run_at=$2,timezone=$3,updated_at=NOW() WHERE id=$1",
[schedule.id, next, HOST_TIMEZONE],
);
tasks.push(task);
}
return tasks;
});
}
}
type TaskRepository = Pick<
AgentTaskStore,
"init" | "recoverInterruptedTasks" | "listQueuedTasks" | "claim" | "updateProgress"
| "complete" | "cancel" | "finishCancellation" | "materializeDueTasks"
> & {
fail(
id: string,
error: string,
errorClass: AgentTaskErrorClass,
nextAttemptAt?: string,
): Promise<unknown>;
};
type AgentTaskExecutor = (
task: AgentTask,
signal: AbortSignal,
turnId: string,
traceId: string,
reportProgress: (progress: AgentTaskProgress) => Promise<void>,
) => Promise<string>;
export type AgentTaskLifecycle = (event: {
task: AgentTask;
turnId: string;
traceId: string;
status: "running" | "completed" | "failed" | "cancelled" | "dead-letter" | "interrupted";
outcome?: string;
error?: string;
errorClass?: AgentTaskErrorClass;
}) => Promise<void>;
type TaskRuntimeSettingsRepository = Pick<AgentRuntimeSettingsStore, "get">;
type ExecutorSettlement =
| { status: "fulfilled"; output: string }
| { status: "rejected"; error: unknown };
type AbortDisposition = {
kind: "timeout" | "cancelled" | "interrupted";
error: Error;
errorClass: AgentTaskErrorClass;
};
const MAX_ABORT_QUIESCENCE_MS = 5_000;
export class AgentTaskRuntime {
private timer?: Timer;
private started = false;
private polling = false;
private active = 0;
private pending: AgentTask[] = [];
private readonly scheduled = new Set<string>();
private readonly activeTasks = new Set<string>();
private readonly controllers = new Map<string, AbortController>();
private readonly executorPromises = new Map<string, Promise<string>>();
private readonly aborts = new Map<string, AbortDisposition>();
private readonly quiescing = new Set<string>();
private readonly jobs = new Set<Promise<void>>();
private lifecycleOperation: Promise<void> = Promise.resolve();
private desiredStarted = false;
private lifecycleGeneration = 0;
private desiredOperation: Promise<void> = Promise.resolve();
private configuration?: AgentRuntimeSettings;
constructor(
readonly store: TaskRepository,
private readonly executeTask: AgentTaskExecutor,
private readonly lifecycle?: AgentTaskLifecycle,
private readonly settings: TaskRuntimeSettingsRepository = agentRuntimeSettings,
) {}
configure(settings: AgentRuntimeSettings) {
if (!Number.isInteger(settings.taskConcurrency) || settings.taskConcurrency < 1) {
throw new RangeError("Task concurrency must be a positive integer");
}
if (!Number.isFinite(settings.taskPollIntervalMs) || settings.taskPollIntervalMs <= 0) {
throw new RangeError("Task poll interval must be positive");
}
if (!Number.isFinite(settings.taskTimeoutMs) || settings.taskTimeoutMs <= 0) {
throw new RangeError("Task timeout must be positive");
}
if (!Number.isFinite(settings.taskStaleAfterMs) || settings.taskStaleAfterMs <= 0) {
throw new RangeError("Task stale threshold must be positive");
}
this.configuration = settings;
this.restartPolling();
this.drain();
}
private configured(): AgentRuntimeSettings {
if (!this.configuration) throw new Error("Task runtime settings are not loaded");
return this.configuration;
}
private restartPolling() {
clearInterval(this.timer);
this.timer = this.started
? setInterval(() => this.poll(), this.configured().taskPollIntervalMs)
: undefined;
}
start(): Promise<void> {
if (this.desiredStarted) return this.desiredOperation;
this.desiredStarted = true;
const generation = ++this.lifecycleGeneration;
this.desiredOperation = this.serializeLifecycle(() => this.startNow(generation));
return this.desiredOperation;
}
private async startNow(generation: number) {
if (this.started) return;
try {
if (this.quiescing.size > 0) {
throw new Error("Task runtime still has an executor quiescing after abort");
}
const settings = await this.settings.get();
if (!this.desiredStarted || generation !== this.lifecycleGeneration) return;
this.configure(settings);
await this.store.init();
const recovered = await this.store.recoverInterruptedTasks(this.configured().taskStaleAfterMs);
if (recovered > 0) appLogger().info("task.recovered", { count: recovered });
const queued = await this.store.listQueuedTasks();
if (!this.desiredStarted || generation !== this.lifecycleGeneration) return;
this.started = true;
for (const task of queued) this.enqueue(task);
this.drain();
this.restartPolling();
} catch (error) {
this.started = false;
this.pending = [];
this.scheduled.clear();
this.restartPolling();
if (generation === this.lifecycleGeneration) this.desiredStarted = false;
throw error;
}
}
stop(): Promise<void> {
if (!this.desiredStarted) return this.lifecycleOperation;
this.desiredStarted = false;
this.started = false;
++this.lifecycleGeneration;
this.desiredOperation = this.serializeLifecycle(() => this.stopNow());
return this.desiredOperation;
}
private async stopNow() {
clearInterval(this.timer);
this.timer = undefined;
this.started = false;
this.pending = [];
const hadActiveExecutors = this.controllers.size > 0;
for (const [id, controller] of this.controllers) {
if (!this.aborts.has(id)) {
this.aborts.set(id, {
kind: "interrupted",
error: new Error("Background task interrupted"),
errorClass: "infrastructure",
});
}
controller.abort(this.aborts.get(id)?.error);
}
await Promise.allSettled([...this.jobs]);
if (hadActiveExecutors && this.configuration && this.quiescing.size === 0) {
await this.store.recoverInterruptedTasks(this.configuration.taskStaleAfterMs);
}
this.scheduled.clear();
}
private serializeLifecycle(operation: () => Promise<void>): Promise<void> {
const next = this.lifecycleOperation.then(operation);
this.lifecycleOperation = next.catch(() => undefined);
return next;
}
enqueue(task: AgentTask) {
if (this.scheduled.has(task.id)) return;
this.scheduled.add(task.id);
this.pending.push(task);
this.drain();
}
async cancel(id: string) {
const cancellation = await this.store.cancel(id);
if (!cancellation) return false;
const controller = this.controllers.get(id);
if (controller) {
const disposition: AbortDisposition = {
kind: "cancelled",
error: new Error("Background task cancelled"),
errorClass: "cancelled",
};
this.aborts.set(id, disposition);
controller.abort(disposition.error);
} else {
this.pending = this.pending.filter((task) => task.id !== id);
this.scheduled.delete(id);
}
return true;
}
private drain() {
while (this.started && this.active < this.configured().taskConcurrency) {
const task = this.pending.shift();
if (!task) return;
this.active++;
this.activeTasks.add(task.id);
const job = this.execute(task)
.catch((error) => {
appLogger().error("task.failed", {
taskId: task.id,
error: error instanceof Error ? error.message : String(error),
});
})
.finally(() => {
this.jobs.delete(job);
if (!this.quiescing.has(task.id)) this.releaseSlot(task.id);
});
this.jobs.add(job);
}
}
private releaseSlot(id: string) {
if (!this.activeTasks.delete(id)) return;
this.active--;
this.scheduled.delete(id);
this.drain();
}
private async execute(task: AgentTask) {
const controller = new AbortController();
this.controllers.set(task.id, controller);
const turnId = crypto.randomUUID();
const traceId = crypto.randomUUID().replaceAll("-", "");
let timeout: Timer | undefined;
let onAbort: (() => void) | undefined;
try {
if (!await this.store.claim(task.id)) return;
await this.notifyLifecycle({ task, turnId, traceId, status: "running" });
const taskTimeoutMs = this.configured().taskTimeoutMs;
const timeoutError = new Error(`Background task timed out after ${taskTimeoutMs}ms`);
const aborted = Promise.withResolvers<{ status: "aborted" }>();
onAbort = () => aborted.resolve({ status: "aborted" });
controller.signal.addEventListener("abort", onAbort, { once: true });
if (controller.signal.aborted) onAbort();
timeout = setTimeout(() => {
if (controller.signal.aborted) return;
const disposition: AbortDisposition = {
kind: "timeout",
error: timeoutError,
errorClass: "timeout",
};
this.aborts.set(task.id, disposition);
controller.abort(disposition.error);
}, taskTimeoutMs);
const executorPromise = Promise.resolve().then(() => this.executeTask(
task,
controller.signal,
turnId,
traceId,
(progress) => this.store.updateProgress(task.id, progress),
));
this.executorPromises.set(task.id, executorPromise);
const settlement = executorPromise.then<ExecutorSettlement, ExecutorSettlement>(
(output) => ({ status: "fulfilled", output }),
(error) => ({ status: "rejected", error }),
);
const first = await Promise.race([settlement, aborted.promise]);
if (first.status === "aborted" || controller.signal.aborted) {
const settled = first.status === "aborted"
? await this.waitForQuiescence(settlement, taskTimeoutMs)
: first;
if (!settled) {
this.quiescing.add(task.id);
const disposition = this.abortDisposition(task.id, controller.signal.reason);
const interruption = this.notifyLifecycle({
task,
turnId,
traceId,
status: "interrupted",
error: `${disposition.error.message}; executor did not acknowledge abort`,
errorClass: disposition.errorClass,
});
this.deferAbortedSettlement(task, turnId, traceId, settlement, interruption);
await interruption;
return;
}
await this.finishAbortedAttempt(
task,
turnId,
traceId,
this.abortDisposition(task.id, controller.signal.reason),
);
return;
}
await this.finishSettledAttempt(task, turnId, traceId, first);
} finally {
clearTimeout(timeout);
if (onAbort) controller.signal.removeEventListener("abort", onAbort);
if (!this.quiescing.has(task.id)) this.clearExecution(task.id);
}
}
private async waitForQuiescence(
settlement: Promise<ExecutorSettlement>,
taskTimeoutMs: number,
): Promise<ExecutorSettlement | undefined> {
let timer: number | undefined;
try {
const expired = new Promise<undefined>((resolve) => {
timer = setTimeout(resolve, Math.min(MAX_ABORT_QUIESCENCE_MS, taskTimeoutMs));
});
return await Promise.race([settlement, expired]);
} finally {
clearTimeout(timer);
}
}
private deferAbortedSettlement(
task: AgentTask,
turnId: string,
traceId: string,
settlement: Promise<ExecutorSettlement>,
interruption: Promise<void>,
) {
void settlement
.then(async () => {
await interruption;
const disposition = this.abortDisposition(
task.id,
this.controllers.get(task.id)?.signal.reason,
);
if (disposition.kind !== "interrupted") {
await this.finishAbortedAttempt(task, turnId, traceId, disposition);
}
})
.catch((error) => {
appLogger().error("task.transition_failed", {
taskId: task.id,
error: error instanceof Error ? error.message : String(error),
});
})
.finally(() => {
this.quiescing.delete(task.id);
this.clearExecution(task.id);
this.releaseSlot(task.id);
});
}
private abortDisposition(id: string, reason: unknown): AbortDisposition {
const disposition = this.aborts.get(id);
if (disposition) return disposition;
return {
kind: "interrupted",
error: reason instanceof Error ? reason : new Error(String(reason ?? "Background task interrupted")),
errorClass: "infrastructure",
};
}
private async finishAbortedAttempt(
task: AgentTask,
turnId: string,
traceId: string,
disposition: AbortDisposition,
) {
if (disposition.kind === "cancelled") {
await this.store.finishCancellation(task.id);
await this.notifyLifecycle({
task,
turnId,
traceId,
status: "cancelled",
error: disposition.error.message,
errorClass: disposition.errorClass,
});
return;
}
if (disposition.kind === "timeout") {
const failure = await this.store.fail(
task.id,
disposition.error.message,
disposition.errorClass,
retryAt(task.attemptCount + 1),
);
if (failure) {
await this.notifyLifecycle({
task,
turnId,
traceId,
status: failure === "dead-letter" ? "dead-letter" : "failed",
error: disposition.error.message,
errorClass: disposition.errorClass,
});
}
return;
}
await this.notifyLifecycle({
task,
turnId,
traceId,
status: "interrupted",
error: disposition.error.message,
errorClass: disposition.errorClass,
});
}
private async finishSettledAttempt(
task: AgentTask,
turnId: string,
traceId: string,
settlement: ExecutorSettlement,
) {
if (settlement.status === "fulfilled") {
const completion = await this.store.complete(task.id, settlement.output);
if (completion === "cancelled") {
await this.notifyLifecycle({
task,
turnId,
traceId,
status: "cancelled",
errorClass: "cancelled",
});
} else if (completion === "completed") {
await this.notifyLifecycle({
task,
turnId,
traceId,
status: "completed",
outcome: settlement.output,
});
}
return;
}
const message = settlement.error instanceof Error
? settlement.error.message
: String(settlement.error);
const errorClass = classifyFailure(settlement.error);
const nextAttemptAt = TRANSIENT_ERROR_CLASSES[errorClass]
? retryAt(task.attemptCount + 1)
: undefined;
const failure = await this.store.fail(task.id, message, errorClass, nextAttemptAt);
if (failure) {
await this.notifyLifecycle({
task,
turnId,
traceId,
status: failure === "dead-letter" ? "dead-letter" : "failed",
error: message,
errorClass,
});
}
}
private async notifyLifecycle(event: Parameters<AgentTaskLifecycle>[0]) {
try {
await this.lifecycle?.(event);
} catch (error) {
appLogger().warn("task.lifecycle_failed", {
taskId: event.task.id,
status: event.status,
error: error instanceof Error ? error.message : String(error),
});
}
}
private clearExecution(id: string) {
this.controllers.delete(id);
this.executorPromises.delete(id);
this.aborts.delete(id);
}
private poll() {
if (this.polling || !this.started) return;
this.polling = true;
void this.tick()
.catch((error) => {
appLogger().error("task.scheduler_failed", { error: error instanceof Error ? error.message : String(error) });
})
.finally(() => {
this.polling = false;
});
}
private async tick() {
for (const task of await this.store.materializeDueTasks()) this.enqueue(task);
for (const task of await this.store.listQueuedTasks()) this.enqueue(task);
}
}