diff --git a/README.md b/README.md index a694d02..4fb6141 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,5 @@ +현재 구현: [질문별 순차 AI 작업 연결](docs/changes/question-plan-dispatch.md). 고정한 자료를 순서대로 기존 영속 작업으로 접수하고, 같은 요청은 완료한 분석을 재사용합니다. 기존 활성 작업1개·합산 예약 예산을 유지하며 브리핑 변환·자동 진행·API/UI·개발 적용은 후속입니다. 집중91/새 전체9,716개·정적/타입/빌드/OpenAPI·독립 최종 수정 검토 C0I0M0를 통과했습니다. + 진행 중: [질문별 자료 선택 영속 계획](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이며 질문 전용 검색·브리핑 변환/표시는 후속입니다. diff --git a/docs/README.md b/docs/README.md index 08497eb..d8e0824 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,3 +1,5 @@ +현재 구현: [질문별 순차 AI 작업 연결](changes/question-plan-dispatch.md). 질문 검색의 고정 선택과 기존 작업/queue를 연결하며 신규 외부 모델 호출이나 앱 활성화는 이번 검증에 포함하지 않습니다. 집중91/새 전체9,716개·정적/타입/빌드/OpenAPI·독립 최종 수정 검토 C0I0M0를 통과했습니다. + 진행 중: [질문별 자료 선택 영속 계획](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이며 질문 전용 검색·브리핑 변환/표시는 후속입니다. diff --git a/docs/changes/question-plan-dispatch.md b/docs/changes/question-plan-dispatch.md new file mode 100644 index 0000000..241155b --- /dev/null +++ b/docs/changes/question-plan-dispatch.md @@ -0,0 +1,34 @@ +# 질문별 자료의 순차 AI 작업 연결 + +2026-10-26 KST. [질문별 자료 선택](question-source-plans.md)의 다음 단계. 최대5자료의 고정 선택을 기존 영속 AI 작업으로 이어서 접수한다. 기존 workspace당 활성 작업1개 제한과 합산 예약 예산을 유지하며 완료된 자식은 재사용한다. caller가 완료 후 같은 요청으로 다시 진행하며 자동 부모workflow·브리핑 변환/API/UI/개발 적용은 후속이다. + +## Global Constraints + +- Owned execution files: NEW packages/agents/src/issuedigest_agents/question_dispatch.py and packages/agents/tests/test_question_dispatch.py; existing packages/storage/src/issuedigest_storage/evidence_jobs.py and packages/storage/tests/test_evidence_jobs.py only the expected-source admission guard and focused tests. Other existing execution files/provider bytes/SDK/workflows/queue/API/UI/migrations/config/dependencies/export/fixtures stay unchanged. Docs allowed. No real provider calls, operational server/DB, main merge or app/default-worker activation; dedicated local synthetic PostgreSQL/nativeDBOS/mockSSE only. +- Existing EvidenceJobRepository.submit gains keyword expected_source_sha256: str|None=None. Exact str lowercase64hex or None before SQL. Within SAME existing gate/auth/load/presentation transaction compare computed source_digest with nonNone expected hash BEFORE prior lookup/budget/newrow; mismatch sanitized ResearchConflict with rollback/no reservation/job/queue. Existing noarg calls behave identically; do not change hashing, budget, active-one-workspace index, job lifecycle, receipts, guards, worker. Compare on retry too. +- New async advance_question_plan(plans:QuestionPlanRepository,jobs:EvidenceJobRepository,*,client:DBOSClient,app_version,token,workspace_id,request_id,question,model,reservation_usd)->QuestionDispatch|None. Caller owns lifecycle/explicit app_version and re-invocation. Require plans.engine is jobs.engine, exact concrete repository classes or documented no arbitrary source-plan injection; validate UUID4/nonblank<=1500 question/exact gpt-6.1-sol/finite exactcentDecimal(0,10]/app_version ASCII1..128 before SQL, including no_match. No new model selection/pricing policy. +- Call existing plans.submit via asyncio.to_thread to get current writer-owned stable plan (do not accept caller source IDs/planDTO). Reconstruct plan via strict JSON raw field checks before use, do not trust model_copy, and match workspace/request/exact question. Existing source membership/currentwriter/allsourcehash guard remains. Zero choices -> QuestionDispatch status no_match, total0/completed0/dispatches(), no jobs/queue/reservation. +- Process fixed sources in order, at most one pending/running job. Derive deterministic child UUID4 using SHA256 fixed namespace 'issuedigest:question-plan-child:v1' + canonical plan_id/snapshot_id/order; set UUID version4 bits, never random on retry; document profile/algorithm. New fixed plan identity means a different request gets different children. Job admission uses original exact question/model/reservation and expected_source_sha256 from plan. No new batch table/migration or hidden persistence; child identities recover deterministically. No reservation flattening/refund/reset on exceptions. +- For each child, await jobs.submit with expected hash offeventloop; None returns None without enqueue/laterchildren. Revalidate current job with jobs.get before trusting status. Verify source_sha256 matches fixed selection, question/model/workspace/snapshot/UUID/status before paid dispatch; no borrowed receipt exposed. Prior completed => append opaque EvidenceDispatch queuedFalse and proceed next. First pending => use unchanged submit_evidence_job with same child request and args (it rechecks current source/hash against admitted input), append ack and stop, regardless races. running => append queuedFalse and stop. failed/cancelled/indeterminate => append queuedFalse and stop blocked; never automatically skip/retry paid execution. Existing one-active-job-per-workspace index and summed monthlybudget stay intact. All children completed => evidence_complete, NOT briefing/semantic certification. +- Frozen QuestionDispatch dataclass (plan_id,status Literal no_match/in_progress/blocked/evidence_complete,completed_count,total_sources,dispatches tuple of existing EvidenceDispatch), only internal identifiers/counters/status/acks; no question/token/source text/hash/receipt/AI facts. Counts reflect validated completed prefix; progress not generation guarantee. Do not poll/sleep/auto-loop/wait for completion; trusted caller invokes again. Do not export through API/UI or register/startworkers. This is sequential resumable admission/dispatch, not automatic parentworkflow or briefing conversion. +- Preserve asyncio.CancelledError, never hidden continuation/nextchild/queue on cancelled SQL waiter. Existing EvidenceDispatchError sanitized IDs may propagate (caller recovers same request); other existing sanitized auth/budget/conflict errors propagate. No token/question/original provider exception logging, no blanket catch swallowing unknown errors. Current per-source transaction can reserve/queue a prefix; errors do not undo dispatched work. Same request reuses children, no other request fallback. +- TDD tests meaningful: expectedhash invalid/mismatch/retry before budget/row writes; defaultoldsubmit behavior; no_match0; onepending stops; actual first completed then second scheduled with deterministic IDs and one reservation each; repeated request no duplicate, finalcomplete0newqueue; blocked terminal stops; revoked access/source/presentation changed; cancellation/ambiguousack preserves pending and no laterchildren; sameengine/invalidinputnoSQL; strict/model_copy tamper. Real PostgreSQL index and DBOSClient/mockSSE integration for >=2sources, no providers. Newfixture cleans own questionplans/jobs and restores monkeypatch modules; no conftest edits. Cover cross-source mutation race between plan and admission with expectedhash rollback. Agent scoped tests/Ruff/format/mypy, controller fullQA/reviews/privacy/PR/Sites/cleanup. No agent subagents. + +DBOS 통합 테스트는 qpd_ 전용 schema를 migrate/config/client에 동일하게 지정한다. client/worker 종료 후 해당 schema만 정리하고 기본 dbos schema/queue가 변경되지 않았음을 검사한다. 운영 설정 변경이 아닌 테스트 격리이며 실패한 첫 전체 실행은 통과로 세지 않는다. + +## Task 1 + +- [x] 위4파일 범위만 TDD로 구현한다. spec/global constraints를 지킨다. 새로운 expected-source guard는 실제 transaction에서 예약 전 적용한다. ordered deterministic children/완료 prefix/첫 미완료 중단/no_match/현재권한/원문/예산/취소/재접수 검사를 포함한다. owned4파일+이 Task1 checkbox만 커밋하고 SDD 보고서에 RED/GREEN/명령/한계/자체검토를 남긴다. + +## Controller gates + +- [x] 독립 태스크/전체/수정 검토·새 전체9,716개/정적/타입/빌드/OpenAPI·기존488실행/이전43증거 보존·privacy·문서/Draft PR113/Sitesv135·자체QA3/volume3 정리. +- [ ] 자동 진행 parentworkflow·다자료 결과의 브리핑 변환·API/UI·실제품질/운영부하·개발 적용(이번 제외). + +## 판단과 비용 + +1. 자료별 순차 접수와 완료 후 재진행으로 기존 활성1개/예산 계약을 유지한다. 자동 연속 실행은 상위 orchestration 후속이며 실패한 자료를 건너뛰지 않는다. +2. plan/source/order 기반 child UUID4를 결정적으로 유도한다. 새batchtable 없이 복구하지만 이 namespace/알고리즘은 이후 바꾸면 안 된다. +3. 예약 전 expected-source guard를 기존 repository에 선택적으로 추가한다. 원문 변경 경쟁을 막지만 기존 저장 코드1개와 검사1개를 수정한다. + +4. 새 native 테스트의 DBOS 저장 공간을 고유 schema로 격리한다. 기본 queue의 다른 앱 소유권과 충돌하지 않지만 테스트 준비/정리가 늘고 전체 재검사가 필요하다. diff --git a/docs/planning/implementation-plan.md b/docs/planning/implementation-plan.md index 73ec783..5f00509 100644 --- a/docs/planning/implementation-plan.md +++ b/docs/planning/implementation-plan.md @@ -1,3 +1,7 @@ +## M3 진행 · 질문별 순차 접수 검증 완료 + +[고정 자료→작업 연결](../changes/question-plan-dispatch.md)은 집중91/새 전체9,716개·정적/타입/빌드/OpenAPI·독립 최종 수정 검토 C0I0M0를 통과했다. 테스트 queue 소유권 간섭은 전용 schema 격리 후 새 DB 전체 재검증으로 해결했다. 기존 활성1개/예산 계약을 보존한다. 같은 요청의 완료 작업은 재사용하고 첫 미완료에서 반환한다. 자료 선택 때의 hash를 예약 전에 비교해 중간 원문 변경을 막는다. 자동 진행·브리핑 변환/표시·개발 적용은 후속이며 실제 제공자/운영/앱 활성화0이다. + ## M3 진행 · 질문별 자료 선택 영속화 [질문 전용 검색 계획](../changes/question-source-plans.md)으로 원래 질문·선택 순서·현재 source hash를 저장한다. 같은 request는 재검색하지 않고 원래 자료를 다시 검증한다. no_match를 현실의 근거 부재·의미 판정으로 표시하지 않는다. 기존485실행/이전52증거·앱/기본worker/main을 유지하며 실제 제공자/예산 예약 변경0이다. 다자료 AI 실행·브리핑 변환/표시·부하 인수·개발 적용이 후속이다. diff --git a/docs/planning/progress.md b/docs/planning/progress.md index f0df95b..da60c8a 100644 --- a/docs/planning/progress.md +++ b/docs/planning/progress.md @@ -1,3 +1,9 @@ +## 검증 완료 · 질문별 순차 AI 작업 연결 · 2026-10-26 + +[계약](../changes/question-plan-dispatch.md)·[검증](../research/question-plan-dispatch-validation.md). 고정 자료의 기대 hash를 예약 전 비교하고 결정적인 자식 요청으로 접수한다. 완료한 작업은 재사용하며 첫 미완료/실패에서 반환한다. 실행df8cd6f/테스트 격리 보완6c6f65b·집중91/새 전체9,716개(Python8,896+JS820)·정적/타입/빌드/OpenAPI·실패/skip/warning0이다. 태스크/전체/최종 수정 검토 C0I0M0다. 최초 전체 실행의 queue 소유권 충돌3건은 새 테스트만 고유 schema로 격리해 해결했고 새 DB에서 전체 재검증했다. 기존488실행/이전43증거 보존·실제 제공자/운영 변경0·앱/기본worker/main 유지다. 자동 진행·브리핑 변환/API/UI·실제 품질/부하·개발 반영은 후속이며 PR/Sites·자체QA 정리는 최종 인계를 따른다. + +최종 인계: [Draft PR113](https://github.com/DigestWorks/IssueDigest/pull/113)·[Sitesv135](https://lawdigest-rebuild-plan.ijaesinii.chatgpt.site/) public/native succeeded. 이전v134와 변경7문서 정확 사본(9a07d6c)을 보존했다. Site source `a4c5a44f81de77b4b6bc8b0abeaef9d425ed0902`, saved `appgprj_6ac612dd42708191a7a0c8377be64d48~appgver_6d7221835ff8819186bbfe4f474d2970`, deployment `appgdep_6adeea3abf4081919988caf0ba5c38b3`다. HTML120/로컬대상3,605/공개텍스트370·비공개60+12항목 제외를 확인했다. 자체QA3/전용volume3을 정확 소유권 확인 후 정리하고 포트 닫힘을 확인했다. 사용자 탭 이동/신규 브라우저 검사0(기존 레이아웃 텍스트 변경)이다. 실제 제공자/운영 변경·앱/기본worker/main 활성화0이며 자동 진행/브리핑/API/UI 인수는 후속이다. 네 판단과 비용은 명세/검증 기록에 남겼다. PR 채팅 첨부는 도구100개 한도로 실패했지만 링크는 유효하다. 원격CI는 로컬 전체 검사와 구분하며 이후 인계는 저장소 원본을 따른다. + ## 검증 완료 · 질문별 자료 선택 영속화 · 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 정리는 최종 인계를 따른다. diff --git a/docs/planning/tasks.md b/docs/planning/tasks.md index 7e708aa..8ec90e8 100644 --- a/docs/planning/tasks.md +++ b/docs/planning/tasks.md @@ -1,3 +1,11 @@ +## 진행 · 질문별 순차 AI 작업 연결 · 2026-10-26 + +- [x] [기대 원문 hash·순차 접수·자식 요청 ID 계약](../changes/question-plan-dispatch.md) 고정. +- [x] 예약 전 source guard·순차 dispatch TDD/실제DBOS 모의 stream 집중91개·정적 검사. +- [x] 독립 검토·새 전체9,716개/정적/타입/빌드/API 계약 검사·기존488실행/이전43증거 보존. +- [x] Draft PR113·Sitesv135/이전v134+정확7문서 보존·자체QA3/전용volume3 정리(최종 인계). +- [ ] 자동 진행·브리핑 변환/API/UI·실제품질/운영부하·개발 적용. + ## 구현·집중 검사 완료 · 질문별 자료 선택 영속화 · 2026-10-26 - [x] [질문 전용 검색·선택/hash 고정·재접수 계약](../changes/question-source-plans.md) 고정. diff --git a/docs/research/question-plan-dispatch-validation.md b/docs/research/question-plan-dispatch-validation.md new file mode 100644 index 0000000..d9621a7 --- /dev/null +++ b/docs/research/question-plan-dispatch-validation.md @@ -0,0 +1,29 @@ +# 질문별 순차 AI 작업 연결 검증 + +2026-10-26 KST. [명세](../changes/question-plan-dispatch.md). 고정한 질문별 선택을 기존 영속 분석 작업으로 접수하는 신뢰 서버용 경계다. 자동 부모 workflow·브리핑 변환·API/UI·개발 적용은 이번 범위가 아니다. 실제 제공자 호출·운영 서버/DB 변경·main/앱/기본 worker 활성화0이다. + +## 구현 계약 + +`advance_question_plan`은 기존 질문 계획을 현재 작성자의 동일 요청으로 재검증한다. 선택한 자료마다 고정 namespace/plan/source/order로 자식 UUID4를 유도하고 예약 전에 기대 원문 hash를 DB transaction 안에서 비교한다. 현재 원문이 다르면 예약·작업 저장 없이 거부한다. 기존 자료별 작업과 현재 권한·원문 검증을 재사용한다. + +완료한 자식은 재사용하고 첫 미완료에 도달하면 반환한다. pending은 기존 queue dispatch를 사용하고 running은 기다리는 상태만 반환하며 failed/cancelled/indeterminate는 이후 자료를 접수하지 않는다. 응답은 ID/상태/개수/queue ack만 담는다. `evidence_complete`는 분석 작업 완료이며 브리핑 완성·사실 인증이 아니다. no_match는 모델·예약·queue를 만들지 않는다. + +상위 caller가 완료 후 같은 요청을 다시 호출해야 진행한다. 일부 자료만 접수한 상태에서 중단될 수 있으며 이미 예약·접수한 작업을 되돌리거나 몰래 재실행하지 않는다. 동시 요청은 같은 자식과 기존 활성1개/합산 예산 제약을 사용한다. 원문이 바뀌었을 때 새 자료로 자동 대체하지 않는다. + +자식 요청 ID 유도 profile은 고정한다: UTF-8 `issuedigest:question-plan-child:v1|{소문자 하이픈 plan UUID}|{소문자 하이픈 snapshot UUID}|{0부터 시작하는 십진수 순서}`의 SHA256 첫16바이트에 UUID version4/표준 variant bit를 설정한다. 이는 비밀값/인증 token이 아닌 멱등성 키다. 이후 변경은 원래 요청의 복구를 깨므로 새 profile과 명시 이행이 필요하다. + +## 검증 + +집중 검사91개와 Ruff/format/mypy가 통과했다. 실제 DBOSClient·등록한 합성 queue·기존 worker·native 모의 SSE로 두 자료의 완료/재접수와 예약2개/모의 HTTP2회를 확인했다. 실제 호출 횟수가 아닌 차단된 모의 요청 수다. 실행df8cd6f/테스트 격리 보완6c6f65b·새 전체9,716개(Python8,896+JS820)·Ruff/format/mypy·타입/빌드/OpenAPI 통과, 실패/skip/warning0이다. 태스크/전체 검토 C0I0M0 후 자체QA에서 발견한 테스트 격리 문제1개를 고쳤고 수정 재검토도 C0I0M0다. 새 테스트→기존 queue검사6개 및 역순1개가 통과했으며 최종 전체 검사는 새 전용 DB에서 다시 실행했다. 처음 실패한 전체 실행은 통과로 세지 않았다. 합성 원문과 전용 PostgreSQL/native DBOS/모의 SSE만 사용하며 실제 제공자 응답·비용·의미 품질을 추정하지 않는다. + +## 판단과 비용 + +1. 자료별 순차 진행으로 기존 활성 작업1개 계약을 유지한다. 자동 연속 실행은 상위 orchestration 후속이며 실패한 자료를 건너뛰지 않는다. +2. 결정적인 자식 UUID4를 사용한다. 새batchtable은 필요 없지만 이 유도 알고리즘/namespace를 바꾸면 복구 계약이 깨진다. +3. 기존 예약 저장에 선택적 expected-source guard를 넣는다. 원문 변경 경쟁을 막지만 저장 코드/검사 각1파일을 수정한다. 나머지 기존488실행파일과 이전43증거는 보존한다. + +첫 전체 실행은 기존 queue 검사3개 실패/8,893개 통과로 종료해 통과로 세지 않았다. 원인은 새 native 테스트가 기본 dbos.queues에 다른 앱 이름으로 같은 고정 queue를 남긴 테스트 간섭이었다. 새 테스트만 고유 qpd_ schema를 사용하고 종료 후 정리하며, 기본 schema/queue 불변을 검사하도록 보완했다. 앱 실행 코드는 변경하지 않았다. 신선한 전용 DB에서 전체 검사를 다시 수행한다. + +4. 테스트 전용 DBOS schema를 격리했다. 기존 queue와의 충돌을 막는 대신 테스트 준비/정리와 전체 재검사 비용이 늘었다. + +최종 인계: [Draft PR113](https://github.com/DigestWorks/IssueDigest/pull/113)·[Sitesv135](https://lawdigest-rebuild-plan.ijaesinii.chatgpt.site/) public/native succeeded. 이전v134와 변경7문서 정확 사본(9a07d6c)을 보존했다. Site source `a4c5a44f81de77b4b6bc8b0abeaef9d425ed0902`, saved `appgprj_6ac612dd42708191a7a0c8377be64d48~appgver_6d7221835ff8819186bbfe4f474d2970`, deployment `appgdep_6adeea3abf4081919988caf0ba5c38b3`다. HTML120/로컬대상3,605/공개텍스트370·비공개60+12항목 제외를 확인했다. 자체QA3/전용volume3을 정확 소유권 확인 후 정리하고 포트 닫힘을 확인했다. 사용자 탭 이동/신규 브라우저 검사0(기존 레이아웃 텍스트 변경)이다. 실제 제공자/운영 변경·앱/기본worker/main 활성화0이며 자동 진행/브리핑/API/UI 인수는 후속이다. 네 판단과 비용은 명세/검증 기록에 남겼다. PR 채팅 첨부는 도구100개 한도로 실패했지만 링크는 유효하다. 원격CI는 로컬 전체 검사와 구분하며 이후 인계는 저장소 원본을 따른다. diff --git a/packages/agents/src/issuedigest_agents/question_dispatch.py b/packages/agents/src/issuedigest_agents/question_dispatch.py new file mode 100644 index 0000000..9d6ab35 --- /dev/null +++ b/packages/agents/src/issuedigest_agents/question_dispatch.py @@ -0,0 +1,238 @@ +"""Explicit, sequential admission of a writer-owned fixed source plan. + +The caller owns the DBOS client and application version and invokes again after +completion. Evidence completion is a transport result, never a briefing or a +semantic certification. A cancelled SQL waiter can leave an admitted child, but +has no hidden continuation to enqueue it or admit another child. +""" + +import asyncio +import json +import re +from dataclasses import dataclass +from decimal import Decimal +from hashlib import sha256 +from typing import Literal +from uuid import UUID + +from dbos import DBOSClient +from issuedigest_domain.question_source_plan import QuestionSourceChoice, QuestionSourcePlan +from issuedigest_storage.evidence_jobs import EvidenceJob, EvidenceJobRepository +from issuedigest_storage.question_plans import QuestionPlanRepository +from issuedigest_storage.repository import StorageCorruption +from pydantic import ValidationError + +from .evidence_dispatch import EvidenceDispatch, submit_evidence_job + + +@dataclass(frozen=True) +class QuestionDispatch: + """Opaque acknowledgements and validated completed-prefix counters only.""" + + plan_id: UUID + status: Literal["no_match", "in_progress", "blocked", "evidence_complete"] + completed_count: int + total_sources: int + dispatches: tuple[EvidenceDispatch, ...] + + +def _id(value: object) -> None: + if type(value) is not UUID or value.version != 4: + raise ValueError("A UUID v4 is required") + + +def _plan( + value: QuestionSourcePlan, workspace: UUID, request: UUID, question: str +) -> QuestionSourcePlan: + """Reconstruct raw fields, including shallow model_copy additions, strictly.""" + try: + if type(value) is not QuestionSourcePlan or set(vars(value)) != set( + QuestionSourcePlan.model_fields + ): + raise ValueError + for field in ("plan_id", "workspace_id", "request_id"): + _id(getattr(value, field)) + if type(value.sources) is not tuple: + raise ValueError + sources = [] + for source in value.sources: + if type(source) is not QuestionSourceChoice or set(vars(source)) != set( + QuestionSourceChoice.model_fields + ): + raise ValueError + _id(source.snapshot_id) + sources.append( + dict(snapshot_id=str(source.snapshot_id), source_sha256=source.source_sha256) + ) + raw = vars(value) | dict( + plan_id=str(value.plan_id), + workspace_id=str(value.workspace_id), + request_id=str(value.request_id), + sources=sources, + ) + plan = QuestionSourcePlan.model_validate_json( + json.dumps(raw, ensure_ascii=False, allow_nan=False), strict=True + ) + if ( + plan.workspace_id != workspace + or plan.request_id != request + or plan.question != question + ): + raise ValueError + return plan + except (ValidationError, ValueError, TypeError, AttributeError): + raise StorageCorruption("Question dispatch plan integrity check failed") from None + + +def _child(plan: UUID, snapshot: UUID, order: int) -> UUID: + """v1 profile: UTF-8 namespace|canonical plan UUID|snapshot UUID|zero-based order. + + Take the first 16 SHA256 bytes and set RFC UUID version 4/variant bits. This + stable profile must never change: it is the durable child recovery identity. + """ + payload = f"issuedigest:question-plan-child:v1|{plan}|{snapshot}|{order}" + return UUID(bytes=sha256(payload.encode("utf-8")).digest()[:16], version=4) + + +def _job( + job: EvidenceJob, + admitted: EvidenceJob, + source: QuestionSourceChoice, + workspace: UUID, + question: str, + model: str, +) -> None: + if ( + type(job) is not EvidenceJob + or type(admitted) is not EvidenceJob + or type(job.job_id) is not UUID + or job.job_id.version != 4 + or job.job_id != admitted.job_id + or job.workspace_id != workspace + or job.snapshot_id != source.snapshot_id + or job.source_sha256 != source.source_sha256 + or type(job.question) is not str + or job.question != question + or type(job.model) is not str + or job.model != model + or type(job.status) is not str + or job.status + not in ("pending", "running", "completed", "failed", "cancelled", "indeterminate") + ): + raise StorageCorruption("Question dispatch job integrity check failed") + + +async def advance_question_plan( + plans: QuestionPlanRepository, + jobs: EvidenceJobRepository, + *, + client: DBOSClient, + app_version: str, + token: str, + workspace_id: UUID, + request_id: UUID, + question: str, + model: str, + reservation_usd: Decimal, +) -> QuestionDispatch | None: + """Admit/acknowledge through the first unfinished child, without paid retries. + + Exact repository classes sharing the very same Engine exclude arbitrary + caller-provided plan repositories. Per-source transactions may commit a + prefix; exceptions never refund reservations or select alternative children. + """ + if ( + type(plans) is not QuestionPlanRepository + or type(jobs) is not EvidenceJobRepository + or plans.engine is not jobs.engine + ): + raise ValueError("Matching concrete question repositories are required") + _id(workspace_id) + _id(request_id) + if ( + type(question) is not str + or not question.strip() + or len(question) > 1500 + or type(model) is not str + or model != "gpt-6.1-sol" + ): + raise ValueError("Invalid question dispatch input") + if ( + type(reservation_usd) is not Decimal + or not reservation_usd.is_finite() + or not Decimal("0") < reservation_usd <= Decimal("10") + or reservation_usd != reservation_usd.quantize(Decimal(".01")) + ): + raise ValueError("Invalid evidence reservation") + if type(app_version) is not str or re.fullmatch(r"[A-Za-z0-9._-]{1,128}", app_version) is None: + raise ValueError("Invalid evidence application version") + saved = await asyncio.to_thread(plans.submit, token, workspace_id, request_id, question) + if saved is None: + return None + plan = _plan(saved, workspace_id, request_id, question) + if not plan.sources: + return QuestionDispatch(plan.plan_id, "no_match", 0, 0, ()) + dispatches: list[EvidenceDispatch] = [] + completed = 0 + for order, source in enumerate(plan.sources): + child = _child(plan.plan_id, source.snapshot_id, order) + admitted = await asyncio.to_thread( + jobs.submit, + token, + workspace_id, + child, + source.snapshot_id, + question, + model=model, + reservation_usd=reservation_usd, + expected_source_sha256=source.source_sha256, + ) + if admitted is None: + return None + current = await asyncio.to_thread(jobs.get, token, workspace_id, admitted.job_id) + if current is None: + return None + _job(current, admitted, source, workspace_id, question, model) + if current.status == "pending": + ack = await submit_evidence_job( + jobs, + client=client, + app_version=app_version, + token=token, + workspace_id=workspace_id, + request_id=child, + snapshot_id=source.snapshot_id, + question=question, + model=model, + reservation_usd=reservation_usd, + ) + if ack is None: + return None + dispatches.append(ack) + # Even a completed acknowledgement race stops this invocation. + status: Literal["in_progress", "blocked"] = ( + "blocked" + if ack.status in ("failed", "cancelled", "indeterminate") + else "in_progress" + ) + return QuestionDispatch( + plan.plan_id, status, completed, len(plan.sources), tuple(dispatches) + ) + dispatches.append( + EvidenceDispatch( + current.job_id, "guarded-evidence-" + str(current.job_id), current.status, False + ) + ) + if current.status == "completed": + completed += 1 + continue + return QuestionDispatch( + plan.plan_id, + "in_progress" if current.status == "running" else "blocked", + completed, + len(plan.sources), + tuple(dispatches), + ) + return QuestionDispatch( + plan.plan_id, "evidence_complete", completed, len(plan.sources), tuple(dispatches) + ) diff --git a/packages/agents/tests/test_question_dispatch.py b/packages/agents/tests/test_question_dispatch.py new file mode 100644 index 0000000..e577516 --- /dev/null +++ b/packages/agents/tests/test_question_dispatch.py @@ -0,0 +1,640 @@ +"""Sequential plan admission uses synthetic SQL; only external queue I/O is replaced.""" + +import asyncio +import importlib +import sys +import threading +from decimal import Decimal +from pathlib import Path +from uuid import uuid4 + +import pytest +from sqlalchemy import event, text + + +@pytest.fixture(scope="module", autouse=True) +def isolated_modules(): + paths = sys.path.copy() + root = Path(__file__).resolve().parents[3] + for path in ("packages/storage/tests", "packages/agents/tests", "apps/worker/tests"): + sys.path.insert(0, str(root / path)) + packages = [importlib.import_module(n) for n in ("issuedigest_agents", "issuedigest_worker")] + before, attrs = set(sys.modules), [vars(p).copy() for p in packages] + yield + for name in set(sys.modules) - before: + if name.startswith(("issuedigest_agents.", "issuedigest_worker.evidence_")): + sys.modules.pop(name, None) + sys.path[:] = paths + for package, original in zip(packages, attrs, strict=True): + for key in set(vars(package)) - original.keys(): + delattr(package, key) + for key, value in original.items(): + setattr(package, key, value) + + +@pytest.fixture +def admission(engine): + from issuedigest_domain import EvidenceBundle + from issuedigest_storage import EvidenceRepository + from issuedigest_storage.evidence_jobs import EvidenceJobRepository + from issuedigest_storage.question_plans import QuestionPlanRepository + from test_evidence_jobs import seed + + from conftest import bundle_data + + with engine.begin() as conn: + conn.execute(text("TRUNCATE question_source_plans, evidence_jobs")) + token, workspace, snapshot = seed(engine) + EvidenceRepository(engine).rebuild_projection(workspace, snapshot) + data = bundle_data() + data["document"]["workspace_id"] = str(workspace) + other = uuid4() + EvidenceRepository(engine).save(other, EvidenceBundle.model_validate(data)) + EvidenceRepository(engine).rebuild_projection(workspace, other) + yield ( + QuestionPlanRepository(engine), + EvidenceJobRepository(engine), + dict( + token=token, + workspace_id=workspace, + request_id=uuid4(), + question=" 앞 ", + model="gpt-6.1-sol", + reservation_usd=Decimal(".50"), + ), + ) + with engine.begin() as conn: + conn.execute(text("TRUNCATE question_source_plans, evidence_jobs")) + + +class QueueClient: + def __init__(self, action=None): + self.calls = [] + self.action = action + + async def enqueue_async(self, options, *args): + self.calls.append((options, args)) + if self.action: + self.action() + + +def module(): + assert importlib.util.find_spec("issuedigest_agents.question_dispatch"), "Missing plan dispatch" + return importlib.import_module("issuedigest_agents.question_dispatch") + + +def advance(admission, client, **changes): + plans, jobs, inputs = admission + return module().advance_question_plan( + plans, jobs, client=client, app_version="question-test-v1", **(inputs | changes) + ) + + +def counts(engine): + with engine.connect() as conn: + return ( + conn.scalar(text("SELECT count(*) FROM evidence_jobs")), + conn.scalar( + text("SELECT coalesce(sum(reservation_count),0) FROM research_budget_months") + ), + ) + + +def complete(jobs, job): + from test_evidence_jobs import completion_receipt + + receipt = completion_receipt(jobs, job) + assert jobs.claim( + job.job_id, + prompt_sha256=receipt["prompt_sha256"], + instructions_sha256=receipt["instructions_sha256"], + ) + jobs.finish( + job.job_id, + receipt=receipt, + known_usage=receipt["usage"], + usage_complete=True, + stream_closed=True, + ) + + +def test_no_match_has_no_jobs_reservations_or_queue(admission, engine): + client = QueueClient() + result = asyncio.run(advance(admission, client, question="?!")) + assert result.status == "no_match" and result.completed_count == result.total_sources == 0 + assert result.dispatches == () and client.calls == [] and counts(engine) == (0, 0) + assert set(vars(result)) == { + "plan_id", + "status", + "completed_count", + "total_sources", + "dispatches", + } + + +def test_first_unfinished_stops_retry_reuses_and_complete_prefix_advances(admission, engine): + plans, jobs, inputs = admission + client = QueueClient() + first = asyncio.run(advance(admission, client)) + assert first.status == "in_progress" and first.total_sources == 2 and first.completed_count == 0 + assert len(first.dispatches) == 1 and first.dispatches[0].queued + assert counts(engine) == (1, 1) + again = asyncio.run(advance(admission, client)) + assert again == first and counts(engine) == (1, 1) + job = jobs.get(inputs["token"], inputs["workspace_id"], first.dispatches[0].job_id) + complete(jobs, job) + second = asyncio.run(advance(admission, client)) + assert second.completed_count == 1 and len(second.dispatches) == 2 + assert not second.dispatches[0].queued and second.dispatches[1].queued + assert second.dispatches[0].job_id != second.dispatches[1].job_id + assert counts(engine) == (2, 2) + complete(jobs, jobs.get(inputs["token"], inputs["workspace_id"], second.dispatches[1].job_id)) + before = len(client.calls) + final = asyncio.run(advance(admission, client)) + assert final.status == "evidence_complete" and final.completed_count == 2 + assert all(not ack.queued for ack in final.dispatches) + assert counts(engine) == (2, 2) and len(client.calls) == before + with engine.connect() as conn: + rows = conn.execute( + text("SELECT request_id,snapshot_id FROM evidence_jobs ORDER BY snapshot_id") + ).all() + assert all(request.version == 4 for request, _ in rows) + assert len({request for request, _ in rows}) == 2 + assert asyncio.run(advance(admission, client)) == final + + +@pytest.mark.parametrize("status", ["running", "failed", "cancelled", "indeterminate"]) +def test_running_or_blocked_never_admits_later_children(admission, engine, status): + _, jobs, inputs = admission + client = QueueClient() + first = asyncio.run(advance(admission, client)) + job_id = first.dispatches[0].job_id + if status == "running": + jobs.claim(job_id, prompt_sha256="a" * 64, instructions_sha256="b" * 64) + elif status == "cancelled": + jobs.cancel(inputs["token"], inputs["workspace_id"], job_id) + else: + jobs.finish(job_id, reason="unknown" if status == "indeterminate" else "output") + client.calls.clear() + result = asyncio.run(advance(admission, client)) + assert result.status == ("in_progress" if status == "running" else "blocked") + assert result.completed_count == 0 and len(result.dispatches) == 1 + assert result.dispatches[0].status == status and not result.dispatches[0].queued + assert client.calls == [] and counts(engine) == (1, 1) + + +@pytest.mark.parametrize( + "field,value", + [ + ("workspace_id", True), + ("request_id", "bad"), + ("question", False), + ("question", " "), + ("question", "x" * 1501), + ("model", "other"), + ("model", True), + ("reservation_usd", 1), + ("reservation_usd", Decimal("NaN")), + ("reservation_usd", Decimal(".001")), + ("reservation_usd", Decimal("0")), + ("reservation_usd", Decimal("10.01")), + ], +) +def test_invalid_inputs_before_any_sql_including_no_match(admission, engine, field, value): + client = QueueClient() + + def forbidden(*args): + pytest.fail("invalid input issued SQL") + + event.listen(engine, "before_cursor_execute", forbidden) + try: + with pytest.raises(ValueError): + asyncio.run(advance(admission, client, **({"question": "?!"} | {field: value}))) + finally: + event.remove(engine, "before_cursor_execute", forbidden) + assert client.calls == [] + + +@pytest.mark.parametrize("version", [None, "", "a b", "한", "a" * 129, "v\n"]) +def test_bad_version_before_sql(admission, engine, version): + plans, jobs, inputs = admission + + def forbidden(*args): + pytest.fail("bad version issued SQL") + + event.listen(engine, "before_cursor_execute", forbidden) + try: + with pytest.raises(ValueError): + asyncio.run( + module().advance_question_plan( + plans, jobs, client=QueueClient(), app_version=version, **inputs + ) + ) + finally: + event.remove(engine, "before_cursor_execute", forbidden) + + +def test_same_engine_and_exact_repositories_required_before_sql(admission, engine): + from issuedigest_storage import create_storage_engine + from issuedigest_storage.evidence_jobs import EvidenceJobRepository + + plans, jobs, inputs = admission + other = create_storage_engine(str(engine.url)) + + def forbidden(*args): + pytest.fail("repository mismatch issued SQL") + + event.listen(engine, "before_cursor_execute", forbidden) + try: + for p, j in ((plans, EvidenceJobRepository(other)), (object(), jobs), (plans, object())): + with pytest.raises(ValueError): + asyncio.run( + module().advance_question_plan( + p, j, client=QueueClient(), app_version="test", **inputs + ) + ) + finally: + event.remove(engine, "before_cursor_execute", forbidden) + other.dispose() + + +@pytest.mark.parametrize( + "change", + ["workspace", "request", "question", "source_type", "source_hash", "extra", "nested_extra"], +) +def test_model_copy_tamper_is_reconstructed_strictly(admission, engine, monkeypatch, change): + from issuedigest_storage.repository import StorageCorruption + + plans, _, inputs = admission + plan = plans.submit( + inputs["token"], inputs["workspace_id"], inputs["request_id"], inputs["question"] + ) + updates = { + "workspace": {"workspace_id": uuid4()}, + "request": {"request_id": uuid4()}, + "question": {"question": False}, + "source_type": {"sources": list(plan.sources)}, + "source_hash": {"sources": (plan.sources[0].model_copy(update={"source_sha256": True}),)}, + "extra": {"receipt": {}}, + "nested_extra": {"sources": (plan.sources[0].model_copy(update={"receipt": {}}),)}, + } + monkeypatch.setattr(plans, "submit", lambda *args: plan.model_copy(update=updates[change])) + client = QueueClient() + with pytest.raises(StorageCorruption): + asyncio.run(advance(admission, client)) + assert client.calls == [] and counts(engine) == (0, 0) + + +@pytest.mark.parametrize("change", ["authority", "title", "source"]) +def test_current_plan_access_and_source_guard(admission, engine, change): + from issuedigest_storage.repository import StorageCorruption + from issuedigest_storage.research import ResearchForbidden + + asyncio.run(advance(admission, QueueClient())) + with engine.begin() as conn: + sql = { + "authority": "UPDATE auth_workspace_memberships SET role='viewer'", + "title": ( + "UPDATE source_presentations SET payload=jsonb_set(payload,'{title}','\"changed\"')" + ), + "source": ( + "UPDATE parsed_representations SET " + "payload=jsonb_set(payload,'{text}','\"changed\"')" + ), + }[change] + conn.execute(text(sql)) + client = QueueClient() + with pytest.raises((StorageCorruption, ResearchForbidden)): + asyncio.run(advance(admission, client)) + assert client.calls == [] and counts(engine) == (1, 1) + + +@pytest.mark.parametrize("completed_prefix", [False, True]) +def test_source_race_after_plan_before_admission_rolls_back( + admission, engine, monkeypatch, completed_prefix +): + from issuedigest_storage import EvidenceRepository + from issuedigest_storage.research import ResearchConflict, ResearchRepository + + plans, jobs, inputs = admission + if completed_prefix: + first = asyncio.run(advance(admission, QueueClient())) + complete( + jobs, jobs.get(inputs["token"], inputs["workspace_id"], first.dispatches[0].job_id) + ) + original = plans.submit + + def changed(*args): + plan = original(*args) + with engine.begin() as conn: + sid = plan.sources[int(completed_prefix)].snapshot_id + bundle = EvidenceRepository._load(conn, inputs["workspace_id"], sid) + presentation = ResearchRepository._presentation( + conn, inputs["workspace_id"], sid, bundle + ) + ResearchRepository._upsert_presentation( + conn, presentation.model_copy(update={"title": "raced"}) + ) + return plan + + monkeypatch.setattr(plans, "submit", changed) + client = QueueClient() + with pytest.raises(ResearchConflict): + asyncio.run(advance(admission, client)) + expected = (1, 1) if completed_prefix else (0, 0) + assert client.calls == [] and counts(engine) == expected + + +def test_ambiguous_queue_and_cancel_preserve_first_pending_without_next(admission, engine): + def lost(): + raise RuntimeError("SECRET") + + with pytest.raises( + importlib.import_module("issuedigest_agents.evidence_dispatch").EvidenceDispatchError + ): + asyncio.run(advance(admission, QueueClient(lost))) + assert counts(engine) == (1, 1) + + def cancel(): + raise asyncio.CancelledError() + + with pytest.raises(asyncio.CancelledError): + asyncio.run(advance(admission, QueueClient(cancel))) + assert counts(engine) == (1, 1) + assert asyncio.run(advance(admission, QueueClient())).dispatches[0].queued + + +def test_cancelled_sql_waiter_never_hidden_enqueues(admission, engine, monkeypatch): + _, jobs, _ = admission + entered, release, committed = threading.Event(), threading.Event(), threading.Event() + original = jobs.submit + + def delayed(*args, **kwargs): + entered.set() + assert release.wait(10) + try: + return original(*args, **kwargs) + finally: + committed.set() + + monkeypatch.setattr(jobs, "submit", delayed) + client = QueueClient() + + async def run(): + task = asyncio.create_task(advance(admission, client)) + assert await asyncio.to_thread(entered.wait, 10) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + release.set() + assert await asyncio.to_thread(committed.wait, 10) + + asyncio.run(run()) + assert client.calls == [] and counts(engine) == (1, 1) + + +def test_revoked_after_admission_and_unknown_error_never_enqueue(admission, engine, monkeypatch): + _, jobs, _ = admission + original = jobs.submit + + def revoke(*args, **kwargs): + job = original(*args, **kwargs) + with engine.begin() as conn: + conn.execute(text("UPDATE auth_workspace_memberships SET active=false")) + return job + + monkeypatch.setattr(jobs, "submit", revoke) + client = QueueClient() + assert asyncio.run(advance(admission, client)) is None + assert client.calls == [] and counts(engine) == (1, 1) + + +def test_real_dbos_mock_sse_two_source_complete_resume( + admission, engine, database_url, monkeypatch +): + import httpx2 as httpx + from dbos import DBOS, DBOSClient + from openai import AsyncOpenAI + from sqlalchemy.schema import DropSchema + from sqlalchemy.sql import quoted_name + from test_evidence_worker import native_body + + _, jobs, inputs = admission + worker = importlib.import_module("issuedigest_worker.evidence_worker") + execute = worker.execute_evidence_job + entered, release = threading.Event(), threading.Event() + requests = [] + schema = "qpd_" + uuid4().hex + + def default_state(): + with engine.connect() as conn: + metadata = conn.execute( + text( + "SELECT c.oid,c.relname,c.relkind FROM pg_class c " + "JOIN pg_namespace n ON n.oid=c.relnamespace " + "WHERE n.nspname='dbos' ORDER BY c.relname" + ) + ).all() + queues = ( + conn.execute(text("SELECT * FROM dbos.queues ORDER BY name")).all() + if conn.scalar(text("SELECT to_regclass('dbos.queues')")) is not None + else None + ) + exists = conn.scalar( + text("SELECT EXISTS (SELECT FROM pg_namespace WHERE nspname='dbos')") + ) + return exists, metadata, queues + + before_default = default_state() + + async def gated(*args, **kwargs): + entered.set() + assert await asyncio.to_thread(release.wait, 20) + return await execute(*args, **kwargs) + + monkeypatch.setattr(worker, "execute_evidence_job", gated) + + async def run(): + def serve(request): + requests.append(request) + return httpx.Response( + 200, headers={"content-type": "text/event-stream"}, stream=native_body() + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(serve)) as http: + provider = AsyncOpenAI(api_key="synthetic", http_client=http, max_retries=0) + client = None + try: + DBOS.migrate(database_url, schema=schema, application_role=engine.url.username) + with engine.connect() as conn: + assert conn.scalar( + text("SELECT EXISTS (SELECT FROM pg_namespace WHERE nspname=:schema)"), + {"schema": schema}, + ) + DBOS( + config=dict( + name="question-dispatch-synthetic", + system_database_url=database_url, + dbos_system_schema=schema, + application_version="question-test-v1", + run_admin_server=False, + ) + ) + client = DBOSClient( + system_database_url=database_url, + application_name="question-dispatch-synthetic", + dbos_system_schema=schema, + ) + DBOS.launch() + await asyncio.to_thread( + importlib.import_module( + "issuedigest_worker.evidence_queue" + ).register_evidence_queue, + jobs, + client=provider, + check_ownership=lambda _: True, + ) + first = await advance(admission, client) + assert await asyncio.to_thread(entered.wait, 10) + again = await advance(admission, client) + assert again == first and counts(engine) == (1, 1) and requests == [] + release.set() + handle = await client.retrieve_workflow_async(first.dispatches[0].workflow_id) + await asyncio.wait_for(handle.get_result(), 20) + second = await advance(admission, client) + assert second.completed_count == 1 and len(second.dispatches) == 2 + assert not second.dispatches[0].queued and second.dispatches[1].queued + assert second.dispatches[0].job_id == first.dispatches[0].job_id + handle = await client.retrieve_workflow_async(second.dispatches[1].workflow_id) + await asyncio.wait_for(handle.get_result(), 20) + + async def forbidden(*args, **kwargs): + pytest.fail("completed plan enqueued") + + monkeypatch.setattr(client, "enqueue_async", forbidden) + final = await advance(admission, client) + assert ( + final.status == "evidence_complete" + and final.completed_count == final.total_sources == 2 + ) + assert all(not ack.queued for ack in final.dispatches) + assert counts(engine) == (2, 2) and len(requests) == 2 + assert all( + jobs.get(inputs["token"], inputs["workspace_id"], ack.job_id).status + == "completed" + for ack in final.dispatches + ) + assert not provider.is_closed() + finally: + release.set() + try: + if client is not None: + client.destroy() + finally: + try: + DBOS.destroy(destroy_registry=True) + finally: + await provider.close() + + try: + asyncio.run(run()) + finally: + # Only this test's freshly generated schema is owned; default dbos is untouched. + with engine.begin() as conn: + conn.execute(DropSchema(quoted_name(schema, True), cascade=True, if_exists=True)) + with engine.connect() as conn: + assert not conn.scalar( + text("SELECT EXISTS (SELECT FROM pg_namespace WHERE nspname=:schema)"), + {"schema": schema}, + ) + assert default_state() == before_default + + +def test_changed_reservation_is_conflict_not_silent_partial(admission, engine): + from issuedigest_storage.research import ResearchConflict + + first = asyncio.run(advance(admission, QueueClient())) + with pytest.raises(ResearchConflict): + asyncio.run(advance(admission, QueueClient(), reservation_usd=Decimal(".75"))) + assert counts(engine) == (1, 1) + assert asyncio.run(advance(admission, QueueClient())) == first + + +def test_unknown_storage_error_and_none_plan_do_not_enqueue(admission, engine, monkeypatch): + plans, _, _ = admission + client = QueueClient() + + def fail(*args): + raise RuntimeError("synthetic unknown") + + monkeypatch.setattr(plans, "submit", fail) + with pytest.raises(RuntimeError, match="synthetic unknown"): + asyncio.run(advance(admission, client)) + monkeypatch.setattr(plans, "submit", lambda *args: None) + assert asyncio.run(advance(admission, client)) is None + assert client.calls == [] and counts(engine) == (0, 0) + + +def test_stable_child_profile_and_new_plan_identity(): + from uuid import UUID + + dispatch = module() + plan = UUID("00000000-0000-4000-8000-000000000001") + snapshot = UUID("00000000-0000-4000-8000-000000000002") + child = dispatch._child(plan, snapshot, 0) + assert str(child) == "5b2fe480-679c-4517-ad39-a3e788d893dc" + assert child.version == 4 + assert dispatch._child(plan, snapshot, 1) != child + assert dispatch._child(snapshot, snapshot, 0) != child + + +@pytest.mark.parametrize( + "field,value", + [ + ("job_id", uuid4()), + ("workspace_id", uuid4()), + ("snapshot_id", uuid4()), + ("question", "changed"), + ("model", "other"), + ("source_sha256", "f" * 64), + ("status", "unknown"), + ], +) +def test_current_job_identity_checked_before_paid_dispatch( + admission, engine, monkeypatch, field, value +): + from dataclasses import replace + + from issuedigest_storage.repository import StorageCorruption + + _, jobs, _ = admission + original = jobs.get + monkeypatch.setattr(jobs, "get", lambda *args: replace(original(*args), **{field: value})) + client = QueueClient() + with pytest.raises(StorageCorruption): + asyncio.run(advance(admission, client)) + assert client.calls == [] and counts(engine) == (1, 1) + + +def test_completed_acknowledgement_race_still_stops_at_first_pending( + admission, engine, monkeypatch +): + _, jobs, _ = admission + original = jobs.submit + calls = 0 + + def raced(*args, **kwargs): + nonlocal calls + calls += 1 + job = original(*args, **kwargs) + if calls == 2: + complete(jobs, job) + return jobs.get(args[0], args[1], job.job_id) + return job + + monkeypatch.setattr(jobs, "submit", raced) + client = QueueClient() + result = asyncio.run(advance(admission, client)) + assert result.status == "in_progress" and len(result.dispatches) == 1 + assert result.dispatches[0].status == "completed" and not result.dispatches[0].queued + assert client.calls == [] and counts(engine) == (1, 1) diff --git a/packages/storage/src/issuedigest_storage/evidence_jobs.py b/packages/storage/src/issuedigest_storage/evidence_jobs.py index c5d33e9..7547965 100644 --- a/packages/storage/src/issuedigest_storage/evidence_jobs.py +++ b/packages/storage/src/issuedigest_storage/evidence_jobs.py @@ -200,6 +200,7 @@ def submit( *, model: str, reservation_usd: Decimal, + expected_source_sha256: str | None = None, ) -> EvidenceJob | None: for value in (workspace_id, request_id, snapshot_id): _uuid4(value) @@ -218,6 +219,11 @@ def submit( or reservation_usd != reservation_usd.quantize(Decimal(".01")) ): raise ValueError("Invalid evidence reservation") + if expected_source_sha256 is not None and ( + type(expected_source_sha256) is not str + or re.fullmatch("[0-9a-f]{64}", expected_source_sha256) is None + ): + raise ValueError("Invalid expected evidence source hash") reservation_usd = reservation_usd.quantize(Decimal(".01")) def save(conn: Connection) -> EvidenceJob | None: @@ -230,6 +236,8 @@ def save(conn: Connection) -> EvidenceJob | None: raise ResearchConflict("Evidence source unavailable") presentation = self._base._presentation(conn, workspace_id, snapshot_id, bundle) hashed = source_digest((snapshot_id, bundle, presentation)) + if expected_source_sha256 is not None and hashed != expected_source_sha256: + raise ResearchConflict("Evidence source changed") identity = digest( dict( snapshot_id=str(snapshot_id), diff --git a/packages/storage/tests/test_evidence_jobs.py b/packages/storage/tests/test_evidence_jobs.py index a423f10..7990daf 100644 --- a/packages/storage/tests/test_evidence_jobs.py +++ b/packages/storage/tests/test_evidence_jobs.py @@ -651,3 +651,50 @@ def worker(): if worker_operation == "claim" else "completed" ) + + +@pytest.mark.parametrize("expected", [True, 1, "", "A" * 64, "a" * 63, "a" * 64 + "\n"]) +def test_expected_source_hash_invalid_before_sql(jobs, engine, expected): + from sqlalchemy import event + + def forbidden(*args): + pytest.fail("invalid expected hash issued SQL") + + event.listen(engine, "before_cursor_execute", forbidden) + try: + with pytest.raises(ValueError): + jobs.submit( + "invalid", + uuid4(), + uuid4(), + uuid4(), + "q", + model="gpt-6.1-sol", + reservation_usd=Decimal(".25"), + expected_source_sha256=expected, + ) + finally: + event.remove(engine, "before_cursor_execute", forbidden) + + +def test_expected_source_mismatch_precedes_budget_and_retry_lookup(jobs, engine): + token, workspace, snapshot = seed(engine) + request = uuid4() + args = (token, workspace, request, snapshot, "q") + options = dict(model="gpt-6.1-sol", reservation_usd=Decimal(".25")) + with pytest.raises(api().ResearchConflict): + jobs.submit(*args, **options, expected_source_sha256="f" * 64) + with engine.connect() as conn: + assert conn.scalar(text("SELECT count(*) FROM evidence_jobs")) == 0 + assert conn.scalar(text("SELECT count(*) FROM research_budget_months")) == 0 + job = jobs.submit(*args, **options) + assert jobs.submit(*args, **options, expected_source_sha256=job.source_sha256) == job + with engine.begin() as conn: + conn.execute(text("UPDATE research_budget_months SET reserved_usd=10,reservation_count=40")) + with pytest.raises(api().ResearchConflict): + jobs.submit(*args, **options, expected_source_sha256="f" * 64) + with engine.connect() as conn: + assert conn.scalar(text("SELECT count(*) FROM evidence_jobs")) == 1 + assert conn.execute( + text("SELECT reserved_usd,reservation_count FROM research_budget_months") + ).one() == (Decimal("10"), 40)