Task queue persist dead letter queue to file and expose /api/tasks/dead letter endpoint

__tests__/api/tasks.test.ts

@@ -1,15 +1,25 @@
 import { beforeEach, describe, expect, it } from "vitest"
+import { mkdtempSync, readFileSync } from "node:fs"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { DELETE as discardDeadLetterTask } from "@/app/api/tasks/dead-letter/[id]/route"
+import { POST as retryDeadLetterTask } from "@/app/api/tasks/dead-letter/[id]/retry/route"
+import { GET as listDeadLetterTasks } from "@/app/api/tasks/dead-letter/route"
 import { GET as getTask } from "@/app/api/tasks/[id]/route"
 import { POST as cancelTask } from "@/app/api/tasks/[id]/cancel/route"
 import { POST as failTask } from "@/app/api/tasks/[id]/fail/route"
 import { POST as retryTask } from "@/app/api/tasks/[id]/retry/route"
 import { GET as listTasks, POST as enqueueTask } from "@/app/api/tasks/route"
-import { resetTaskQueueForTests } from "@/lib/agent-runtime/task-queue"
+import { loadDeadLetterQueueForTests, resetTaskQueueForTests } from "@/lib/agent-runtime/task-queue"
 
 const context = (id: string) => ({ params: Promise.resolve({ id }) })
 
 describe("/api/tasks", () => {
-  beforeEach(() => resetTaskQueueForTests())
+  beforeEach(() => {
+    const tempDir = mkdtempSync(join(tmpdir(), "open-stellar-task-dlq-"))
+    process.env.TASK_DLQ_FILE = join(tempDir, "task-dlq.json")
+    resetTaskQueueForTests()
+  })
 
   it("enqueues and retrieves a prioritized task", async () => {
     const createRes = await enqueueTask(new Request("http://localhost/api/tasks", {
@@ -83,4 +93,67 @@ describe("/api/tasks", () => {
     expect(retry.task.status).toBe("pending")
     expect(retry.task.retryCount).toBe(0)
   })
+
+  it("persists dead-letter entries and reloads them from file", async () => {
+    const createRes = await enqueueTask(new Request("http://localhost/api/tasks", {
+      method: "POST",
+      body: JSON.stringify({ id: "persisted-task", type: "agent.persist", maxRetries: 0, payload: { sector: "beta" } }),
+      headers: { "Content-Type": "application/json" },
+    }))
+    expect(createRes.status).toBe(201)
+
+    const failRes = await failTask(new Request("http://localhost/api/tasks/persisted-task/fail", {
+      method: "POST",
+      body: JSON.stringify({ error: "permanent" }),
+    }), context("persisted-task"))
+    expect(failRes.status).toBe(200)
+
+    const persisted = JSON.parse(readFileSync(process.env.TASK_DLQ_FILE!, "utf8"))
+    expect(persisted).toHaveLength(1)
+    expect(persisted[0]).toMatchObject({
+      task: { id: "persisted-task", status: "dead-letter" },
+      errorMessage: "permanent",
+      retryCount: 0,
+    })
+
+    resetTaskQueueForTests()
+    loadDeadLetterQueueForTests()
+    const listRes = await listDeadLetterTasks(new Request("http://localhost/api/tasks/dead-letter"))
+    const list = await listRes.json()
+    expect(list.tasks[0].task.id).toBe("persisted-task")
+  })
+
+  it("lists, retries, and discards dead-letter tasks via dedicated endpoints", async () => {
+    for (const id of ["dlq-1", "dlq-2"]) {
+      await enqueueTask(new Request("http://localhost/api/tasks", {
+        method: "POST",
+        body: JSON.stringify({ id, type: "agent.dead", maxRetries: 0 }),
+        headers: { "Content-Type": "application/json" },
+      }))
+      await failTask(new Request(`http://localhost/api/tasks/${id}/fail`, {
+        method: "POST",
+        body: JSON.stringify({ error: `${id} failed` }),
+      }), context(id))
+    }
+
+    const listRes = await listDeadLetterTasks(new Request("http://localhost/api/tasks/dead-letter?limit=1&offset=1"))
+    const list = await listRes.json()
+    expect(listRes.status).toBe(200)
+    expect(list.tasks).toHaveLength(1)
+    expect(list.pagination.total).toBe(2)
+    expect(list.tasks[0]).toMatchObject({ errorMessage: "dlq-2 failed", retryCount: 0 })
+
+    const retryRes = await retryDeadLetterTask(new Request("http://localhost/api/tasks/dead-letter/dlq-1/retry", { method: "POST" }), context("dlq-1"))
+    const retried = await retryRes.json()
+    expect(retryRes.status).toBe(200)
+    expect(retried.task.status).toBe("pending")
+
+    const discardRes = await discardDeadLetterTask(new Request("http://localhost/api/tasks/dead-letter/dlq-2", { method: "DELETE" }), context("dlq-2"))
+    const discarded = await discardRes.json()
+    expect(discardRes.status).toBe(200)
+    expect(discarded.task.status).toBe("dead-letter")
+
+    const notFoundRes = await discardDeadLetterTask(new Request("http://localhost/api/tasks/dead-letter/missing", { method: "DELETE" }), context("missing"))
+    expect(notFoundRes.status).toBe(404)
+  })
 })

app/api/tasks/dead-letter/[id]/retry/route.ts

@@ -0,0 +1,16 @@
+import { NextResponse } from "next/server"
+import { retryDeadLetterTask } from "@/lib/agent-runtime/task-queue"
+
+type TaskRouteContext = {
+  params: Promise<{ id: string }>
+}
+
+export async function POST(_req: Request, context: TaskRouteContext) {
+  const { id } = await context.params
+  try {
+    return NextResponse.json({ ok: true, task: retryDeadLetterTask(id) })
+  } catch (error) {
+    const message = error instanceof Error ? error.message : "Failed to retry dead-letter task"
+    return NextResponse.json({ ok: false, error: message }, { status: message === "Task not found" ? 404 : 400 })
+  }
+}

app/api/tasks/dead-letter/[id]/route.ts

@@ -0,0 +1,16 @@
+import { NextResponse } from "next/server"
+import { discardDeadLetterTask } from "@/lib/agent-runtime/task-queue"
+
+type TaskRouteContext = {
+  params: Promise<{ id: string }>
+}
+
+export async function DELETE(_req: Request, context: TaskRouteContext) {
+  const { id } = await context.params
+  try {
+    return NextResponse.json({ ok: true, task: discardDeadLetterTask(id) })
+  } catch (error) {
+    const message = error instanceof Error ? error.message : "Failed to discard dead-letter task"
+    return NextResponse.json({ ok: false, error: message }, { status: message === "Task not found" ? 404 : 400 })
+  }
+}

app/api/tasks/dead-letter/route.ts

@@ -0,0 +1,21 @@
+import { NextResponse } from "next/server"
+import { listDeadLetterEntries } from "@/lib/agent-runtime/task-queue"
+
+export const dynamic = "force-dynamic"
+
+export async function GET(req: Request) {
+  const url = new URL(req.url)
+  const offset = Number(url.searchParams.get("offset") ?? 0)
+  const limit = Number(url.searchParams.get("limit") ?? 50)
+  const { entries, total, offset: normalizedOffset, limit: normalizedLimit } = listDeadLetterEntries({ offset, limit })
+
+  return NextResponse.json({
+    ok: true,
+    tasks: entries,
+    pagination: {
+      total,
+      offset: normalizedOffset,
+      limit: normalizedLimit,
+    },
+  })
+}

lib/agent-runtime/task-queue.ts

@@ -1,3 +1,5 @@
+import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"
+import { dirname } from "node:path"
 import { DISTRICTS } from "@/lib/data"
 import type { DistrictId } from "@/lib/types"
 
@@ -23,6 +25,13 @@ export interface QueuedTask {
   deadLetteredAt?: string
 }
 
+export interface DeadLetterTaskEntry {
+  task: QueuedTask
+  failedAt: string
+  errorMessage: string
+  retryCount: number
+}
+
 export interface EnqueueTaskInput {
   id?: string
   type: string
@@ -61,6 +70,9 @@ const PRIORITY_WEIGHT: Record<QueuedTaskPriority, number> = {
 }
 
 const RETRY_BACKOFF_SECONDS = [5, 15, 45, 135] as const
+function getDeadLetterQueuePath(): string {
+  return process.env.TASK_DLQ_FILE ?? "/.data/task-dlq.json"
+}
 
 function nextTaskId(): string {
   queueState.sequence += 1
@@ -72,6 +84,10 @@ export function resetTaskQueueForTests(): void {
   queueState.sequence = 0
 }
 
+export function loadDeadLetterQueueForTests(): void {
+  loadDeadLetterQueue()
+}
+
 export function isQueuedTaskPriority(value: unknown): value is QueuedTaskPriority {
   return value === "critical" || value === "high" || value === "normal" || value === "low"
 }
@@ -91,6 +107,69 @@ export function isValidTaskDistrict(value: unknown): value is DistrictId {
   return typeof value === "string" && DISTRICTS.some((district) => district.id === value)
 }
 
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value)
+}
+
+function toDeadLetterEntry(task: QueuedTask): DeadLetterTaskEntry {
+  return {
+    task,
+    failedAt: task.deadLetteredAt ?? task.updatedAt,
+    errorMessage: task.error ?? "Task failed",
+    retryCount: task.retryCount,
+  }
+}
+
+function persistDeadLetterQueue(): void {
+  const dlqFilePath = getDeadLetterQueuePath()
+  const entries = [...queueState.tasks.values()]
+    .filter((task) => task.status === "dead-letter")
+    .map(toDeadLetterEntry)
+  mkdirSync(dirname(dlqFilePath), { recursive: true })
+  const tempPath = `${dlqFilePath}.${process.pid}.${Date.now()}.tmp`
+  writeFileSync(tempPath, `${JSON.stringify(entries, null, 2)}\n`, "utf8")
+  renameSync(tempPath, dlqFilePath)
+}
+
+function hydrateTask(value: unknown): QueuedTask | null {
+  if (!isRecord(value)) return null
+  if (typeof value.id !== "string" || typeof value.type !== "string") return null
+  if (!isQueuedTaskPriority(value.priority) || value.status !== "dead-letter") return null
+  const now = new Date().toISOString()
+  return {
+    id: value.id,
+    type: value.type,
+    payload: isRecord(value.payload) ? value.payload : {},
+    priority: value.priority,
+    targetAgentId: typeof value.targetAgentId === "string" ? value.targetAgentId : undefined,
+    targetDistrict: isValidTaskDistrict(value.targetDistrict) ? value.targetDistrict : undefined,
+    targetCapability: typeof value.targetCapability === "string" ? value.targetCapability : undefined,
+    retryCount: typeof value.retryCount === "number" ? value.retryCount : 0,
+    maxRetries: typeof value.maxRetries === "number" ? value.maxRetries : 0,
+    createdAt: typeof value.createdAt === "string" ? value.createdAt : now,
+    scheduledFor: typeof value.scheduledFor === "string" ? value.scheduledFor : undefined,
+    status: "dead-letter",
+    result: value.result,
+    error: typeof value.error === "string" ? value.error : undefined,
+    updatedAt: typeof value.updatedAt === "string" ? value.updatedAt : now,
+    deadLetteredAt: typeof value.deadLetteredAt === "string" ? value.deadLetteredAt : undefined,
+  }
+}
+
+function loadDeadLetterQueue(): void {
+  const dlqFilePath = getDeadLetterQueuePath()
+  if (!existsSync(dlqFilePath)) return
+  const parsed: unknown = JSON.parse(readFileSync(dlqFilePath, "utf8"))
+  if (!Array.isArray(parsed)) return
+  for (const entry of parsed) {
+    const taskValue = isRecord(entry) && "task" in entry ? entry.task : entry
+    const task = hydrateTask(taskValue)
+    if (task) queueState.tasks.set(task.id, task)
+  }
+}
+
+loadDeadLetterQueue()
+
 function assertNonEmpty(value: string, field: string): string {
   const trimmed = value.trim()
   if (!trimmed) throw new Error(`${field} is required`)
@@ -165,6 +244,18 @@ export function listDeadLetterTasks(): QueuedTask[] {
   return listTasks({ status: "dead-letter", includeDeadLetter: true })
 }
 
+export function listDeadLetterEntries(options: { offset?: number; limit?: number } = {}): {
+  entries: DeadLetterTaskEntry[]
+  total: number
+  offset: number
+  limit: number
+} {
+  const offset = Math.max(0, Math.floor(options.offset ?? 0))
+  const limit = Math.min(100, Math.max(1, Math.floor(options.limit ?? 50)))
+  const entries = listDeadLetterTasks().map(toDeadLetterEntry)
+  return { entries: entries.slice(offset, offset + limit), total: entries.length, offset, limit }
+}
+
 export function cancelTask(id: string): QueuedTask {
   const task = queueState.tasks.get(id)
   if (!task) throw new Error("Task not found")
@@ -181,6 +272,7 @@ export function failTask(id: string, error: string): QueuedTask {
   if (task.retryCount >= task.maxRetries) {
     const dead = { ...task, status: "dead-letter" as const, error, updatedAt: now.toISOString(), deadLetteredAt: now.toISOString() }
     queueState.tasks.set(id, dead)
+    persistDeadLetterQueue()
     return dead
   }
   const backoffSeconds = RETRY_BACKOFF_SECONDS[Math.min(task.retryCount, RETRY_BACKOFF_SECONDS.length - 1)]
@@ -209,5 +301,15 @@ export function retryDeadLetterTask(id: string): QueuedTask {
     updatedAt: new Date().toISOString(),
   }
   queueState.tasks.set(id, retry)
+  persistDeadLetterQueue()
   return retry
 }
+
+export function discardDeadLetterTask(id: string): QueuedTask {
+  const task = queueState.tasks.get(id)
+  if (!task) throw new Error("Task not found")
+  if (task.status !== "dead-letter") throw new Error("Only dead-letter tasks can be discarded")
+  queueState.tasks.delete(id)
+  persistDeadLetterQueue()
+  return task
+}