Flow Manager

Назначение

Сервис создания, хранения, версионирования и исполнения процессных пайплайнов (flows). Оркестрация пайплайнов с поддержкой Starlark-скриптов, in-memory тестового запуска и real-time стриминга событий через NATS.

Порты

• HTTP: 8099
• БД: flow_manager_db

Ключевые возможности

• CRUD пайплайнов с контролем доступа по владельцу
• Версионированные шаблоны — каждое сохранение создаёт иммутабельную версию
• Модель графа: DAG из нод, связанных рёбрами (edges)
• In-memory тестовый запуск (без записи в БД) с потоковой передачей событий через NATS
• Production-запуск с сохранением в БД и отслеживанием статуса
• Тестирование отдельных нод (code, job) через /try/...
• Ограничение параллельных тестовых запусков: максимум 50 одновременных
• Starlark-исполнитель: print(), setVar(), getVar(), enumerate(), fetch(), b64encode() / b64decode(), json_encode() / json_decode(), setupProcess(), runProcess(), общий ctx
• Артефакты: глобальный модуль artefacts — загрузка, извлечение, создание ZIP-архивов, сохранение в песочницу
• Файловая система: глобальный модуль files — чтение, запись, удаление файлов во временном каталоге, экспорт в песочницу
• Двойная аутентификация: JWT (пользователи) и Service Token (сервисы)
• Swagger UI

Статусы инстансов

Статус	Описание
idle	Создан, ожидает запуска
pending	Подготовка к запуску
running	Выполняется
waiting	Пауза на waitNode (delay/until/event); resume через таймер, NATS или POST /pipelines/run/:id/resume
completed	Успешно завершён
failed	Завершён с ошибкой
cancelled	Отменён

Типы нод

Control

Нода	Описание
startNode	Точка входа пайплайна. Не выполняет никакой логики, только помечает начало выполнения.

Input

Нода	Описание
inputButtonNode	Вход по нажатию кнопки. В тестовом режиме — pass-through (выполняется без логики).
inputEventNode	Триггер: старт instance по NATS subject (data.subject). Payload → ctx.trigger.
inputFormNode	Вход по отправке формы. В тестовом режиме — pass-through.
inputHttpRequestNode	Триггер: POST /api/v1/hooks/pipelines/:id (+ X-Hook-Secret или ?secret=). Async 202 {instance_id}; payload → ctx.trigger (body, json, method, query, опционально headers).

Display

Нода	Описание
displayTextNode	Текст-аннотация на canvas (runtime no-op / лог).
displayMediaNode	Image/audio/video на canvas (url или file_id).
logMessageNode	Пишет message (с {{var}}) в логи ноды.

Transform

Нода	Описание
transformCodeNode	Выполняет Starlark-скрипт (Python-подобный язык) над общим контекстом. Доступные функции описаны ниже в разделе transformCodeNode.
transformSetVariableNode	Устанавливает одну или несколько переменных в контексте из JSON-источника (data.source).
transformExtractArtifactNode	Извлекает артефакты из выполненного jobNode. В тестовом режиме — pass-through.
transformMergeProcessNode	Синхронизация: ожидает завершения всех параллельных веток runProcess перед продолжением.
transformCollectProcessNode	Как merge, но собирает из каждой ветки только указанные ключи ctx в список (output_key, по умолчанию process.results).
jsonPathNode	Извлекает поля из всего ctx (engine: jsonpath|jq, выражение от корня контекста).

Logic

Нода	Описание
logicConditionNode	Вычисляет Starlark-выражения как условия и направляет выполнение на соответствующий выходной порт (input-N). Если ни одно условие не выполнено — переход на порт error.
logicSwitchNode	Узел маршрутизации/переключения. В тестовом режиме — pass-through.
waitNode	Пауза: mode=delay|until|event. Production — park (waiting) + resume; test — blocking delay (cap 30s).

Platform

Нода	Описание
createTaskNode / updateTaskNode / changeTaskStatusNode	Task Tracker CRUD / status transition.
createDocumentNode	Создаёт документ.
sendNotificationNode	POST notifications /send по шаблону.
runSubflowNode	Nested Prepare+Start другого пайплайна; depth ≤ 3.

Network

Нода	Описание
httpRequestNode	Исходящий GET/POST с allowlist/SSRF (WEB_DOMAIN_ALLOWLIST).
fetchUrlNode	GET URL → текст.
webSearchNode	Поиск (SearXNG и т.п.) без Agent.

Output

