ss-tools/specs/research-task-execution-architecture.md

# Task Execution Architecture — superset-tools

**Date:** 2026-07-02  
**Purpose:** Comprehensive audit of task execution paths — what runs through TaskManager, what bypasses it, and why.  
**Context:** User noted translation tasks are missing from `/reports` (Task Status Center), which only shows tasks from the generic TaskManager pipeline.

---

## 1. TaskManager Pipeline (the unified path)

### Core files

| File | Purpose |
|------|---------|
| `src/core/task_manager/manager.py` | Thin facade composing Graph, EventBus, Lifecycle |
| `src/core/task_manager/graph.py` | In-memory Task registry with CRUD, pagination, filters |
| `src/core/task_manager/lifecycle.py` | State machine: PENDING→RUNNING→SUCCESS/FAILED/WAITING |
| `src/core/task_manager/event_bus.py` | Async log buffer, persistence flush, WebSocket fan-out |
| `src/core/task_manager/context.py` | TaskContext container passed to plugin.execute() |
| `src/core/task_manager/models.py` | Task, TaskStatus, LogEntry, LogFilter (Pydantic) |
| `src/core/plugin_loader.py` | Filesystem-based PluginBase discovery and registration |
| `src/core/plugin_base.py` | ABC PluginBase with id, name, execute, get_schema |
| `src/core/scheduler.py` | APScheduler service (backup, validation, translation jobs) |
| `src/core/async_job_runner.py` | Bridge: sync APScheduler ↔ async event loop |
| `src/dependencies.py` | Singleton factory for TaskManager, PluginLoader, SchedulerService |
| `src/api/routes/tasks.py` | REST API: POST/GET /api/tasks, WebSocket status/logs |

### Pipeline flow

```
POST /api/tasks {"plugin_id": "...", "params": {...}}
  │
  ▼
TaskManager.create_task(plugin_id, params)           [manager.py:306]
  │
  ▼
JobLifecycle.create_task(plugin_id, params)          [lifecycle.py:96]
  ├─ PluginLoader.has_plugin(plugin_id)  →  raise ValueError if missing
  ├─ Task(plugin_id=..., params=..., status=PENDING)
  ├─ TaskGraph.add_task(task)                        [graph.py:125]
  ├─ TaskPersistenceService.persist_task(task)       [lifecycle.py:111]
  └─ returns Task object
  │
  ▼
asyncio.create_task( lifecycle._run_task(task_id) )  [manager.py:314]
  │
  ▼
JobLifecycle._run_task(task_id)                      [lifecycle.py:128]
  ├─ TaskGraph.get_task(task_id)
  ├─ PluginLoader.get_plugin(task.plugin_id)
  ├─ task.status = RUNNING; persisted; broadcast_status
  ├─ Creates TaskContext(task_id, add_log_fn, params) [context.py:70]
  ├─ Inspects plugin.execute() signature:
  │   └─ If accepts `context`: plugin.execute(params, context=context)
  │     If sync: wrapped in asyncio.to_thread()
  │     If async: awaited directly
  ├─ On success: task.result = result; task.status = SUCCESS
  ├─ On failure: task.status = FAILED
  ├─ Finally: task.finished_at, flush_task_logs(), persist_task(), broadcast
  └─ Additional: broadcasts dataset.updated for "dataset-mapper"/"llm_documentation"
```

### TaskStatus values

- `PENDING`, `RUNNING`, `SUCCESS`, `FAILED`, `AWAITING_MAPPING`, `AWAITING_INPUT`

### Registered plugin_id values (PluginBase subclasses)

All discovered by PluginLoader scanning `backend/src/plugins/`. Classes inheriting `PluginBase` are instantiated and registered by their `id` property:

| plugin_id | PluginBase subclass | Source file |
|-----------|-------------------|-------------|
| `superset-backup` | BackupPlugin | `backup.py` |
| `superset-migration` | MigrationPlugin | `migration.py` |
| `search-datasets` | SearchPlugin | `search.py` |
| `dataset-mapper` | MapperPlugin | `mapper.py` |
| `system-debug` | DebugPlugin | `debug.py` |
| `maintenance_banner_apply` | MaintenanceBannerPlugin | `maintenance_banner.py` |
| `git-integration` | GitPlugin | `git_plugin.py` |
| `llm_dashboard_validation` | DashboardValidationPlugin | `llm_analysis/plugin.py` |
| `llm_documentation` | DocumentationPlugin | `llm_analysis/plugin.py` |

