Menu
popagent
publicLatest change f81c1154ece5a040b42b3624c8429fbf56c15cfa - Remediate failed background 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,
type AgentWorkflowDetail,
type AgentWorkflowSummary,
} from "./api-types";
import { autonomyActivity, type AutonomyActivityGate } from "./autonomy-runtime";
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",
remediation_for_task_id AS "remediationForTaskId",
prompt, model, status, output, error,
steps_completed AS "stepsCompleted", progress, activity, workflow,
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"`;
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 TASK_ACTIVITY_LIMIT = 256;
const SELF_UPDATE_TASK_LOCK = "popagent-self-update-task";
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,
};
function workflowSummary(detail: AgentWorkflowDetail): AgentWorkflowSummary {
return {
runId: detail.runId,
correlationId: detail.correlationId,
state: detail.state,
currentPhase: detail.currentPhase,
phases: detail.phases.map(({ id, label, state, evidence }) => ({
id,
label,
state,
evidence: evidence.slice(0, 12),
})),
};
}
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 (names.includes("selfupdateprepareerror")) return "workflow";
if (names.includes("selfupdateresearcherror")) return "research";
if (names.includes("selfupdatedecideerror")) return "decision";
if (names.includes("selfupdateimplementerror")) return "implementation";
if (names.includes("selfupdateinspectchangeerror")) return "inspection";
if (names.includes("selfupdateverifyerror")) return "verification";
if (names.includes("selfupdatereviewerror")) return "review";
if (names.includes("selfupdatecommiterror")) return "commit";
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());
constructor(private readonly activity: AutonomyActivityGate = autonomyActivity) {}
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','build-maintenance')),
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','build-maintenance')),
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,
activity JSONB NOT NULL DEFAULT '[]'::jsonb,
workflow JSONB,
workflow_detail JSONB,
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,
heartbeat_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.none("ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS heartbeat_at TIMESTAMPTZ");
await this.storage.db.none("ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS activity JSONB NOT NULL DEFAULT '[]'::jsonb");
await this.storage.db.tx(async (db) => {
const migrationId = "2026-08-15-task-remediation-v1";
await db.one("SELECT pg_advisory_xact_lock(hashtext($1))", [migrationId]);
if (await db.oneOrNone("SELECT 1 FROM popagent_data_migrations WHERE id=$1", [migrationId])) return;
await db.none(`
ALTER TABLE popagent_agent_tasks
ADD COLUMN IF NOT EXISTS remediation_for_task_id TEXT
REFERENCES popagent_agent_tasks(id) ON DELETE SET NULL;
CREATE UNIQUE INDEX IF NOT EXISTS popagent_agent_tasks_remediation_parent_idx
ON popagent_agent_tasks (remediation_for_task_id)
WHERE remediation_for_task_id IS NOT NULL;
`);
await db.none("INSERT INTO popagent_data_migrations (id) VALUES ($1) ON CONFLICT DO NOTHING", [migrationId]);
});
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;
ALTER TABLE popagent_agent_tasks ADD COLUMN IF NOT EXISTS heartbeat_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]);
});
await this.storage.db.tx(async (db) => {
const migrationId = "2026-08-15-build-maintenance-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
DROP CONSTRAINT IF EXISTS popagent_agent_schedules_source_check;
ALTER TABLE popagent_agent_tasks
DROP CONSTRAINT IF EXISTS popagent_agent_tasks_source_check;
ALTER TABLE popagent_agent_schedules
ADD CONSTRAINT popagent_agent_schedules_source_check
CHECK (source IN ('user','self-update','build-maintenance'));
ALTER TABLE popagent_agent_tasks
ADD CONSTRAINT popagent_agent_tasks_source_check
CHECK (source IN ('user','self-update','build-maintenance'));
CREATE UNIQUE INDEX IF NOT EXISTS popagent_agent_schedules_build_maintenance_workspace_idx
ON popagent_agent_schedules (workspace_id)
WHERE source='build-maintenance';
`);
await db.none("INSERT INTO popagent_data_migrations (id) VALUES ($1)", [migrationId]);
});
await this.storage.db.tx(async (db) => {
const migrationId = "2026-08-15-autonomous-workflow-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_tasks
ADD COLUMN IF NOT EXISTS workflow JSONB,
ADD COLUMN IF NOT EXISTS workflow_detail JSONB;
`);
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 getTask(id: string): Promise<AgentTask | null> {
await this.init();
return this.storage.db.oneOrNone<AgentTask>(
`SELECT ${TASK_COLUMNS} FROM popagent_agent_tasks WHERE id=$1`,
[id],
);
}
async getWorkflow(id: string): Promise<AgentWorkflowDetail | null> {
await this.init();
const row = await this.storage.db.oneOrNone<{ workflowDetail: AgentWorkflowDetail }>(
`SELECT workflow_detail AS "workflowDetail"
FROM popagent_agent_tasks
WHERE id=$1 AND source='self-update'`,
[id],
);
return row?.workflowDetail ?? null;
}
async automationStatus() {
await this.init();
const settings = await this.storage.db.one<{
enabled: boolean;
idleImprovementEnabled: boolean;
idleWorkspaceIds: string[];
}>(`SELECT enabled,idle_improvement_enabled AS "idleImprovementEnabled",
idle_workspace_ids AS "idleWorkspaceIds"
FROM popagent_autonomy_settings WHERE singleton`);
const activeTask = await this.storage.db.oneOrNone<AgentTask>(
`SELECT ${TASK_COLUMNS} FROM popagent_agent_tasks
WHERE source IN ('self-update','build-maintenance') AND status IN ('queued','running','cancelling')
ORDER BY created_at LIMIT 1`,
);
return {
state: !settings.enabled || !settings.idleImprovementEnabled || !settings.idleWorkspaceIds.length
? "paused"
: activeTask ? "improving" : "active",
activeTask,
selectedWorkspaceIds: settings.idleWorkspaceIds,
} as const;
}
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 resolve(id: string, output: string): Promise<"completed" | "active" | "missing"> {
await this.init();
return this.storage.db.tx(async (db) => {
const completed = await db.oneOrNone<{ id: string }>(
`UPDATE popagent_agent_tasks
SET status='completed',output=$2,error=NULL,last_error_class=NULL,
next_attempt_at=NULL,dead_lettered_at=NULL,progress='Completed',completed_at=NOW()
WHERE id=$1 AND status IN ('failed','cancelled','dead-letter')
RETURNING id`,
[id, output],
);
if (completed) return "completed";
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,heartbeat_at=NOW()
WHERE status='running'
AND COALESCE(heartbeat_at,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 createRemediationTask(task: AgentTask, error: string, errorClass: AgentTaskErrorClass, model: string): Promise<AgentTask | undefined> {
if (task.remediationForTaskId) return undefined;
await this.init();
const prompt = `Diagnose and remediate failed task ${task.id}.
Workspace: ${task.workspaceId}
Failure class: ${errorClass}
Failure: ${error.slice(0, 4_000)}
Inspect the failed task and repository evidence. Fix a contained code or configuration defect when safe, retry transient work when appropriate, or resolve the original task with concrete evidence when it is stale or already fixed. If operator action is required, report the exact blocker. Do not create another remediation task.`;
return (await this.storage.db.oneOrNone<AgentTask>(
`INSERT INTO popagent_agent_tasks
(id, session_id, workspace_id, source, remediation_for_task_id, prompt, model, status, max_attempts)
VALUES ($1,$2,$3,'user',$4,$5,$6,'queued',3)
ON CONFLICT (remediation_for_task_id) WHERE remediation_for_task_id IS NOT NULL DO NOTHING
RETURNING ${TASK_COLUMNS}`,
[crypto.randomUUID(), task.sessionId, task.workspaceId, task.id, prompt, model],
)) ?? undefined;
}
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(),heartbeat_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,
activity=CASE WHEN $4::jsonb IS NULL THEN activity
ELSE (CASE WHEN jsonb_array_length(activity) >= $5 THEN activity - 0 ELSE activity END) || $4::jsonb
END,
workflow=COALESCE($6::jsonb,workflow),
workflow_detail=COALESCE($7::jsonb,workflow_detail),
heartbeat_at=NOW()
WHERE id=$1 AND status IN ('running','cancelling')`,
[
id,
update.stepsCompleted,
update.progress,
update.activity ? JSON.stringify([update.activity]) : null,
TASK_ACTIVITY_LIMIT,
update.workflow ? JSON.stringify(workflowSummary(update.workflow)) : null,
update.workflow ? JSON.stringify(update.workflow) : null,
],
);
}
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(),heartbeat_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 IN ('self-update','build-maintenance') AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN 'queued'
WHEN source IN ('self-update','build-maintenance') 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 IN ('self-update','build-maintenance') AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN $4
ELSE NULL
END,
started_at=CASE
WHEN source IN ('self-update','build-maintenance') AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN NULL
ELSE started_at
END,
progress=CASE
WHEN source IN ('self-update','build-maintenance') 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 IN ('self-update','build-maintenance') 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 IN ('self-update','build-maintenance') AND $4::timestamptz IS NOT NULL AND attempt_count < max_attempts THEN NULL
ELSE NOW()
END,
dead_lettered_at=CASE
WHEN source IN ('self-update','build-maintenance') AND $4::timestamptz IS NOT NULL AND attempt_count >= max_attempts THEN NOW()
ELSE NULL
END,
heartbeat_at=NOW()
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},latest.status AS "latestTaskStatus",
latest.heartbeat_at::text AS "latestHeartbeatAt"
FROM popagent_agent_schedules AS schedules
LEFT JOIN LATERAL (
SELECT status,heartbeat_at FROM popagent_agent_tasks
WHERE schedule_id=schedules.id ORDER BY created_at DESC LIMIT 1
) latest ON TRUE
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;
const created = await this.storage.db.one<{ id: string }>(
`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 id`,
[crypto.randomUUID(), input.sessionId ?? null, input.workspaceId ?? DEFAULT_WORKSPACE_ID,
input.name, input.prompt, input.model, maxAttempts, input.cron, HOST_TIMEZONE, input.enabled, next],
);
return (await this.listSchedules()).find((schedule) => schedule.id === created.id)!;
}
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;
const updated = await this.storage.db.oneOrNone<{ id: string }>(
`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 id`,
[id, input.workspaceId ?? null, input.name, input.prompt, input.model,
input.maxAttempts ?? null, input.cron, HOST_TIMEZONE, input.enabled, next],
);
return updated ? (await this.listSchedules()).find((schedule) => schedule.id === id) : 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;
const schedule = await this.storage.db.one<{ id: string }>(
`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 id`,
[crypto.randomUUID(), input.workspaceId, input.name, input.prompt, input.model,
input.maxAttempts, input.cron, HOST_TIMEZONE, input.enabled, next],
);
return (await this.listSchedules(input.workspaceId)).find((item) => item.id === schedule.id)!;
}
async upsertBuildMaintenanceSchedule(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;
const schedule = await this.storage.db.one<{ id: string }>(
`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,'build-maintenance',$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (workspace_id) WHERE source='build-maintenance'
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 id`,
[crypto.randomUUID(), input.workspaceId, input.name, input.prompt, input.model,
input.maxAttempts, input.cron, HOST_TIMEZONE, input.enabled, next],
);
return (await this.listSchedules(input.workspaceId)).find((item) => item.id === schedule.id)!;
}
async disableBuildMaintenanceSchedules(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='build-maintenance' AND enabled
AND NOT (workspace_id=ANY($1::text[]))`,
[exceptWorkspaceIds],
);
}
async queueSelfUpdate(workspaceId: string): Promise<AgentTask | undefined> {
await this.init();
return this.storage.db.tx(async (db) => {
await db.one("SELECT pg_advisory_xact_lock(hashtext($1))", [SELF_UPDATE_TASK_LOCK]);
return (await db.oneOrNone<AgentTask>(
`INSERT INTO popagent_agent_tasks
(id,session_id,schedule_id,workspace_id,source,prompt,model,status,max_attempts)
SELECT $1,NULL,id,workspace_id,'self-update',prompt,model,'queued',max_attempts
FROM popagent_agent_schedules
WHERE source='self-update' AND workspace_id=$2 AND enabled
AND NOT EXISTS (
SELECT 1 FROM popagent_agent_tasks
WHERE source='self-update' AND status IN ('queued','running','cancelling')
)
RETURNING ${TASK_COLUMNS}`,
[crypto.randomUUID(), workspaceId],
)) ?? undefined;
});
}
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 && this.activity.isIdle) {
await db.one("SELECT pg_advisory_xact_lock(hashtext($1))", [SELF_UPDATE_TASK_LOCK]);
schedules.push(...await db.any<AgentSchedule>(
`SELECT ${SCHEDULE_COLUMNS} FROM popagent_agent_schedules AS schedules
WHERE schedules.source IN ('self-update','build-maintenance')
AND schedules.enabled
AND schedules.next_run_at <= NOW()
AND NOT EXISTS (
SELECT 1 FROM popagent_agent_tasks
WHERE source IN ('self-update','build-maintenance')
AND status IN ('queued','running','cancelling')
)
AND NOT EXISTS (
SELECT 1 FROM popagent_agent_schedules
WHERE source='user' AND enabled AND next_run_at <= NOW()
)
AND EXISTS (
SELECT 1 FROM popagent_autonomy_settings AS settings
WHERE singleton AND enabled
AND (
(schedules.source='self-update'
AND self_update_enabled
AND idle_improvement_enabled
AND idle_workspace_ids ? schedules.workspace_id)
OR
(schedules.source='build-maintenance'
AND idle_improvement_enabled
AND idle_workspace_ids ? schedules.workspace_id)
)
FOR UPDATE
)
ORDER BY schedules.next_run_at, schedules.id
LIMIT 1
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>;
createRemediationTask?: AgentTaskStore["createRemediationTask"];
};
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" | "progress" | "completed" | "failed" | "cancelled" | "dead-letter" | "interrupted";
progress?: AgentTaskProgress;
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 activityLeases = new Map<string, () => void>();
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,
private readonly activity: AutonomyActivityGate = autonomyActivity,
) {}
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 autonomous = task.source === "self-update" || task.source === "build-maintenance";
const lease = autonomous
? this.activity.tryEnterAutonomous(task.id)
: this.activity.enterBackground(task.id);
if (!lease) return;
this.activityLeases.set(task.id, lease);
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,
async (progress) => {
await this.store.updateProgress(task.id, progress);
await this.notifyLifecycle({ task, turnId, traceId, status: "progress", 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,
});
if (failure !== "queued") {
await this.enqueueRemediation(task, disposition.error.message, 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,
});
if (failure !== "queued") await this.enqueueRemediation(task, message, errorClass);
}
}
private async enqueueRemediation(task: AgentTask, error: string, errorClass: AgentTaskErrorClass) {
if (task.remediationForTaskId || !this.store.createRemediationTask) return;
try {
const remediation = await this.store.createRemediationTask(
task,
error,
errorClass,
this.configured().defaultModel,
);
if (remediation) this.enqueue(remediation);
} catch (remediationError) {
appLogger().error("task.remediation_failed", {
taskId: task.id,
error: remediationError instanceof Error ? remediationError.message : String(remediationError),
});
}
}
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);
const lease = this.activityLeases.get(id);
if (lease) {
this.activityLeases.delete(id);
lease();
}
}
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);
}
}