Нода	Описание
outputHttpNode	HTTP-выход (отправка ответа). В тестовом режиме — pass-through.
outputEmailNode	Email-выход (отправка письма). В тестовом режиме — pass-through.
cliAgentNode	Экспорт сгенерированного артефакта или файла из контекста на один из подключенных CLI Агентов (agent_id, export_path, output_key).

Сервисы

Нода	Описание
serviceNode	Универсальный коннектор из каталога integration_services. Автор заполняет только inputs. HTTP/SMTP/IMAP и extract в ctx скрыты в YAML.

Сырой HTTP остаётся в Network (httpRequestNode).

Job

Нода	Описание
jobNode	Создаёт и мониторит задачу через API command-manager. Ожидает завершения (таймаут 5 минут, интервал опроса 2 сек). Собирает артефакты по завершении.

AI

Нативные ноды через AiRouterClient (формы в UI). Для кастомной логики по-прежнему доступен Code-нода и up.ai_router в Starlark.

Нода	Описание
aiChatCompletionNode	Chat completion: prompt → ai.response (+ ai.usage). Поддержка {{var}} из ctx. Перед вызовом — Guard egress scan (Authorization: Service).
aiRagSearchNode	Поиск по knowledge base (scopes + query) → ai.results.
aiRagIngestNode	Индексация документа в scope → ai.document_id, ai.chunk_count.
aiAgentNode	execution_selector=direct сохраняет синхронный POST /agent/run; published_agent принимает agent_slug и optional numeric version, вызывает AI Router /agents/:slug/runs и возвращает ai.run_id. Flow Manager не создаёт Command Manager job напрямую. Guard egress scan сохраняется.
aiClassifyNode	Классификатор (Bosun / capability=decision): POST /api/v1/decide → ai.answers, ai.decision. Guard egress scan.
aiSummarizeNode	Суммаризация: extractive / llm / hybrid. Поля content, method, strategy (head_tail|head|tail), max_chars, threshold, instructions, provider/model. → ai.summary (+ ai.summarize meta). Guard egress scan.

Палитра AI больше не предлагает отдельный узел Providers (aiProvidersNode остаётся в runtime для старых графов, в UI не создаётся).

При ошибке ноды пишут сообщение в ai.error и падают как failed.

Retry

Нода	Описание
retryNode	Останавливает ветку после N проходов через ноду (лимит в data.retries).

---

Контекст выполнения

Каждый пайплайн имеет общий контекст (execution context) — хранилище переменных, передаваемое между нодами. При переходе к следующей ноде контекст клонируется, что позволяет ветвление с изолированными состояниями.

Автоматические переменные

Переменная	Описание
artefacts.previous	Список ID артефактов от последнего jobNode
artefacts.all	Все артефакты, накопленные с начала пайплайна
ai.response	Текст ответа AI Chat / Agent
ai.usage	Usage tokens
ai.results	Результаты RAG Search
ai.document_id / ai.chunk_count	Результат RAG Ingest
ai.answers / ai.decision	Ответ Классификатора (decide)
ai.summary	Текст после aiSummarizeNode
ai.summarize	Мета суммаризации (method, original_chars, result_chars, …)
ai.providers / ai.provider	Список / обновлённый провайдер (legacy aiProvidersNode)
ai.tool_calls	Вызовы tools от Agent
ai.error	Ошибка последней AI-ноды

Рабочая директория

Каждый инстанс пайплайна получает временную директорию ARTIFACTS_TMP_DIR/<instance_id>. Директория удаляется после выполнения ноды. Модули files и artefacts работают только внутри этой директории.

---

transformCodeNode

Нода выполняет Starlark-скрипт (Python-подобный sandboxed-язык) в контексте текущего выполнения пайплайна. Скрипт имеет доступ к общему контексту (ctx), глобальным модулям (files, artefacts) и набору встроенных функций (print, setVar/getVar, enumerate, fetch, b64encode/b64decode, json_encode/json_decode, setupProcess/runProcess).

Полное описание всех функций, контекста и хелперов — в Starlark API.

Диаграмма зависимостей (редактор)

Кнопка «Зависимости» в шапке редактора открывает майнд-карту зависимостей текущего процесса (read-only). Строится клиентом из узлов шаблона, без отдельного API:

• группы по типам узлов: jobNode.data.id → Команды, envNode.data.env_id → Окружения, serviceNode.data.service_id → Сервисы, runSubflowNode.data.pipeline_id → Подпроцессы, provider_id (AI-узлы) → AI-провайдеры;
• имена резолвятся лениво через публичные API соответствующих сервисов (учитывается доступность сервиса в инсталляции);
• связи «команда → окружение» выводятся из props команды;
• клик по узлу открывает сущность (команда / окружение / подпроцесс) в новой вкладке;
• доступен экспорт карты в Mermaid mindmap.

