jUdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z m Z ddlmZddlmZmZmZddlmZej,eZd Zd Zd Zej8d k7ZerddlZiZd e d<ejBZ"d"dZ#e d#dZ$d$dZ%d%dZ&d&d'dZ'd&d(dZ(d&d)dZ)d*dZ*d+dZ+d,dZ,dd d-dZ-ddd d.dZ.d/dZ/ddd d0dZ0 d1d Z1d'd!Z2y)2u기획서 project-level 분석 service. 기획서는 project-level 자원이므로 분석 결과도 episode 와 무관하게 ``projects/{project_id}/checkpoints/planning_doc_analysis/manifest.json`` 에 저장. 기존 ``PlanningDocAnalysisStep`` (StepRunner) 은 episode 단위 cascade 호환을 위해 유지하되, 본 service 가 만든 project-level checkpoint 가 있으면 그대로 mirror (LLM 재호출 없음). 업로드 직후 백그라운드 dispatch 로 사용: from app.services.planning_doc_analysis_service import ( dispatch_planning_doc_analysis_async, ) dispatch_planning_doc_analysis_async(project_id) ## race 안전 — per-project lock + multi-layer guard 핵심 race: 업로드 → 분석 시작 → 사용자가 즉시 delete + 재업로드 시, 이전 background thread (A) 가 늦게 끝나서 새 SOT (B) 를 덮어쓰는 시나리오. status write / manifest promote / clear / dispatch queued write 가 모두 "check-then-act" 라 단순 atomic write 만으로는 race window 가 남음: A: is_current_job(A) → True (status=A) B: clear → dispatch → status=B A: atomic_write status={state:running, job:A} ← B 의 도장 덮어씀 근본 해법은 critical section 직렬화. ``_project_lock(project_id)`` 가 fcntl.flock + threading.Lock combo 로 same/cross-process 모두 보호. LLM call 은 길어서 lock 밖 — write/check/cleanup 만 lock 안. 추가 안전망 (lock 으로 닫지 못한 edge — 옛 schema, race-free 호출 등): L1. ``_write_status``/``_write_manifest`` 의 ``expected_job_id`` 가드 — lock 안에서 호출되면 redundant 한 double check (안전 차원). L2. manifest payload 에 ``job_id`` + ``source_hash`` 동봉 → reader 검증. L3. promote = staging file (``manifest.{job_id}.json``) 에 atomic write 후 lock 안에서 ``os.replace`` — cross-job manifest.json 보호. L4. ``load_project_checkpoint`` 가 manifest.source_hash 와 현재 disk source hash 비교 → 다르면 reject (cleanup unlink 제거 — reader 의 늦은 unlink 가 fresh manifest 를 죽이는 race 차단). 다음 dispatch promote 또는 clear 가 stale 자연 정리. ) annotationsN)contextmanager)datetimetimezone)Path)DictIteratorOptional)settingsz status.json manifest.jsonz.lockWindowszDict[str, threading.Lock] _thread_locksct5tj|}|tj}|t|<|cdddS#1swYyxYwN)_thread_locks_guardrget threadingLock) project_idlks `/Users/manta/Documents/Projects/TheRoad-I1/backend/app/services/planning_doc_analysis_service.py_get_thread_lockrSs?    z * :!B(*M* %   s 6AAc #Kt|}|jd} tret|}|j dd|t z }t |d}tj|jtjd|D tj|jtj |j|j!y#t$r$}tjd|dd|Yd}~Jd}~wwxYw#t$rYLwxYw#| tj|jtjn0#t$r$}tjd|dd|Yd}~nd}~wwxYw |jn#t$rYnwxYw|j!wxYww)uKper-project critical section — same/cross process 모두 직렬화. threading.Lock + fcntl.flock combo: - threading.Lock: 같은 process 내 thread 들 빠르게 직렬화. - fcntl.flock (POSIX): uvicorn workers / --reload 일시 중첩 등 cross-process race 방어 (file descriptor advisory lock). LLM call 같은 long-running operation 은 lock 밖에서 호출 — write / check / cleanup 만 critical section. 분석 자체가 lock 잡으면 사용자 재업로드 시 새 dispatch 가 이전 분석 끝날 때까지 막히는 UX 손해. NTparentsexist_okwz2_project_lock fcntl unlock failed (project=%s): %s)racquire _IS_POSIXget_project_checkpoint_dirmkdir _LOCK_FILEopenfcntlflockfilenoLOCK_EXLOCK_UNOSErrorloggerwarningcloserelease)rtlockfd lock_pathexcs r _project_lockr4\sm Z (E MMO A *:6A GGD4G 0JIY$A KK EMM 2 =  AHHJ 6      HrNC    =  AHHJ 6 HrNC       sGA/D&G2C'DG' D0D GDG D# G"D##G&F?*2EF? F &FF?F  F?FF? F+(F?*F++F??GcFttj|z dz dz S)u<project-level planning_doc_analysis checkpoint 디렉토리. checkpointsplanning_doc_analysis)rr projects_dir)rs rr!r!s1 X " "#     " "c|jjdd|j|jdt j j dddz} |jtj|dd d tjt|t|y#t$r3|jr! |j#t $rYwxYwwxYw) utemp file + os.replace 로 atomic JSON write. 부분 쓰기/동시 read 에 의한 corrupt JSON 방지. tmp 파일에 uuid 접미사를 붙여 동시 다중 thread 충돌을 회피 (현재는 job_id 가드로 단일 writer 가 보장되지만 안전 차원 dual layer). Tr.Nrz.tmpF) ensure_asciiindentutf-8encoding)parentr" with_suffixsuffixuuiduuid4hex write_textjsondumpsosreplacestr Exceptionexistsunlinkr*)pathpayloadtmps r_atomic_write_jsonrTs KKdT2   4;;1TZZ\-=-=bq-A,B$)GG HC   JJwU1 =   3s8SY'  ::<       s1 AB22C. CC. C)&C.(C))C.cx tj}ttj|z dz dz }|j rR|j d |jd5 t fddD]}|j | ddd|j d d }|d dl m }|}d} d dlm}|j!|j#|j$|k(j'} | r| j(xsdnd} |j | j+d|r|j-|j/S#1swYxYw#t$r6}tjd|dd ||j d Yd}~d}~wwxYw#|r|j-wwxYw)uT기획서 source 의 결정적 hash. PDF bytes (streaming) + 'PDF_END|TEXT:' separator + planning_doc_text 의 sha256 결합. PDF 가 없거나 text 가 비어 있어도 hex digest 가 항상 return (empty source 도 결정적 값). reader/writer 가 같은 입력으로 같은 hash 를 만들도록 separator 명시. assetsplanning_doc.pdfsPDF:rbc&jdS)Ni)read)r0srz%compute_source_hash..s !&&-r9r9Nz5compute_source_hash: PDF read failed (project=%s): %srs s|TEXT:Fr SessionLocalTProjectRegistryr?)hashlibsha256rr r8rOupdater$iterr*r+r,app.core.databaser]app.models.catalogr_queryfilteridfirstplanning_doc_textencoder- hexdigest) rdbhpdf_pathchunkr3own_dbr]r_projecttextr0s @rcompute_source_hashrusy A X " "#j08;>PP   &t$!"7=EHHUO>%HHY F z2 ^ 6 HH_ % VO&&*4 5 UW  5<))/R W%&  HHJ ;;=7%$ & NNG2A  HH_ % %  &,  HHJ s=E"+&EE"?A5F$EE"" F!++FF!$F9ct||}|#|jds|jdryttj|z dz dz }|j ryd}|dd lm}|}d} dd lm }|j|j|j|k(j}|r?t|jxsd j!d k\r |r|j#yy|r|j# ttj|z d z dz }|j r-|j%D]} | dz dz j syy#|r|j#wwxYw)u기획서 존재 여부 단일 판정 — 어느 한 곳이라도 있으면 True. 검사 순서 (우선순위): 1. project-level checkpoint — ``load_project_checkpoint`` 로 검증된 data (source_hash 일치) + ``available_sections`` 비어 있지 않을 때만. empty/stale/error manifest 는 false-positive 차단. 2. PDF on disk (assets/planning_doc.pdf) 3. ProjectRegistry.planning_doc_text (≥100 chars) 4. legacy episode-level checkpoint (옛 프로젝트 호환) db: caller 세션. None 이면 SessionLocal 일회 사용. `_if_planning_doc` (applicability) 와 `_project_has_planning_doc` (dispatcher) 가 동일 판정을 공유하도록 단일 source. rnavailable_sections charactersTrVrWFrr\r^r`dr6episodesr7r )load_project_checkpointrrr r8rOrer]rfr_rgrhrirjlenrkstripr-iterdir) rrncp_datarprrr]r_rs legacy_baseep_dirs rproject_has_planning_docrsp$&jR8G ;;+ , L0I X " "#j08;>PP F z2 ^ 6 HH_ % VO&&*4 5 UW  sG55;BBDEL  HHJ  HHJ X " "#j0=@:M!))+F00?BJJL,   HHJ s :A-E//Fct|tz }|jsy tj|j d}t|tsy|jd}t|tr7|r5 t||}||k7r"tjd |dd|dd |dd y|jd }t|tr|SdS#t tjf$r }tjd|Yd}~yd}~wwxYw#t$r&}tjd|dd||}Yd}~d}~wwxYw) uY저장된 project-level 분석 결과 — source_hash 검증 통과 시에만 반환. 검증 로직 (L4 — stale resurrection 최종 안전망): - manifest.source_hash 가 있고 현재 disk source 의 hash 와 다르면 stale (delete 후 다른 source 로 재업로드 race 등) → ``None`` reject. **cleanup unlink 는 의도적으로 하지 않음** — hash 계산 도중 다른 thread 가 fresh manifest 를 promote 하면, reader 가 fresh 를 unlink 하는 race 가 생김. disk 정리는 다음 dispatch promote (덮어쓰기) 또는 ``clear_project_checkpoint`` 가 담당. - manifest.source_hash 가 없으면 (옛 schema) 검증 skip + 정상 read. 옛 schema (source_hash 없는 manifest) 는 한 번 정상 read 된 후, 새 분석 트리거 시 새 schema 로 덮어씀. Nr?r@z/planning_doc project checkpoint read failed: %s source_hashrwudplanning_doc source_hash compute failed (project=%s): %s — skipping reader hash check (read as-is)ruoplanning_doc manifest source_hash mismatch (stale): project=%s manifest=%s current=%s — rejecting (no unlink) data)r!_MANIFEST_FILErOrIloads read_textr*JSONDecodeErrorr+r, isinstancedictrrMrurN)rrncprawr3 manifest_hash current_hashrs rr|r|s: $J /. @B 99;jjw78 c4 GGM*M-%- ).zbAL L ( NNC2A cr 2L"4E   776?DdD)43t3C T)) *H#N ) NN:2A  )L  )s/%C? DD1D  D ED>>Ec8t|tz }|jsddiS tj|j dS#t tjf$r3}tjd|dd|dd t|d cYd}~Sd}~wwxYw) u현재 분석 진행 상태. {state: idle|queued|running|done|error, ...} corrupt JSON 은 silent ``idle`` 대신 ``error/status_unreadable`` 로 surface — 운영자가 disk 손상을 인지하도록. stateidler?r@uGplanning_doc status read failed (project=%s): %s — surfacing as errorNrerrorstatus_unreadable)rrdetail) r! _STATUS_FILErOrIrrr*rr+r,rM)rrr3s r load_statusrIs $J /, >B 99;  Tzz",,,899 T)) *T U rNC !+>#c(SS Ts$A B&(BBBc>t|tz }|jsy tj|j d}t|tr%|jd}t|tr|SdSy#t tjf$rYywxYw)u?현재 disk status.json 의 job_id (없거나 손상 시 None).Nr?r@job_id) r!rrOrIrrr*rrrrrM)rrrvs r_read_status_job_idr\s #J /, >B 99;jjw78#t GGH q#&q0D0  T)) *s%BBBc(|yt|}||k(S)ujob_id 가 현재 status.json 의 job_id 와 일치하면 True. job_id=None 으로 호출되면 guard 를 비활성 (sync 직접 호출용). T)r)rrcurrents r_is_current_jobrks! ~!*-G f r9expected_job_idc|1t||s%tjd|dd|t|yt t |t z |y)ustatus.json atomic write. expected_job_id 가 주어지면 현재 disk job_id 와 일치할 때만 쓰고 stale 이면 drop. Returns True iff 실제로 write 됨. NzOplanning_doc status write dropped (stale job_id): project=%s self=%s current=%srFT)rr+inforrTr!r)rrRrs r _write_statusrvsT"?:+W  ! rNO-@-L 1*= LgV r9rrcHt|tz }|||tjtj j d}|t|||St||s%tjd|dd|t|y|jd|dz }t||t||s6tjd|dd|t| |jy t#j$t'|t'||S#t$rYyt$r!}tj!d||Yd}~yd}~wwxYw#t$rK}tj!d |dd| |jn#ttf$rYnwxYwYd}~yd}~wwxYw) u{manifest 를 staging file → promote 패턴으로 atomic 게시. 이전의 "write-then-conditional-unlink" 는 race window 가 있었음: A: status check OK → atomic replace manifest.json (A data) B: 그 사이 clear + 새 dispatch + write manifest.json (B data) A: status check FAIL → unconditional out.unlink() → **B 의 fresh manifest 까지 삭제** staging 패턴: 1. ``manifest.{job_id}.json`` 에 atomic write (job-isolated) 2. status.job_id 가 자기 것인지 재확인 - 자기 것 → ``os.replace(staging, manifest.json)`` 으로 promote - 아니면 → 자기 staging 만 unlink (manifest.json 은 건드리지 않음) 3. promote 후 동시 다른 job 이 새 promote 해도 자기 staging 은 이미 소비된 후 → 안전. manifest.json 의 owner 는 마지막 promote 가 결정. L1: 1차 status 가드 (early bail-out — staging write 비용 절약). L2: payload 에 ``job_id`` + ``source_hash`` → reader 가 L4 검증 가능. L3: promote 직전 status 재확인 + cross-job manifest.json 보호. expected_job_id=None 호출 (sync 직접 caller) 은 race-free 가정 → 단순 atomic write. caller 는 자기 책임으로 호출 직전에 status/source 정합성 확인. Returns 실제로 promote 된 Path (drop / staging-only cleanup 시 None). )rrr generated_atNuOplanning_doc manifest write dropped (1차 stale): project=%s self=%s current=%srz manifest.z.jsonuQplanning_doc manifest promote dropped (2차 stale): project=%s self=%s current=%su.planning_doc staging cleanup failed: %s — %su7planning_doc manifest promote failed: project=%s — %s)r!rrnowrutc isoformatrTrr+rrrBrPFileNotFoundErrorr*r,rKrLrM)rrrroutrRstagingr3s r_write_manifestrsB %Z 0> AC!" X\\2<<> G3(  : 7 ] rNO-@-L jjY&7u==Gw( : 7 _ rNO-@-L   NN   3w<S* J+!      NN@'3       E rNC   NN !7+    sZD,(E  E !E )EE  F!F1FFFFFFF!c*t|5t|}|js dddyttfD])}||z }|js |j +|jdD]'}|jtk(r |j ) dddy#t $r!}tjd||Yd}~d}~wwxYw#t $r!}tjd||Yd}~d}~wwxYw#1swYyxYw)u기획서 교체/삭제 시 호출 — manifest+status+staging file 모두 제거. staging file (``manifest.{job_id}.json``) 는 in-flight 분석이 promote 하기 전 임시 파일. clear 시 이전 dispatch 의 잔재를 같이 청소. ``_project_lock`` 안에서 수행 — 동시에 발생하는 dispatch / status write / promote 와 직렬화되어 partial state (예: status 만 남고 manifest 잔존) 가 노출되지 않음. Nu+clear planning checkpoint failed: %s — %szmanifest.*.jsonu(clear planning staging failed: %s — %s) r4r!rOrrrPr*r+r,globname)rr1rpr3r0s rclear_project_checkpointrs z " &z 2xxz  # "$\2DDAxxzZHHJ 3)*Avv' S + # "ZNN#PRSUXYYZ SI1cRR S# # "sdD $D B/(*D C#D / C8CD CD  D%D<D DD  Dr queued_atc nddlm}ddlm}m}m}ddlm}ddlm }ddl m } tjtjj!} t#|||| |||||||  S#t$$r} t'|5t)|d||| tjtjj!t+| d | d d d n #1swYnxYwn0#t$$r$} t,j/d |d d | Yd } ~ nd } ~ wwxYwt,j/d|d d | d } ~ wwxYw)u 동기 실행 — 기획서를 LLM 으로 분석 + project-level checkpoint 저장. 구조: write/check 만 ``_project_lock`` 안에서 직렬화, LLM call 은 lock 밖. concurrent dispatch / clear 가 status SOT 를 뒤집어도 자기 phase 시작 시 ``_is_current_job(job_id)`` 로 안전하게 bail out. Phase 흐름: 1. **lock**: running status 마킹 + bail-out 체크 + LLM input 수집 (PDF bytes, text, source_hash) 2. **lock 밖**: LLM call (PDF multimodal → text fallback) 3. **lock**: bail-out 체크 + manifest staging→promote + done/error status 마킹 job_id: dispatch 시 발급된 식별자. ``None`` 이면 ``_is_current_job`` / ``_write_*`` 의 stale-write guard 만 비활성화됨 (caller 가 race-free 보장한다는 가정). ``_project_lock`` 자체는 항상 유지 — 동시 sync 호출이나 clear 와 SOT mutation 직렬화는 그대로 보장. queued_at: dispatch 시점 timestamp (status payload 에 보존). rr\)_ANALYSIS_SCHEMA _EMPTY_RESULT_SYSTEM_PROMPTr^)call_structured)load_project_llm_config) rr started_at session_localschema empty_result system_promptproject_registry_clsrrrrrrr finished_atrrNz@planning_doc_analysis error-status write failed (project=%s): %srz-planning_doc_analysis (project=%s) failed: %s)rer] app.core.steps.planning_doc_steprrrrfr_app.modules.llm.llm_clientr&app.services.analysis_dispatch_servicerrrrrr!_run_planning_doc_analysis_phasesrNr4rrMr+ exception) rrrr]rrrr_rrrr3e2s rrun_planning_doc_analysis_syncrs+4/ 3:Nhll+557J(0 !&#&(!0+$;    z*!("(%.&0'/||HLL'A'K'K'M!$S %+ +**    R2A     ; rNC  5sUA++ D45 C$AC C$C C$#D/$ D-D D/ DD//D4c  |} t|5|Kt||s?tjd|dd|t ||cddd| j S| j |j|j|k(j} | r | jnd} ttj|z dz dz }|j}| | |}t|| }t!|d|||d| |s| rt#| j%d krtjd |ddt'|||| t!|d |||t)j*t,j.j1dd| |cddd| j S|r|j3}nd}ddd| j d}iddl}|j7|j9d}dddddd|idg}tjd|ddt#| | d|||xsdd}|X rVt#| j%d k\r:d| }tjd|ddt#| | d|||xsdd}t|5|?t||s3tjd |dd|t |||n|cdddS|wtj=d!|ddt'||| t!|d"|||t)j*t,j.j1d#d$| |cdddSdd%lm }||}||d&<t'||| }| |cdddSt!|d |||t)j*t,j.j1t#|jCd'gt#|jCd(g|d)| tjd*|ddt#|jCd'gt#|jCd(g|||cdddS#1swYxYw#| j wxYw#t:$r!}tj=d|Yd}~hd}~wwxYw#1swYyxYw)+urun_planning_doc_analysis_sync 의 본체 — outer try/except 가 status error 마킹을 보장하므로 본 함수는 정상 흐름만 책임. NzPplanning_doc_analysis bail at phase 1 (stale job): project=%s self=%s current=%srrVrWrwrunning)rrrrrrzuJplanning_doc_analysis (project=%s): 기획서 없음 — 빈 결과 저장rdoneT)rrrrremptyrr?rtuI이 기획서 PDF를 분석하여 구조화된 정보를 추출하세요.)typert image_urlurlzdata:application/pdf;base64,)rrz=planning_doc_analysis (project=%s): PDF multimodal (%d bytes)r7g?)stepr user_promptresponse_schemaproject_config temperaturez/PDF multimodal failed, falling back to text: %su_다음 기획서를 분석하여 구조화된 정보를 추출하세요. ## 기획서 전문 z  :AG55dMX**+%%  oo'G"9Z!H -jR@K &$!*",  !' !S)<)<)>%?#%E `rN  $* !'"(%.&0'/||HLL'A'K'K'M!% %+ $}' &P  $//1  M'P  F""9-44W=c  $9'C     K rNC N  $,+' &19T F    ##% &# - &&3_ 6   J rNC . !('("-5   z "  oj&&I KKC2A(;J(G  $/6\ # " > NNN2A  L &K  $$!*",#+<< #=#G#G#I@ !'  ? # "J .f5'/#$ "   =g # "h   &('||HLL9CCE!&**\2">?!$VZZ0CR%H!I$ #    4 rN  <, -  .3 4   W # "K' &P  @  NNA3   : # "ss Q74Q* Q7"D3Q* Q7/Q*Q77R .8R90A/R9)%R9CR9*Q4/Q77R  R6R11R69Sc tjjtjt j jt5tdddddfd}tj|ddddddd }|jS#1swYFxYw) uy업로드 직후 백그라운드 thread 로 분석 실행 + ``job_id`` 반환. 동작: - ``_project_lock`` 안에서 uuid ``job_id`` 발급 + ``status.json`` 에 ``{state:queued, job_id}`` 도장. concurrent dispatch 와 직렬화되어 마지막 dispatch 의 job_id 가 SOT 가 됨. - background thread 는 자기 ``job_id`` 를 들고 ``run_..._sync`` 호출. phase 마다 자기 job_id 가 still current 인지 lock 안에서 확인. 응답 지연 회피 + 결과는 ``checkpoints/planning_doc_analysis/manifest.json``. 진행 상태는 ``status.json`` 으로 polling 가능. queued)rrrNc| ty#t$r }tjd|Yd}~yd}~wwxYw)Nrz3planning_doc_analysis background thread crashed: %s)rrNr+r)r3rrrs r_runz2dispatch_planning_doc_analysis_async.._runVs?  *6Y   LLEs   s ;6;zplanning_doc_analysis-r-T)targetrdaemon) rErFrGrrrrrr4rrThreadstart)rrtrrs` @@r$dispatch_planning_doc_analysis_asyncr>sZZ\  F X\\*446I z "j"#   #  %j!n%5Qvbqzl C A GGI M1 # "s B44B=)rrMreturnzthreading.Lock)rrMrzIterator[None])rrMrr)rQrrRrrNoner)rrMrrM)rrMrbool)rrMrzOptional[Dict])rrMrr)rrMr Optional[str])rrMrrrr)rrMrRrrrrr) rrMrrrrrrrzOptional[Path])rrMrr)rrMrrrrrr) rrMrrrrrrMrrMrr)3__doc__ __future__rrarIloggingrKplatformrrE contextlibrrrpathlibrtypingrr r app.core.configr getLogger__name__r+rrr#systemr r%r__annotations__rrrr4r!rTrurr|rrrrrrrrrr9rrs*X#   %'++$   8 $     HOO  *  ,. (-$inn&$$N0+\:z54pT& &*  #   4&*!% XX X# X  X  XvSD!# NN N N  Nbdd d d  dd dN)r9