/** * A2A Task Manager — Full lifecycle management for A2A tasks. * * State machine: submitted → working → completed | failed | cancelled * * Features: * - UUID v4 task IDs * - In-memory storage with optional SQLite persistence * - Event logging for each state transition * - TTL with configurable expiration (default 5 min) * - Concurrent task limit */ import { randomUUID } from "crypto"; // ============ Types ============ export type TaskState = "submitted" | "working" | "completed" | "failed" | "cancelled"; export interface TaskInput { skill: string; messages: Array<{ role: string; content: string }>; metadata?: Record; } export interface TaskArtifact { type: "text" | "json" | "error"; content: string; } export interface TaskEvent { timestamp: string; state: TaskState; message?: string; } export interface A2ATask { id: string; skill: string; state: TaskState; input: TaskInput; artifacts: TaskArtifact[]; events: TaskEvent[]; metadata: Record; createdAt: string; updatedAt: string; expiresAt: string; /** * GHSA-jcm5-6wpp-wjj8: principal that created the task (hashed API key). * `undefined` = created under the keyless local-first posture — such tasks * stay visible to every caller, matching the pre-owner behavior. Tasks WITH * an owner are only returned/cancelled/listed for the same owner. */ owner?: string; } export interface TaskListFilter { state?: TaskState; skill?: string; limit?: number; offset?: number; } export interface A2ATaskStats { counts: Record; total: number; activeStreams: number; lastTaskAt: string | null; } // ============ Valid Transitions ============ const VALID_TRANSITIONS: Record = { submitted: ["working", "failed", "cancelled"], working: ["completed", "failed", "cancelled"], completed: [], failed: [], cancelled: [], }; // ============ Task Manager ============ export class A2ATaskManager { private tasks = new Map(); private readonly ttlMs: number; private cleanupInterval: ReturnType; private activeStreams = 0; constructor(ttlMinutes: number = 5) { this.ttlMs = ttlMinutes * 60 * 1000; this.cleanupInterval = setInterval(() => this.cleanupExpired(), 60_000); if ( this.cleanupInterval && typeof this.cleanupInterval === "object" && "unref" in this.cleanupInterval ) { (this.cleanupInterval as { unref?: () => void }).unref?.(); } } createTask(input: TaskInput, owner?: string): A2ATask { const now = new Date(); const task: A2ATask = { id: randomUUID(), skill: input.skill, state: "submitted", input, artifacts: [], events: [{ timestamp: now.toISOString(), state: "submitted" }], metadata: input.metadata || {}, createdAt: now.toISOString(), updatedAt: now.toISOString(), expiresAt: new Date(now.getTime() + this.ttlMs).toISOString(), ...(owner !== undefined ? { owner } : {}), }; this.tasks.set(task.id, task); return task; } /** * Owner scoping (GHSA-jcm5-6wpp-wjj8): a task carrying an owner is visible * only to that owner. Ownerless tasks (keyless posture, or created before * this field existed) stay visible to everyone — no behavior change there. */ private isVisibleTo(task: A2ATask, owner?: string): boolean { return task.owner === undefined || task.owner === owner; } getTask(taskId: string, owner?: string): A2ATask | undefined { const task = this.tasks.get(taskId); if (task && new Date(task.expiresAt) < new Date()) { if (task.state === "submitted" || task.state === "working") { this.updateTask(taskId, "failed", undefined, "Task expired"); } } const current = this.tasks.get(taskId); if (!current || !this.isVisibleTo(current, owner)) return undefined; return current; } updateTask( taskId: string, state: TaskState, artifacts?: TaskArtifact[], message?: string ): A2ATask { const task = this.tasks.get(taskId); if (!task) throw new Error(`Task ${taskId} not found`); const valid = VALID_TRANSITIONS[task.state]; if (!valid.includes(state)) { throw new Error(`Invalid transition: ${task.state} → ${state}`); } const now = new Date().toISOString(); task.state = state; task.updatedAt = now; task.events.push({ timestamp: now, state, message }); if (artifacts) task.artifacts.push(...artifacts); return task; } cancelTask(taskId: string, owner?: string): A2ATask { // Owner check BEFORE the mutation (GHSA-jcm5-6wpp-wjj8): a caller must not // cancel another principal's task by id. Uses the same not-found error as // a missing task so an IDOR probe cannot distinguish "exists but not // yours" from "does not exist". const task = this.tasks.get(taskId); if (!task || !this.isVisibleTo(task, owner)) { throw new Error(`Task ${taskId} not found`); } return this.updateTask(taskId, "cancelled", undefined, "Cancelled by client"); } countTasks(filter?: Pick): number { let tasks = [...this.tasks.values()]; if (filter?.state) tasks = tasks.filter((t) => t.state === filter.state); if (filter?.skill) tasks = tasks.filter((t) => t.skill === filter.skill); return tasks.length; } listTasks(filter?: TaskListFilter, owner?: string): A2ATask[] { let tasks = [...this.tasks.values()]; // GHSA-jcm5-6wpp-wjj8: when an owner scope is supplied, owned tasks of // other principals are hidden; ownerless tasks remain visible (posture). if (owner !== undefined) tasks = tasks.filter((t) => this.isVisibleTo(t, owner)); if (filter?.state) tasks = tasks.filter((t) => t.state === filter.state); if (filter?.skill) tasks = tasks.filter((t) => t.skill === filter.skill); tasks.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); const offset = Math.max(0, filter?.offset || 0); const limit = typeof filter?.limit === "number" && Number.isFinite(filter.limit) ? Math.max(1, Math.floor(filter.limit)) : 50; return tasks.slice(offset, offset + limit); } beginStream() { this.activeStreams += 1; } endStream() { this.activeStreams = Math.max(0, this.activeStreams - 1); } getStats(): A2ATaskStats { const counts: Record = { submitted: 0, working: 0, completed: 0, failed: 0, cancelled: 0, }; let lastTaskAt: string | null = null; for (const task of this.tasks.values()) { counts[task.state] += 1; const updatedAt = new Date(task.updatedAt).getTime(); if (!Number.isFinite(updatedAt)) continue; if (!lastTaskAt || updatedAt > new Date(lastTaskAt).getTime()) { lastTaskAt = task.updatedAt; } } return { counts, total: this.tasks.size, activeStreams: this.activeStreams, lastTaskAt, }; } private cleanupExpired() { const now = new Date(); for (const [id, task] of this.tasks) { if ( new Date(task.expiresAt) < now && task.state !== "completed" && task.state !== "failed" && task.state !== "cancelled" ) { task.state = "failed"; task.updatedAt = now.toISOString(); task.events.push({ timestamp: now.toISOString(), state: "failed", message: "TTL expired" }); } // Remove terminal tasks older than 2x TTL if ( ["completed", "failed", "cancelled"].includes(task.state) && now.getTime() - new Date(task.updatedAt).getTime() > this.ttlMs * 2 ) { this.tasks.delete(id); } } } destroy() { clearInterval(this.cleanupInterval); } } // Singleton const globalForA2A = globalThis as unknown as { _a2aTaskManager?: A2ATaskManager }; export function getTaskManager(): A2ATaskManager { if (!globalForA2A._a2aTaskManager) { globalForA2A._a2aTaskManager = new A2ATaskManager(); } return globalForA2A._a2aTaskManager; }