---
name: pipeline-service
description: "Pipeline execution engine — task scheduling, step DAG execution, human task interaction, and LLM bridging."
---

# pipeline_service

Core execution engine that drives pipeline task lifecycle: scheduling, DAG step execution, human-in-the-loop interaction, and LLM integration.

## Architecture

```
pipeline_service/
├── executor.py            # Task/step scheduling engine
├── storage.py             # Database CRUD (pipeline_tasks, task_steps, artifacts)
├── llm_bridge.py          # Unified LLM API call interface
├── agent_loop.py          # AI Agent multi-turn execution loop
├── human.py               # Human task interaction (approval, input)
├── intent_classifier.py   # Natural language intent recognition
├── state.py               # Task/step state machine
├── step_registry.py       # Step type handler registry
└── init.py                # ServerEnv registration
```

## Data Model

| Table | Purpose |
|-------|---------|
| `pipeline_tasks` | Task instances (id, pipeline_id, status, version, params) |
| `pipeline_task_steps` | Step execution records (task_id, step_name, state, input, output) |
| `pipeline_artifacts` | Step input/output artifacts |
| `pipeline_human_tasks` | Human-in-the-loop tasks (approval, input forms) |
| `pipeline_step_types` | Registered step type handlers |

## Key Functions (registered via ServerEnv)

- `submit_task()` — Create and start a new task
- `get_task_detail()` — Task state + all steps
- `get_task_steps()` — Steps for a task
- `control_task()` — Pause/resume/cancel
- `restart_task()` — Restart completed/failed task
- `list_tasks()` — Task list with filters
- `llm_call()` — LLM API call
- `call_llm()` — SDLC handler interface (delegates to llm_call)

## Pitfalls

- **DBPools init**: Must check `db.databases` and load from `config.databases` if empty. DBPools() is NOT a singleton.
- **sor.R() → sqlExe**: Never use `sor.R('table', where, sort)` 3-arg. Use `sqlExe("SELECT ... WHERE ... ORDER BY", params)`.
- **dir() → vars()**: Use `vars(rec)` not `dir(rec)` for attribute iteration — `dir()` includes methods.
- **dict access → getattr**: sqlor returns DictObject, not dict. Use `getattr(rec, 'field')` not `rec['field']`.
- **pipeline_service must be pip installed**: After git pull on server, run `pip install --upgrade`.
SKILLEOF