popagent

Latest change 6915a66ad654763c6197f3ed1ca4a5756f35dc92 - Fix autonomous task lineage contract by Ólafur Búi Ólafsson

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,
    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 (
           '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([
      "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,
      })).toBeUndefined();
      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",
        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 second' 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);
  });

  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" });
    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()).toContainEqual(expect.objectContaining({
      id: task.id,
      attemptCount: 1,
      maxAttempts: 3,
      nextAttemptAt: null,
      lastErrorClass: null,
      deadLetteredAt: null,
    }));
  });

  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("dead-letters exhausted transient failures", async () => {
    const task = await store.createTask({
      prompt: "Exhaust one attempt",
      model: TEST_MODEL,
      maxAttempts: 1,
    });
    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,
      "temporary gateway failure",
      "infrastructure",
      new Date(Date.now() + 1_000).toISOString(),
    )).toBe("dead-letter");
    expect(await store.listTasks()).toContainEqual(expect.objectContaining({
      id: task.id,
      status: "dead-letter",
      attemptCount: 1,
      maxAttempts: 1,
      lastErrorClass: "infrastructure",
      nextAttemptAt: null,
      deadLetteredAt: expect.any(String),
      progress: "Dead-lettered after 1 of 1 attempts (infrastructure)",
    }));
  });

  test("caps restart recovery at the task attempt limit", async () => {
    const task = await store.createTask({
      prompt: "Do not recover forever",
      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(0);
    expect(await store.listTasks()).toContainEqual(expect.objectContaining({
      id: task.id,
      status: "dead-letter",
      attemptCount: 1,
      maxAttempts: 1,
      lastErrorClass: "infrastructure",
      deadLetteredAt: expect.any(String),
    }));
  });

  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("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("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: "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();
      }
    }
  });
});