j(2UdZddlmZddlZddlZddlZddlmZmZddlm Z ddl m Z m Z e dZ e dZd Zd ed <d Zd ed <ddZddd ddZ d ddZddZdddddddd ddZy)u:utils/audit_chairman_recovery.py — chairman 수동 복구 audit log. task-2471 1차 hardening 회귀 가드의 일환으로 헤임달이 작성. task-2472 보강: schema 6필드 → 10필드 확장. 기존 코드 backward-compatible 유지. Chairman (회장) 가 RECOVERABLE_BLOCKED 상태의 task 를 수동으로 복구할 때 사용하는 ``memory/orchestration-audit/chairman-manual-recovery.jsonl`` 의 schema 정의 + atomic append 헬퍼. Schema (each JSONL line) — task-2472 확장 10필드:: { "task_id": str, "actor": str, "action": str, "input_state": str, # 기존 from_state 호환 "output_state": str, # 기존 to_state 호환 "approval_evidence": dict | None, "result": str, # "ok" | "rejected" "reason": str, "timestamp": str, # 기존 ts 호환 "evidence_hash": str, # sha256 # 하위호환 필드 "from_state": str, "to_state": str, "ts": str, "evidence_paths": [str], "detail": {"sha256_before": ..., "sha256_after": ...}, } Usage (기존 append_recovery 호환):: 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"], ) Usage (신규 record_recovery_audit):: from utils.audit_chairman_recovery import record_recovery_audit p = record_recovery_audit( task_id="task-2472", actor="thor", action="manual_recovery", input_state="RECOVERABLE_BLOCKED", output_state="MERGING", approval_evidence={"approved_by": "chairman", "ts": "...", "evidence_path": "..."}, result="ok", reason="branch protection 해제 확인", sha256_before="abc...", sha256_after="def...", ) 원자적 append (``os.O_APPEND``) — 동시 호출 race-safe. 디렉토리는 ``mkdir(parents=True, exist_ok=True)`` 로 생성. ) annotationsN)datetimetimezone)Path)OptionalSequencez9memory/orchestration-audit/chairman-manual-recovery.jsonlz/memory/orchestration-audit/state-recovery.jsonltask_idts from_stateto_statereasonevidence_pathsztuple[str, ...]RECOVERY_RECORD_KEYS) r actoraction input_state output_stateapproval_evidenceresultr timestamp evidence_hashEXTENDED_RECORD_KEYScftjtjj dS)z(ISO 8601 UTC (``YYYY-MM-DDTHH:MM:SSZ``).z%Y-%m-%dT%H:%M:%SZ)rnowrutcstrftime4/home/jay/workspace/utils/audit_chairman_recovery.py_now_isor!is! << % . ./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``). task_id 필수ufrom_state 필수uto_state 필수WORKSPACE_ROOT/home/jay/workspaceTparentsexist_okr F ensure_ascii utf-8) isinstancestrstrip ValueErrorrosenvirongetAUDIT_JSONL_PATHparentmkdirr!listjsondumpsopenO_WRONLYO_APPENDO_CREATwriteencodeclose) r r r rrr r" work_roottargetrecordlinefds r append_recoveryrHns|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). r%r&rr.)encodingN) rr3r4r5r6existsr<r1appendr:loadsJSONDecodeError)r"rCrDoutfrFs r read_recoveriesrRsT ')>?I) )F ==? C S7 +q D::B??CC ctj|dd}tj|j dj S)uRsha256(json.dumps(record, sort_keys=True, ensure_ascii=False)).hexdigest() 반환.TF) sort_keysr+r.)r:r;hashlibsha256rA hexdigest)rE serializeds r rrs7FdGJ >>*++G4 5 ? ? AArok)rrr sha256_before sha256_after audit_pathr"c t|tr|js tdt|tr|js tdt|tr|js tdt|tr|js tdt|tr|js td| xs)t t j jdd} | | tz } n| js| | z n| } | jjdd t}i}|||d <| | |d <|j|j|j|j|j||t|tr|n t|||j|j|t|tr|jd d gngd }|r||d<t|}i|d|i}tj |ddz}t j"t| t j$t j&zt j(zd} t j*||j-dt j.|| S#t j.|wxYw)u?10필드 schema로 chairman recovery audit 기록. 기존 append_recovery() backward-compatible 유지. 신규 필드: approval_evidence, result, reason, sha256_before/after, evidence_hash. Parameters ---------- task_id: ``task-NNNN`` 형식. actor: 요청자 식별자. action: 수행한 action (예: "manual_recovery", "state_repair"). input_state, output_state: 상태 전이 출발/도착. approval_evidence: chairman approval evidence dict. 형식: {"approved_by": str, "evidence_path": str, "ts": str} 또는 None. result: "ok" | "rejected". reason: 복구 사유 (인간 가독). sha256_before, sha256_after: state 파일 sha256 (before/after repair). detail로 묶여 evidence_hash에 포함. audit_path: 미지정 시 STATE_RECOVERY_JSONL_PATH (state-recovery.jsonl). chairman-manual-recovery.jsonl 도 호환 유지. workspace: Workspace root. Returns ------- Path 실제 append 된 파일 경로. r$u actor 필수u action 필수uinput_state 필수uoutput_state 필수r%r&Tr'r[r\ evidence_pathrZ) r rrrrrrrrr r r rdetailrFr*r,r-r.)r/r0r1r2rr3r4r5STATE_RECOVERY_JSONL_PATH is_absoluter7r8r!dictrr:r;r<r=r>r?r@rArB)r rrrrrrrr[r\r]r"rCrDrr` base_recordev_hashrErFrGs r record_recovery_auditrfsnd gs #7==?)** eS !(( fc "&,,.)) k3 '{/@/@/B-.. lC ( 0B0B0D.//T ')>?I66/9/E/E/GZ'Z MMt4 IF "/!-~==?,,."((*$**,.&vs3&V!'') &&(+T2 " "?B 7 8#K( & HK(G 6 6_g 6F ::f5 1D 8D VbkkBKK7"**De LB T[[)*  M  s %KK')returnr0)r r0r r0r r0rr0rz Sequence[str]r Optional[str]r"Optional[Path]rgr)N)r"rirgz list[dict])rErcrgr0)r r0rr0rr0rr0rr0rzOptional[dict]rr0rr0r[rhr\rhr]rir"rirgr)__doc__ __future__rrUr:r3rrpathlibrtypingrrr6rar__annotations__rr!rHrRrrfrrr ros;x# '%ST!!RS)o )o E $? ???  ? " ? ?? ?F!%@B)-#'"&!% $p p p  p  p  p&p p p!p ppp pr