"""GROUNDING-V2 §4a — 조사 기록의 **정본**은 append-only DB 행이다. ★계약: 「조사는 별도 durable record」. canon 평문 칸이나 `review_notes` 에 쓰면 뒤 단계가 그 칸을 덮어 **감사가 사라진다** — 실제로 그 판이 한 번 있었다. """ import hashlib import itertools import json import pytest from app.modules.pipeline.grounding_claims import ( DELTA_EFFECT_DIFF, DELTA_EFFECT_NO_DIFF, DELTA_UNRESOLVED, DELTA_YES, GAP_NOT_FOUND, GAP_TIME_CAPPED, LIMIT_ADMISSION, GAP_TRANSMISSION_FAILED, INPUT_IDENTITY_FIELDS, PROVENANCE_FIELDS, PROVENANCE_INPUT_BINDING, PROVENANCE_OUTCOME_FIELDS, STATUS_COMPLETED, STATUS_RETRYABLE, STATUS_UNRESOLVED_TERMINAL, TRANSMISSION_FAILED, TRANSMISSION_OK, build_gap, build_revision_row, canonical_research_input, canonical_rows, plan_admission, research_input_hash, revision_content_hash, validate_claim, ) #: ★hash 칸은 **실제 hash 모양**이어야 한다. 전에는 `"ph1"` 같은 자리끼움을 #: 썼고, 그래서 자리끼움이 통과하는 결함을 시험이 **정답으로 못박고 있었다**. def _h(seed: str) -> str: return hashlib.sha256(seed.encode()).hexdigest()[:16] SID = "rs_abc" RI = dict(era="1983년", region="대한민국", claims_pack_version="claims/1.0", prompt_raw_hash=_h("prompt"), schema_raw_hash=_h("schema"), policy_version="pol_v1", policy_contract_hash=_h("policy"), provider="openai", model="gpt-5.6-sol", model_version="1") KW = dict(subject_payload_hashes={SID: _h("pay"), "rs_z": _h("pay_z")}, research_inputs=RI) #: ★provenance 는 **같은 호출을 가리킨다** — 겹치는 좌표는 `RI` 와 같아야 한다. PROV = dict(provider=RI["provider"], model=RI["model"], model_version=RI["model_version"], prompt_pack_version=RI["claims_pack_version"], prompt_locator="_base/grounding_claims_search/1.0/system.md", schema_locator="_base/grounding_claims_search/1.0/claims_schema.json", prompt_raw_hash=RI["prompt_raw_hash"], schema_raw_hash=RI["schema_raw_hash"], payload_hash=_h("payload"), local_trace_id="trace_1", policy_version=RI["policy_version"], #: ★정상 호출은 응답 ID 로 선다. 전송 실패 갈래는 전용 class 가 잰다. transmission_status="ok", provider_request_id="req_ok") BASE = dict(project_id="p1", episode_id="e1", research_subject_id=SID, subject_payload_hash=_h("pay"), research_inputs=RI, created_at="2026-08-30T12:00:00Z", attempt_id="att_1", provenance=PROV) def _bound_prov(ri, **over): """★입력 신원에 **결속된** provenance. 겹치는 좌표를 손으로 맞추면 시험마다 갈리고, 갈리면 결속 규칙 자체를 못 잰다.""" from app.modules.pipeline.grounding_claims import PROVENANCE_INPUT_BINDING return {**PROV, **{pk: ri[ik] for pk, ik in PROVENANCE_INPUT_BINDING}, **over} def _claim(**over): c, why = validate_claim( {"statement_native": "1983년 승차권은 등사 인쇄였다", "required": True, "sources": ["https://example.org/a"], "delta_effect": DELTA_EFFECT_DIFF, **over}, subject_id=SID) assert c is not None, why return c class TestThreeIdentitiesAreNotOne: """★하나로 뭉치면 **같은 입력의 두 결과**를 못 남긴다 (Codex). 같은 조사 입력인데 `delta=yes` 와 `no` 가 나올 수 있다. 그건 **다른 시도** 이지 다른 조사가 아니다 — 유일성을 입력에만 걸면 두 번째가 안 들어간다. """ def test_the_same_input_gives_the_same_input_hash(self): a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[], **BASE) assert a["research_input_hash"] == b["research_input_hash"] def test_a_different_result_gives_a_different_content_hash(self): a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[], **BASE) assert a["revision_content_hash"] != b["revision_content_hash"] def test_two_attempts_of_the_same_input_are_different_rows(self): a = build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": "att_1"}) b = build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": "att_2"}) assert a["id"] != b["id"] assert a["research_input_hash"] == b["research_input_hash"] assert a["revision_content_hash"] == b["revision_content_hash"] def test_a_retry_after_a_time_cap_does_not_collide(self): """★상한에 걸린 1차와 완주한 2차는 **다른 호출**이다. 호출부가 새 `attempt_id` 를 발급하고, 그러면 두 행이 다 남는다. """ capped = build_revision_row(claims=[_claim()], time_capped=True, **{**BASE, "attempt_id": "att_1"}) done = build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": "att_2"}) assert capped["status"] == STATUS_RETRYABLE assert done["status"] == STATUS_COMPLETED assert capped["id"] != done["id"] def test_attempt_id_is_required_not_derived(self): """★결과에서 뽑으면 **같은 결과를 두 번 실제 호출**했을 때 비용·trace 가 다른데 한 행으로 접혀 두 번째 기록이 사라진다 (Codex).""" for bad in ("", " ", None): with pytest.raises(ValueError): build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": bad}) def test_the_same_call_twice_is_the_same_row(self): """★같은 시도 id 를 두 번 쓰면 같은 행 — append-only 가 쓰레기로 안 찬다.""" a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[_claim()], **BASE) assert a["id"] == b["id"] def test_the_row_id_hashes_the_whole_attempt_id(self): """★앞 8자만 쓰면 `attempt-111A` 와 `attempt-111B` 가 **PK 충돌**한다.""" a = build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": "attempt-111A"}) b = build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": "attempt-111B"}) assert a["id"] != b["id"] @pytest.mark.parametrize("field,other", [ ("era", "1990년"), ("region", "일본"), ("claims_pack_version", "claims/2.0"), ("policy_version", "pol_v2"), ("prompt_raw_hash", _h("ph2")), ("provider", "google")]) def test_changing_the_input_changes_the_input_hash(self, field, other): """★판이 바뀌면 **provenance 도 같이 바뀐다** — 겹치는 좌표는 결속돼 있어서, 한쪽만 바꾸면 저장 자체가 안 된다.""" a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[_claim()], **{ **BASE, "research_inputs": {**RI, field: other}, "provenance": _bound_prov({**RI, field: other})}) assert a["research_input_hash"] != b["research_input_hash"] def test_changing_the_source_text_changes_the_input_hash(self): a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[_claim()], **{ **BASE, "subject_payload_hash": _h("다른 원문")}) assert a["research_input_hash"] != b["research_input_hash"] def test_the_content_hash_ignores_row_order(self): """★같은 claims 를 다른 순서로 담았다고 「다른 결과」가 되면, 같은 결과가 다른 판으로 기록돼 감사가 갈린다 (Codex).""" x = _claim() y = _claim(statement_native="둘째 사실") assert revision_content_hash([x, y], [], DELTA_YES) \ == revision_content_hash([y, x], [], DELTA_YES) def test_the_content_hash_ignores_gap_order(self): g1 = build_gap(SID, reason=GAP_NOT_FOUND, query="가") g2 = build_gap(SID, reason=GAP_NOT_FOUND, query="나") assert revision_content_hash([], [g1, g2], DELTA_UNRESOLVED) \ == revision_content_hash([], [g2, g1], DELTA_UNRESOLVED) @pytest.mark.parametrize("claims,kw,expect", [ ([], {}, STATUS_UNRESOLVED_TERMINAL), ("one", {}, STATUS_COMPLETED), ("one", {"time_capped": True}, STATUS_RETRYABLE), ([], {"time_capped": True}, STATUS_RETRYABLE), ]) def test_the_status_is_derived_from_gaps_and_delta(self, claims, kw, expect): """★★`status` 를 semantic hash 에서 뺀 근거 — **유도되는 값**이다. ★솔직하게: 「hash 에 status 가 있다」는 것 자체는 시험으로 못 태운다. 같은 claims·같은 gaps·같은 delta 인데 status 만 다른 행은 **만들어질 수 없기** 때문이다. 일부러 되돌려 넣어 봐도 아무 시험도 안 빨개졌다. 그래서 잠그는 것은 hash 모양이 아니라 **그 유도가 성립한다는 것**이다. 이게 깨지면 status 는 독립된 값이 되고, 그때는 hash 에 다시 넣어야 한다. """ rows = [_claim()] if claims == "one" else [] row = build_revision_row(claims=rows, **{**BASE, **kw}) gaps = json.loads(row["gaps_json"]) capped = any(g["reason"] == GAP_TIME_CAPPED for g in gaps) broken = any(g["reason"] == GAP_TRANSMISSION_FAILED for g in gaps) derived = (STATUS_RETRYABLE if (capped or broken) else STATUS_UNRESOLVED_TERMINAL if row["delta"] == DELTA_UNRESOLVED else STATUS_COMPLETED) assert row["status"] == expect assert derived == row["status"] def test_two_attempts_of_the_same_capped_result_hash_the_same(self): """★같은 결과는 같은 semantic hash — 그래도 **행은 갈린다**(attempt).""" a = build_revision_row(claims=[], **{ **BASE, "attempt_id": "att_a", "time_capped": True}) b = build_revision_row(claims=[], **{ **BASE, "attempt_id": "att_b", "time_capped": True}) assert a["revision_content_hash"] == b["revision_content_hash"] assert a["id"] != b["id"] def test_a_time_capped_result_still_differs_from_a_complete_one(self): """★positive control — gap 이 다르므로 semantic hash 도 다르다. 위 시험이 「status 를 뺐으니 다 같아진다」로 읽히면 안 된다. """ capped = build_revision_row(claims=[_claim()], **{ **BASE, "time_capped": True}) done = build_revision_row(claims=[_claim()], **BASE) assert capped["revision_content_hash"] != done["revision_content_hash"] class TestStatusSeparatesCappedFromTerminal: """★`unresolved` 하나로 뭉치면 **시간에 잘린 것**과 **다 보고도 못 정한 것** 이 같아진다. 앞의 것은 다음 판에서 다시 사야 한다 (Codex). """ def test_a_decided_row_is_completed(self): row = build_revision_row(claims=[_claim()], **BASE) assert row["delta"] == DELTA_YES and row["status"] == STATUS_COMPLETED def test_nothing_found_is_terminal_not_retryable(self): row = build_revision_row(claims=[], **BASE) assert row["delta"] == DELTA_UNRESOLVED assert row["status"] == STATUS_UNRESOLVED_TERMINAL def test_an_explicit_time_cap_is_retryable(self): row = build_revision_row(claims=[_claim()], time_capped=True, **BASE) assert row["status"] == STATUS_RETRYABLE def test_a_time_capped_gap_is_retryable(self): g = build_gap(SID, reason=GAP_TIME_CAPPED, limit_kind=LIMIT_ADMISSION) row = build_revision_row(claims=[_claim()], gaps=[g], **BASE) assert row["status"] == STATUS_RETRYABLE def test_a_conflict_is_terminal_not_retryable(self): both = [_claim(), _claim(statement_native="안 변했다", required=False, delta_effect=DELTA_EFFECT_NO_DIFF)] row = build_revision_row(claims=both, **BASE) assert row["status"] == STATUS_UNRESOLVED_TERMINAL class TestResumeReadsRowsNotASingleHash: """★행 순서에 따라 재구매 여부가 바뀌면 안 된다.""" def test_a_completed_row_stops_the_rebuy(self): row = build_revision_row(claims=[_claim()], **BASE) got = plan_admission([SID, "rs_z"], **KW, done_rows=[row]) assert got["admitted"] == ["rs_z"] def test_a_retryable_row_is_bought_again(self): row = build_revision_row(claims=[_claim()], time_capped=True, **BASE) got = plan_admission([SID], **KW, done_rows=[row]) assert got["admitted"] == [SID] and got["retryable"] == [SID] def test_a_changed_pack_makes_it_pending_again(self): row = build_revision_row(claims=[_claim()], **BASE) got = plan_admission([SID], done_rows=[row], **{ **KW, "research_inputs": {**RI, "claims_pack_version": "claims/2.0"}}) assert got["admitted"] == [SID] def test_every_row_order_gives_the_same_answer(self): rows = [build_revision_row(claims=[_claim()], **{**BASE, "attempt_id": "att_1"}), build_revision_row(claims=[], **{**BASE, "attempt_id": "att_2"}), build_revision_row(claims=[_claim()], time_capped=True, **{**BASE, "attempt_id": "att_3"})] outs = {tuple(plan_admission([SID], **KW, done_rows=list(p))["admitted"]) for p in itertools.permutations(rows)} assert len(outs) == 1, outs class TestOnlyValidatedRowsReachTheRecord: """★「정본에 안 넣는다」고 주석에 써 놓고 **원본을 그대로 저장**했다 (Codex).""" def test_a_malformed_claim_is_not_in_the_record(self): fake = {"claim_id": "cl_fake", "research_subject_id": SID, "kind": "fact", "required": True, "delta_effect": DELTA_EFFECT_DIFF, "sources": ["https://u.org/a"], "statement_native": "x"} row = build_revision_row(claims=[fake], **BASE) assert row["delta"] == DELTA_UNRESOLVED assert json.loads(row["claims_json"]) == [] assert json.loads(row["audit_json"])["rejected"] def test_good_and_bad_together_keeps_only_the_good(self): fake = {"claim_id": "cl_fake", "research_subject_id": SID, "kind": "fact", "required": True, "sources": ["https://u.org/a"], "statement_native": "x"} row = build_revision_row(claims=[_claim(), fake], **BASE) assert len(json.loads(row["claims_json"])) == 1 def test_the_stored_rows_are_normalised(self): row = build_revision_row( claims=[_claim(statement_native=" 띄어 쓰기 ")], **BASE) assert json.loads(row["claims_json"])[0]["statement_native"] == \ "띄어 쓰기" def test_a_malformed_gap_is_caught_too(self): row = build_revision_row( claims=[_claim()], gaps=[{"research_subject_id": SID, "reason": "몰라"}], **BASE) assert row["delta"] == DELTA_UNRESOLVED assert json.loads(row["gaps_json"]) == [] def test_the_raw_payload_is_only_kept_if_the_caller_passes_it(self): """★내가 「원시 응답을 audit 에 저장한다」고 보고했는데 **셌다** — 실제로는 거절 **사유**만 담고 원본은 호출부가 넘겨야 남는다 (Codex).""" fake = {"claim_id": "cl_fake", "statement_native": "x", "required": True, "sources": ["https://u.org/a"]} plain = build_revision_row(claims=[fake], **BASE) assert "raw" not in json.loads(plain["audit_json"]) with_raw = build_revision_row(claims=[fake], audit={"raw": [fake]}, **BASE) assert json.loads(with_raw["audit_json"])["raw"] == [fake] class TestTheTableIsAppendOnlyAndAdditive: def test_the_model_has_the_three_identity_columns(self): from app.models.project import GroundingResearchRevision as G cols = set(G.__table__.columns.keys()) for c in ("research_input_hash", "revision_content_hash", "attempt_id", "status", "delta", "delta_reason", "claims_json", "gaps_json", "audit_json", "created_at"): assert c in cols, c def test_uniqueness_is_per_attempt_not_per_input(self): """★입력에만 걸면 같은 입력의 다른 결과가 막힌다.""" from app.models.project import GroundingResearchRevision as G uq = [c for c in G.__table__.constraints if c.__class__.__name__ == "UniqueConstraint"] assert uq, "유일성 제약이 없다" cols = {c.name for c in list(uq)[0].columns} assert cols == {"project_id", "episode_id", "research_subject_id", "research_input_hash", "attempt_id"} def test_the_migration_is_additive_only(self): from pathlib import Path src = (Path(__file__).resolve().parents[2] / "alembic" / "versions" / "011_grounding_research_revision.py").read_text(encoding="utf-8") for banned in ("drop_column", "alter_column", 'op.execute("UPDATE', 'op.execute("DELETE'): assert banned not in src, banned for want in ("research_input_hash", "revision_content_hash", "attempt_id", "status"): assert want in src, want def test_the_record_does_not_live_in_canon_or_review_notes(self): import inspect from app.modules.pipeline import grounding_claims as gc src = inspect.getsource(gc) assert "review_notes" not in src assert "entity_canon" not in src class TestTheEnvelopeFieldsAreRequired: """★`bool("false")` 는 True 다 — 이 판에서 `required`·`gradable`· `required_discriminator` 로 이미 **세 번** 데인 부류다 (Codex). """ @pytest.mark.parametrize("bad", ["false", "true", 1, 0, None]) def test_time_capped_must_be_a_real_bool(self, bad): with pytest.raises(ValueError): build_revision_row(claims=[_claim()], time_capped=bad, **BASE) @pytest.mark.parametrize("bad", ["", " ", None]) def test_created_at_is_required(self, bad): """★언제 조사했는지 없는 기록은 되짚을 수 없다.""" kw = {**BASE, "created_at": bad} with pytest.raises(ValueError): build_revision_row(claims=[_claim()], **kw) def test_a_real_bool_and_a_real_time_pass(self): row = build_revision_row(claims=[_claim()], time_capped=True, **BASE) assert row["status"] == STATUS_RETRYABLE assert row["created_at"] == "2026-08-30T12:00:00Z" class TestATimeCapNeverStoresADecidedDelta: """★시간 초과는 계약상 `unresolved` 다. 상한에 걸렸는데 `delta=yes` 로 저장되면 「다 보고 조사하기로 했다」가 거짓이 된다 — 아직 다 안 봤다 (Codex). """ def test_a_capped_run_is_unresolved_even_with_a_good_claim(self): row = build_revision_row(claims=[_claim()], time_capped=True, **BASE) assert row["delta"] == DELTA_UNRESOLVED assert row["status"] == STATUS_RETRYABLE def test_the_cap_is_recorded_as_a_gap(self): """★사유가 안 남으면 나중에 「왜 미확정인가」를 못 되짚는다.""" row = build_revision_row(claims=[_claim()], time_capped=True, **BASE) gaps = json.loads(row["gaps_json"]) assert any(g["reason"] == GAP_TIME_CAPPED for g in gaps) assert any(g["required_discriminator"] for g in gaps) def test_an_uncapped_run_still_decides(self): """★positive control — 상한이 아니면 그대로 판정한다.""" assert build_revision_row(claims=[_claim()], **BASE)["delta"] == DELTA_YES class TestOneCanonicalizerForHashAndStorage: """★따로 정렬하면 hash 는 같은데 **저장 bytes 가 달라진다** — 같은 결과가 다른 바이트로 남아 감사가 갈린다 (Codex). """ def test_claim_order_changes_neither_hash_nor_bytes(self): x, y = _claim(), _claim(statement_native="둘째 사실") a = build_revision_row(claims=[x, y], **BASE) b = build_revision_row(claims=[y, x], **BASE) assert a["revision_content_hash"] == b["revision_content_hash"] assert a["claims_json"] == b["claims_json"] assert a["id"] == b["id"] def test_source_order_inside_a_claim_does_not_matter(self): """★출처를 적은 **순서**가 결과를 바꾸면 안 된다.""" p = _claim(sources=["https://a.org/1", "https://b.org/2"]) q = _claim(sources=["https://b.org/2", "https://a.org/1"]) assert revision_content_hash([p], [], DELTA_YES) \ == revision_content_hash([q], [], DELTA_YES) def test_the_canonicalizer_sorts_sources(self): got = canonical_rows([{"sources": ["https://b.org", "https://a.org"]}]) assert got[0]["sources"] == ["https://a.org", "https://b.org"] class TestProvenanceIsRequiredAndSeparate: """★빈 provenance 행을 만들면 그 행들은 **영원히 못 되짚는다** (Codex). 그리고 `revision_content_hash` 에는 **안 넣는다** — 같은 claims 는 생산자가 달라도 같은 semantic hash 가 맞다. """ def test_every_field_is_required(self): for k in PROVENANCE_FIELDS: with pytest.raises(ValueError): build_revision_row(claims=[_claim()], **{**BASE, "provenance": {**PROV, k: ""}}) @pytest.mark.parametrize("bad", [{}, None, "x", 7]) def test_a_missing_envelope_is_refused(self, bad): with pytest.raises(ValueError): build_revision_row(claims=[_claim()], **{**BASE, "provenance": bad}) def test_it_is_stored_as_its_own_column(self): row = build_revision_row(claims=[_claim()], **BASE) stored = json.loads(row["provenance_json"]) assert set(stored) == set(PROVENANCE_FIELDS) | {"provider_request_id"} def test_the_producer_does_not_change_the_semantic_hash(self): """★같은 claims 를 다른 생산자가 낸 것은 **같은 결과**다. ★전에는 provenance 의 `model` 만 갈아 끼워 이것을 보였는데, 그건 입력 신원과 어긋난 조합이라 **애초에 만들어질 수 없는 판**이었다 — 그 시험이 결속 없음을 정답으로 못박고 있었다 (Codex). 이제 **양쪽을 같이** 바꾼다. """ other = {**RI, "provider": "google", "model": "gemini-3.1-pro-preview"} a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[_claim()], **{ **BASE, "research_inputs": other, "provenance": _bound_prov(other)}) # 무엇을 조사했나 = 다르다 (다른 검색기로 다시 산다) assert a["research_input_hash"] != b["research_input_hash"] # 무엇이 나왔나 = 같다 (같은 claims 는 생산자가 달라도 같은 결과) assert a["revision_content_hash"] == b["revision_content_hash"] def test_the_model_column_exists(self): from app.models.project import GroundingResearchRevision as G assert "provenance_json" in G.__table__.columns.keys() class TestATransmissionFailureCanActuallyBeStored: """★`STATUS_RETRYABLE` 의 「전송 실패」 갈래가 **닿을 수 있어야** 뜻이 있다. 전에는 `provider_request_id` 를 늘 요구해서, 주석은 「전송 실패면 그 사실」이라 적어 놓고 코드는 전송 실패를 **아예 저장할 수 없었다**. 규칙을 주석과 코드 두 곳에 적으면 한쪽만 고쳐진다 — 여기서는 코드가 안 고쳐졌다. ★행이 아예 안 남으면 「무엇이 왜 실패했나」를 못 되짚는다. 다시 사는 것은 행이 없어도 되지만(admission 이 못 본 것으로 센다), **되짚는 것은 안 된다**. """ def _prov(self, **kw): """★전송 실패 판은 응답 ID 를 **안 받은** 것이 기본이다.""" base = {k: v for k, v in PROV.items() if k != "provider_request_id"} if "transmission_error" in kw and "transmission_status" not in kw: kw["transmission_status"] = TRANSMISSION_FAILED return {**base, **kw} def test_a_broken_transmission_is_stored_as_retryable(self): row = build_revision_row(claims=[], **{ **BASE, "provenance": self._prov( transmission_error="ConnectionResetError: peer closed")}) assert row["status"] == STATUS_RETRYABLE assert row["delta"] == DELTA_UNRESOLVED assert [g["reason"] for g in json.loads(row["gaps_json"])] == [ GAP_TRANSMISSION_FAILED] def test_the_failure_text_survives_into_the_gap(self): """★무엇이 깨졌는지 안 남으면 되짚을 수 없다.""" row = build_revision_row(claims=[], **{ **BASE, "provenance": self._prov( transmission_error="ReadTimeout after 120s")}) assert "ReadTimeout" in json.loads(row["gaps_json"])[0]["note"] def test_a_partial_result_from_a_broken_call_is_not_a_conclusion(self): """★★깨진 전송에서 나온 출처 붙은 claim 이 있어도 `yes` 가 안 된다. 응답 ID 를 이미 받은 뒤 스트림이 끊기는 것은 **정상적인 실패 모양**이라 둘 다 채워진다. 그때 부분 산출을 그대로 접으면 「다 보고 정했다」가 거짓이 된다. """ row = build_revision_row(claims=[_claim()], **{ **BASE, "provenance": self._prov( provider_request_id="req_9", transmission_error="stream broke")}) assert row["status"] == STATUS_RETRYABLE assert row["delta"] == DELTA_UNRESOLVED def test_a_clean_call_with_only_a_request_id_still_concludes(self): """★positive control — 실패 좌표가 없으면 결론이 그대로 난다. 이 시험이 안 빨개지면 위의 것들은 아무것도 안 재고 있는 것이다. """ row = build_revision_row(claims=[_claim()], **{ **BASE, "provenance": self._prov( transmission_status=TRANSMISSION_OK, provider_request_id="req_1")}) assert row["status"] == "completed" assert row["delta"] == "yes" @pytest.mark.parametrize("st,missing", [ (TRANSMISSION_OK, "provider_request_id"), (TRANSMISSION_FAILED, "transmission_error")]) def test_the_outcome_must_match_what_the_status_says(self, st, missing): """★결말이 말한 것과 좌표가 안 맞으면 두 칸이 서로 다른 말을 한다.""" with pytest.raises(ValueError): build_revision_row(claims=[], **{ **BASE, "provenance": self._prov(transmission_status=st)}) @pytest.mark.parametrize("bad", ["", "OK", "성공", "timeout", "abandoned"]) def test_an_unknown_transmission_status_is_refused(self, bad): """★enum 밖을 받으면 그 값은 아무 갈래에도 안 걸려 조용히 지나간다. ★`timeout`·`abandoned` 를 지금 안 만든 것은 **쓰는 호출자가 없어서**다. 사유는 `transmission_error` 자유 문장으로 남는다. """ with pytest.raises(ValueError): build_revision_row(claims=[], **{ **BASE, "provenance": self._prov( transmission_status=bad, provider_request_id="req_1")}) def test_surrounding_whitespace_is_not_a_different_status(self): """★공백은 값이 아니다 — 여기서 갈리면 호출부가 못 맞춘다.""" row = build_revision_row(claims=[], **{ **BASE, "provenance": self._prov( transmission_status=" ok ", provider_request_id="req_1")}) assert json.loads(row["provenance_json"])["transmission_status"] == "ok" def test_the_always_known_fields_are_still_all_required(self): """★전송이 깨져도 **무엇으로 어떻게 불렀는지는 안다.** 그건 계속 필수다.""" for k in PROVENANCE_FIELDS: with pytest.raises(ValueError): build_revision_row(claims=[], **{ **BASE, "provenance": self._prov( transmission_error="boom", **{k: ""})}) def test_the_outcome_pair_is_not_in_the_always_required_list(self): """★두 목록이 겹치면 「둘 중 하나」 규칙이 조용히 「둘 다」가 된다.""" assert not (set(PROVENANCE_FIELDS) & set(PROVENANCE_OUTCOME_FIELDS)) def test_a_retryable_row_is_bought_again_next_round(self): """★★끝점 — 저장된 그 행을 admission 에 그대로 먹인다. 손으로 만든 dict 가 아니라 **writer 가 낸 행**을 먹여야, 두 모듈이 `status` 를 같은 이름으로 부르는지까지 잰다. """ row = build_revision_row(claims=[], **{ **BASE, "provenance": self._prov(transmission_error="boom")}) d = plan_admission( [BASE["research_subject_id"]], subject_payload_hashes={ BASE["research_subject_id"]: BASE["subject_payload_hash"]}, research_inputs=BASE["research_inputs"], done_rows=[row]) assert d["admitted"] == [BASE["research_subject_id"]] assert d["retryable"] == [BASE["research_subject_id"]] assert d["already_done"] == [] class TestProvenanceAndInputsPointAtTheSameCall: """★★겹치는 좌표가 갈리면 **어느 쪽이 정본인지 아무도 모른다** (Codex). 실측 — 입력에 `provider=openai`, provenance 에 `provider=other` 를 넣어도 저장됐다. 그러면 「무엇으로 조사했나」와 「무엇을 조사했나」가 서로 다른 말을 하고, `research_input_hash` 로 재사용을 가르는 규칙이 뜻을 잃는다. ★그리고 이 결함을 `test_the_producer_does_not_change_the_semantic_hash` 가 **정답으로 못박고 있었다** — provenance 만 갈아 끼워 같은 hash 를 확인했는데, 그건 애초에 만들어질 수 없는 조합이었다. """ @pytest.mark.parametrize("prov_key,input_key", list(PROVENANCE_INPUT_BINDING)) def test_a_drifting_coordinate_is_refused(self, prov_key, input_key): with pytest.raises(ValueError, match="다른 호출"): build_revision_row(claims=[_claim()], **{ **BASE, "provenance": {**PROV, prov_key: _h("딴것")}}) def test_the_bound_baseline_still_stores(self): """★positive control — 맞춰 놓으면 그대로 저장된다.""" assert build_revision_row(claims=[_claim()], **BASE)["status"] \ == "completed" def test_every_bound_coordinate_names_a_real_field_on_both_sides(self): """★표에 오타가 있으면 그 축은 **늘 통과**한다 — 없는 칸끼리 비교라서.""" for pk, ik in PROVENANCE_INPUT_BINDING: assert pk in PROVENANCE_FIELDS, pk assert ik in INPUT_IDENTITY_FIELDS, ik def test_the_producer_may_still_differ_where_it_is_free_to(self): """★★같은 claims 는 **같은 semantic hash** 다 — 결속되지 않은 좌표에서. 전 시험은 결속된 좌표(provider)를 갈아 끼워 「생산자가 달라도 같다」를 보였는데, 그 조합은 이제 아예 저장이 안 된다. 자유로운 좌표 (trace ID·응답 ID·payload hash)로 다시 잰다. """ a = build_revision_row(claims=[_claim()], **BASE) b = build_revision_row(claims=[_claim()], **{ **BASE, "attempt_id": "att_2", "provenance": { **PROV, "local_trace_id": "trace_9", "provider_request_id": "req_9", "payload_hash": _h("다른 payload")}}) assert a["revision_content_hash"] == b["revision_content_hash"] assert a["id"] != b["id"] def test_the_schema_locator_is_kept(self): """★프롬프트만 남기고 schema 를 빼면 **절반만 되짚힌다** — 별도 파일이다.""" assert "schema_locator" in PROVENANCE_FIELDS stored = json.loads( build_revision_row(claims=[_claim()], **BASE)["provenance_json"]) assert stored["schema_locator"].endswith(".json") class TestTheStoredRowCanRecomputeItsOwnInputHash: """★★`research_input_hash` 는 **한 방향**이다 — 되풀어 볼 수 없다. 그러니 12칸이 **전부 행 어딘가에** 있어야 나중에 「이 조사가 정말 그 입력이었나」를 검산할 수 있다. 실측에서 `subject_payload_hash` 와 `policy_contract_hash` 둘만 어디에도 없었다 — 그 둘은 제가 「중복이라 안 넣겠다」고 거부했던 칸이다. 거부가 틀렸다. ★칼럼과 provenance 를 **둘 다** 세는 이유: 일곱 칸은 `PROVENANCE_INPUT_BINDING` 이 입력과 같음을 보장하므로 provenance 에 있으면 되찾아진다. 결속이 없으면 provenance 값은 다른 호출의 것일 수 있어 검산에 못 쓴다. """ def test_the_envelope_holds_every_identity_field(self): row = build_revision_row(claims=[_claim()], **BASE) got = json.loads(row["research_input_json"]) assert sorted(got) == sorted(INPUT_IDENTITY_FIELDS) def test_the_row_actually_recomputes_the_same_hash(self): """★★끝점 — 있다는 것과 **같은 값**이라는 것은 다르다. 저장된 봉투에서 12칸을 꺼내 hash 를 다시 계산해 저장된 것과 맞춘다. 어느 한 칸이라도 다른 값을 적어 두면 여기서 갈린다. """ row = build_revision_row(claims=[_claim()], **BASE) parts = json.loads(row["research_input_json"]) assert research_input_hash(**parts) == row["research_input_hash"] def test_the_hash_is_computed_from_the_stored_bytes_themselves(self): """★★따로 접으면 저장된 봉투와 hash 가 갈린다 — `canonical_rows` 와 같은 규칙이다. 여기서는 **문자열 자체**를 hash 해서 맞춘다.""" row = build_revision_row(claims=[_claim()], **BASE) direct = hashlib.sha256( row["research_input_json"].encode()).hexdigest()[:24] assert direct == row["research_input_hash"] def test_adding_a_thirteenth_field_would_be_stored_too(self): """★칸을 더했을 때 **저장을 따로 고쳐야 한다면** 언젠가 안 고친다. 봉투가 `INPUT_IDENTITY_FIELDS` 를 그대로 도니, 여기가 늘면 저장도 는다. """ assert set(json.loads( canonical_research_input( research_subject_id=SID, subject_payload_hash=_h("pay"), **RI) )) == set(INPUT_IDENTITY_FIELDS) def test_the_model_has_the_envelope_column(self): from app.models.project import GroundingResearchRevision as G assert "research_input_json" in G.__table__.columns.keys() class TestProvenanceDoesNotSwallowUnknownFields: """★★모르는 칸을 조용히 버리면, 넘긴 쪽은 **남았다고 믿는다**. 실측 — `cost_usd` 를 넘겨도 오류 없이 사라졌다 (Codex). 검색 배선에서 새 감사 좌표를 넘겼다고 믿고 기록이 유실될 현실적인 경로다. """ def test_an_unknown_field_is_refused(self): with pytest.raises(ValueError, match="모르는 칸"): build_revision_row(claims=[_claim()], **{ **BASE, "provenance": {**PROV, "cost_usd": "0.014"}}) def test_the_error_names_the_field(self): """★무엇을 버렸는지 안 말하면 호출부가 못 고친다.""" with pytest.raises(ValueError, match="cost_usd"): build_revision_row(claims=[_claim()], **{ **BASE, "provenance": {**PROV, "cost_usd": "0.014"}}) def test_the_known_envelope_still_passes(self): """★positive control — 허용 칸만이면 그대로 저장된다.""" assert build_revision_row(claims=[_claim()], **BASE)["provenance_json"] class TestTheShapeSearchActuallyReturns: """★★★**검색이 실제로 돌려주는 모양**으로 잰다. 이 파일의 다른 시험은 전부 `_claim()` — 즉 `validate_claim` 을 이미 거쳐 `claim_id` 가 붙은 것 — 으로 재고 있었다. 그런데 production 은 `_save_revisions` 에서 **provider 응답의 raw claim 을 그대로** 넘긴다. 그리고 `grounding_claims_search` 스키마에는 `claim_id` 칸이 **없다**. 옛 규칙은 「호출부가 준 `claim_id` 와 계산값이 다르면 오염」을 **무조건** 걸었다. 안 준 것도 `None != 계산값` 이라 오염이 됐다 — 그래서 **모든 실제 조사 결과가 `unresolved` 로 닫혔고**, §4b 의 `research_required → reference_required` 강제가 한 번도 안 걸렸다. ★조립부(`_claim()`)에서만 재서 안 보였다. """ #: 검색 스키마가 실제로 내는 칸만 — `claim_id` 는 없다. RAW = {"statement_native": "1983년 승차권은 등사 인쇄였다", "kind": "fact", "required": True, "sources": ["https://example.org/a"], "delta_effect": DELTA_EFFECT_DIFF, "discriminator_family": "print"} def test_a_raw_search_claim_is_not_malformed(self): row = build_revision_row(claims=[dict(self.RAW)], **BASE) assert row["delta"] == "yes", \ "★검색이 주는 모양이 오염으로 처리됐다 — 조사 결과가 영영 안 선다" assert len(json.loads(row["claims_json"])) == 1 def test_it_gets_the_computed_claim_id(self): row = build_revision_row(claims=[dict(self.RAW)], **BASE) stored = json.loads(row["claims_json"])[0] assert stored["claim_id"] == _claim()["claim_id"] def test_a_wrong_claim_id_is_still_malformed(self): """★안 준 것만 봐준다 — **틀리게 준 것**은 그대로 막는다.""" row = build_revision_row( claims=[{**self.RAW, "claim_id": "cl_wrong"}], **BASE) assert row["delta"] == DELTA_UNRESOLVED def test_the_status_is_completed_not_unresolved(self): """★`_research_forced_short_ids` 는 `completed` + `delta=yes` 만 본다.""" row = build_revision_row(claims=[dict(self.RAW)], **BASE) assert row["status"] == "completed" class TestOneCallableDecidesForBothSides: """★측정 도구가 **같은 함수**를 타야 한다 (Codex). 도구는 오염된 줄을 조용히 버리고 남은 것으로 `yes` 를 냈고, 저장 경로는 하나라도 깨지면 통째로 `unresolved` 였다. 그러면 도구가 「맞음」으로 센 것을 production 은 미확정으로 쓴다. """ def test_the_helper_closes_on_malformed_like_the_record_does(self): from app.modules.pipeline.grounding_claims import sanitize_and_decide good = {"statement_native": "1983년 승차권은 등사 인쇄였다", "kind": "fact", "required": True, "sources": ["https://example.org/a"], "delta_effect": DELTA_EFFECT_DIFF} bad = {"statement_native": "x", "required": "네", "sources": []} out = sanitize_and_decide([good, bad], [], research_subject_id=SID) assert out["delta"] == DELTA_UNRESOLVED assert out["malformed"], "오염을 안 셌다" def test_the_record_and_the_helper_agree(self): from app.modules.pipeline.grounding_claims import sanitize_and_decide raw = [{"statement_native": "1983년 승차권은 등사 인쇄였다", "kind": "fact", "required": True, "sources": ["https://example.org/a"], "delta_effect": DELTA_EFFECT_DIFF}] row = build_revision_row(claims=[dict(raw[0])], **BASE) out = sanitize_and_decide(raw, [], research_subject_id=SID) assert row["delta"] == out["delta"]