WebSocket Events

Flow Manager публикует события в NATS. Gateway подписывается на эти subjects и транслирует их клиентам через WebSocket.

• Тестовые запуски: pipeline.test.<pipe_uuid>.node.*
• Production Run: pipeline.run.<instance_id>.node.*
• Платформенные: job.created, job.changed, executor.status, runner.change

Полное описание всех событий с пейлоадами — в WebSocket Events.

Конфигурация

Переменная	По умолчанию	Описание
PORT	8099	Порт сервера
HOST	0.0.0.0	Адрес сервера
GIN_MODE	debug	Режим Gin (debug/release)
DB_HOST	localhost	Адрес PostgreSQL
DB_PORT	5432	Порт PostgreSQL
DB_USER	flow_manager_user	Пользователь PostgreSQL
DB_PASSWORD	—	Пароль PostgreSQL
DB_NAME	flow_manager_db	Имя базы данных
DB_SSL_MODE	disable	Режим SSL
JWT_SECRET	—	Секрет для JWT
JWT_ALGORITHM	HS256	Алгоритм JWT
JWT_TTL	24h	Время жизни JWT
SERVICE_TOKEN	—	Токен межсервисного взаимодействия
MESSAGING_URL	nats://localhost:4222	URL NATS
ARTIFACTS_TMP_DIR	./tmp/flow-manager	Каталог для временных файлов артефактов
SANDBOX_DIR	./sandbox	Постоянный каталог для files.export() и artefacts.save() (структура: {SANDBOX_DIR}/{pipeline_id}/)
TEST_PIPELINE_TTL	10m	Время жизни тестового пайплайна в памяти
LOG_LEVEL	info	Уровень логирования
LOG_FORMAT	json	Формат логов
INTEGRATION_PACK_KEY	—	Ключ AES-256-GCM для .usint (32 байта или base64; иначе SHA-256)
INTEGRATION_PACK_SIGNING_KEY	—	Опциональный HMAC подписи пакета
INTEGRATIONS_REGISTRY_URL	https://unishift.ru/api/integrations	Публичный каталог пакетов

Интеграции («Сервисы»)

Каталог коннекторов в таблице integration_services (одна YAML на строку). Список для редактора — только UI-поля (slug, name, icon, inputs).

Метод	Путь	Назначение
GET	/api/v1/integration-services	UI-каталог
GET	/api/v1/integration-services/:slug	Поля формы
GET	/api/v1/integration-services/registry	Прокси каталога unishift.ru
POST	/api/v1/integration-services/import	Импорт .usint (admin)
POST	/api/v1/integration-services/install-from-registry	Скачать и импортировать slug

Set A — seed YAML в миграции 007_integration_services. Set B — packs/usint/<pack-id>/ в корне репозитория → flow-manager pack seal … -o *.usint.

Формат .usint: magic USINT1, version, plaintext length, nonce 12 байт, AES-256-GCM. Сайт: HTML /integrations, API GET /api/integrations (+ /:slug, /:slug/download). Шаблоны: {{inputs.key}}, {{ctx.key}}, {{env.KEY}}. Секреты и свой хост — в Настройки → Окружения (env-manager), затем process env. Список API отдаёт env_keys (имена, не значения).

Полный каталог GitLab, требования к токену, параметры всех нод и webhook-рецепты: GitLab-интеграция. Входящий GitLab webhook принимается HTTP-триггером, после чего workflow фильтрует событие и запускает нужные service-ноды.

Сценарии интеграции с Concierge

Flow Manager выступает оркестратором внешней коммуникации Concierge:

1. Входящие нотификации в Concierge:
   • Вебхук внешней системы принимается в ноде inputHttpRequestNode или готовом коннекторе serviceNode.
   • Пайплайн трансформирует данные и вызывает POST /api/v1/external/notifications в сервисе Concierge с ServiceTokenAuth.
2. Обратный форвардинг ответов:
   • Ответы пользователей/модераторов публикуются в NATS на тему unishift.concierge.message_replied.
   • В пайплайн подключается inputEventNode с подпиской на этот subject, после чего httpRequestNode или serviceNode отправляет отклики в целевую CRM/банк.
3. Provisioning чатов и партнёров:
   • Нода вызова HTTP в пайплайн вызывается на POST /api/v1/external/provision-chat для автоматического онбординга новых клиентов и создания чатов поддержки.