diff --git a/docs/integration/db-access-handoff.md b/docs/integration/db-access-handoff.md index b105966..84a1040 100644 --- a/docs/integration/db-access-handoff.md +++ b/docs/integration/db-access-handoff.md @@ -77,10 +77,10 @@ psql "postgresql://:@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 diff --git a/docs/integration/pipeline-deploy.md b/docs/integration/pipeline-deploy.md index a7fc5b1..f4cd1ef 100644 --- a/docs/integration/pipeline-deploy.md +++ b/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` 팩토리). ## 로컬 실행 (개발) diff --git a/docs/integration/pipeline-service-api-contract.md b/docs/integration/pipeline-service-api-contract.md index 3fae182..e898e45 100644 --- a/docs/integration/pipeline-service-api-contract.md +++ b/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 소비자 없음).
**[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). | diff --git a/ipe/v2/db/__init__.py b/ipe/v2/db/__init__.py index 0d41c88..9dc579e 100644 --- a/ipe/v2/db/__init__.py +++ b/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", ] diff --git a/ipe/v2/db/migrations/versions/0004_replace_algorithm_with_junction.py b/ipe/v2/db/migrations/versions/0004_replace_algorithm_with_junction.py new file mode 100644 index 0000000..d2165ee --- /dev/null +++ b/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") diff --git a/ipe/v2/db/persistence.py b/ipe/v2/db/persistence.py index 0fa2c56..59931d0 100644 --- a/ipe/v2/db/persistence.py +++ b/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 diff --git a/ipe/v2/db/schema.py b/ipe/v2/db/schema.py index 8c3cc3d..097416e 100644 --- a/ipe/v2/db/schema.py +++ b/ipe/v2/db/schema.py @@ -1,4 +1,4 @@ -"""DB 스키마 SSOT — 정규화 3 테이블 (계약 §3, 파이프라인 소유 DDL). +"""DB 스키마 SSOT — 정규화 4 테이블 (계약 §3, 파이프라인 소유 DDL). JSON 컬럼은 ``JSON().with_variant(JSONB, "postgresql")`` 로 포터블: 운영 PostgreSQL 에서는 jsonb, 단위테스트 sqlite 에서는 일반 JSON(TEXT) 로 매핑된다. @@ -6,6 +6,7 @@ - ``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, diff --git a/ipe/v2/difficulty.py b/ipe/v2/difficulty.py index 2d723d4..51d7662 100644 --- a/ipe/v2/difficulty.py +++ b/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 [], }, } diff --git a/tests/v2/db/test_persistence.py b/tests/v2/db/test_persistence.py index 47f2bf2..78c1c37 100644 --- a/tests/v2/db/test_persistence.py +++ b/tests/v2/db/test_persistence.py @@ -12,7 +12,14 @@ from sqlalchemy import create_engine, func, select from sqlalchemy.engine import Engine -from ipe.v2.db import generation_requests, init_schema, persist_run, problems, test_cases +from ipe.v2.db import ( + generation_requests, + init_schema, + persist_run, + problem_algorithms, + problems, + test_cases, +) def _engine(tmp_path: Path) -> Engine: @@ -22,7 +29,11 @@ def _engine(tmp_path: Path) -> Engine: def _body( - *, run_id: str, final_status: str = "success", with_package: bool = True + *, + run_id: str, + final_status: str = "success", + with_package: bool = True, + composition: list[str] | None = None, ) -> dict[str, Any]: package: dict[str, Any] | None = None if with_package: @@ -60,6 +71,9 @@ def _body( "package_version": "1.0", "mode": "p2", "hidden_algorithm": "dijkstra", + "composition": ( + ["union_find", "toposort"] if composition is None else composition + ), "timing": {"max_golden_elapsed_ms": 40}, }, } @@ -92,7 +106,6 @@ def test_persist_success_maps_all_tables(tmp_path: Path) -> None: assert prob["status"] == "draft" assert prob["time_limit_ms"] == 1000 # max(1000, 40*3=120) assert prob["internal_meta"]["hidden_algorithm"] == "dijkstra" - assert prob["algorithm"] == "dijkstra" # 1급 컬럼 승격 (meta 와 동일값) cases = ( c.execute( @@ -160,3 +173,49 @@ def test_persist_without_package_logs_only(tmp_path: Path) -> None: ) assert req["final_status"] == "fail_verification" assert req["problem_id"] is None # 감사 로그만 + + +def test_persist_writes_algorithm_junction(tmp_path: Path) -> None: + """problem_algorithms: 코어(role='core') + 합성(role='composition') N:M 적재.""" + eng = _engine(tmp_path) + pid = persist_run(eng, _body(run_id="algo")) # composition=[union_find, toposort] + + assert pid is not None + with eng.connect() as c: + rows = ( + c.execute( + select(problem_algorithms).where( + problem_algorithms.c.problem_id == pid + ) + ) + .mappings() + .all() + ) + pairs = {(r["algorithm"], r["role"]) for r in rows} + assert ("dijkstra", "core") in pairs # 코어 = hidden_algorithm + assert ("union_find", "composition") in pairs + assert ("toposort", "composition") in pairs + assert len(rows) == 3 # 코어 1 + 합성 2 + + +def test_persist_algorithm_junction_core_only_when_no_composition( + tmp_path: Path, +) -> None: + """P1(합성 없음): 코어 1행만 적재된다.""" + eng = _engine(tmp_path) + pid = persist_run(eng, _body(run_id="p1", composition=[])) + + assert pid is not None + with eng.connect() as c: + rows = ( + c.execute( + select(problem_algorithms).where( + problem_algorithms.c.problem_id == pid + ) + ) + .mappings() + .all() + ) + assert len(rows) == 1 + assert rows[0]["algorithm"] == "dijkstra" + assert rows[0]["role"] == "core"