feat: add DAG engine issues (0007a-e) and feature flag

Desglose del sistema de orquestacion propio para reemplazar Dagu: - 0007a: core puro (parse, validate, topo sort) - 0007b: process manager (spawn, wait, kill) - 0007c: execution store (SQLite) - 0007d: scheduler (cron parser, ticker) - 0007e: app CLI que compone todo Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-08 01:18:19 +02:00
parent 5c712bb974
commit d9b448a07b
7 changed files with 641 additions and 0 deletions
@@ -0,0 +1,115 @@
+# 0007c — Execution store: persistencia de estado
+
+## Metadata
+
+| Campo | Valor |
+|-------|-------|
+| **ID** | 0007c |
+| **Estado** | pendiente |
+| **Prioridad** | alta |
+| **Tipo** | feature |
+
+## Dependencias
+
+| ID | Título | Estado | Requerido |
+|----|--------|--------|-----------|
+| 0007a | Funciones core del DAG engine | pendiente | Si |
+
+**Bloqueada por:** `#0007a`
+
+**Desbloquea:** `#0007e`
+
+---
+
+## Objetivo
+
+Funciones para persistir el estado de ejecuciones de DAGs en SQLite: que DAG se ejecuto, cuando, que steps corrieron, resultado de cada step, logs. Permite historial, reintentos y debugging.
+
+## Contexto
+
+- Cada ejecucion de un DAG genera un `run` con multiples `step_results`
+- Similar a `operations.db` pero especifico para el DAG engine
+- La BD vive en el directorio de la app del scheduler (no en raiz)
+- Debe soportar consultas tipo: "ultimas 10 ejecuciones de X", "steps fallidos"
+
+## Arquitectura
+
+```
+functions/infra/
+├── dag_store_init.go       — NEW: crea schema SQLite para runs/steps
+├── dag_store_init.md
+├── dag_store_run.go        — NEW: CRUD de runs (create, update status, list, get)
+├── dag_store_run.md
+├── dag_store_step.go       — NEW: CRUD de step results dentro de un run
+├── dag_store_step.md
+
+types/infra/
+├── dag_run.md              — NEW: id, dag_name, status, started_at, finished_at, trigger
+├── dag_step_result.md      — NEW: run_id, step_name, status, exit_code, stdout, stderr, duration_ms
+```
+
+### Patron pure core / impure shell
+
+- `infra/` — Todas impuras (SQLite I/O)
+- Schema sencillo: `dag_runs` + `dag_step_results`
+
+## Tareas
+
+### Fase 1: Tipos y schema
+
+- [ ] **1.1** Definir `DagRun` — id (ULID), dag_name, dag_path, status (pending/running/success/failed/cancelled), started_at, finished_at, trigger (manual/schedule/api)
+- [ ] **1.2** Definir `DagStepResult` — id, run_id, step_name, status, exit_code, stdout, stderr, started_at, finished_at, duration_ms
+- [ ] **1.3** Schema SQLite: `dag_runs`, `dag_step_results` con indices
+
+### Fase 2: Funciones
+
+- [ ] **2.1** `dag_store_init` — crea/migra la BD SQLite
+- [ ] **2.2** `dag_store_run` — create_run, update_run_status, get_run, list_runs (con filtros)
+- [ ] **2.3** `dag_store_step` — insert_step_result, list_steps_for_run
+
+### Fase 3: Tests
+
+- [ ] **3.1** Ciclo completo: init → create run → insert steps → update status → query
+- [ ] **3.2** Queries: ultimas N ejecuciones, ejecuciones fallidas, steps de un run
+
+### Fase 4: Cleanup
+
+- [ ] `fn index` y verificar IDs
+
+---
+
+## Ejemplo de uso
+
+```go
+db := dag_store_init("/path/to/scheduler.db")
+
+run := dag_store_create_run(db, "my_pipeline", "manual")
+// run.ID = "01HXZ..."
+
+dag_store_insert_step(db, run.ID, DagStepResult{
+    StepName:   "fetch_data",
+    Status:     "success",
+    ExitCode:   0,
+    Stdout:     "fetched 1000 rows",
+    DurationMs: 1234,
+})
+
+dag_store_update_status(db, run.ID, "success")
+
+// Consultar
+runs := dag_store_list_runs(db, "my_pipeline", 10) // ultimas 10
+```
+
+## Decisiones de diseno
+
+- **SQLite por DAG engine, no por DAG**: una sola BD para todas las ejecuciones, no una por cada DAG
+- **ULID para run IDs**: ordenables por tiempo, unicos sin coordinacion
+- **stdout/stderr en BD**: para steps cortos. Para output grande, guardar path a archivo de log
+
+## Criterios de aceptacion
+
+- [ ] Schema se crea correctamente
+- [ ] CRUD completo funciona
+- [ ] Queries con filtros funcionan
+- [ ] Tests pasan
+- [ ] Indexado en registry.db