pipeline-service/skill/SKILL.md

name	description
pipeline-service	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