Menu
popagent
publicLatest change 5e361abf2b93831addde65d670bfbec2a201b11c - feat: run autonomous work when idle by AkurAI Build
import { afterAll, beforeAll, describe, expect, test, vi } from "bun:test";
import type { AgentRuntimeSettings, AgentTask, AgentTaskErrorClass } from "./api-types";
import { initializeEvolutionStorage } from "./autonomy-settings";
import { createTaskLifecycleHooks } from "./hook-lifecycle";
import { HookBlockedError, HookRuntime, type HookEvent } from "./hooks";
import { AgentTaskRuntime, AgentTaskStore } from "./tasks";
const TEST_MODEL = "test/model";
const runtimeSettings = (
overrides: Partial<AgentRuntimeSettings> = {},
) => ({
get: async (): Promise<AgentRuntimeSettings> => ({
defaultModel: TEST_MODEL,
modelSource: "configured",
supervisorMaxSteps: 24,
specialistMaxSteps: 12,
toolConcurrency: 3,
delegationContextMessages: 12,
delegationResultCharacters: 16_000,
maxProcessorRetries: 3,
finalResponseFeedback: "Return a final response",
delegationFailureFeedback: "Delegate {{agentId}} failed: {{error}}",
delegationResultTruncationMarker: "\n[truncated]",
taskConcurrency: 1,
taskPollIntervalMs: 60_000,
taskTimeoutMs: 900_000,
taskStaleAfterMs: 86_400_000,
updatedAt: "",
...overrides,
}),
});
const store = new AgentTaskStore();
const taskIds: string[] = [];
const scheduleIds: string[] = [];
beforeAll(() => store.init());
afterAll(async () => {
for (const id of taskIds) await store.storage.db.none("DELETE FROM popagent_agent_tasks WHERE id = $1", [id]);
for (const id of scheduleIds) await store.deleteSchedule(id);
});
describe("AgentTaskStore", () => {
test("applies the retry schema migration idempotently", async () => {
await new AgentTaskStore().init();
await new AgentTaskStore().init();
expect(await store.storage.db.one<{ count: number }>(
`SELECT COUNT(*)::int AS count FROM popagent_data_migrations
WHERE id='2026-08-14-agent-task-retry-v1'`,
)).toEqual({ count: 1 });
const columns = await store.storage.db.any<{ columnName: string }>(
`SELECT column_name AS "columnName"
FROM information_schema.columns
WHERE table_schema=current_schema() AND table_name='popagent_agent_tasks'
AND column_name IN (
'activity','attempt_count','max_attempts','next_attempt_at','last_error_class','dead_lettered_at','heartbeat_at'
)
ORDER BY column_name`,
);
expect(columns.map((column) => column.columnName)).toEqual([
"activity",
"attempt_count",
"dead_lettered_at",
"heartbeat_at",
"last_error_class",
"max_attempts",
"next_attempt_at",
]);
expect(await store.storage.db.one<{ count: number }>(
`SELECT COUNT(*)::int AS count
FROM information_schema.columns
WHERE table_schema=current_schema()
AND table_name='popagent_agent_schedules'
AND column_name='max_attempts'`,
)).toEqual({ count: 1 });
});
test("persists immutable source ownership and materializes trusted self-update lineage", async () => {
await initializeEvolutionStorage();
const workspaceId = `self-update-source-${crypto.randomUUID()}`;
const previous = await store.storage.db.one<{ enabled: boolean; selfUpdateEnabled: boolean }>(
`SELECT enabled,self_update_enabled AS "selfUpdateEnabled"
FROM popagent_autonomy_settings WHERE singleton`,
);
let scheduleId: string | undefined;
let materializedTaskId: string | undefined;
try {
const [first, contender] = await Promise.all([
store.upsertSelfUpdateSchedule({
workspaceId,
name: "System source schedule",
prompt: "Maintain source",
model: TEST_MODEL,
maxAttempts: 2,
cron: "0 3 * * *",
enabled: true,
}),
new AgentTaskStore().upsertSelfUpdateSchedule({
workspaceId,
name: "System source schedule",
prompt: "Maintain source",
model: TEST_MODEL,
maxAttempts: 2,
cron: "0 3 * * *",
enabled: true,
}),
]);
scheduleId = first.id;
expect(contender.id).toBe(first.id);
expect(first.source).toBe("self-update");
expect(await store.storage.db.one<{ count: number }>(
`SELECT COUNT(*)::int AS count FROM popagent_data_migrations
WHERE id='2026-08-14-task-source-v1'`,
)).toEqual({ count: 1 });
expect((await store.storage.db.one<{ indexDefinition: string }>(
`SELECT indexdef AS "indexDefinition" FROM pg_indexes
WHERE schemaname=current_schema()
AND indexname='popagent_agent_schedules_self_update_workspace_idx'`,
)).indexDefinition).toContain("WHERE (source = 'self-update'::text)");
expect(await store.updateSchedule(first.id, {
workspaceId,
name: "Attempted user edit",
prompt: "Attempted user edit",
model: TEST_MODEL,
maxAttempts: 1,
cron: "0 4 * * *",
enabled: false,
})).toEqual(expect.objectContaining({
id: first.id,
source: "self-update",
name: "System source schedule",
prompt: "Maintain source",
model: TEST_MODEL,
maxAttempts: 2,
cron: "0 4 * * *",
enabled: false,
}));
expect(await store.upsertSelfUpdateSchedule({
workspaceId,
name: "Updated system source schedule",
prompt: "Inspect concrete candidates",
model: TEST_MODEL,
maxAttempts: 2,
cron: "15 5 * * *",
enabled: true,
})).toEqual(expect.objectContaining({
id: first.id,
name: "Updated system source schedule",
prompt: "Inspect concrete candidates",
cron: "0 4 * * *",
enabled: false,
}));
expect(await store.deleteSchedule(first.id)).toBe(false);
await expect(store.createTask({
workspaceId,
scheduleId: first.id,
prompt: "Attempted source forgery",
model: TEST_MODEL,
})).rejects.toThrow();
await new AgentTaskStore().init();
expect((await store.listSchedules(workspaceId))[0]).toEqual(expect.objectContaining({
id: first.id,
source: "self-update",
cron: "0 4 * * *",
enabled: false,
}));
await store.updateSchedule(first.id, {
workspaceId,
name: first.name,
prompt: first.prompt,
model: first.model,
maxAttempts: first.maxAttempts,
cron: "0 4 * * *",
enabled: true,
});
await store.storage.db.none(
"UPDATE popagent_autonomy_settings SET enabled=true,self_update_enabled=false WHERE singleton",
);
await store.storage.db.none(
"UPDATE popagent_agent_schedules SET next_run_at=NOW() + INTERVAL '1 day' WHERE id=$1",
[first.id],
);
expect((await store.materializeDueTasks()).some((item) => item.scheduleId === first.id)).toBe(false);
await store.storage.db.none(
"UPDATE popagent_autonomy_settings SET self_update_enabled=true,idle_improvement_enabled=true,idle_workspace_ids=jsonb_build_array($1::text) WHERE singleton",
[workspaceId],
);
const materialized = (await store.materializeDueTasks()).find((item) => item.scheduleId === first.id);
materializedTaskId = materialized?.id;
expect(materialized).toEqual(expect.objectContaining({
source: "self-update",
workspaceId,
scheduleId: first.id,
maxAttempts: 2,
}));
} finally {
if (materializedTaskId) {
await store.storage.db.none("DELETE FROM popagent_agent_tasks WHERE id=$1", [materializedTaskId]);
}
if (scheduleId) {
await store.storage.db.none("DELETE FROM popagent_agent_schedules WHERE id=$1", [scheduleId]);
}
await store.storage.db.none(
`UPDATE popagent_autonomy_settings
SET enabled=$1,self_update_enabled=$2 WHERE singleton`,
[previous.enabled, previous.selfUpdateEnabled],
);
}
});
test("does not rewrite existing schedule timing after the retry migration", async () => {
const schedule = await store.createSchedule({
name: "Migration timing schedule",
prompt: "Keep this occurrence",
model: TEST_MODEL,
cron: "* * * * *",
enabled: true,
});
scheduleIds.push(schedule.id);
await store.storage.db.none(
"UPDATE popagent_agent_schedules SET next_run_at=NOW() - INTERVAL '1 hour' WHERE id=$1",
[schedule.id],
);
const before = await store.storage.db.one<{ nextRunAt: string }>(
`SELECT next_run_at::text AS "nextRunAt"
FROM popagent_agent_schedules WHERE id=$1`,
[schedule.id],
);
await new AgentTaskStore().init();
expect(await store.storage.db.one<{ nextRunAt: string }>(
`SELECT next_run_at::text AS "nextRunAt"
FROM popagent_agent_schedules WHERE id=$1`,
[schedule.id],
)).toEqual(before);
await store.updateSchedule(schedule.id, {
name: schedule.name,
workspaceId: schedule.workspaceId,
prompt: schedule.prompt,
model: schedule.model,
maxAttempts: schedule.maxAttempts,
cron: schedule.cron,
enabled: false,
});
});
test("creates a governed autonomous workflow task on demand", async () => {
const task = await store.createWorkflowTask({
workspaceId: "default",
prompt: "Implement with specialist handoffs",
model: TEST_MODEL,
});
try {
expect(task).toMatchObject({
source: "self-update",
status: "queued",
prompt: "Implement with specialist handoffs",
});
} finally {
await store.storage.db.none("DELETE FROM popagent_agent_tasks WHERE id=$1", [task.id]);
}
});
test("persists and transitions a background task", async () => {
const task = await store.createTask({ prompt: "Persistence test", model: "cx/gpt-5.6-sol" });
taskIds.push(task.id);
expect(task.status).toBe("queued");
expect(await store.claim(task.id)).toBe(true);
expect((await store.listTasks()).find((item) => item.id === task.id)?.heartbeatAt).toBeString();
await store.updateProgress(task.id, {
stepsCompleted: 3,
progress: "Using webSearch",
activity: {
turnId: "turn-1",
runId: "run-1",
agentId: "researcher",
agentName: "Researcher",
iteration: 3,
maxIterations: 12,
isFinal: false,
finishReason: "tool-calls",
text: "Inspecting current behavior",
tools: ["webSearch"],
},
});
const heartbeat = (await store.listTasks()).find((item) => item.id === task.id)?.heartbeatAt;
expect(heartbeat).toBeString();
await store.complete(task.id, "TASK_COMPLETE");
expect((await store.listTasks()).find((item) => item.id === task.id)?.heartbeatAt).not.toBe(heartbeat);
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "completed",
output: "TASK_COMPLETE",
stepsCompleted: 3,
recoveryCount: 0,
progress: "Completed",
}));
expect((await store.listTasks()).find((item) => item.id === task.id)?.activity).toEqual([
expect.objectContaining({ agentName: "Researcher", tools: ["webSearch"] }),
]);
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
attemptCount: 1,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
}));
});
test("creates one durable remediation task per terminal failure", async () => {
const failed = await store.createTask({ prompt: "Fail for remediation", model: TEST_MODEL });
taskIds.push(failed.id);
const first = await store.createRemediationTask(failed, "broken output", "validation", "repair/model");
expect(first).toEqual(expect.objectContaining({
workspaceId: failed.workspaceId,
model: "repair/model",
remediationForTaskId: failed.id,
status: "queued",
}));
if (first) taskIds.push(first.id);
expect(await store.createRemediationTask(failed, "same failure", "validation", "repair/model")).toBeUndefined();
expect(await store.createRemediationTask(first!, "recursive failure", "unknown", "repair/model")).toBeUndefined();
});
test("deletes only terminal tasks", async () => {
const task = await store.createTask({ prompt: "Terminal deletion test", model: TEST_MODEL });
taskIds.push(task.id);
expect(await store.deleteTerminalTask(task.id)).toBe("active");
expect(await store.claim(task.id)).toBe(true);
await store.fail(task.id, "Expected failure", "unknown");
expect(await store.deleteTerminalTask(task.id)).toBe("deleted");
taskIds.splice(taskIds.indexOf(task.id), 1);
expect(await store.deleteTerminalTask(task.id)).toBe("missing");
});
test("resolves failed work with operator evidence", async () => {
const task = await store.createTask({ prompt: "Resolve failed task", model: TEST_MODEL });
taskIds.push(task.id);
expect(await store.claim(task.id)).toBe(true);
await store.fail(task.id, "Expected failure", "unknown");
expect(await store.resolve(task.id, "Reviewed and resolved inline.")).toBe("completed");
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "completed",
output: "Reviewed and resolved inline.",
error: null,
progress: "Completed",
}));
});
test("does not resolve active or missing work", async () => {
const task = await store.createTask({ prompt: "Keep active task", model: TEST_MODEL });
taskIds.push(task.id);
expect(await store.resolve(task.id, "Not terminal")).toBe("active");
expect(await store.resolve(crypto.randomUUID(), "Missing")).toBe("missing");
});
test("keeps active cancellation observable until execution stops", async () => {
const task = await store.createTask({ prompt: "Cancellation test", model: TEST_MODEL });
taskIds.push(task.id);
expect(await store.claim(task.id)).toBe(true);
expect(await store.cancel(task.id)).toBe("cancelling");
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "cancelling",
progress: "Cancelling",
lastErrorClass: "cancelled",
nextAttemptAt: null,
completedAt: null,
}));
await store.finishCancellation(task.id);
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "cancelled",
progress: "Cancelled",
lastErrorClass: "cancelled",
nextAttemptAt: null,
completedAt: expect.any(String),
}));
});
test("increments recovery counts when interrupted work is requeued", async () => {
await store.recoverInterruptedTasks(86_400_000);
const task = await store.createTask({ prompt: "Recovery test", model: TEST_MODEL });
taskIds.push(task.id);
expect(task.recoveryCount).toBe(0);
expect(await store.claim(task.id)).toBe(true);
expect(await store.recoverInterruptedTasks(86_400_000)).toBe(1);
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "queued",
progress: expect.stringContaining("Retry scheduled after infrastructure interruption"),
recoveryCount: 1,
startedAt: null,
}));
});
test("recovers a freshly started task even when it was queued long ago", async () => {
const task = await store.createTask({ prompt: "Old queue, fresh execution", model: TEST_MODEL });
taskIds.push(task.id);
await store.storage.db.none(
"UPDATE popagent_agent_tasks SET created_at = NOW() - INTERVAL '2 days' WHERE id = $1",
[task.id],
);
expect(await store.claim(task.id)).toBe(true);
expect(await store.recoverInterruptedTasks(60_000)).toBe(1);
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "queued",
progress: expect.stringContaining("Retry scheduled after infrastructure interruption"),
}));
});
test("durably delays transient retries without changing task boundaries", async () => {
const workspaceId = `workspace-${crypto.randomUUID()}`;
const retryTime = new Date(Date.now() + 60_000).toISOString();
const schedule = await store.createSchedule({
workspaceId,
name: "Retry boundary schedule",
prompt: "Scheduled source",
model: TEST_MODEL,
cron: "0 9 * * *",
enabled: true,
});
scheduleIds.push(schedule.id);
const task = await store.createTask({
sessionId: "retry-session",
workspaceId,
scheduleId: schedule.id,
prompt: "Retry the same work",
model: TEST_MODEL,
maxAttempts: 3,
});
taskIds.push(task.id);
await store.storage.db.none("UPDATE popagent_agent_tasks SET source='self-update' WHERE id=$1", [task.id]);
expect(await store.claim(task.id)).toBe(true);
expect(await store.fail(task.id, "provider capacity exhausted", "provider", retryTime)).toBe("queued");
const persisted = (await store.listTasks(workspaceId)).find((item) => item.id === task.id);
expect(persisted).toEqual(expect.objectContaining({
sessionId: "retry-session",
scheduleId: schedule.id,
workspaceId,
prompt: "Retry the same work",
model: TEST_MODEL,
status: "queued",
error: "provider capacity exhausted",
attemptCount: 1,
maxAttempts: 3,
lastErrorClass: "provider",
startedAt: null,
completedAt: null,
deadLetteredAt: null,
progress: expect.stringContaining("Retry scheduled after provider failure"),
}));
expect(Date.parse(persisted?.nextAttemptAt ?? "")).toBeGreaterThan(Date.now());
expect((await store.listQueuedTasks()).some((item) => item.id === task.id)).toBe(false);
expect(await store.claim(task.id)).toBe(false);
await store.storage.db.none(
"UPDATE popagent_agent_tasks SET next_attempt_at=NOW() WHERE id=$1",
[task.id],
);
expect((await store.listQueuedTasks()).some((item) => item.id === task.id)).toBe(true);
expect(await store.claim(task.id)).toBe(true);
await store.complete(task.id, "recovered");
});
test("never retries permanent failures even when given a future attempt", async () => {
const task = await store.createTask({ prompt: "Invalid work", model: TEST_MODEL });
taskIds.push(task.id);
expect(await store.claim(task.id)).toBe(true);
expect(await store.fail(
task.id,
"invalid task input",
"validation",
new Date(Date.now() + 60_000).toISOString(),
)).toBe("failed");
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "failed",
lastErrorClass: "validation",
nextAttemptAt: null,
deadLetteredAt: null,
progress: "Failed without retry (validation)",
completedAt: expect.any(String),
}));
});
test("keeps transient failures queued after the configured attempt budget", async () => {
const task = await store.createTask({
prompt: "Recover without operator attention",
model: TEST_MODEL,
maxAttempts: 1,
});
taskIds.push(task.id);
expect(await store.claim(task.id)).toBe(true);
expect(await store.fail(
task.id,
"temporary gateway failure",
"infrastructure",
new Date(Date.now() + 1_000).toISOString(),
)).toBe("queued");
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "queued",
attemptCount: 1,
maxAttempts: 2,
lastErrorClass: "infrastructure",
nextAttemptAt: expect.any(String),
deadLetteredAt: null,
completedAt: null,
progress: "Retry scheduled after infrastructure failure (attempt 1 of 2)",
}));
});
test("keeps interrupted work recoverable after the configured attempt budget", async () => {
const task = await store.createTask({
prompt: "Recover after restart",
model: TEST_MODEL,
maxAttempts: 1,
});
taskIds.push(task.id);
expect(await store.claim(task.id)).toBe(true);
expect(await store.recoverInterruptedTasks(86_400_000)).toBe(1);
expect(await store.listTasks()).toContainEqual(expect.objectContaining({
id: task.id,
status: "queued",
attemptCount: 1,
maxAttempts: 2,
lastErrorClass: "infrastructure",
nextAttemptAt: expect.any(String),
deadLetteredAt: null,
completedAt: null,
}));
});
test("persists, pauses, resumes, and deletes a recurring schedule", async () => {
const schedule = await store.createSchedule({
name: "Persistence schedule",
prompt: "Scheduled test",
model: "cx/gpt-5.6-sol",
maxAttempts: 4,
cron: "0 9 * * *",
enabled: true,
});
scheduleIds.push(schedule.id);
expect(schedule).toEqual(expect.objectContaining({ maxAttempts: 4, nextRunAt: expect.any(String) }));
const paused = await store.updateSchedule(schedule.id, { ...schedule, enabled: false });
expect(paused).toEqual(expect.objectContaining({ enabled: false, nextRunAt: null }));
const resumed = await store.updateSchedule(schedule.id, { ...schedule, enabled: true });
expect(resumed).toEqual(expect.objectContaining({ enabled: true }));
expect(await store.deleteSchedule(schedule.id)).toBe(true);
scheduleIds.splice(scheduleIds.indexOf(schedule.id), 1);
});
test("scopes tasks and schedules to their repository workspace", async () => {
const workspaceId = `workspace-${crypto.randomUUID()}`;
const task = await store.createTask({
workspaceId,
prompt: "Workspace task",
model: TEST_MODEL,
});
const schedule = await store.createSchedule({
workspaceId,
name: "Workspace schedule",
prompt: "Workspace schedule task",
model: TEST_MODEL,
cron: "0 9 * * *",
enabled: true,
});
taskIds.push(task.id);
scheduleIds.push(schedule.id);
expect(await store.listTasks(workspaceId)).toEqual([
expect.objectContaining({ id: task.id, workspaceId }),
]);
expect(await store.listSchedules(workspaceId)).toEqual([
expect.objectContaining({ id: schedule.id, workspaceId }),
]);
expect((await store.listTasks("another-workspace")).some((item) => item.id === task.id)).toBe(false);
});
test("permits only one active autonomous task across workspaces", async () => {
await initializeEvolutionStorage();
const workspaceIds = [`singleton-a-${crypto.randomUUID()}`, `singleton-b-${crypto.randomUUID()}`];
const previous = await store.storage.db.one<{
enabled: boolean;
selfUpdateEnabled: boolean;
idleImprovementEnabled: boolean;
idleWorkspaceIds: string[];
}>(`SELECT enabled,self_update_enabled AS "selfUpdateEnabled",
idle_improvement_enabled AS "idleImprovementEnabled",
idle_workspace_ids AS "idleWorkspaceIds"
FROM popagent_autonomy_settings WHERE singleton`);
const selfUpdateScheduleIds: string[] = [];
const selfUpdateTaskIds: string[] = [];
try {
for (const workspaceId of workspaceIds) {
const schedule = await store.upsertSelfUpdateSchedule({
workspaceId,
name: `Autonomous ${workspaceId}`,
prompt: `Improve ${workspaceId}`,
model: TEST_MODEL,
maxAttempts: 1,
cron: "* * * * *",
enabled: true,
});
selfUpdateScheduleIds.push(schedule.id);
}
await store.storage.db.none(
`UPDATE popagent_autonomy_settings
SET enabled=true,self_update_enabled=true,idle_improvement_enabled=true,
idle_workspace_ids=$1::jsonb
WHERE singleton`,
[JSON.stringify(workspaceIds)],
);
await store.storage.db.none(
"UPDATE popagent_agent_schedules SET next_run_at=NOW() - INTERVAL '1 second' WHERE id=ANY($1::text[])",
[selfUpdateScheduleIds],
);
const materialized = await store.materializeDueTasks();
selfUpdateTaskIds.push(...materialized.filter((task) => task.source === "self-update").map((task) => task.id));
expect(materialized.filter((task) => task.source === "self-update")).toHaveLength(1);
expect(await store.materializeDueTasks()).toEqual([]);
await store.storage.db.none("DELETE FROM popagent_agent_tasks WHERE id=ANY($1::text[])", [selfUpdateTaskIds]);
const contender = new AgentTaskStore();
await contender.init();
const queued = await Promise.all([
store.queueSelfUpdate(workspaceIds[0]!),
contender.queueSelfUpdate(workspaceIds[1]!),
]);
selfUpdateTaskIds.push(...queued.filter((task) => task !== undefined).map((task) => task.id));
expect(queued.filter((task) => task !== undefined)).toHaveLength(1);
} finally {
for (const id of selfUpdateTaskIds) {
await store.storage.db.none("DELETE FROM popagent_agent_tasks WHERE id=$1", [id]);
}
await store.storage.db.none("DELETE FROM popagent_agent_schedules WHERE id=ANY($1::text[])", [selfUpdateScheduleIds]);
await store.storage.db.none(
`UPDATE popagent_autonomy_settings
SET enabled=$1,self_update_enabled=$2,idle_improvement_enabled=$3,
idle_workspace_ids=$4::jsonb
WHERE singleton`,
[previous.enabled, previous.selfUpdateEnabled, previous.idleImprovementEnabled, JSON.stringify(previous.idleWorkspaceIds)],
);
}
});
test("materializes one task per due schedule occurrence", async () => {
const schedule = await store.createSchedule({
name: "Due schedule",
prompt: "Due test",
model: "cx/gpt-5.6-sol",
maxAttempts: 2,
cron: "* * * * *",
enabled: true,
});
scheduleIds.push(schedule.id);
const contender = new AgentTaskStore();
await contender.init();
await store.storage.db.none("UPDATE popagent_agent_schedules SET next_run_at = NOW() - INTERVAL '1 second' WHERE id = $1", [schedule.id]);
const batches = await Promise.all([store.materializeDueTasks(), contender.materializeDueTasks()]);
const materialized = batches.flat().filter((task) => task.scheduleId === schedule.id);
taskIds.push(...materialized.map((task) => task.id));
expect(materialized).toHaveLength(1);
expect(materialized[0]).toEqual(expect.objectContaining({
scheduleId: schedule.id,
workspaceId: schedule.workspaceId,
prompt: schedule.prompt,
model: schedule.model,
attemptCount: 0,
maxAttempts: 2,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
}));
const listed = (await store.listSchedules()).find((item) => item.id === schedule.id);
expect(listed).toEqual(expect.objectContaining({ lastRunAt: expect.any(String), nextRunAt: expect.any(String), latestTaskStatus: "queued" }));
});
});
describe("AgentTaskRuntime", () => {
test("recovers every queued task with bounded execution concurrency", async () => {
const queued = Array.from({ length: 105 }, (_, index): AgentTask => ({
id: `queued-${index}`,
sessionId: null,
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: `Task ${index}`,
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: new Date(index).toISOString(),
startedAt: null,
completedAt: null,
}));
const completed: string[] = [];
const firstBatchStarted = Promise.withResolvers<void>();
const release = Promise.withResolvers<void>();
const allCompleted = Promise.withResolvers<void>();
let active = 0;
let peak = 0;
let started = 0;
let recovered = false;
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => {
recovered = true;
return 0;
},
listQueuedTasks: async () => queued,
claim: async () => true,
updateProgress: async () => undefined,
complete: async (id: string) => {
completed.push(id);
if (completed.length === queued.length) allCompleted.resolve();
return "completed" as const;
},
fail: async () => undefined,
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(repository, async (task) => {
active++;
peak = Math.max(peak, active);
started++;
if (started === 3) firstBatchStarted.resolve();
await release.promise;
active--;
return `completed ${task.prompt}`;
}, undefined, runtimeSettings({ taskConcurrency: 3 }));
for (const task of queued) runtime.enqueue(task);
try {
await runtime.start();
await firstBatchStarted.promise;
expect(recovered).toBe(true);
expect(started).toBe(3);
expect(peak).toBe(3);
release.resolve();
await allCompleted.promise;
expect(completed).toHaveLength(queued.length);
expect(new Set(completed).size).toBe(queued.length);
expect(peak).toBe(3);
} finally {
release.resolve();
await runtime.stop();
}
});
test("polls for retries that become eligible after startup", async () => {
const task: AgentTask = {
id: "delayed-retry",
sessionId: "retry-session",
scheduleId: null,
source: "user",
workspaceId: "retry-workspace",
prompt: "Resume when due",
model: TEST_MODEL,
status: "queued",
output: null,
error: "provider unavailable",
stepsCompleted: 0,
progress: "Retry scheduled",
recoveryCount: 0,
attemptCount: 1,
maxAttempts: 3,
nextAttemptAt: new Date(Date.now() + 1).toISOString(),
lastErrorClass: "provider",
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const completed = Promise.withResolvers<void>();
let reads = 0;
let claimed = false;
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => ++reads === 1 ? [] : [task],
claim: async () => {
if (claimed) return false;
claimed = true;
return true;
},
updateProgress: async () => undefined,
complete: async () => {
completed.resolve();
return "completed" as const;
},
fail: async () => undefined,
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(
repository,
async () => "resumed",
undefined,
runtimeSettings({ taskPollIntervalMs: 5 }),
);
try {
await runtime.start();
await completed.promise;
expect(reads).toBeGreaterThan(1);
expect(claimed).toBe(true);
} finally {
await runtime.stop();
}
});
test("lets stop win while startup recovery is still in flight", async () => {
const queuedTask: AgentTask = {
id: "startup-stop-task",
sessionId: null,
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "Must not start",
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const initializationStarted = Promise.withResolvers<void>();
const finishInitialization = Promise.withResolvers<void>();
let recoveries = 0;
let executions = 0;
let materializations = 0;
const repository = {
init: async () => {
initializationStarted.resolve();
await finishInitialization.promise;
},
recoverInterruptedTasks: async () => {
recoveries++;
return 0;
},
listQueuedTasks: async () => [queuedTask],
claim: async () => true,
updateProgress: async () => undefined,
complete: async () => "completed" as const,
fail: async () => undefined,
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => {
materializations++;
return [];
},
};
const runtime = new AgentTaskRuntime(
repository,
async () => {
executions++;
return "must not execute";
},
undefined,
runtimeSettings({ taskPollIntervalMs: 1 }),
);
vi.useFakeTimers();
try {
const starting = Promise.all([runtime.start(), runtime.start()]);
await initializationStarted.promise;
const stopping = runtime.stop();
finishInitialization.resolve();
await Promise.all([starting, stopping]);
vi.advanceTimersByTime(5);
expect(recoveries).toBe(1);
expect(executions).toBe(0);
expect(materializations).toBe(0);
} finally {
finishInitialization.resolve();
vi.useRealTimers();
}
});
test("emits a correlated lifecycle for autonomous task execution", async () => {
const task: AgentTask = {
id: "lifecycle-task",
sessionId: "session-1",
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "Complete lifecycle",
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const finished = Promise.withResolvers<void>();
const events: HookEvent[] = [];
const hookRuntime = new HookRuntime({
config: {
version: 1,
hooks: {
TaskStart: [{ id: "start", type: "http", url: "https://hooks.example/task-start" }],
TaskStop: [{ id: "stop", type: "http", url: "https://hooks.example/task-stop" }],
},
},
transport: async (_handler, event) => {
events.push(event);
if (event.eventName === "TaskStop") finished.resolve();
return { schemaVersion: 1, outcome: "pass" };
},
});
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => [],
claim: async () => true,
updateProgress: async () => undefined,
complete: async () => "completed" as const,
fail: async () => undefined,
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(
repository,
async () => "TASK_COMPLETE",
createTaskLifecycleHooks(hookRuntime),
runtimeSettings(),
);
runtime.enqueue(task);
try {
await runtime.start();
await finished.promise;
expect(events.map((event) => event.eventName)).toEqual(["TaskStart", "TaskStop"]);
expect(events.map((event) => event.detail.status)).toEqual(["running", "completed"]);
expect(new Set(events.map((event) => event.turnId)).size).toBe(1);
} finally {
await runtime.stop();
}
});
test("keeps durable completion terminal when a lifecycle observer throws", async () => {
const task: AgentTask = {
id: "observer-failure-task",
sessionId: null,
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "Complete despite observer failure",
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const terminalObserved = Promise.withResolvers<void>();
const lifecycleEvents: Array<{ status: string; traceId: string; outcome?: string }> = [];
let durableStatus: AgentTask["status"] = "queued";
let failures = 0;
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => [],
claim: async () => {
durableStatus = "running";
return true;
},
updateProgress: async () => undefined,
complete: async () => {
durableStatus = "completed";
return "completed" as const;
},
fail: async () => {
failures++;
durableStatus = "failed";
},
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(
repository,
async () => "durable output",
async (event) => {
lifecycleEvents.push({
status: event.status,
traceId: event.traceId,
outcome: event.outcome,
});
if (event.status === "completed") terminalObserved.resolve();
throw new Error(`observer failed after ${event.status}`);
},
runtimeSettings(),
);
runtime.enqueue(task);
try {
await runtime.start();
await terminalObserved.promise;
await runtime.stop();
expect(durableStatus as AgentTask["status"]).toBe("completed");
expect(failures).toBe(0);
expect(lifecycleEvents.map((event) => event.status)).toEqual(["running", "completed"]);
expect(new Set(lifecycleEvents.map((event) => event.traceId)).size).toBe(1);
expect(lifecycleEvents[0]?.traceId).toMatch(/^[a-f0-9]{32}$/);
expect(lifecycleEvents[1]?.outcome).toBe("durable output");
} finally {
await runtime.stop();
}
});
test("finalizes cancellation only after the active executor acknowledges abort", async () => {
const task: AgentTask = {
id: "cancellation-task",
sessionId: null,
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "Cancel active work",
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const started = Promise.withResolvers<void>();
const acknowledgeAbort = Promise.withResolvers<void>();
const stopped = Promise.withResolvers<void>();
let status: AgentTask["status"] = "queued";
let executorSettled = false;
let failures = 0;
const lifecycleStatuses: string[] = [];
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => [],
claim: async () => {
status = "running";
return true;
},
updateProgress: async () => undefined,
complete: async () => undefined,
fail: async () => { failures++; },
cancel: async () => {
status = "cancelling";
return "cancelling" as const;
},
finishCancellation: async () => {
expect(executorSettled).toBe(true);
status = "cancelled";
},
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(
repository,
async (_task, signal) => {
started.resolve();
await new Promise<void>((resolve) => {
if (signal.aborted) resolve();
else signal.addEventListener("abort", () => resolve(), { once: true });
});
await acknowledgeAbort.promise;
executorSettled = true;
throw signal.reason;
},
async (event) => {
lifecycleStatuses.push(event.status);
if (event.status === "cancelled") stopped.resolve();
},
runtimeSettings(),
);
runtime.enqueue(task);
try {
await runtime.start();
await started.promise;
expect(await runtime.cancel(task.id)).toBe(true);
expect(status as AgentTask["status"]).toBe("cancelling");
expect(executorSettled).toBe(false);
expect(lifecycleStatuses).not.toContain("cancelled");
acknowledgeAbort.resolve();
await stopped.promise;
expect(status as AgentTask["status"]).toBe("cancelled");
expect(failures).toBe(0);
} finally {
acknowledgeAbort.resolve();
await runtime.stop();
}
});
test("holds the slot and retry eligibility until a timed-out executor settles", async () => {
const task: AgentTask = {
id: "timeout-task",
sessionId: null,
scheduleId: null,
source: "user",
workspaceId: "default",
prompt: "Acknowledge timeout late",
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const followingTask: AgentTask = {
...task,
id: "after-timeout-task",
prompt: "Wait for the occupied slot",
};
const firstStarted = Promise.withResolvers<void>();
const interrupted = Promise.withResolvers<void>();
const acknowledgeAbort = Promise.withResolvers<void>();
const failure = Promise.withResolvers<{
error: string;
errorClass: AgentTaskErrorClass;
nextAttemptAt?: string;
}>();
const followingStarted = Promise.withResolvers<void>();
const retryStarted = Promise.withResolvers<void>();
let timeoutStatus: AgentTask["status"] = "queued";
let followingStatus: AgentTask["status"] = "queued";
let attempts = 0;
let activeExecutors = 0;
let peakExecutors = 0;
let firstSettled = false;
let followingDidStart = false;
let failureCalls = 0;
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => [
...(timeoutStatus === "queued" ? [task] : []),
...(followingStatus === "queued" ? [followingTask] : []),
],
claim: async (id: string) => {
if (id === task.id && timeoutStatus === "queued") {
timeoutStatus = "running";
return true;
}
if (id === followingTask.id && followingStatus === "queued") {
followingStatus = "running";
return true;
}
return false;
},
updateProgress: async () => undefined,
complete: async (id: string) => {
if (id === task.id) timeoutStatus = "completed";
else followingStatus = "completed";
return "completed" as const;
},
fail: async (
_id: string,
error: string,
errorClass: AgentTaskErrorClass,
nextAttemptAt?: string,
) => {
failureCalls++;
expect(firstSettled).toBe(true);
timeoutStatus = "queued";
failure.resolve({ error, errorClass, nextAttemptAt });
return "queued" as const;
},
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(
repository,
async (current, signal) => {
activeExecutors++;
peakExecutors = Math.max(peakExecutors, activeExecutors);
try {
if (current.id === task.id) {
attempts++;
if (attempts === 1) {
firstStarted.resolve();
await new Promise<void>((resolve) => {
if (signal.aborted) resolve();
else signal.addEventListener("abort", () => resolve(), { once: true });
});
await acknowledgeAbort.promise;
firstSettled = true;
return "late first attempt";
}
retryStarted.resolve();
return "retry completed";
}
followingDidStart = true;
followingStarted.resolve();
return "following completed";
} finally {
activeExecutors--;
}
},
async (event) => {
if (event.task.id === task.id && event.status === "interrupted") interrupted.resolve();
},
runtimeSettings({ taskConcurrency: 1, taskPollIntervalMs: 1, taskTimeoutMs: 10 }),
);
runtime.enqueue(task);
runtime.enqueue(followingTask);
try {
await runtime.start();
await firstStarted.promise;
await interrupted.promise;
expect(timeoutStatus as AgentTask["status"]).toBe("running");
expect(failureCalls).toBe(0);
expect(followingDidStart).toBe(false);
expect(attempts).toBe(1);
acknowledgeAbort.resolve();
expect(await failure.promise).toEqual({
error: "Background task timed out after 10ms",
errorClass: "timeout",
nextAttemptAt: expect.any(String),
});
await followingStarted.promise;
await retryStarted.promise;
expect(peakExecutors).toBe(1);
expect(attempts).toBe(2);
} finally {
acknowledgeAbort.resolve();
await runtime.stop();
}
});
test("queues one remediation after a terminal failure and never for remediation failures", async () => {
const remediated: string[] = [];
const remediationQueued = Promise.withResolvers<void>();
const remediationCompleted = Promise.withResolvers<void>();
let parentResolutionAttempted = false;
const failed = (id: string, remediationForTaskId?: string): AgentTask => ({
id,
sessionId: null,
scheduleId: null,
source: "user",
remediationForTaskId,
workspaceId: "workspace",
prompt: "fail",
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 1,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
});
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => [],
claim: async () => true,
updateProgress: async () => undefined,
complete: async (id: string) => {
if (id !== "successful-remediation") return undefined;
remediationCompleted.resolve();
return "completed" as const;
},
fail: async () => "failed" as const,
resolve: async () => {
parentResolutionAttempted = true;
return "completed" as const;
},
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
createRemediationTask: async (task: AgentTask) => {
remediated.push(task.id);
remediationQueued.resolve();
return undefined;
},
};
const runtime = new AgentTaskRuntime(
repository,
async (task) => {
if (task.id === "successful-remediation") return "repaired";
throw new Error("broken");
},
undefined,
runtimeSettings(),
);
runtime.enqueue(failed("original"));
runtime.enqueue(failed("remediation", "original"));
runtime.enqueue(failed("successful-remediation", "original"));
await runtime.start();
try {
await remediationQueued.promise;
await remediationCompleted.promise;
expect(parentResolutionAttempted).toBe(false);
expect(remediated).toEqual(["original"]);
} finally {
await runtime.stop();
}
});
test("retries only conservatively classified transient failures", async () => {
const scenarios: Array<{
name: string;
error: Error;
errorClass: AgentTaskErrorClass;
retryable: boolean;
}> = [
{
name: "provider",
error: Object.assign(new Error("provider capacity exhausted"), {
name: "APICallError",
statusCode: 503,
}),
errorClass: "provider",
retryable: true,
},
{
name: "rate limit",
error: Object.assign(new Error("too many requests"), { statusCode: 429 }),
errorClass: "rate-limit",
retryable: true,
},
{
name: "browser",
error: new Error("Chromium browser disconnected"),
errorClass: "browser",
retryable: true,
},
{
name: "database",
error: Object.assign(new Error("database connection exception"), { code: "08006" }),
errorClass: "database",
retryable: true,
},
{
name: "infrastructure",
error: Object.assign(new Error("socket failed"), { code: "ECONNRESET" }),
errorClass: "infrastructure",
retryable: true,
},
{
name: "provider timeout",
error: Object.assign(new Error("request exceeded its deadline"), { name: "TimeoutError" }),
errorClass: "timeout",
retryable: true,
},
{
name: "gateway response-header timeout",
error: new Error("[504]: Upstream request did not return response headers after 600000ms (codex/gpt-5.6-luna-medium)"),
errorClass: "timeout",
retryable: true,
},
{
name: "hook denial",
error: new HookBlockedError("blocked by policy", "policy"),
errorClass: "hook-denial",
retryable: false,
},
{
name: "invalid input",
error: new RangeError("invalid task input"),
errorClass: "validation",
retryable: false,
},
{
name: "permission",
error: new Error("workspace access denied"),
errorClass: "permission",
retryable: false,
},
{
name: "missing configuration",
error: new Error("Model provider is not configured"),
errorClass: "configuration",
retryable: false,
},
{
name: "unknown",
error: new Error("deterministic implementation failure"),
errorClass: "unknown",
retryable: false,
},
];
for (const [index, scenario] of scenarios.entries()) {
const failure = Promise.withResolvers<{
errorClass: AgentTaskErrorClass;
nextAttemptAt?: string;
}>();
const task: AgentTask = {
id: `classification-${index}`,
sessionId: "classification-session",
scheduleId: null,
source: "user",
workspaceId: "classification-workspace",
prompt: `Classify ${scenario.name}`,
model: TEST_MODEL,
status: "queued",
output: null,
error: null,
stepsCompleted: 0,
progress: null,
recoveryCount: 0,
attemptCount: 0,
maxAttempts: 3,
nextAttemptAt: null,
lastErrorClass: null,
deadLetteredAt: null,
createdAt: "",
startedAt: null,
completedAt: null,
};
const repository = {
init: async () => undefined,
recoverInterruptedTasks: async () => 0,
listQueuedTasks: async () => [],
claim: async () => true,
updateProgress: async () => undefined,
complete: async () => undefined,
fail: async (
_id: string,
_error: string,
errorClass: AgentTaskErrorClass,
nextAttemptAt?: string,
) => {
failure.resolve({ errorClass, nextAttemptAt });
},
cancel: async () => "cancelled" as const,
finishCancellation: async () => undefined,
materializeDueTasks: async () => [],
};
const runtime = new AgentTaskRuntime(
repository,
async () => { throw scenario.error; },
undefined,
runtimeSettings(),
);
runtime.enqueue(task);
try {
await runtime.start();
const result = await failure.promise;
expect(result.errorClass).toBe(scenario.errorClass);
if (scenario.retryable) {
expect(Date.parse(result.nextAttemptAt ?? "")).toBeGreaterThan(Date.now());
} else {
expect(result.nextAttemptAt).toBeUndefined();
}
} finally {
await runtime.stop();
}
}
});
});