---
name: task_manager
kind: function
lang: py
domain: core
version: "1.0.0"
purity: impure
signature: "class TaskManager: create_task(task_type: str, metadata: dict | None = None) -> str; get_task(task_id: str) -> Task | None; update_task(task_id: str, ...) -> None; complete_task(task_id: str, result: dict) -> None; fail_task(task_id: str, error: str) -> None; list_tasks(task_type: str | None = None) -> list[dict]; cleanup_old_tasks(max_age_hours: int = 24) -> None"
description: "Gestor de tareas background en memoria, thread-safe, con patron singleton. Permite crear, actualizar, completar y limpiar tareas de larga duracion. Patron tipico: hilo background actualiza progreso via update_task(), frontend hace polling via get_task()."
tags: [task, manager, singleton, thread-safe, background, async, polling, progress]
uses_functions: []
uses_types: [task_py_core, task_status_py_core]
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [threading, uuid, datetime]
params: []
output: "Instancia singleton thread-safe para gestionar tareas de larga duración"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/core/task_manager.py"
---

## Ejemplo

```python
manager = TaskManager()

# Crear tarea
task_id = manager.create_task("graph_build", metadata={"nodes": 500})

# Hilo background actualiza progreso
manager.update_task(task_id, status=TaskStatus.PROCESSING, progress=25, message="Cargando datos...")
manager.update_task(task_id, progress=75, message="Construyendo grafo...")
manager.complete_task(task_id, result={"graph_id": "abc-123"})

# Frontend hace polling
task = manager.get_task(task_id)
print(task.to_dict())
# {"task_id": "...", "status": "completed", "progress": 100, ...}

# Listar todas las tareas de un tipo
tasks = manager.list_tasks(task_type="graph_build")

# Limpiar tareas antiguas (> 24h)
manager.cleanup_old_tasks(max_age_hours=24)
```

## Notas

Singleton con double-checked locking: la primera instancia se crea una sola vez aunque multiples hilos llamen a `TaskManager()` simultaneamente.

El lock `_tasks_lock` protege el diccionario interno `_tasks` en todas las operaciones de lectura y escritura.

`update_task()` solo actualiza los campos que recibe como no-None, lo que permite actualizaciones parciales sin sobreescribir el estado existente.

`cleanup_old_tasks()` solo elimina tareas en estado terminal (COMPLETED o FAILED). Las tareas PENDING o PROCESSING no se eliminan independientemente de su edad.

El `error_type` referencia `error_go_core` como convencion del registry — los errores reales en Python se lanzan como excepciones, no se retornan.
