diff --git a/README.md b/README.md index 3464d82e..a694d020 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ +진행 중: [질문별 자료 선택 영속 계획](docs/changes/question-source-plans.md). 질문만으로 저장 색인에서 최대5자료를 선택하고 같은 요청은 원래 선택과 현재 원문을 재검증합니다. 실제 AI/예산 예약/개발 활성화는 없으며 다자료 실행·브리핑 변환/표시는 후속입니다. + 현재 구현: [근거 작업 접수·queue 전달](docs/changes/evidence-job-dispatch.md). 신규23/새전체9,595개(Python8,775+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토/수정 재검토를 통과했습니다. 저장→명시 버전 queue 전달·동일 ID 재접수와 권한/원문 재검사를 연결했습니다. 실제 제공자 호출0·현재 앱 활성화0이며 질문 전용 검색·브리핑 변환/표시는 후속입니다. 현재 구현: [영속 근거 작업·비동기 worker](docs/changes/durable-evidence-worker.md). 신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토 C0I0M0를 통과했습니다. SQL 송신 의도·중단 후 무재호출·공유 예약 예산·현재 권한/원문 검증을 연결했습니다. 이번 실제 AI 호출0회이며 서비스 요청·브리핑 변환·개발 배포는 후속입니다. diff --git a/docs/README.md b/docs/README.md index 26afb50b..08497eb2 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,3 +1,5 @@ +진행 중: [질문별 자료 선택 영속 계획](changes/question-source-plans.md). 질문만으로 저장 색인에서 최대5자료를 선택하고 같은 요청은 원래 선택과 현재 원문을 재검증합니다. 실제 AI/예산 예약/개발 활성화는 없으며 다자료 실행·브리핑 변환/표시는 후속입니다. + 현재 구현: [근거 작업 접수·queue 전달](changes/evidence-job-dispatch.md). 신규23/새전체9,595개(Python8,775+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토/수정 재검토를 통과했습니다. 저장→명시 버전 queue 전달·동일 ID 재접수와 권한/원문 재검사를 연결했습니다. 실제 제공자 호출0·현재 앱 활성화0이며 질문 전용 검색·브리핑 변환/표시는 후속입니다. 현재 구현: [영속 근거 작업·비동기 worker](changes/durable-evidence-worker.md). 신규57/새전체9,572개(Python8,752+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 검토 C0I0M0를 통과했습니다. SQL 송신 의도·중단 후 무재호출·공유 예약 예산·현재 권한/원문 검증을 연결했습니다. 이번 실제 AI 호출0회이며 서비스 요청·브리핑 변환·개발 배포는 후속입니다. diff --git a/docs/changes/question-source-plans.md b/docs/changes/question-source-plans.md new file mode 100644 index 00000000..5f143c58 --- /dev/null +++ b/docs/changes/question-source-plans.md @@ -0,0 +1,36 @@ +# 질문만으로 선택한 근거 자료의 영속 계획 + +2026-10-26 KST. [작업 접수·queue](evidence-job-dispatch.md) 다음 D019 검색 경계다. 사용자는 질문만 제출하며 서버가 현재 workspace의 승인된 저장 색인에서 최대5자료를 선택한다. 같은 request 재접수는 처음 선택한 원문/hash를 그대로 재검증하고 검색을 새로 하지 않는다. 이 변경은 영속 검색 계획을 구현한다. AI 실행 예약·다자료 실행 orchestration·브리핑 변환/표시·개발 활성화는 후속이다. + +## Global Constraints + +- 신규5파일과 아래 명시한 기존 migration 검사1개만 변경: `packages/domain/src/issuedigest_domain/question_source_plan.py`, `packages/domain/tests/test_question_source_plan.py`, `packages/storage/src/issuedigest_storage/question_plans.py`, `packages/storage/migrations/versions/0011_question_source_plans.py`, `packages/storage/tests/test_question_plans.py`. 기존484 실행 파일·이전52증거·AI 모델/SDK/workflow/queue·API/UI·exports/dependencies/config는 그대로 둔다. 기존 `packages/storage/tests/test_evidence_jobs.py`의 migration 검사만 동일 연결/트랜잭션의 실제0010 revision upgrade/downgrade 작업을 검사하도록 수정한다. 향후 head 전체를 되돌리거나 version stamp로 우회하지 않는다. 나머지 기존484파일은 보존한다. 문서 변경 허용. 실제 제공자 호출·운영 서버/DB·main 병합/앱 활성화0. +- Domain frozen strict extra-forbid models: `QuestionSourceChoice(snapshot_id: UUID4, source_sha256: lower64hex)`; `QuestionSourcePlan(profile='question-source-plan-v1', retrieval_profile='question-topics-ko-v2', plan_id:UUID4, workspace_id:UUID4, request_id:UUID4, question:strict str nonblank<=1500 preserved, status:Literal['selected','no_match'], reason:Literal['no_terms','no_candidates']|None, sources:tuple[QuestionSourceChoice,...] max5)`. selected requires1..5 unique snapshot IDs/reasonNone; no_match requires0 and explicit reason. No source text/receipt/model facts/actor/private credentials in domain DTO. Exact UUID4/type checks; strict JSON roundtrip accepted, bool/coercion/unknown model_copy fields rejected by storage read reconstruction. No optional parent context in this version; no UI parent mixing or browser source ID input. +- `QuestionPlanRepository(engine)` with `submit(token, workspace_id, request_id, question)->QuestionSourcePlan|None`, `get(token, workspace_id, plan_id)->QuestionSourcePlan|None`. Inputs exact UUID4/nonblank string<=1500 before SQL, no normalization/truncation/rewording. Same current session/member rules as ResearchRepository: submit owner/editor, get current read member; original submitter must still be current writer before any saved result is returned. No cross-workspace lookup/result. Narrow reuse existing ResearchRepository._actor/_member/_presentation/_transaction, EvidenceRepository._load, evidence_jobs.source_digest/_lock_jobs. Acquire same evidence-job advisory transaction gate BEFORE any auth/user/plan locks, consistently on every new operation. SQL only, never provider I/O; this may serialize bounded searches with new-job SQL operations and must be documented as throughput cost. No new independent lock order or auth classification. +- Use new separate MetaData/table `question_source_plans`: UUID plan/workspace/actor/request columns; exact question, input hash, domain payload JSONB + canonical payload hash (including original actor identity so row/payload tampering rejects). Unique workspace+actor+request. Additional minimal fixed columns/constraints implementation choice. Domain payload must match top-level plan/workspace/request/question exactly. Hash uses sorted compact UTF8 JSON allow_nan=False; hash is corruption detection, not signature. Import must not change legacy global metadata registry or perform DB/network/model work. +- New submission: authorize; atomically check prior actor/request; if prior exact question/profile input matches, validate saved payload and ALL chosen current original sources/presentations against stored source hashes then return it WITHOUT retrieval call or duplicate row. Changed same request question or invalid/tampered source/hash/metadata must reject, never reselect/replace. Concurrent identical requests produce one plan. Different actors/workspaces remain separate under existing shared initial validation scope, not personal/public reclassification. A new request can choose newer corpus; same request no_match remains no_match after index additions. +- With no prior, call unchanged `research_retrieval.question_terms(question)` and `retrieve_snapshot_ids(conn, workspace_id, terms)` on current indexed evidence. No planner/model/embedder/web/legacy ResearchRun creation. Empty terms stores no_match/no_terms with retrieval0. Empty IDs stores no_match/no_candidates. Otherwise preserve exact returned order,1..5 unique UUID4 IDs; malformed helper result fails closed. Load/verify every canonical source/presentation in same transaction, hash exactly source_digest tuple(snapshot,bundle,presentation). Existing helper already respects current workspace/latest policy intake/bounded topic search; do not modify or claim semantic relevance/coverage/absence from no_match. On saved read, revalidate only saved originals, not latest search results. Never query another workspace, create budget reservation/research run/evidence job/queue step, or consume paid model. Known no_match isn't a claim no real-world evidence exists. +- Reconstruct stored DTO with strict raw shape/field values, check input/payload/row identities and source hashes on every get and idempotent submit. Unknown fields, missing fields, wrong actual types/bool, duplicate/oversized choices, swapped actor/workspace/request/plan/question/status/reason/profile/hash, canonical bundle tampering, source deletion or title metadata mutation reject with sanitized StorageCorruption/ResearchConflict/ResearchForbidden/AuthenticationDenied as appropriate. No raw question/source/exception/credential logging. No receipt/data exposure in fixed errors. No shallow model_copy trust. No query/repair/downgrade of corrupted plans. +- Migration revision `0011_question_source_plans`, down_revision `0010_evidence_jobs`; upgrade creates only new table/constraints/index, downgrade drops only it. No old schema/data changes/import writes. New tests clean new table before/after (root fixture doesn't know it), never edit conftest or past fixtures. Test migration preservation includes existing evidence_jobs and research budget rows/old metadata snapshot. +- TDD using dedicated local PostgreSQL synthetic sources/index/identities; no provider/server/private fixtures. RED missing modules, GREEN meaningful domain/raw guards + real existing lexical index integration, deterministic duplicate request concurrency, current role/session/workspace/source/presentation rejection, persisted no_match and plan reuse without search/budget/run/job mutations, tamper/rehashed corruption, old/evidence job migration and lock order preservation. Agent owns scoped tests/Ruff/format/mypy; controller full QA/reviews/historical audits/publication. No agent subagents. + +저장 계획의 get/재접수에서도 기존 `_presentation`의 지연 복구를 허용한다. presentation 행만 없고 canonical 원문이 존재하면 기존 경로로 presentation을 재구성하며, 재구성한 원문/presentation의 hash가 저장값과 일치할 때만 원래 계획을 반환한다. 원문이 없거나 변경됐거나 재구성한 presentation hash가 다르면 계속 거부한다. presentation만 삭제한 경우의 별도 집중 검사는 비차단 후속으로 남기며 이번 검증 통과 항목으로 세지 않는다. + +## Task 1: 질문별 검색 선택 영속화 + +- [x] 신규5파일을 TDD로 구현한다. 위 global constraints를 읽고 그대로 따른다. 정확 질문과 ordered current sources·source_digest를 고정하고 재접수 때 검색하지 않는다. 역할·동시성·상태/타입/행/원문 변조·no_match·기존 budget/run/job 무변경·migration/import 격리 검사를 포함한다. 기존 migration 검사1개 예외 외484파일을 보존한다. scoped 정적 검사와 집중 테스트 후 owned5파일+이 추적 명세의 Task1 체크박스만 커밋하고 ignored SDD 보고서에 정확 RED/GREEN/명령/인터페이스/한계/자체검토를 남긴다. + +## Controller gates + +- [x] 독립 태스크/전체 검토·전체9,657개 검사·기존484실행+기존 migration 검사1개 검토/이전52증거 보존·privacy·문서/Draft PR112/Sitesv134·자체QA 정리. 비차단 단독 표시 정보 삭제 검사는 후속. +- [ ] 이 계획의 다자료 AI 실행 연결·브리핑 변환/표시·운영 정책/부하·개발 적용. 이번 활성화 없음. + +## 판단과 영향 + +1. 새 질문은 기존 문자 기반 저장 색인에서 최대5자료를 선택한다. 추가 AI 비용 없이 D019 질문 검색을 연결하지만 의미 검색/관련성·현실의 근거 유무를 보장하지 않는다. +2. 같은 request의 자료 선택과 no_match를 고정한다. 재접수 일관성을 얻지만 새 자료 반영은 새로운 request가 필요하다. +3. 기존 새 작업의 SQL gate를 공유한다. 사용자/자료 잠금 순서 역전을 막지만 검색 중 다른 새 작업 SQL도 기다릴 수 있어 부하 인수가 필요하다. + +4. 새 migration으로 드러난 기존0010 검사 가정을 수정한다. 같은 연결에서0010의 실제 작업만 검사하여 최신 head를 훼손하지 않는다. 기존 검사1개를 바꾸는 비용이 있으며 나머지484파일과 실행 코드는 보존하고 전체 migration 경로는 별도 검사한다. + +5. 기존 표시 정보 복원을 유지한다. 원문과 복원 hash가 같으면 표시 정보 삭제만으로는 실패하지 않는다. 이 동작을 문서화했고 표시 정보 단독 삭제 회귀 검사는 비차단 후속으로 남는다. diff --git a/docs/planning/implementation-plan.md b/docs/planning/implementation-plan.md index fc69b7b0..73ec783c 100644 --- a/docs/planning/implementation-plan.md +++ b/docs/planning/implementation-plan.md @@ -1,3 +1,7 @@ +## M3 진행 · 질문별 자료 선택 영속화 + +[질문 전용 검색 계획](../changes/question-source-plans.md)으로 원래 질문·선택 순서·현재 source hash를 저장한다. 같은 request는 재검색하지 않고 원래 자료를 다시 검증한다. no_match를 현실의 근거 부재·의미 판정으로 표시하지 않는다. 기존485실행/이전52증거·앱/기본worker/main을 유지하며 실제 제공자/예산 예약 변경0이다. 다자료 AI 실행·브리핑 변환/표시·부하 인수·개발 적용이 후속이다. + ## M3 검증 완료 · 근거 작업 접수·queue 전달 [접수·dispatch 계약](../changes/evidence-job-dispatch.md)·[검증 범위](../research/evidence-job-dispatch-validation.md). 신규23/새전체9,595개(Python8,775+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·독립 최종 수정 재검토 C0I0M0다. 실제 DBOSClient/queue·모의 SSE에서 동일 ID/예약1/HTTP1을 조건별로 확인했다. 실행 순서에 기대던 테스트는 결정적 gate로 보완했다. 기존481실행/이전46증거와 앱/기본worker/main은 유지한다. 질문 전용 검색·다자료/브리핑 변환/표시·운영 정책/부하·개발 적용은 다음이다. diff --git a/docs/planning/progress.md b/docs/planning/progress.md index 31210a7a..f0df95bc 100644 --- a/docs/planning/progress.md +++ b/docs/planning/progress.md @@ -1,3 +1,9 @@ +## 검증 완료 · 질문별 자료 선택 영속화 · 2026-10-26 + +[계약](../changes/question-source-plans.md)·[검증](../research/question-source-plans-validation.md). 질문만으로 기존 저장 색인에서 최대5자료를 선택하고 원문/제목 hash를 고정한다. 같은 요청은 재검색하지 않고 현재 권한·원문을 재검증하며 no_match도 유지한다. 실행9ecdce0/기존 검사 보완14e1585·신규62/전체9,657개(Python8,837+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0이다. 태스크 C0I0M1/전체 C0I0M2 후 문서 보완 재검토 C0I0M0이며 비차단 후속1개(표시 정보 단독 삭제 집중 검사)를 남겼다. 기존484실행은 그대로이고 승인한 migration 검사1개만 수정했으며 이전52증거를 보존했다. 실제 제공자/운영 변경0·앱/기본worker/main 유지다. 다자료 AI 실행·브리핑 표시·개발 적용은 후속이며 PR/Sites·자체QA 정리는 최종 인계를 따른다. + +최종 인계: [Draft PR112](https://github.com/DigestWorks/IssueDigest/pull/112)·[Sitesv134](https://lawdigest-rebuild-plan.ijaesinii.chatgpt.site/) public/native succeeded. 이전v133과 변경7문서의 정확 사본(7cf3552)을 보존했다. Site source `3704954703d97a9f2872d7a4022431ee9dc5ebcd`, saved `appgprj_6ac612dd42708191a7a0c8377be64d48~appgver_a85308e949cc8191b6ca110a73193d04`, deployment `appgdep_6aded7187e8481919a25f09f7239ef3d`다. HTML119/로컬대상3,536/공개텍스트365·비공개60+12항목 제외를 확인했다. 자체QA1/전용volume1 정리와 포트 닫힘을 확인했다. 사용자 탭 이동·신규 브라우저 검사0(기존 레이아웃의 텍스트 변경)이며 앱/기본worker/main 활성화0이다. 비차단 표시 정보 단독 삭제 검사1개는 후속이며 다섯 판단과 비용은 명세/검증 기록에 남겼다. 원격CI는 로컬 전체 검사와 구분한다. PR 채팅 첨부는 도구의100개 한도로 실패했으며 링크는 유효하다. 이후 인계는 저장소 원본을 따른다. + ## 검증 완료 · 근거 작업 접수·DBOS queue 전달 · 2026-10-26 [새 계약](../changes/evidence-job-dispatch.md)·[검증 기록](../research/evidence-job-dispatch-validation.md). 실행afac2d7/테스트보완b6cbd98·신규23/새전체9,595개(Python8,775+JS820)·정적 검사/빌드/OpenAPI·실패/skip/warning0·최종 수정 재검토 C0I0M0·미해결0다. 태스크/전체 검토의 minor1(중복 접수 테스트의 실행 순서 의존)은 실제 완료 선행 RED2건으로 재현하고 결정적 대기/해제로 보완했다. 최초 전체 중단 실행은 통과로 세지 않고 보존했다. 기존481실행/이전46증거·앱/기본worker/main을 유지하며 실제 제공자/운영 서버 변경0이다. 질문 전용 검색·브리핑 변환/표시는 후속이며 PR/Sites/자체QA 정리는 최종 인계를 따른다. diff --git a/docs/planning/tasks.md b/docs/planning/tasks.md index 5f8da30b..7e708aae 100644 --- a/docs/planning/tasks.md +++ b/docs/planning/tasks.md @@ -1,3 +1,14 @@ +## 구현·집중 검사 완료 · 질문별 자료 선택 영속화 · 2026-10-26 + +- [x] [질문 전용 검색·선택/hash 고정·재접수 계약](../changes/question-source-plans.md) 고정. +- [x] strict 계획 DTO·repository/migration·기존 lexical 검색 연결 TDD. +- [x] 역할/자료/동시성·변조·재검색0·no_match·예산/run/job 무변경 집중62개+승인된 기존 migration 검사1개(합계63개)·scoped 정적 검사. +- [x] 기존 파일484개 변경 없이 보존·승인된 기존 migration 검사1개만 수정·이전 증거52개 보존. +- [x] 독립 최종 검토 C0I0M2 비차단·lazy presentation 복구 계약 문서화. +- [ ] 전체QA·PR/Sites·자체QA 정리. +- [ ] presentation만 삭제한 saved get/retry 집중 검사(비차단 후속). +- [ ] 다자료 AI 실행·브리핑 변환/표시·운영 부하·개발 반영. + ## 검증 완료 · 근거 작업 접수·queue 전달 · 2026-10-26 - [x] [접수/queue 경계·명시 버전·ack 유실 계약](../changes/evidence-job-dispatch.md) 고정. diff --git a/docs/research/question-source-plans-validation.md b/docs/research/question-source-plans-validation.md new file mode 100644 index 00000000..7cef1eef --- /dev/null +++ b/docs/research/question-source-plans-validation.md @@ -0,0 +1,31 @@ +# 질문별 자료 선택 영속 계획 검증 + +2026-10-26 KST. [명세](../changes/question-source-plans.md)에 따라 질문만으로 기존 저장 색인의 자료를 선택하고 같은 요청의 선택을 보존하는 내부 계약이다. API/UI·다자료 AI 실행·브리핑 표시·개발 활성화는 후속이며 실제 제공자 호출·운영 서버/DB 변경은 0회다. + +## 구현 + +`QuestionPlanRepository.submit`은 현재 권한을 확인하고 기존 문자 기반 검색에서 최대5자료를 선택한다. 질문·순서·원문과 제목의 hash를 저장한다. 같은 사용자/요청은 검색하지 않고 저장한 선택과 현재 권한·원문을 재검증한다. 입력 변경·원문 삭제/변경·저장 행 변조는 재선택 없이 거부한다. `get`도 현재 읽기 권한과 최초 작성자의 현재 쓰기 권한을 확인한다. hash는 손상 탐지이며 서명이 아니다. + +저장 계획의 get/재접수에서도 기존 `_presentation`의 지연 복구를 허용한다. presentation 행만 없고 canonical 원문이 존재하면 기존 경로로 presentation을 재구성하며, 재구성한 원문/presentation의 hash가 저장값과 일치할 때만 원래 계획을 반환한다. 원문이 없거나 변경됐거나 재구성한 presentation hash가 다르면 계속 거부한다. presentation만 삭제한 경우의 별도 집중 검사는 비차단 후속으로 남기며 이번 검증 통과 항목으로 세지 않는다. + +검색어가 없거나 색인 결과가 없으면 명시적 no_match를 저장한다. 이후 자료가 추가돼도 같은 요청은 바뀌지 않으며 새 요청에서 검색한다. 이는 제한된 저장 검색 결과이며 현실에 근거가 없다는 뜻이나 의미 관련성 인증이 아니다. + +별도 metadata와0011 migration으로 새 테이블만 추가한다. 기존 작업 SQL gate를 권한 잠금보다 먼저 획득해 잠금 순서를 유지한다. AI·예약·research run·evidence job·queue 생성은 없다. + +## 검증 + +신규 집중 검사62개를 합성 PostgreSQL에서 통과했다. 실제 저장 색인·현재 세션/역할·원문/제목 hash·동일 요청 동시성·검색 재호출 금지·negative cache·raw/rehashed 변조·기존 예산/run/job 무변경을 확인했다. 실행9ecdce0/기존 검사 보완14e1585·전체9,657개(Python8,837+JS820)·Ruff/format/mypy·타입/빌드/OpenAPI가 통과했고 실패/skip/warning0이다. 태스크 검토 C0I0M1·전체 검토 C0I0M2 후 문서 보완f11df42의 재검토 C0I0M0이며, 표시 정보 단독 삭제 집중 검사는 비차단 후속1개다. 실행 파일은 전체 검사 시작 이후 바꾸지 않았다. + +새 migration을 추가하자 기존0010 검사가 최신 head 전체를 downgrade하고0010까지만 복구하는 가정이 실패했다. 실제 RED를 확인한 뒤 동일 연결/트랜잭션에서0010 revision의 실제 upgrade/downgrade만 검사하도록 기존 검사1개를 수정했다. version stamp나 schema 우회 없이 최신 테이블/version을 유지한다. 전체 migration 경로는 별도 검사한다. 원래485파일의 사전 hash를 보존하고 이 예외 외484파일·이전52증거의 동일성을 확인한다. + +## 판단·비용·후속 + +1. 기존 문자 검색을 사용해 추가 AI 비용을 없앴다. 의미 관련성·자료 완전성은 보장하지 않는다. +2. 요청별 선택과 no_match를 고정했다. 새 자료 반영은 새 요청이 필요하다. +3. 작업 SQL gate를 공유했다. 잠금 순서를 유지하지만 검색 중 다른 작업이 기다릴 수 있어 운영 부하 인수가 필요하다. +4. 기존 migration 검사1개를 해당 revision 작업에 한정했다. 검사 파일 수정 비용이 있으며 기존484파일 보존과 전체 경로 검증을 별도로 유지한다. +5. 동일 hash의 표시 정보 복원을 유지한다. 표시 정보 삭제만으로는 실패하지 않으며, 이 경우만의 집중 회귀 검사는 후속으로 남는다. + +선택한 여러 자료를 영속 AI 작업으로 연결하고 결과를 브리핑으로 변환하는 경로는 아직 연결하지 않았다. 현재 앱·기본 worker·main은 유지한다. + +최종 인계: [Draft PR112](https://github.com/DigestWorks/IssueDigest/pull/112)·[Sitesv134](https://lawdigest-rebuild-plan.ijaesinii.chatgpt.site/) public/native succeeded. 이전v133과 변경7문서의 정확 사본(7cf3552)을 보존했다. Site source `3704954703d97a9f2872d7a4022431ee9dc5ebcd`, saved `appgprj_6ac612dd42708191a7a0c8377be64d48~appgver_a85308e949cc8191b6ca110a73193d04`, deployment `appgdep_6aded7187e8481919a25f09f7239ef3d`다. HTML119/로컬대상3,536/공개텍스트365·비공개60+12항목 제외를 확인했다. 자체QA1/전용volume1 정리와 포트 닫힘을 확인했다. 사용자 탭 이동·신규 브라우저 검사0(기존 레이아웃의 텍스트 변경)이며 앱/기본worker/main 활성화0이다. 비차단 표시 정보 단독 삭제 검사1개는 후속이며 다섯 판단과 비용은 명세/검증 기록에 남겼다. 원격CI는 로컬 전체 검사와 구분한다. PR 채팅 첨부는 도구의100개 한도로 실패했으며 링크는 유효하다. 이후 인계는 저장소 원본을 따른다. diff --git a/packages/domain/src/issuedigest_domain/question_source_plan.py b/packages/domain/src/issuedigest_domain/question_source_plan.py new file mode 100644 index 00000000..139c0f2e --- /dev/null +++ b/packages/domain/src/issuedigest_domain/question_source_plan.py @@ -0,0 +1,40 @@ +"""Immutable, question-only selection from the stored literal evidence index.""" + +from typing import Literal, Self + +from pydantic import UUID4, BaseModel, ConfigDict, Field, model_validator + + +class QuestionSourceChoice(BaseModel): + model_config = ConfigDict(strict=True, frozen=True, extra="forbid") + + snapshot_id: UUID4 + source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$", min_length=64, max_length=64) + + +class QuestionSourcePlan(BaseModel): + model_config = ConfigDict(strict=True, frozen=True, extra="forbid") + + profile: Literal["question-source-plan-v1"] = "question-source-plan-v1" + retrieval_profile: Literal["question-topics-ko-v2"] = "question-topics-ko-v2" + plan_id: UUID4 + workspace_id: UUID4 + request_id: UUID4 + question: str = Field(min_length=1, max_length=1500) + status: Literal["selected", "no_match"] + reason: Literal["no_terms", "no_candidates"] | None + sources: tuple[QuestionSourceChoice, ...] = Field(max_length=5) + + @model_validator(mode="after") + def consistent_selection(self) -> Self: + if not self.question.strip(): + raise ValueError("Question must be nonblank") + ids = tuple(source.snapshot_id for source in self.sources) + if len(ids) != len(set(ids)): + raise ValueError("Sources must be unique") + if self.status == "selected": + if not self.sources or self.reason is not None: + raise ValueError("Invalid selected plan") + elif self.sources or self.reason is None: + raise ValueError("Invalid no-match plan") + return self diff --git a/packages/domain/tests/test_question_source_plan.py b/packages/domain/tests/test_question_source_plan.py new file mode 100644 index 00000000..23a4400f --- /dev/null +++ b/packages/domain/tests/test_question_source_plan.py @@ -0,0 +1,73 @@ +"""Strict immutable question selection contracts, without source or actor data.""" + +import importlib +from uuid import uuid4 + +import pytest +from pydantic import ValidationError + + +def api(): + name = "issuedigest_domain.question_source_plan" + assert importlib.util.find_spec(name), "Missing question selection domain" + return importlib.import_module(name) + + +def fields(): + return dict( + plan_id=uuid4(), + workspace_id=uuid4(), + request_id=uuid4(), + question=" exact ", + status="selected", + reason=None, + sources=(api().QuestionSourceChoice(snapshot_id=uuid4(), source_sha256="a" * 64),), + ) + + +def test_roundtrip_preserves_question_and_freezes(): + plan = api().QuestionSourcePlan(**fields()) + assert plan.question == " exact " + assert api().QuestionSourcePlan.model_validate_json(plan.model_dump_json(), strict=True) == plan + with pytest.raises(ValidationError): + plan.question = "changed" + + +@pytest.mark.parametrize( + "key,value", + [ + ("question", True), + ("question", " "), + ("question", "a" * 1501), + ("plan_id", str(uuid4())), + ("workspace_id", True), + ("status", "pending"), + ("reason", "no_terms"), + ("sources", ()), + ("actor_id", uuid4()), + ("profile", "other"), + ("retrieval_profile", "other"), + ], +) +def test_rejects_invalid_plan(key, value): + data = fields() + data[key] = value + with pytest.raises(ValidationError): + api().QuestionSourcePlan(**data) + + +def test_choice_and_state_constraints(): + cls = api().QuestionSourcePlan + data = fields() + for choices in (data["sources"] * 2, data["sources"] * 6, list(data["sources"])): + with pytest.raises(ValidationError): + cls(**(data | {"sources": choices})) + for reason in ("no_terms", "no_candidates"): + assert ( + cls(**(data | {"status": "no_match", "reason": reason, "sources": ()})).reason == reason + ) + with pytest.raises(ValidationError): + cls(**(data | {"status": "no_match", "sources": ()})) + for hashed in ("A" * 64, "a" * 63, True): + with pytest.raises(ValidationError): + api().QuestionSourceChoice(snapshot_id=uuid4(), source_sha256=hashed) diff --git a/packages/storage/migrations/versions/0011_question_source_plans.py b/packages/storage/migrations/versions/0011_question_source_plans.py new file mode 100644 index 00000000..dfa38217 --- /dev/null +++ b/packages/storage/migrations/versions/0011_question_source_plans.py @@ -0,0 +1,17 @@ +"""Isolated durable question-only source plans.""" + +from alembic import op +from issuedigest_storage.question_plans import question_source_plans + +revision = "0011_question_source_plans" +down_revision = "0010_evidence_jobs" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + question_source_plans.create(op.get_bind(), checkfirst=False) + + +def downgrade() -> None: + question_source_plans.drop(op.get_bind(), checkfirst=False) diff --git a/packages/storage/src/issuedigest_storage/question_plans.py b/packages/storage/src/issuedigest_storage/question_plans.py new file mode 100644 index 00000000..23cb05ae --- /dev/null +++ b/packages/storage/src/issuedigest_storage/question_plans.py @@ -0,0 +1,212 @@ +"""Durable lexical selections; no models, reservations or execution activation. + +All operations share the evidence-job SQL transaction gate before auth locks. +This also serializes bounded new searches with short evidence-job transactions; +throughput and operational load acceptance remain separate work. +Canonical hashes detect corruption, not replacement of all data and its hashes. +""" + +import json +from collections.abc import Mapping +from typing import Any +from uuid import UUID, uuid4 + +from issuedigest_domain.question_source_plan import QuestionSourceChoice, QuestionSourcePlan +from pydantic import ValidationError +from sqlalchemy import Column, MetaData, String, Table, UniqueConstraint, insert, select +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.dialects.postgresql import UUID as PGUUID +from sqlalchemy.engine import Connection, Engine + +from . import research_retrieval +from .evidence_jobs import EvidenceJobRepository, digest, source_digest +from .repository import EvidenceRepository, StorageCorruption +from .research import ResearchConflict, ResearchForbidden, ResearchRepository + +metadata = MetaData() +question_source_plans = Table( + "question_source_plans", + metadata, + *[ + Column(name, PGUUID(as_uuid=True), primary_key=name == "plan_id", nullable=False) + for name in ("plan_id", "workspace_id", "actor_id", "request_id") + ], + *[ + Column(name, String, nullable=False) + for name in ("question", "input_sha256", "payload_sha256") + ], + Column("payload", JSONB, nullable=False), + UniqueConstraint("workspace_id", "actor_id", "request_id"), +) + + +def _id(value: object) -> None: + if type(value) is not UUID or value.version != 4: + raise ValueError("A UUID v4 is required") + + +def _input(actor: UUID, workspace: UUID, request: UUID, question: str) -> str: + return digest( + dict( + actor_id=str(actor), + workspace_id=str(workspace), + request_id=str(request), + question=question, + profile="question-source-plan-v1", + retrieval_profile="question-topics-ko-v2", + ) + ) + + +class QuestionPlanRepository: + def __init__(self, engine: Engine): + self.engine = engine + self._base = ResearchRepository(engine) + + @staticmethod + def _source(conn: Connection, workspace: UUID, snapshot: UUID) -> str: + bundle = EvidenceRepository._load(conn, workspace, snapshot) + if bundle is None: + raise ResearchConflict("Question plan source unavailable") + try: + presentation = ResearchRepository._presentation(conn, workspace, snapshot, bundle) + except (ValidationError, ValueError, TypeError): + raise StorageCorruption("Question plan source changed") from None + return source_digest((snapshot, bundle, presentation)) + + @staticmethod + def _dto(conn: Connection, row: Mapping[Any, Any]) -> QuestionSourcePlan: + try: + for field in ("plan_id", "workspace_id", "actor_id", "request_id"): + _id(row[field]) + raw = row["payload"] + if type(raw) is not dict or set(raw) != set(QuestionSourcePlan.model_fields): + raise ValueError + if type(raw["sources"]) is not list or any( + type(source) is not dict or set(source) != set(QuestionSourceChoice.model_fields) + for source in raw["sources"] + ): + raise ValueError + plan = QuestionSourcePlan.model_validate_json( + json.dumps(raw, ensure_ascii=False, allow_nan=False), strict=True + ) + if ( + plan.plan_id != row["plan_id"] + or plan.workspace_id != row["workspace_id"] + or plan.request_id != row["request_id"] + or type(row["question"]) is not str + or plan.question != row["question"] + or row["payload_sha256"] != digest(dict(actor_id=str(row["actor_id"]), payload=raw)) + or row["input_sha256"] + != _input(row["actor_id"], plan.workspace_id, plan.request_id, plan.question) + ): + raise ValueError + except (ValidationError, ValueError, TypeError, KeyError): + raise StorageCorruption("Question plan integrity check failed") from None + if not ResearchRepository._member(conn, row["actor_id"], plan.workspace_id, write=True): + raise ResearchForbidden("Question plan authority unavailable") + for source in plan.sources: + if ( + QuestionPlanRepository._source(conn, plan.workspace_id, source.snapshot_id) + != source.source_sha256 + ): + raise StorageCorruption("Question plan source changed") + return plan + + def submit( + self, token: str, workspace_id: UUID, request_id: UUID, question: str + ) -> QuestionSourcePlan | None: + _id(workspace_id) + _id(request_id) + if type(question) is not str or not question.strip() or len(question) > 1500: + raise ValueError("Invalid question plan input") + + def save(conn: Connection) -> QuestionSourcePlan | None: + EvidenceJobRepository._lock_jobs(conn) + actor = self._base._actor(conn, token, workspace_id, write=True) + if actor is None: + return None + prior = ( + conn.execute( + select(question_source_plans) + .where( + question_source_plans.c.workspace_id == workspace_id, + question_source_plans.c.actor_id == actor, + question_source_plans.c.request_id == request_id, + ) + .with_for_update() + ) + .mappings() + .one_or_none() + ) + if prior is not None: + plan = self._dto(conn, prior) + if plan.question != question: + raise ResearchConflict("Question plan request changed") + return plan + terms = research_retrieval.question_terms(question) + ids = ( + research_retrieval.retrieve_snapshot_ids(conn, workspace_id, terms) if terms else () + ) + try: + if type(ids) is not tuple or len(ids) > 5 or len(set(ids)) != len(ids): + raise ValueError + for identity in ids: + _id(identity) + except (TypeError, ValueError): + raise StorageCorruption("Question plan search result invalid") from None + sources = tuple( + QuestionSourceChoice( + snapshot_id=identity, source_sha256=self._source(conn, workspace_id, identity) + ) + for identity in ids + ) + plan = QuestionSourcePlan( + plan_id=uuid4(), + workspace_id=workspace_id, + request_id=request_id, + question=question, + status="selected" if sources else "no_match", + reason=None if sources else "no_candidates" if terms else "no_terms", + sources=sources, + ) + payload = plan.model_dump(mode="json") + conn.execute( + insert(question_source_plans).values( + plan_id=plan.plan_id, + workspace_id=workspace_id, + actor_id=actor, + request_id=request_id, + question=question, + input_sha256=_input(actor, workspace_id, request_id, question), + payload=payload, + payload_sha256=digest(dict(actor_id=str(actor), payload=payload)), + ) + ) + return plan + + return self._base._transaction(save) + + def get(self, token: str, workspace_id: UUID, plan_id: UUID) -> QuestionSourcePlan | None: + _id(workspace_id) + _id(plan_id) + + def read(conn: Connection) -> QuestionSourcePlan | None: + EvidenceJobRepository._lock_jobs(conn) + if self._base._actor(conn, token, workspace_id) is None: + return None + row = ( + conn.execute( + select(question_source_plans) + .where( + question_source_plans.c.workspace_id == workspace_id, + question_source_plans.c.plan_id == plan_id, + ) + .with_for_update() + ) + .mappings() + .one_or_none() + ) + return self._dto(conn, row) if row is not None else None + + return self._base._transaction(read) diff --git a/packages/storage/tests/test_evidence_jobs.py b/packages/storage/tests/test_evidence_jobs.py index 843b93ea..a423f10c 100644 --- a/packages/storage/tests/test_evidence_jobs.py +++ b/packages/storage/tests/test_evidence_jobs.py @@ -243,27 +243,37 @@ def test_atomic_duplicate_claims(jobs, engine): def test_migration_only_new_table_and_old_rows(jobs, engine, database_url): - from alembic import command + from alembic.migration import MigrationContext + from alembic.operations import Operations + from alembic.script import ScriptDirectory from sqlalchemy import inspect from conftest import migration_config token, workspace, snapshot = seed(engine) - with engine.connect() as conn: - old = conn.execute(text("SELECT snapshot_id,workspace_id FROM evidence_snapshots")).all() config = migration_config(database_url) - before = set(inspect(engine).get_table_names()) - command.downgrade(config, "0009_policy_intake") - try: - assert set(inspect(engine).get_table_names()) == before - {"evidence_jobs"} - with engine.connect() as conn: - assert ( - conn.execute(text("SELECT snapshot_id,workspace_id FROM evidence_snapshots")).all() - == old - ) - finally: - command.upgrade(config, "0010_evidence_jobs") - assert set(inspect(engine).get_table_names()) == before + revision = ScriptDirectory.from_config(config).get_revision("0010_evidence_jobs") + with engine.begin() as conn: + old = conn.execute(text("SELECT snapshot_id,workspace_id FROM evidence_snapshots")).all() + version = conn.execute(text("SELECT version_num FROM alembic_version")).all() + before = set(inspect(conn).get_table_names()) + with Operations.context(MigrationContext.configure(conn)): + revision.module.downgrade() + try: + assert set(inspect(conn).get_table_names()) == before - {"evidence_jobs"} + assert ( + conn.execute( + text("SELECT snapshot_id,workspace_id FROM evidence_snapshots") + ).all() + == old + ) + assert ( + conn.execute(text("SELECT version_num FROM alembic_version")).all() == version + ) + finally: + revision.module.upgrade() + assert set(inspect(conn).get_table_names()) == before + assert conn.execute(text("SELECT version_num FROM alembic_version")).all() == version def test_source_metadata_change_rejects_same_request(jobs, engine): diff --git a/packages/storage/tests/test_question_plans.py b/packages/storage/tests/test_question_plans.py new file mode 100644 index 00000000..4c6bce5a --- /dev/null +++ b/packages/storage/tests/test_question_plans.py @@ -0,0 +1,518 @@ +"""Synthetic PostgreSQL selection durability, authorization and corruption guards.""" + +import importlib +from uuid import uuid4 + +import pytest +from issuedigest_storage import EvidenceRepository, IdentityRepository, SessionRepository +from sqlalchemy import select, text, update +from test_evidence_jobs import seed +from test_research_retrieval import save + + +def api(): + name = "issuedigest_storage.question_plans" + assert importlib.util.find_spec(name), "Missing question selection repository" + return importlib.import_module(name) + + +@pytest.fixture +def plans(engine): + with engine.begin() as conn: + conn.execute(text("TRUNCATE question_source_plans")) + yield api().QuestionPlanRepository(engine) + with engine.begin() as conn: + conn.execute(text("TRUNCATE question_source_plans")) + + +def test_repository_exists(): + assert api().QuestionPlanRepository + + +def test_index_selection_is_ordered_fixed_and_has_no_paid_side_effects(plans, engine, monkeypatch): + token, workspace, snapshot = seed(engine) + repo = EvidenceRepository(engine) + repo.rebuild_projection(workspace, snapshot) + other = save(engine, workspace, "앞 앞 앞") + with engine.connect() as conn: + unchanged = { + table: conn.execute(text("SELECT * FROM " + table)).all() + for table in ("evidence_jobs", "research_runs", "research_budget_months") + } + request = uuid4() + plan = plans.submit(token, workspace, request, " 앞 ") + assert plan.question == " 앞 " and plan.status == "selected" + assert tuple(s.snapshot_id for s in plan.sources) == tuple(sorted((snapshot, other), key=str)) + save(engine, workspace, "앞") + + def forbidden(*args): + pytest.fail("saved plan searched again") + + monkeypatch.setattr(api().research_retrieval, "retrieve_snapshot_ids", forbidden) + assert plans.submit(token, workspace, request, " 앞 ") == plan + assert plans.get(token, workspace, plan.plan_id) == plan + with pytest.raises(api().ResearchConflict): + plans.submit(token, workspace, request, "앞") + with engine.connect() as conn: + for table, rows in unchanged.items(): + assert conn.execute(text("SELECT * FROM " + table)).all() == rows + assert conn.scalar(text("SELECT count(*) FROM question_source_plans")) == 1 + + +@pytest.mark.parametrize("question,reason", [("?!", "no_terms"), ("희귀주제", "no_candidates")]) +def test_no_match_is_durable_after_new_corpus(plans, engine, monkeypatch, question, reason): + token, workspace, _ = seed(engine) + request = uuid4() + if reason == "no_terms": + monkeypatch.setattr( + api().research_retrieval, + "retrieve_snapshot_ids", + lambda *args: pytest.fail("empty terms searched"), + ) + plan = plans.submit(token, workspace, request, question) + assert plan.status == "no_match" and plan.reason == reason and plan.sources == () + save(engine, workspace, "희귀주제") + assert plans.submit(token, workspace, request, question) == plan + if reason == "no_candidates": + assert plans.submit(token, workspace, uuid4(), question).status == "selected" + + +@pytest.mark.parametrize( + "column,value", + [ + ("question", "changed"), + ("actor_id", uuid4()), + ("input_sha256", "f" * 64), + ("payload_sha256", "f" * 64), + ("request_id", uuid4()), + ], +) +def test_row_tampering_rejects(plans, engine, column, value): + token, workspace, _ = seed(engine) + plan = plans.submit(token, workspace, uuid4(), "?!") + with engine.begin() as conn: + conn.execute(update(api().question_source_plans).values(**{column: value})) + with pytest.raises(api().StorageCorruption): + plans.get(token, workspace, plan.plan_id) + + +@pytest.mark.parametrize("change", ["extra", "missing", "bool", "profile", "state", "source", "id"]) +def test_rehashed_invalid_payload_rejects(plans, engine, change): + token, workspace, _ = seed(engine) + plan = plans.submit(token, workspace, uuid4(), "?!") + payload = plan.model_dump(mode="json") + if change == "extra": + payload["receipt"] = {} + elif change == "missing": + del payload["reason"] + elif change == "bool": + payload["question"] = True + elif change == "profile": + payload["profile"] = "changed" + elif change == "state": + payload["status"] = "selected" + elif change == "source": + payload["sources"] = [dict(snapshot_id=str(uuid4()), source_sha256="a" * 64)] + else: + payload["plan_id"] = str(uuid4()) + with engine.begin() as conn: + actor = conn.scalar(select(api().question_source_plans.c.actor_id)) + conn.execute( + update(api().question_source_plans).values( + payload=payload, + payload_sha256=api().digest(dict(actor_id=str(actor), payload=payload)), + ) + ) + with pytest.raises(api().StorageCorruption): + plans.get(token, workspace, plan.plan_id) + + +def test_current_reader_writer_and_workspace_rules(plans, engine): + from issuedigest_storage.auth_schema import auth_workspace_memberships + from issuedigest_storage.sessions import AuthenticationDenied + + token, workspace, _ = seed(engine) + plan = plans.submit(token, workspace, uuid4(), "?!") + viewer = uuid4() + identity = IdentityRepository(engine) + identity.register_identity(viewer, "synthetic", str(viewer)) + identity.set_membership(viewer, workspace, "viewer") + vt = SessionRepository(engine).issue_session(viewer).token + assert plans.get(vt, workspace, plan.plan_id) == plan + assert plans.get(vt, uuid4(), plan.plan_id) is None + with pytest.raises(api().ResearchForbidden): + plans.submit(vt, workspace, uuid4(), "q") + with pytest.raises(AuthenticationDenied): + plans.get("invalid", workspace, plan.plan_id) + with engine.begin() as conn: + conn.execute( + update(auth_workspace_memberships) + .where(auth_workspace_memberships.c.user_id != viewer) + .values(role="viewer") + ) + with pytest.raises(api().ResearchForbidden): + plans.get(vt, workspace, plan.plan_id) + + +@pytest.mark.parametrize("damage", ["title", "bundle", "delete"]) +def test_original_source_revalidated_on_get_and_retry(plans, engine, damage): + from issuedigest_storage.research import ResearchRepository + + token, workspace, snapshot = seed(engine) + EvidenceRepository(engine).rebuild_projection(workspace, snapshot) + request = uuid4() + plan = plans.submit(token, workspace, request, "앞") + with engine.begin() as conn: + if damage == "title": + bundle = EvidenceRepository._load(conn, workspace, snapshot) + presentation = ResearchRepository._presentation(conn, workspace, snapshot, bundle) + ResearchRepository._upsert_presentation( + conn, presentation.model_copy(update={"title": "changed"}) + ) + elif damage == "bundle": + conn.execute( + text( + "UPDATE parsed_representations SET " + "payload=jsonb_set(payload,'{text}', '\"changed\"')" + ) + ) + else: + for table in ( + "evidence_search_units", + "evidence_search_projections", + "source_presentations", + ): + conn.execute( + text("DELETE FROM " + table + " WHERE snapshot_id=:s"), {"s": snapshot} + ) + conn.execute( + text("DELETE FROM snapshot_citations WHERE snapshot_id=:s"), {"s": snapshot} + ) + conn.execute(text("DELETE FROM outbox WHERE snapshot_id=:s"), {"s": snapshot}) + conn.execute( + text("DELETE FROM evidence_snapshots WHERE snapshot_id=:s"), {"s": snapshot} + ) + for call in ( + lambda: plans.get(token, workspace, plan.plan_id), + lambda: plans.submit(token, workspace, request, "앞"), + ): + with pytest.raises((api().StorageCorruption, api().ResearchConflict)): + call() + + +@pytest.mark.parametrize( + "result", + [(True,), (str(uuid4()),), (uuid4(),) * 2, tuple(uuid4() for _ in range(6)), [uuid4()]], +) +def test_malformed_retrieval_fails_closed(plans, engine, monkeypatch, result): + token, workspace, _ = seed(engine) + monkeypatch.setattr(api().research_retrieval, "retrieve_snapshot_ids", lambda *args: result) + with pytest.raises(api().StorageCorruption): + plans.submit(token, workspace, uuid4(), "topic") + with engine.connect() as conn: + assert conn.scalar(text("SELECT count(*) FROM question_source_plans")) == 0 + + +def test_migration_and_import_leave_old_rows_and_registry_untouched(plans, engine, database_url): + from decimal import Decimal + + from alembic import command + from issuedigest_storage.evidence_jobs import EvidenceJobRepository + from issuedigest_storage.schema import metadata as old_metadata + from sqlalchemy import inspect + + from conftest import migration_config + + token, workspace, snapshot = seed(engine) + job = EvidenceJobRepository(engine).submit( + token, + workspace, + uuid4(), + snapshot, + "q", + model="gpt-6.1-sol", + reservation_usd=Decimal(".25"), + ) + plans.submit(token, workspace, uuid4(), "?!") + old_registry = set(old_metadata.tables) + assert "question_source_plans" not in old_registry + importlib.reload(api()) + assert set(old_metadata.tables) == old_registry + with engine.connect() as conn: + old = { + table: conn.execute(text("SELECT * FROM " + table)).all() + for table in ("evidence_jobs", "research_budget_months", "evidence_snapshots") + } + before = set(inspect(engine).get_table_names()) + config = migration_config(database_url) + command.downgrade(config, "0010_evidence_jobs") + try: + assert set(inspect(engine).get_table_names()) == before - {"question_source_plans"} + with engine.connect() as conn: + for table, rows in old.items(): + assert conn.execute(text("SELECT * FROM " + table)).all() == rows + finally: + command.upgrade(config, "0011_question_source_plans") + assert set(inspect(engine).get_table_names()) == before + assert EvidenceJobRepository(engine).get(token, workspace, job.job_id).status == "pending" + with engine.begin() as conn: + conn.execute(text("DELETE FROM evidence_jobs WHERE job_id=:j"), {"j": job.job_id}) + + +@pytest.mark.parametrize("operation", ["submit", "get"]) +def test_shared_job_gate_precedes_auth_and_serializes_duplicate_submission( + plans, engine, operation +): + from concurrent.futures import ThreadPoolExecutor + from threading import Event, get_ident + from time import monotonic + + from issuedigest_storage.evidence_jobs import EvidenceJobRepository + from sqlalchemy import event + + token, workspace, _ = seed(engine) + request = uuid4() + existing = plans.submit(token, workspace, request, "?!") if operation == "get" else None + held, release, started = Event(), Event(), Event() + ids = {} + + def before(conn, cursor, statement, parameters, context, executemany): + thread = get_ident() + if thread == ids.get("second"): + ids.setdefault("second_pid", conn.connection.driver_connection.info.backend_pid) + started.set() + if thread in (ids.get("first"), ids.get("second")): + first_statement = ids.setdefault((thread, "first_sql"), statement) + assert "pg_advisory_xact_lock" in first_statement + + def after(conn, cursor, statement, parameters, context, executemany): + if ( + get_ident() == ids.get("first") + and "auth_workspace_memberships" in statement + and not held.is_set() + ): + ids["first_pid"] = conn.connection.driver_connection.info.backend_pid + held.set() + assert release.wait(10) + + def first(): + ids["first"] = get_ident() + return ( + plans.get(token, workspace, existing.plan_id) + if existing + else plans.submit(token, workspace, request, "?!") + ) + + def second(): + ids["second"] = get_ident() + if existing: + + def check(conn): + EvidenceJobRepository._lock_jobs(conn) + return plans._base._actor(conn, token, workspace) + + return plans._base._transaction(check) + return plans.submit(token, workspace, request, "?!") + + event.listen(engine, "before_cursor_execute", before) + event.listen(engine, "after_cursor_execute", after) + try: + with ThreadPoolExecutor(max_workers=2) as pool: + one = pool.submit(first) + try: + assert held.wait(5) + two = pool.submit(second) + assert started.wait(5) + with engine.connect() as observer: + deadline = monotonic() + 5 + while ids["first_pid"] not in observer.scalar( + text("SELECT pg_blocking_pids(:p)"), {"p": ids["second_pid"]} + ): + assert monotonic() < deadline + Event().wait(0.01) + release.set() + result, duplicate = one.result(10), two.result(10) + assert result is not None and duplicate is not None + if not existing: + assert result == duplicate + finally: + release.set() + finally: + event.remove(engine, "before_cursor_execute", before) + event.remove(engine, "after_cursor_execute", after) + with engine.connect() as conn: + assert conn.scalar(text("SELECT count(*) FROM question_source_plans")) == 1 + + +@pytest.mark.parametrize( + "field,value", + [ + ("workspace_id", True), + ("request_id", "not-a-uuid"), + ("question", True), + ("question", " "), + ("question", "x" * 1501), + ], +) +def test_invalid_inputs_fail_before_sql(plans, engine, field, value): + from sqlalchemy import event + + data = dict(token="invalid", workspace_id=uuid4(), request_id=uuid4(), question="q") + data[field] = value + + def forbidden(*args): + pytest.fail("invalid input issued SQL") + + event.listen(engine, "before_cursor_execute", forbidden) + try: + with pytest.raises(ValueError): + plans.submit(**data) + with pytest.raises(ValueError): + plans.get("invalid", uuid4(), True) + finally: + event.remove(engine, "before_cursor_execute", forbidden) + + +def test_different_actors_are_separate_but_current_members_can_read(plans, engine): + token, workspace, _ = seed(engine) + actor = uuid4() + identity = IdentityRepository(engine) + identity.register_identity(actor, "synthetic", str(actor)) + identity.set_membership(actor, workspace, "editor") + second_token = SessionRepository(engine).issue_session(actor).token + request = uuid4() + one = plans.submit(token, workspace, request, "?!") + two = plans.submit(second_token, workspace, request, "?!") + assert one.plan_id != two.plan_id + assert plans.get(second_token, workspace, one.plan_id) == one + assert plans.get(token, workspace, two.plan_id) == two + assert plans.get(token, workspace, uuid4()) is None + + +@pytest.mark.parametrize( + "damage", + ["duplicate", "oversized", "source_bool", "source_extra", "source_hash", "reason_bool"], +) +def test_rehashed_selected_shapes_and_hashes_are_checked(plans, engine, damage): + token, workspace, snapshot = seed(engine) + EvidenceRepository(engine).rebuild_projection(workspace, snapshot) + plan = plans.submit(token, workspace, uuid4(), "앞") + raw = plan.model_dump(mode="json") + if damage == "duplicate": + raw["sources"] *= 2 + elif damage == "oversized": + raw["sources"] = [dict(snapshot_id=str(uuid4()), source_sha256="a" * 64) for _ in range(6)] + elif damage == "source_bool": + raw["sources"][0]["snapshot_id"] = True + elif damage == "source_extra": + raw["sources"][0]["quote"] = "forbidden" + elif damage == "source_hash": + raw["sources"][0]["source_sha256"] = "f" * 64 + else: + raw["reason"] = False + with engine.begin() as conn: + actor = conn.scalar(select(api().question_source_plans.c.actor_id)) + conn.execute( + update(api().question_source_plans).values( + payload=raw, payload_sha256=api().digest(dict(actor_id=str(actor), payload=raw)) + ) + ) + with pytest.raises(api().StorageCorruption): + plans.get(token, workspace, plan.plan_id) + + +@pytest.mark.parametrize( + "change", ["revoke_session", "expire_session", "idle_session", "disable_user", "remove_member"] +) +def test_current_session_and_membership_are_required(plans, engine, change): + from issuedigest_storage.sessions import AuthenticationDenied + + token, workspace, _ = seed(engine) + request = uuid4() + plan = plans.submit(token, workspace, request, "?!") + with engine.begin() as conn: + statement = { + "revoke_session": "UPDATE auth_sessions SET revoked=true", + "expire_session": ( + "UPDATE auth_sessions SET issued_at=clock_timestamp()-interval '3 days', " + "last_activity_at=clock_timestamp()-interval '2 days', " + "expires_at=clock_timestamp()-interval '1 day'" + ), + "idle_session": ( + "UPDATE auth_sessions SET issued_at=clock_timestamp()-interval '9 days', " + "expires_at=clock_timestamp()+interval '4 days', " + "last_activity_at=clock_timestamp()-interval '8 days'" + ), + "disable_user": "UPDATE auth_users SET enabled=false", + "remove_member": "UPDATE auth_workspace_memberships SET active=false", + }[change] + conn.execute(text(statement)) + for call in ( + lambda: plans.get(token, workspace, plan.plan_id), + lambda: plans.submit(token, workspace, request, "?!"), + ): + if "session" in change: + with pytest.raises(AuthenticationDenied): + call() + else: + assert call() is None + + +def test_saved_originals_do_not_follow_new_index_or_other_workspace(plans, engine): + token, workspace, snapshot = seed(engine) + EvidenceRepository(engine).rebuild_projection(workspace, snapshot) + alien = save(engine, uuid4(), "앞") + request = uuid4() + plan = plans.submit(token, workspace, request, "앞") + assert tuple(s.snapshot_id for s in plan.sources) == (snapshot,) + assert alien not in tuple(s.snapshot_id for s in plan.sources) + newer = save(engine, workspace, "앞") + assert plans.submit(token, workspace, request, "앞") == plan + assert set(s.snapshot_id for s in plans.submit(token, workspace, uuid4(), "앞").sources) == { + snapshot, + newer, + } + + +def test_real_index_caps_selection_at_five_in_stable_order(plans, engine): + token, workspace, _ = seed(engine) + ids = tuple(save(engine, workspace, "공통주제") for _ in range(7)) + plan = plans.submit(token, workspace, uuid4(), "공통주제") + assert tuple(source.snapshot_id for source in plan.sources) == tuple(sorted(ids, key=str)[:5]) + + +def test_shallow_model_copy_cannot_smuggle_unknown_fields(plans, engine): + token, workspace, _ = seed(engine) + plan = plans.submit(token, workspace, uuid4(), "?!") + copied = plan.model_copy(update={"question": False}) + raw = copied.model_dump(mode="json", warnings=False) + with engine.begin() as conn: + actor = conn.scalar(select(api().question_source_plans.c.actor_id)) + conn.execute( + update(api().question_source_plans).values( + payload=raw, payload_sha256=api().digest(dict(actor_id=str(actor), payload=raw)) + ) + ) + with pytest.raises(api().StorageCorruption): + plans.submit(token, workspace, plan.request_id, "?!") + + +def test_same_actor_request_in_two_workspaces_is_separate(plans, engine): + token, workspace, _ = seed(engine) + from issuedigest_storage.auth_schema import auth_sessions + from issuedigest_storage.sessions import token_digest + + with engine.connect() as conn: + actor = conn.scalar( + select(auth_sessions.c.user_id).where( + auth_sessions.c.token_sha256 == token_digest(token) + ) + ) + other = uuid4() + IdentityRepository(engine).set_membership(actor, other, "owner") + request = uuid4() + first = plans.submit(token, workspace, request, "?!") + second = plans.submit(token, other, request, "?!") + assert first.plan_id != second.plan_id + assert plans.get(token, other, first.plan_id) is None + assert plans.get(token, workspace, second.plan_id) is None