feat(v2): 알고리즘 분류 N:M 정션 — problems.algorithm → problem_algorithms (계약 v3.0)

docs/integration/db-access-handoff.md

@@ -77,10 +77,10 @@ psql "postgresql://<DB_USER>:<PASSWORD>@127.0.0.1:16380/fly-db"
 
 ---
 
-## 3. 스키마 (3 테이블)
+## 3. 스키마 (4 테이블)
 
 파이프라인이 alembic으로 소유/관리한다(`ipe/v2/db/`). 백엔드는 **읽기 전용으로 가정**하고
-DDL을 변경하지 않는다. 현재 적용 버전: `0003_add_difficulty_column` (head).
+DDL을 변경하지 않는다. 현재 적용 버전: `0004_algorithm_junction` (head).
 
 ```sql
 CREATE TABLE problems (
@@ -92,7 +92,6 @@ CREATE TABLE problems (
     constraints       JSONB       NOT NULL,              -- [{name,min_value,max_value,description}]
     samples           JSONB       NOT NULL,              -- [{input_text,expected_output,description}]
     internal_meta     JSONB       NOT NULL,              -- ⚠️ 내부 전용(응시자 비노출): hidden_algorithm/composition/qa 등
-    algorithm         VARCHAR(64) NULL,                  -- 알고리즘 분류(은닉 코어=시드, 예 'dijkstra'). internal_meta.hidden_algorithm 승격 — 쿼리/필터용. ⚠️ 응시자 비노출 (mig 0002, indexed)
     difficulty        VARCHAR(64) NULL,                  -- BOJ 티어 라벨(예 'Gold IV'). meta.difficulty.label 승격. 옵션(--with-difficulty 켠 경우만 채워짐) (mig 0003, indexed)
     solution_code     TEXT        NULL,                  -- ⚠️ 내부 정해코드(응시자 비노출)
     solution_language VARCHAR(16) NULL,                  -- 예: 'python'
@@ -111,6 +110,14 @@ CREATE TABLE test_cases (
 );
 CREATE INDEX ix_test_cases_problem_id ON test_cases (problem_id);
 
+CREATE TABLE problem_algorithms (                        -- 문제 ↔ 알고리즘 분류 N:M (알고리즘 필터링용)
+    problem_id VARCHAR(36) NOT NULL REFERENCES problems(id) ON DELETE CASCADE,
+    algorithm  VARCHAR(64) NOT NULL,                     -- TargetAlgorithm 값(예 'dijkstra', 19종)
+    role       VARCHAR(16) NOT NULL,                     -- 'core'(은닉 코어) | 'composition'(합성 기법)
+    PRIMARY KEY (problem_id, algorithm)                  -- ⚠️ 응시자 비노출 (내부 분류)
+);
+CREATE INDEX ix_problem_algorithms_algorithm ON problem_algorithms (algorithm);
+
 CREATE TABLE generation_requests (                       -- 생성 감사 로그(백엔드는 보통 불필요)
     idempotency_key VARCHAR(64) PRIMARY KEY,
     seed            VARCHAR(64) NOT NULL,                -- 시드 알고리즘 hint
@@ -125,10 +132,10 @@ CREATE TABLE generation_requests (                       -- 생성 감사 로그
 ```
 
 ### 핵심 규약
-- **응시자 비노출 컬럼**: `internal_meta`, `solution_code`(+`solution_language`),
-  `algorithm`(은닉 코어). 응시자 API로 절대 내보내지 말 것. 채점/검수 내부에서만 사용.
-  `difficulty`(티어 라벨)는 노출 가능하나(백엔드 product 정책) `internal_meta.difficulty`
-  의 reasoning/factors 는 정해 단서가 될 수 있어 비노출.
+- **응시자 비노출**: `internal_meta`, `solution_code`(+`solution_language`),
+  `problem_algorithms` 전체(은닉 코어+합성). 응시자 API로 절대 내보내지 말 것 — 채점/검수
+  /내부 필터에서만 사용. `difficulty`(티어 라벨)는 노출 가능하나(백엔드 product 정책)
+  `internal_meta.difficulty` 의 reasoning/factors 는 정해 단서가 될 수 있어 비노출.
 - **status 생애주기**: 파이프라인은 `draft`로 적재. **`published` 승격은 백엔드 책임**
   (응시자에게는 `published`만 노출 권장).
 - **채점**: `test_cases.input` → 응시자 코드 실행 → stdout과 `expected`를 **줄단위
@@ -159,6 +166,16 @@ FROM test_cases
 WHERE problem_id = :problem_id
 ORDER BY seq;
 