---

## 2. Scheduler Service

SchedulerService (`src/core/scheduler.py`) manages three types of jobs via APScheduler:

| Job Type | Trigger Mechanism | Uses TaskManager? |
|----------|------------------|-------------------|
| Backup (`backup_{env_id}`) | `task_manager.create_task("superset-backup")` via AsyncJobRunner.run() | **Yes** |
| Translation (`translate_{schedule_id}`) | Direct call to execute_scheduled_translation() → TranslationOrchestrator | **No** |
| Validation (`validation_{policy_id}`) | `task_manager.create_task("llm_dashboard_validation")` via AsyncJobRunner.run() | **Yes** |

---

## 3. Translation System (standalone, bypasses TaskManager)

Translation tasks use a **separate, parallel execution pipeline**. They never go through `TaskManager.create_task()` or `PluginBase.execute()`.

### Key files

| File | Purpose |
|------|---------|
| `src/plugins/translate/orchestrator.py` | TranslationOrchestrator — run lifecycle coordination |
| `src/plugins/translate/orchestrator_planner.py` | TranslationPlanner — plan generation |
| `src/plugins/translate/orchestrator_runner.py` | TranslationStageRunner — execution, retry, cancel |
| `src/plugins/translate/orchestrator_sql.py` | SQL INSERT orchestrator |
| `src/plugins/translate/scheduler.py` | TranslationScheduler CRUD + execute_scheduled_translation() |
| `src/api/routes/translate/_run_routes.py` | POST /api/translate/jobs/{job_id}/run |
| `src/api/routes/translate/_schedule_routes.py` | Translation schedule CRUD |

### Translation execution flow

```
POST /api/translate/jobs/{job_id}/run                   [_run_routes.py:32]
  │
  ▼
TranslationOrchestrator(db, config_manager, username)   [orchestrator.py:46]
  │
  ▼
TranslationPlanner.plan_run(job_id)                     [orchestrator_planner.py]
  ├─ Creates TranslationRun DB row (status=PENDING)
  └─ Returns TranslationRun object
  │
  ▼
asyncio.create_task( _background_execute() )            [_run_routes.py:123]
  │  (separate DB session, separate orchestrator)
  ▼
TranslationOrchestrator.execute_run(bg_run)             [orchestrator.py:89]
  │
  ▼
TranslationStageRunner.execute_run(run)                 [orchestrator_runner.py:45]
  │
  ▼
TranslationExecutionEngine.execute_run(run)
  ├─ Fetches data from source
  ├─ Creates batches
  ├─ Calls LLM for translation (per-batch)
  ├─ Generates SQL INSERT statements
  ├─ Submits SQL to Superset SQL Lab
  └─ Records results in TranslationRun/TranslationBatch/TranslationRecord DB rows
```

### Translation scheduling flow (also bypasses)

```
SchedulerService.load_schedules()                       [scheduler.py:69]
  ├─ Queries TranslationSchedule table for is_active=True
  └─ scheduler.add_job(
        execute_scheduled_translation,                  [translate/scheduler.py:278]
        CronTrigger(...)
     )
  │
  ▼ (on trigger)
execute_scheduled_translation(schedule_id, job_id, ...)
  ├─ TranslationOrchestrator(db, config_manager, "scheduler")
  ├─ orch.start_run(job_id=job_id, is_scheduled=True)
  ├─ orch.execute_run(run) via AsyncJobRunner.run()
  └─ TranslationRun.status set in DB
```

### Key differences: TaskManager vs Translation

