üidUdZddlmZddlZddlZddlZddlmZmZddlm Z ddl m Z m Z e dZ dZd ed <dd Zddd  dd Z d ddZy)uutils/audit_chairman_recovery.py — chairman 수동 복구 audit log. task-2471 1차 hardening 회귀 가드의 일환으로 헤임달이 작성. Chairman (회장) 가 RECOVERABLE_BLOCKED 상태의 task 를 수동으로 복구할 때 사용하는 ``memory/orchestration-audit/chairman-manual-recovery.jsonl`` 의 schema 정의 + atomic append 헬퍼. Schema (each JSONL line):: { "task_id": str, "ts": ISO 8601 UTC, "from_state": str, "to_state": str, "reason": str, "evidence_paths": [str] } Usage:: from utils.audit_chairman_recovery import append_recovery p = append_recovery( task_id="task-2471", from_state="RECOVERABLE_BLOCKED", to_state="MERGING", reason="branch protection 해제 확인", evidence_paths=[".tasks/evidence/task-2471/recovery.json"], ) 원자적 append (``os.O_APPEND``) — 동시 호출 race-safe. 디렉토리는 ``mkdir(parents=True, exist_ok=True)`` 로 생성. ) annotationsN)datetimetimezone)Path)OptionalSequencez9memory/orchestration-audit/chairman-manual-recovery.jsonltask_idts from_stateto_statereasonevidence_pathsztuple[str, ...]RECOVERY_RECORD_KEYScftjtjj dS)z(ISO 8601 UTC (``YYYY-MM-DDTHH:MM:SSZ``).z%Y-%m-%dT%H:%M:%SZ)rnowrutcstrftimeN/home/jay/workspace/.worktrees/task-2487-dev2/utils/audit_chairman_recovery.py_now_isor=s! << % . ./C DDr)r workspacect|tr|js tdt|tr|js tdt|tr|js td|xs)t t j jdd}|tz }|jjdd|j|xs t|j|jt|tr|n t||r t|ngd} tj| d d z} t jt|t j t j"zt j$zd } t j&| | j)d t j*| |S#t j*| wxYw)uJSONL 한 줄 atomic append. Parameters ---------- task_id: ``task-NNNN`` 형식. 빈 문자열이면 ``ValueError``. from_state, to_state: 상태 전이의 출발/도착. 빈 문자열이면 ``ValueError``. reason: 복구 사유 (인간 가독). 빈 문자열도 허용 (회장 재량). evidence_paths: 근거 파일 경로 list. 비어 있으면 빈 리스트로 기록. ts: ISO 8601 timestamp. 미지정 시 현재 UTC. workspace: Workspace root. 기본 ``Path(os.environ['WORKSPACE_ROOT'])`` 또는 ``/home/jay/workspace``. Returns ------- Path 실제 append 된 파일 경로 (``workspace/AUDIT_JSONL_PATH``). utask_id 필수ufrom_state 필수uto_state 필수WORKSPACE_ROOT/home/jay/workspaceT)parentsexist_okr F) ensure_ascii iutf-8) isinstancestrstrip ValueErrorrosenvirongetAUDIT_JSONL_PATHparentmkdirrlistjsondumpsopenO_WRONLYO_APPENDO_CREATwriteencodeclose) r r r rrr r work_roottargetrecordlinefds rappend_recoveryr;Bs|B gs #7==?)** j# &j.>.>.@,-- h $HNN,<*++T ')>?I) )F MMt4==?HJ &&(NN$&vs3&V2@$~.b F ::f5 1D 8D VbkkBKK7"**De LB T[[)*  M  s '%G##G:c|xs)ttjjdd}|tz }|j sgSg}|j dd5}|D]:}|j}|s |jtj|< ddd|S#tj$rY]wxYw#1swY|SxYw)uJSONL 파일 전체 읽어서 dict list 반환. 파일 부재 시 빈 리스트. 파싱 실패 라인은 skip (graceful). rrrr!)encodingN) rr&r'r(r)existsr/r$appendr-loadsJSONDecodeError)rr6r7outfr9s rread_recoveriesrEsT ')>?I) )F ==? C S7 +q D::B??CC )returnr#)r r#r r#r r#rr#rz Sequence[str]r z Optional[str]rOptional[Path]rFr)N)rrGrFz list[dict])__doc__ __future__rr-r&os.pathrrpathlibrtypingrrr)r__annotations__rr;rErrrrNs B# '%ST)oE $? ???  ? " ? ?? ?F!%r