+-- 알고리즘 분류 필터: 'segtree' 가 코어든 합성이든 포함된 published 문제
+SELECT DISTINCT p.id, p.title
+FROM problems p
+JOIN problem_algorithms pa ON pa.problem_id = p.id
+WHERE pa.algorithm = 'segtree' AND p.status = 'published'
+ORDER BY p.title;
+
+-- 한 문제의 알고리즘 구성 (코어 + 합성)
+SELECT algorithm, role FROM problem_algorithms WHERE problem_id = :problem_id;
+
 -- 내부 검수용: 정해코드 + 숨은 알고리즘
 SELECT solution_language, solution_code, internal_meta
 FROM problems

docs/integration/pipeline-deploy.md

@@ -1,6 +1,6 @@
 # 파이프라인 API 서버 배포 가이드 (Slice 2)
 
-계약: [pipeline-service-api-contract.md](pipeline-service-api-contract.md) (v2.0).
+계약: [pipeline-service-api-contract.md](pipeline-service-api-contract.md) (v3.0).
 구현: `ipe/v2/api.py` (`create_app` 팩토리).
 
 ## 로컬 실행 (개발)

docs/integration/pipeline-service-api-contract.md

@@ -1,11 +1,12 @@
-# IPE 파이프라인 ↔ 서비스 백엔드 API 계약 (v2.0)
+# IPE 파이프라인 ↔ 서비스 백엔드 API 계약 (v3.0)
 
-> **문서 상태**: 확정 — 2026-06-22 (v2.0), 파이프라인 측 작성.
+> **문서 상태**: 확정 — 2026-06-22 (v3.0), 파이프라인 측 작성.
 > **대상 독자**: 서비스 백엔드(BOJ 유사) 구현 개발자.
 > **변경 절차**: 본 문서가 계약의 단일 진실원천. 필드 추가는 minor(하위호환),
 > 제거/의미 변경은 major + 양측 합의.
-> **⚠ v2.0 (breaking)**: `mode` enum 이 `hidden`/`direct` → `p1`/`p2` 로 의미 변경(=major).
-> 백엔드 미연동 시점이라 양측 합의 없이 반영 — 기존 v1.0 소비자 없음. 전체 변경이력 §7.
+> **⚠ v3.0 (breaking)**: 알고리즘 분류가 `problems.algorithm` 스칼라 → `problem_algorithms`
+> N:M 정션(코어+합성 전부, role 구분 — §3.4). (v2.0: `mode` `hidden`/`direct`→`p1`/`p2`.)
+> 백엔드 미연동 시점이라 양측 합의 없이 반영 — 소비자 없음. 전체 변경이력 §7.
 
 ---
 
@@ -205,7 +206,9 @@ two_sum, segtree, fenwick, heap, sieve, string_match
 필드 규약:
 
 - **`meta.hidden_algorithm`/`meta.composition` 은 유저에게 절대 노출 금지**
-  (노출 = 은닉·유출 방지 설계 무력화). DB 내부 컬럼으로만.
+  (노출 = 은닉·유출 방지 설계 무력화). 내부 운영 DB 에서는 알고리즘 분류 필터링용으로
+  `problem_algorithms` 정션(코어=role `core`, 합성=role `composition`)에 적재된다(§3.4) —
+  이 역시 내부 전용.
 - **`solution.golden_code` 은 내부 검수·재현용 정해 — 응시자에게 절대 노출 금지**
   (채점은 `test_suite` 줄 단위 exact-match 라 정해 코드 불요; DB 내부 컬럼·감사용).
   `package_version` 무변경 `1.0` 에 **additive** 로 추가 — 기존 백엔드는 무시해도 안전.
