Pipeline Studio API (P0a/P0b/P1)

범위: 사용자가 조립·분기·버전관리하는 적재 파이프라인의 작성(authoring) 백엔드(P0a) + 실행 엔진(P0b). 코드 샌드박스(P2)·GUI 빌더(P3)는 후속 단계. 본 문서는 /api/v1/ps/* REST API와 Dagster 인터프리터 실행을 다룬다.

• §1–6 (P0a): 작성·배포·롤백 백엔드 (gend-api).
• §7–10 (P0b): 실행 엔진(dag_executor) + 핸들러 + run 원장 + 트리거/콜백 API + 배포 검증.
• §11 (P1): async fan-out 병렬 · MetadataSource meta.* 조인 · 멀티스토어 retention · stale-run GC.

1. 개요

Pipeline Studio는 비정형 적재 파이프라인을 GUI 없이도 REST로 조립할 수 있게 한다. 파이프라인은 노드(스텝) + 엣지(분기/병렬) 그래프(graph_json)로 표현되며, 편집용 Draft와 배포 시 동결되는 불변 Revision을 분리해 안전한 롤백을 보장한다.

2. 핵심 개념

개념	설명
PipelineDefinition	논리적 파이프라인 컨테이너. current_revision_id로 활성 리비전을 가리킴
Draft	편집 중 그래프(graph_json). 자유 수정 가능
Revision	배포 시점에 동결된 불변 스냅샷. 롤백 = 옛 리비전 활성화
StepLibraryItem	재사용 스텝 정본. shell/python 타입은 approval_status=pending(승인 필요), 그 외 approved
ConnectorCatalogItem	AIM/EAI 등 외부 HTTP 호출 대상(http_api 스텝이 참조). 자격증명은 Vault 참조
MetadataSource	문서↔메타 매핑 테이블 바인딩 (분기 조건 meta.* + 보관기간 출처)

스냅샷-at-deploy (재사용 모델)

라이브러리 스텝은 편집 가능한 "정본"이지만, 배포 시 그 정의가 리비전에 스냅샷된다. 따라서 배포 후 라이브러리를 수정해도 이미 배포된 리비전은 불변 — 재현성·롤백 안전성이 보장된다.

워크스페이스 격리

모든 리소스는 호출자의 tenant_slug → Workspace로 스코프된다. 타 테넌트 리소스 접근은 404(존재 은닉), 미존재 slug는 403. admin/단일테넌트는 unfiltered.

3. 데이터 모델

4. API 레퍼런스 (/api/v1/ps)

모든 엔드포인트는 JWT 인증 필요(미인증 시 401).

4.1 Connector 카탈로그

POST /api/v1/ps/connectors
{ "name": "aim", "base_url": "https://aim.internal/api",
  "auth_vault_ref": "aim/token", "egress_allow": ["aim.internal:443"] }
→ 201 { "id": "...", "name": "aim", "base_url": "...", "egress_allow": [...], "created_at": "..." }

GET /api/v1/ps/connectors → 200 [ ... ]   # 워크스페이스 스코프

• 같은 워크스페이스 내 이름 중복 → 409. (NULL workspace의 admin 항목은 이름 중복 허용)

4.2 스텝 라이브러리

POST /api/v1/ps/steps
{ "name": "std-chunk", "type": "builtin.chunk", "config": {"strategy": "recursive"} }
→ 201 { ..., "approval_status": "approved" }

POST /api/v1/ps/steps  { "type": "shell", ... } → 201 { "approval_status": "pending" }  # 코드 스텝은 승인 필요
GET /api/v1/ps/steps → 200 [ ... ]

4.3 파이프라인 + Draft

POST /api/v1/ps/pipelines  { "name": "doc-kr" }  → 201 { "id": "...", "current_revision_id": null }
GET  /api/v1/ps/pipelines  → 200 [ ... ]          # 워크스페이스 스코프
GET  /api/v1/ps/pipelines/{id}                    → 200 | 404
GET  /api/v1/ps/pipelines/{id}/draft → 200 { "graph": {"nodes": [], "edges": []} }
PUT  /api/v1/ps/pipelines/{id}/draft
{ "graph": { "nodes": [{"node_id":"p","type":"builtin.parse","config":{}},
                        {"node_id":"c","type":"builtin.chunk","config":{}}],
             "edges": [{"from":"p","to":"c"}] } }
→ 200   # 비순환·builtin순서·참조 검증 통과 시. 위반 시 422 (cycle/order/unknown node/duplicate)

4.4 Deploy / Rollback / Revision

POST /api/v1/ps/pipelines/{id}/deploy  { "note": "first" }
→ 201 { "id": "...", "revision_number": 1, "status": "active", ... }
   # 검증 → library_ref 스냅샷 → 불변 Revision 생성, current_revision_id 갱신
   # 미승인 코드 스텝/미존재 library_ref/그래프 무효 → 422, 동시 deploy 경쟁 → 409

POST /api/v1/ps/pipelines/{id}/rollback  { "revision_id": "<old>" }  → 200 (current 되돌림) | 404
GET  /api/v1/ps/pipelines/{id}/revisions/{rid} → 200 { "graph": {...스냅샷...} }

4.5 MetadataSource 바인딩

POST /api/v1/ps/pipelines/{id}/metadata-source
{ "table_ref": "hive.gov.doc_meta", "join_key": "doc_id",
  "field_map": {"retention_days":"보관기간","doc_type":"종류"}, "retention_column": "retention_days" }
→ 201 { ... }

5. 배포·롤백 흐름

6. 배포 검증 (실 prod, 2026-06-04)

P0a 백엔드는 AKS prod(gend-api:0.2.20260604-99a5b49)에 배포·검증됨:

• 라우트 등록 — 라이브 앱에 9개 /api/v1/ps/* 라우트 등록 확인 (connectors · steps · pipelines · draft · deploy · rollback · revisions · metadata-source).
• 스키마 자동 생성 — pod 기동 시 init_db create_all 이 6개 ps_* 테이블 생성 확인 (ps_pipeline_definitions / ps_pipeline_drafts / ps_pipeline_revisions / ps_step_library_items / ps_connector_catalog / ps_metadata_sources).
• 인증 보호 — 미인증 요청 → 401 (404/500 아님 = 라이브 + 보호됨).
• 실 PostgreSQL ORM 검증 — INSERT+SELECT 트랜잭션(롤백)으로 확인: UUID PK 생성 · created_at timezone-aware · JSON 컬럼(egress_allow_json) 라운드트립 · NOT NULL JSON server_default '{}' 적용. (SQLite 단위테스트가 못 잡는 PG 고유 동작 포함)

단위/통합 테스트: 53개(router/service 40 + graph 검증 13) — 크로스테넌트 격리(전 엔드포인트)·스냅샷 불변성·deploy 정확성 포함.

7. 실행 엔진 (P0b) — dag_executor

P0a가 저장만 하던 graph_json을 P0b가 실제 실행한다. 엔진은 인터프리터 op 패턴이다 — 리비전마다 Dagster 그래프를 동적 등록하지 않고, 고정 단일 op dag_executor(고정 job pipeline_studio_exec_job)가 run_config로 받은 graph_json을 위상정렬로 순회하며 조건 분기를 평가하고, 노드 타입별 핸들러를 기존 GenD 서비스로 디스패치한다.

노드 핸들러 (10종)

핸들러 시그니처는 async (node, upstream, ctx) -> dict. 각 핸들러는 기존 서비스를 재사용한다.

노드 타입	동작	출력
builtin.parse	parsers.parse_file	text · metadata · filename
builtin.chunk	chunking.chunk_text	chunks
builtin.embed	주입된 임베더	chunks · embeddings
builtin.vector	벡터 백엔드 upsert(부재 시 graceful-0+warn)	upserted
builtin.graph	그래프 적재(ArangoDB, P1a #1773 실배선)	graph_nodes
http_api	Connector 카탈로그 + httpx(AIM/EAI)	response · 매핑
s3_op	S3 pull(boto3, action="pull")	file_bytes · filename
retention	테이블 모드=RetentionExecutor / 메타 모드(P1, #1895)=MultiStoreRetentionExecutor(벡터·그래프·테이블 파생물, S3 raw 원본 보존)	deleted_rows·by_store·success
shell / python	P2 샌드박스 전까지 NotImplementedError (deploy가 미승인 차단)	—

노드 IO는 위상순으로 흐른다: s3_op → builtin.parse → builtin.chunk → builtin.embed → builtin.vector. 다운스트림 노드는 in-edge 소스 노드 출력을 병합한 upstream dict를 받는다. P1(#1893): 한 노드에서 갈라지는 독립 분기(예: builtin.vector ∥ builtin.graph)는 위상 레벨별 병렬 실행된다 — 아래 §11.1.

조건 분기

edge.condition(문자열)은 안전 제한 평가기(AST 화이트리스트, eval 금지)로 판정한다. 노출 컨텍스트: meta(파서 추출 metadata) · doc_type(확장자) · file(filename/size) · node(노드별 출력). 한 노드는 도달 가능한 in-edge 중 조건 통과가 1개 이상일 때 실행되고, 전부 거짓이면 스킵된다.

run 원장 (격리)

기존 ingestion 6-stage 원장을 generalize하지 않고 신규 격리 테이블 ps_pipeline_runs / ps_pipeline_run_nodes를 둔다(회귀 위험 회피). 노드 콜백은 node_status_json(denormalized) + append-only 이벤트 행을 갱신하고, run 종료 콜백은 succeeded/failed로 terminal 전이(멱등).

8. 실행 API (P0b)

# 실행 트리거 — 활성 리비전의 graph_json 으로 Dagster 잡 launch
POST /api/v1/ps/pipelines/{pipeline_id}/runs
{ "file_path": "s3://bucket/doc.pdf" }
→ 202 { "id": "<run_id>", "status": "running", "dagster_run_id": "..." }
   # 활성 리비전 없음 → 422 (deploy 먼저), 파이프라인 미존재/타테넌트 → 404

# run 원장 조회 — 워크스페이스 fence (타테넌트 → 404)
GET /api/v1/ps/runs/{run_id}
→ 200 { "id": "...", "status": "running|succeeded|failed",
        "node_status_json": {"parse": "ok", "chunk": "ok"}, "failed_node_id": null, ... }

내부 콜백 (Dagster → gend-api, fail-closed)

# JWT 없음 · X-Internal-Token 필수. 빈/불일치 토큰 → 403
POST /api/v1/ps/internal/run-nodes/{run_id}/status   # 노드 단위
{ "node_id": "parse", "status": "running|ok|failed", "error_msg": null }

POST /api/v1/ps/internal/runs/{run_id}/status        # run 종료
{ "status": "succeeded|failed", "error_msg": null }
→ 200 { "ok": true }   # 토큰 미설정/불일치/헤더 누락 → 403, 미존재 run → 404

# P1(#1895) — 멀티스토어 retention 실행 (retention 핸들러 콜백)
POST /api/v1/ps/internal/retention/execute
{ "config": { "meta_source_id": "<uuid>", "retention_column": "retention_days",
              "retention_days": 365, "stores": ["vector","graph","table"], "collection": "DocumentChunks" },
  "workspace_id": "<uuid|null>" }
→ 200 { "deleted_rows": N, "by_store": {"vector":n,"graph":n,"table":n}, "success": true, "errors": [] }

# P1(#1895) — stale-run GC sweep (pipeline_studio_gc_sensor 가 300s 주기 호출)
POST /api/v1/ps/internal/runs/gc
{ "max_age_minutes": 120 }
→ 200 { "reaped": ["<run_id>", ...] }   # running + ended_at IS NULL + started_at < now-N → failed (멱등)

내부 엔드포인트는 _protected_routers 밖(JWT 면제)이지만 X-Internal-Token fail-closed로 게이트된다. 토큰은 Dagster 런타임이 gend-api 와 공유하는 서비스-투-서비스 신뢰.

9. 배포·실행 흐름 (P0b)

10. 배포 검증 (실 prod, 2026-06-05)

P0b는 AKS prod에 배포·검증됨 (gend-api:0.2.20260605-f528dc4 · dagster:0.2.20260605-f528dc4d):

• gend-api — 신규 라우트(/pipelines/:id/runs · /runs/:id · /internal/run-nodes|runs/:id/status) 등록 + 실 PostgreSQL ps_pipeline_runs/ps_pipeline_run_nodes 생성 확인. 내부 콜백 무토큰/오토큰 → 403, run-create 무JWT → 401 (감사 HMAC 체인 기록).
• Dagster (webserver+daemon) — pipeline_studio_exec_job 등록(22 잡 중) + 핸들러 10종 로드 + 이미지 내 gend_api 존재 확인.
• 테스트: pipelines 35 + apps/api run-ledger/trigger/seed 65 + 5차원 적대 리뷰(blocker 2건 수정).

11. P1 확장 (#1893 / #1894 / #1895, 2026-06-07)

P0b 의 "위상순 순차 + 테이블-only retention + GC 부재" 한계를 해소.

11.1 Async fan-out 병렬 실행 (#1893)

run_graph 는 위상 정렬을 레벨(level) 로 분할(_topo_levels, Kahn)하고, 같은 레벨(상호 의존 없음)의 노드를 asyncio.gather 로 동시 실행한다.

• 동시성 상한: asyncio.Semaphore(max(1, max_concurrency)). op 이 GEND_PS_FANOUT_MAX(기본 8) env 로 주입. 0/음수는 deadlock 방지로 1로 클램프.
• fan-in 배리어: 레벨 내 모든 in-flight 핸들러가 끝난 뒤 다음 레벨 진행. 한 노드 실패 시 형제를 취소하지 않고 완료를 기다린 뒤(부분 commit 일관성) to_run 순서상 첫 예외를 re-raise → run-terminal failed.
• 공유 상태 race-free: outputs/cond_ctx/reachable 변이는 gather 바깥(레벨 사이, 직렬)에서만. 조건 평가는 직전 레벨까지 확정된 컨텍스트 스냅샷만 읽어 결정적.
• 비순환 검증 iterative 전환: pipeline_graph._assert_acyclic 가 재귀 DFS → 명시 스택 iterative DFS. 수천 노드 선형 체인 deploy 가 RecursionError 로 깨지지 않음(메시지·예외타입 불변).

11.2 MetadataSource meta.* 조인 풍부화 (#1894)

바인딩만 되던 MetadataSource(§4.5)를 실행 흐름에 배선 — 외부 메타 테이블(예 iceberg.silver.doc_meta)을 join_key 로 조인해 부서/분류 등을 노드 컨텍스트·청크 메타에 주입.

# 실행 트리거에 join_value 추가 (메타 테이블 join_key 에 매칭, 보통 원본 doc_id)
POST /api/v1/ps/pipelines/{pipeline_id}/runs
{ "file_path": "s3://gend-ingestion/doc.pdf", "join_value": "DOC-42" }

흐름: API create_run 이 바인딩된 spec(table_ref/join_key/field_map)+join_value 를 seed 에 주입 → op 이 graph 실행 직전 SELECT <field_map cols> FROM <table_ref> WHERE <join_key>=:join_value LIMIT 1 을 Trino 로 1회 조회 → base_ctx["meta"] 풍부화(명시 seed meta 우선) → 조건 평가기(meta.*, 무변경)·builtin.vector(VectorChunk.metadata)·builtin.graph(GraphDocument.metadata) 가 사용.

• SQL injection 안전: 식별자(table/join_key/cols)는 " quote+doubling, join_value 는 '→'' 리터럴 escape. 전용 테스트(' OR 1=1 --)로 회귀 가드.
• graceful: join_value 미제공 / 메타 행 부재 / gend_api·Trino 실패 → 빈 meta(조건 falsey, 적재는 종전대로). 미바인딩 파이프라인은 1바이트도 동작 불변.

11.3 멀티스토어 retention (#1895)

retention 노드의 메타 모드 — MultiStoreRetentionExecutor 가 만료 문서의 파생물을 스토어별로 정리.

• 만료 식별자: MetadataSource.table_ref 를 WHERE <retention_column> < now - interval 'N' day 로 Trino 조회 → join_key 값을 적재 시와 동일 UUID5 규칙(_ps_document_id)으로 document_id 도출.
• 삭제 대상: vector(Weaviate delete_by_document_id) · graph(Arango delete_document) · table(RetentionExecutor.execute_policy). S3 raw 원본은 절대 삭제하지 않음(보존).
• best-effort(원자성 X): 한 스토어 실패가 다른 스토어를 막지 않고 success=False+errors[] 로 표면화(설정된 sink 실패는 fail-loud). 반환 {deleted_rows, by_store, success, errors}.

11.4 stale-run GC (#1895)

run-terminal 콜백 유실·op 크래시·OOM 으로 status="running"+ended_at IS NULL 로 고착된 orphan run 을 주기 정리.

• pipeline_studio_gc_sensor(minimum_interval_seconds=300, ingestion_retry 패턴 — RunRequest 미발행, 내부 토큰)가 POST /api/v1/ps/internal/runs/gc 호출 → sweep_stale_runs(max_age_minutes=120) 가 대상 run 을 mark_run_finished("failed", …)(terminal-guard 멱등).

11.5 prod 검증 (2026-06-07~08)

양면 배포 — gend-api 자동(docker-build.yml, #1893 pipeline_graph + #1894/#1895 service/router), Dagster 배치 dagster:0.2.20260607-333dbc37(daemon+webserver kubectl set image).

• gend-api: /ps/internal/retention/execute·/ps/internal/runs/gc 라우트 등록 + MultiStoreRetentionExecutor·sweep_stale_runs import OK.
• Dagster: import gend_pipelines.definitions: OK · pipeline_studio_gc_sensor defs.sensors 등록 확인 · daemon/webserver pod 2/2 restarts=0.
• 기능 E2E: daemon→gend-api POST /runs/gc → 200 {"reaped":[]}(sweep 실동작) · 무토큰 /retention/execute → 403(fail-closed). 멀티스토어 retention 실삭제·메타조인·chat registry 키 송신은 운영자 setup(provider/메타소스 등록·바인딩) 후 기능 검증 권장.

12. 제약 & 후속

• P1 완료(§11): async fan-out 병렬(#1893) · 비순환 판정 iterative(#1893) · MetadataSource meta.* 조인(#1894) · 멀티스토어 retention(#1895) · stale-run GC(#1895). builtin.vector/graph 실 backend(Weaviate/Arango)는 P1a(#1773)로 배선·검증됨.
• 잔여: shell/python 샌드박스 실행(P2, Kata) · GUI 빌더(P3) · retention anonymize/archive action(현재 placeholder=delete) · 멀티스토어 retention 실삭제 prod 기능 검증(운영자 메타소스 setup 필요).