| Feature | TaskManager (PluginBase) | Translation Runs |
|---------|------------------------|------------------|
| State model | Pydantic `Task` in memory (SQL persistence) | SQLAlchemy `TranslationRun` in DB |
| Logger | `TaskContext.logger` → EventBus → WebSocket push | `TranslationEventLog` → DB rows |
| WebSocket | `/ws/logs/{task_id}` (push) + `/ws/task-events` | `/ws/translate/run/{run_id}` (poll, 1s interval) |
| Execution model | `plugin.execute(params, context)` | `TranslationOrchestrator.execute_run(run)` |
| Pause/Resume | Built-in (AWAITING_INPUT, AWAITING_MAPPING) | Not supported |
| Cancellation | `TaskManager.cancel_task()` | `TranslationStageRunner.cancel_run()` |
| Discovery | PluginLoader filesystem scan | Hardcoded orchestrator class |
| Results in `/reports` | Yes | **No** |

---

## 4. Complete Audit: All Execution Paths Bypassing TaskManager

### 🔴 Critical bypasses (full task execution, NOT in TaskManager)

| # | Path | File:Line | Launcher | Work Done | State Tracking |
|---|------|-----------|----------|-----------|----------------|
| **A1** | Manual translation run | `_run_routes.py:123` | `asyncio.create_task(_background_execute())` | LLM translation, SQL generation, Superset API | `TranslationRun` table (SQLAlchemy) |
| **A2** | Scheduled translation run | `translate/scheduler.py:278` | APScheduler → AsyncJobRunner.run() | LLM translation, SQL generation, Superset API | `TranslationRun` table (SQLAlchemy) |

### 🟡 Semi-bypasses (blocking HTTP, could be TaskManager async)

| # | Path | File:Line | Work Done | Notes |
|---|------|-----------|-----------|-------|
| **D1** | Retry failed batches | `_run_routes.py:140` | LLM calls + SQL gen | Blocks HTTP response, no 202 Accepted |
| **D2** | Retry SQL insert | `_run_routes.py:168` | Superset SQL submit | Blocks HTTP response, no 202 Accepted |

### 🟢 Non-task execution paths (should NOT be in TaskManager)

| # | Path | File:Line | Work Done | Reason for staying out |
|---|------|-----------|-----------|----------------------|
| **A3** | Agent LLM title generation | `agent/app.py:488` | `asyncio.create_task(generate_llm_title(...))` | Best-effort, sub-second, non-critical. No business state beyond title text. |
| **B1** | EventBus async flusher | `event_bus.py:72` | Flush log buffer to DB every 2s | Internal TaskManager infrastructure |
| **B2** | TaskLogger fire-forget log writes | `task_logger.py:100` | Async log delivery to EventBus | Internal TaskManager infrastructure |
| **C1-C5** | WebSocket event consumers | `app.py:664,770,824,866,895` | Relay events to browser clients | Event relay, not task execution |
| **E** | Thread pool executors | `utils/executors.py:50` | 3× ThreadPoolExecutor | Infra for blocking I/O offloading |

---

## 5. Impact Summary

```
TaskManager (unified)
├─ backup          ✅
├─ migration       ✅
├─ llm_validation  ✅
├─ llm_documentation ✅
├─ dataset-mapper  ✅
├─ search-datasets ✅
├─ git-integration ✅
├─ maintenance     ✅
├─ debug           ✅
│
└─ translation     ❌  ← bypasses completely (A1 + A2)
     dataset review?  need to verify
```

### If translation were unified:

1. Create `TranslatePlugin extends PluginBase` with `id = "translate-run"` or similar
2. Its `execute(params, context)` method would:
   - Receive `job_id` and `run_id` from params
   - Open its own DB session
   - Call `TranslationOrchestrator(db, ...).execute_run(run)`
   - Report progress via `context.logger` (→ automatic WebSocket push)
3. The scheduler would call `task_manager.create_task("translate-run", {schedule_id, job_id})` instead of `execute_scheduled_translation()`
4. Manual POST would call `task_manager.create_task()` instead of the orchestrator directly
5. Translation tasks would automatically appear in `/reports` with log streaming, status broadcasts, and cancel support

### Key benefit:
- Single API: `POST /api/tasks` for ALL background work
- Single monitoring: `/reports` shows ALL tasks including translations
- Unified WebSocket: push-based logs instead of poll-based
- Elimination of ~200 lines of duplicated concurrency/DB-session/stale-run cleanup code