@@ -249,6 +252,10 @@ two_sum, segtree, fenwick, heap, sieve, string_match
      constraints jsonb, samples jsonb, internal_meta jsonb, status
      draft|review|published, time_limit_ms, created_at)`
    - `test_cases(problem_id, seq, input text, expected text, category)`
+   - `problem_algorithms(problem_id, algorithm, role)` — 문제↔알고리즘 N:M
+     (PK `(problem_id, algorithm)`). `role` = `core`(은닉 코어) | `composition`(합성 기법),
+     둘 다 §2.4 enum 어휘. **알고리즘 분류 필터링**용(특정 기법이 코어든 합성이든 포함된
+     문제 조회). 응시자 비노출. v3.0 에서 구 `problems.algorithm` 스칼라(코어만)를 대체.
    - `generation_requests(idempotency_key, seed, mode, job_id, final_status,
      attempts, raw_package jsonb, created_at)`
 
@@ -265,7 +272,7 @@ two_sum, segtree, fenwick, heap, sieve, string_match
   attempts 증가).
 - **스키마 소유 = 파이프라인**(alembic): 배포 시 `IPE_DB_URL=… alembic upgrade head`
   로 테이블 생성/이행. 위 §3-4 스키마가 실제 구현(`problems`/`test_cases`/
-  `generation_requests`)이며 SSOT 는 `ipe/v2/db/schema.py`. 서비스 백엔드는 이 테이블을
+  `problem_algorithms`/`generation_requests`)이며 SSOT 는 `ipe/v2/db/schema.py`. 서비스 백엔드는 이 테이블을
   **읽기 전용**으로 소비(특히 `problems.status='draft'` 만, `test_cases` 는 success
   문제만 채점 — §2.5 규약).
 - 매핑: `problems.internal_meta`=패키지 `meta` 전체(내부전용), `solution_code`=정해
@@ -321,6 +328,7 @@ two_sum, segtree, fenwick, heap, sieve, string_match
 
 | 버전 | 일자 | 변경 |
 |---|---|---|
+| **v3.0** | 2026-06-22 | **[breaking]** 알고리즘 분류가 `problems.algorithm` 스칼라(코어 1개) → `problem_algorithms` N:M 정션으로 교체(§3.4). 코어(role `core`) + 합성(role `composition`, §2.1 P2) 전부 행으로 — 백엔드가 합성 기법까지 필터 가능. 패키지(`meta.hidden_algorithm`/`composition`)는 불변, DB 적재 형상만 변경. 백엔드 미연동 시점 반영(소비자 없음). |
 | **v2.0** | 2026-06-22 | **[breaking]** `mode` enum `hidden`/`direct` → `p1`/`p2` (§2.1). 의미 변경=major. `p1`=단일·공개·QA 3종 / `p2`=합성·은닉·QA 4종 — 모드가 합성/은닉/지문/QA관점 4노브를 한 번에 결정. 패키지 `meta.mode`·`meta.composition`·`meta.qa.verdicts` 도 모드에 종속. 백엔드 미연동 시점 반영(기존 v1.0 소비자 없음).<br>**[additive]** `meta.difficulty`(RFC R4 사후 calibration, 옵션 — `--with-difficulty`/`IPE_WITH_DIFFICULTY` 켤 때만) + 직접 적재 DB 의 `problems.algorithm`·`problems.difficulty` 1급 컬럼. 키/컬럼 부재 시 무시 안전(하위호환). |
 | v1.0 | 2026-06-12 | 최초 확정 — 3 엔드포인트(§2), ProblemPackage 스키마(§2.5), 직접 적재 모드(§3.5). |
 

ipe/v2/db/__init__.py

@@ -8,13 +8,20 @@
 """
 
 from .persistence import init_schema, persist_run
-from .schema import generation_requests, metadata, problems, test_cases
+from .schema import (
+    generation_requests,
+    metadata,
+    problem_algorithms,
+    problems,
+    test_cases,
+)
 
 __all__ = [
     "generation_requests",
     "init_schema",
     "metadata",
     "persist_run",
+    "problem_algorithms",
     "problems",
     "test_cases",
 ]

ipe/v2/db/migrations/versions/0004_replace_algorithm_with_junction.py

@@ -0,0 +1,71 @@
+"""replace problems.algorithm scalar → problem_algorithms (N:M 정션)
+
+Revision ID: 0004_algorithm_junction
+Revises: 0003_add_difficulty
+Create Date: 2026-06-22
+
+알고리즘 분류를 단일 스칼라(코어만)에서 N:M 정션으로 교체 — 합성 기법(composition)
+까지 필터 가능하도록(계약 v3.0, breaking). 코어(reduction_core)=role 'core', 합성
+기법=role 'composition'. 둘 다 ``internal_meta``(hidden_algorithm / composition, jsonb)
+에서 백필 후 구 ``problems.algorithm`` 스칼라 컬럼을 제거한다.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0004_algorithm_junction"
+down_revision: str | None = "0003_add_difficulty"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "problem_algorithms",
+        sa.Column(
+            "problem_id",
+            sa.String(36),
+            sa.ForeignKey("problems.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("algorithm", sa.String(64), nullable=False),  # TargetAlgorithm 값
+        sa.Column("role", sa.String(16), nullable=False),  # 'core' | 'composition'
+        sa.PrimaryKeyConstraint("problem_id", "algorithm"),
+    )
+    op.create_index(
+        "ix_problem_algorithms_algorithm", "problem_algorithms", ["algorithm"]
+    )
+    # 백필 — 코어: internal_meta(jsonb).hidden_algorithm (PostgreSQL).
+    op.execute(
+        "INSERT INTO problem_algorithms (problem_id, algorithm, role) "
+        "SELECT id, internal_meta->>'hidden_algorithm', 'core' FROM problems "
+        "WHERE internal_meta->>'hidden_algorithm' IS NOT NULL"
+    )
+    # 백필 — 합성: internal_meta.composition (jsonb array). 코어와 중복이면 skip.
+    op.execute(
+        "INSERT INTO problem_algorithms (problem_id, algorithm, role) "
+        "SELECT id, jsonb_array_elements_text(internal_meta->'composition'), "
+        "'composition' FROM problems "
+        "WHERE jsonb_typeof(internal_meta->'composition') = 'array' "
+        "ON CONFLICT (problem_id, algorithm) DO NOTHING"
+    )
+    # 구 스칼라 컬럼 제거 — 정션이 대체.
+    op.drop_index("ix_problems_algorithm", table_name="problems")
+    op.drop_column("problems", "algorithm")
+
+
+def downgrade() -> None:
+    op.add_column("problems", sa.Column("algorithm", sa.String(64), nullable=True))
+    op.create_index("ix_problems_algorithm", "problems", ["algorithm"])
+    # 코어(role='core')만 스칼라로 복원 — 합성 정보는 internal_meta 에 잔존.
+    op.execute(
+        "UPDATE problems SET algorithm = ("
+        "SELECT pa.algorithm FROM problem_algorithms pa "
+        "WHERE pa.problem_id = problems.id AND pa.role = 'core' LIMIT 1)"
+    )
+    op.drop_index("ix_problem_algorithms_algorithm", table_name="problem_algorithms")
+    op.drop_table("problem_algorithms")

ipe/v2/db/persistence.py

@@ -14,7 +14,13 @@
 from sqlalchemy import insert, select, update
 from sqlalchemy.engine import Connection, Engine
 
-from .schema import generation_requests, metadata, problems, test_cases
+from .schema import (
+    generation_requests,
+    metadata,
+    problem_algorithms,
+    problems,
+    test_cases,
+)
 
 # TL 산정 (계약 §4): time_limit_ms = max(하한, max_golden_elapsed_ms × 배수).
 _TL_MULTIPLIER = 3
@@ -34,10 +40,35 @@ def _time_limit_ms(package: dict[str, Any]) -> int | None:
     return max(_TL_FLOOR_MS, round(float(max_ms) * _TL_MULTIPLIER))
 
 
+def _insert_problem_algorithms(
+    conn: Connection, problem_id: str, meta: dict[str, Any]
+) -> None:
+    """알고리즘 분류 N:M 적재 — 코어(role='core') + 합성(role='composition').
+
+    둘 다 ``internal_meta`` 출처(hidden_algorithm / composition, TargetAlgorithm 어휘).
+    코어가 합성에도 들어가 있으면 중복 PK 회피로 코어 우선 1회만.
+    """
+    rows: list[dict[str, str]] = []
+    seen: set[str] = set()
+    core = meta.get("hidden_algorithm")
+    if core:
+        rows.append({"problem_id": problem_id, "algorithm": core, "role": "core"})
+        seen.add(core)
+    for tech in meta.get("composition") or []:
+        if tech and tech not in seen:
+            rows.append(
+                {"problem_id": problem_id, "algorithm": tech, "role": "composition"}
+            )
+            seen.add(tech)
+    if rows:
+        conn.execute(insert(problem_algorithms), rows)
+
+
 def _insert_problem(conn: Connection, package: dict[str, Any], now: datetime) -> str:
     problem = package["problem"]
     io_contract = problem.get("io_contract") or {}
     solution = package.get("solution") or {}
+    meta = package.get("meta") or {}
     problem_id = str(uuid.uuid4())
     conn.execute(
         insert(problems).values(
@@ -48,12 +79,9 @@ def _insert_problem(conn: Connection, package: dict[str, Any], now: datetime) ->
             output_format=io_contract.get("output_format", ""),
             constraints=problem.get("constraints", []),
             samples=problem.get("sample_testcases", []),
-            internal_meta=package.get("meta", {}),
-            algorithm=(package.get("meta") or {}).get("hidden_algorithm"),
+            internal_meta=meta,
             # difficulty: meta.difficulty.label (사후 calibration, RFC R4) 승격. 미주석이면 None.
-            difficulty=((package.get("meta") or {}).get("difficulty") or {}).get(
-                "label"
-            ),
+            difficulty=(meta.get("difficulty") or {}).get("label"),
             solution_code=solution.get("golden_code"),
             solution_language=solution.get("language"),
             status="draft",
@@ -76,6 +104,7 @@ def _insert_problem(conn: Connection, package: dict[str, Any], now: datetime) ->
                 for i, c in enumerate(cases)
             ],
         )
+    _insert_problem_algorithms(conn, problem_id, meta)
     return problem_id
 
 

ipe/v2/db/schema.py

@@ -1,11 +1,12 @@
-"""DB 스키마 SSOT — 정규화 3 테이블 (계약 §3, 파이프라인 소유 DDL).
+"""DB 스키마 SSOT — 정규화 4 테이블 (계약 §3, 파이프라인 소유 DDL).
 
 JSON 컬럼은 ``JSON().with_variant(JSONB, "postgresql")`` 로 포터블: 운영 PostgreSQL
 에서는 jsonb, 단위테스트 sqlite 에서는 일반 JSON(TEXT) 로 매핑된다.
 
 - ``problems``: 출하 문제 1건 (지문+io+제약+샘플+정해+내부메타+TL). status 는
   ``draft``/``review``/``published`` — 파이프라인은 ``draft`` 로 적재, 승격은 백엔드.
 - ``test_cases``: 채점셋 케이스 (problem_id FK, 줄단위 exact-match 채점용).
+- ``problem_algorithms``: 문제 ↔ 알고리즘 N:M (코어+합성, role 구분 — 백엔드 분류 필터).
 - ``generation_requests``: 생성 요청 감사 로그 (idempotency_key PK, raw_package 원문).
 """
 
@@ -41,12 +42,8 @@
     Column("samples", _JSON, nullable=False),  # [{input_text,expected_output,...}]
     # internal_meta: hidden_algorithm/composition/qa 등 — 응시자 비노출 (계약 §2.5)
     Column("internal_meta", _JSON, nullable=False),
-    # algorithm: 문제의 알고리즘 분류(은닉 코어 = 시드, 예: 'dijkstra'). internal_meta.
-    # hidden_algorithm 과 동일값을 쿼리·필터·집계 편의를 위해 1급 컬럼으로 승격(응시자
-    # 비노출 — 내부 운영 DB). 적재 시 필수 기록.
-    Column("algorithm", String(64), nullable=True, index=True),
     # difficulty: BOJ 티어 라벨(예: 'Gold IV'). meta.difficulty.label 에서 승격 — 쿼리·
-    # 필터·집계 편의로 1급 컬럼(algorithm 과 동형). 사후 calibration(RFC R4) 산출,
+    # 필터·집계 편의로 1급 컬럼(승격 패턴). 사후 calibration(RFC R4) 산출,
     # 응시자 비노출(internal). 전체 report 는 internal_meta.difficulty 에 보존.
     Column("difficulty", String(64), nullable=True, index=True),
     Column("solution_code", Text, nullable=True),  # 내부 정해 (응시자 비노출)
@@ -73,6 +70,22 @@
     Column("category", String(64), nullable=True),
 )
 
+# 문제 ↔ 알고리즘 분류 N:M (계약 v3.0). 코어(reduction_core)=role 'core' 1행 +
+# 합성(composition)=role 'composition' N행. 둘 다 TargetAlgorithm(19종) 어휘.
+# 백엔드 알고리즘 필터링용 — 구 problems.algorithm 스칼라(코어만)를 대체. 응시자 비노출.
+problem_algorithms = Table(
+    "problem_algorithms",
+    metadata,
+    Column(
+        "problem_id",
+        String(36),
+        ForeignKey("problems.id", ondelete="CASCADE"),
+        primary_key=True,
+    ),
+    Column("algorithm", String(64), primary_key=True, index=True),  # TargetAlgorithm 값
+    Column("role", String(16), nullable=False),  # 'core' | 'composition'
+)
+
 generation_requests = Table(
     "generation_requests",
     metadata,

ipe/v2/difficulty.py

@@ -224,7 +224,7 @@ def _row_to_package(row: Any) -> dict[str, Any]:
             "language": row["solution_language"],
         },
         "meta": {
-            "hidden_algorithm": row["algorithm"],
+            "hidden_algorithm": meta.get("hidden_algorithm"),
             "composition": meta.get("composition") or [],
         },
     }