popagent

Latest change 6edc7e928c57dab23c0973ed3f7ee70b3ebb72d5 - Initial popagent baseline by AkurAI Build

import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import type { AgentRuntimeSettings, AgentTask } from "./api-types";
import { createTaskLifecycleHooks } from "./hook-lifecycle";
import { 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,
    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("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);
    await store.updateProgress(task.id, { stepsCompleted: 3, progress: "Using webSearch" });
    await store.complete(task.id, "TASK_COMPLETE");
    expect(await store.listTasks()).toContainEqual(expect.objectContaining({
      id: task.id,
      status: "completed",
      output: "TASK_COMPLETE",
      stepsCompleted: 3,
      recoveryCount: 0,
      progress: "Completed",
    }));
  });

  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",
      completedAt: null,
    }));
    await store.finishCancellation(task.id);
    expect(await store.listTasks()).toContainEqual(expect.objectContaining({
      id: task.id,
      status: "cancelled",
      progress: "Cancelled",
      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: "Queued after restart",
      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: "Queued after restart",
    }));
  });

  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", cron: "0 9 * * *", enabled: true });
    scheduleIds.push(schedule.id);
    expect(schedule.nextRunAt).not.toBeNull();
    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("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", 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((await store.listSchedules()).find((item) => item.id === schedule.id)).toEqual(
      expect.objectContaining({ lastRunAt: expect.any(String), nextRunAt: expect.any(String) }),
    );
  });
});

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,
      workspaceId: "default",
      prompt: `Task ${index}`,
      model: TEST_MODEL,
      status: "queued",
      output: null,
      error: null,
      stepsCompleted: 0,
      progress: null,
      recoveryCount: 0,
      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("emits a correlated lifecycle for autonomous task execution", async () => {
    const task: AgentTask = {
      id: "lifecycle-task",
      sessionId: "session-1",
      scheduleId: null,
      workspaceId: "default",
      prompt: "Complete lifecycle",
      model: TEST_MODEL,
      status: "queued",
      output: null,
      error: null,
      stepsCompleted: 0,
      progress: null,
      recoveryCount: 0,
      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("finalizes cancellation after the active executor acknowledges abort", async () => {
    const task: AgentTask = {
      id: "cancellation-task",
      sessionId: null,
      scheduleId: null,
      workspaceId: "default",
      prompt: "Cancel active work",
      model: TEST_MODEL,
      status: "queued",
      output: null,
      error: null,
      stepsCompleted: 0,
      progress: null,
      recoveryCount: 0,
      createdAt: "",
      startedAt: null,
      completedAt: null,
    };
    const started = Promise.withResolvers<void>();
    const allowFinalization = Promise.withResolvers<void>();
    const stopped = Promise.withResolvers<void>();
    let status: AgentTask["status"] = "queued";
    const currentStatus = (): AgentTask["status"] => status;
    const repository = {
      init: async () => undefined,
      recoverInterruptedTasks: async () => 0,
      listQueuedTasks: async () => [],
      claim: async () => {
        status = "running";
        return true;
      },
      updateProgress: async () => undefined,
      complete: async () => undefined,
      fail: async () => undefined,
      cancel: async () => {
        status = "cancelling";
        return "cancelling" as const;
      },
      finishCancellation: async () => {
        await allowFinalization.promise;
        status = "cancelled";
      },
      materializeDueTasks: async () => [],
    };
    const runtime = new AgentTaskRuntime(
      repository,
      async () => {
        started.resolve();
        return new Promise<string>(() => undefined);
      },
      async (event) => {
        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(currentStatus()).toBe("cancelling");
      allowFinalization.resolve();
      await stopped.promise;
      expect(currentStatus()).toBe("cancelled");
    } finally {
      allowFinalization.resolve();
      await runtime.stop();
    }
  });

  test("fails tasks that exceed their runtime deadline", async () => {
    const failure = Promise.withResolvers<string>();
    let aborted = false;
    const task: AgentTask = {
      id: "timeout-task",
      sessionId: null,
      scheduleId: null,
      workspaceId: "default",
      prompt: "Never completes",
      model: TEST_MODEL,
      status: "queued",
      output: null,
      error: null,
      stepsCompleted: 0,
      progress: null,
      recoveryCount: 0,
      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) => { failure.resolve(error); },
      cancel: async () => "cancelled" as const,
      finishCancellation: async () => undefined,
      materializeDueTasks: async () => [],
    };
    const runtime = new AgentTaskRuntime(repository, async (_task, signal) => {
      signal.addEventListener("abort", () => { aborted = true; }, { once: true });
      return new Promise<string>(() => undefined);
    }, undefined, runtimeSettings({ taskTimeoutMs: 10 }));
    runtime.enqueue(task);

    try {
      await runtime.start();
      expect(await failure.promise).toBe("Background task timed out after 10ms");
      expect(aborted).toBe(true);
    } finally {
      await runtime.stop();
    }
  });
});