# -*- coding: utf-8 -*- """anu_v3.batch_hold_adjudicator — batch-level consolidated HOLD adjudicator (회장 BATCH_LEVEL_HOLD_ADJUDICATION 시스템화 Track A · task-2610). 회장 verbatim 목표: 병렬 작업 중 Codex HIGH/HOLD 가 발생해도 개별 track 만 보고 회장에게 멈추지 말고, batch 전체 context 를 모은 뒤 Critical7 인지 non-critical remediation 인지 자동 분류하고, Critical7 이 아니면 ANU-Codex loop 로 자동 remediation 하여 all-settled 까지 진행한다. 본 모듈의 단일 책임 (Track A): * 모든 track 의 상태(collector 기록·independent-ANU verdict·Track B Critical7 분류·dependency·dispatch)를 **하나로 모아** consolidated adjudication 을 수행한다. * 분류 taxonomy (회장 verbatim 7종): AUTHORITATIVE_PASS · HOLD_CANDIDATE · AUTO_REMEDIATION_HOLD · CHAIR_HOLD · WAITING_FOR_DEPENDENCY · NOT_STARTED_BY_DESIGN · DISPATCH_NOT_RECEIVED * **개별 collector 는 HOLD_CANDIDATE 만 기록**한다. collector 가 최종 분류(AUTHORITATIVE_PASS/CHAIR_HOLD 등)를 자칭하면 그 자칭은 *무시* 되고 본 batch-level adjudicator 가 재도출한다 (collector self-finalization 금지 — §3/§5). * **최종 분류는 본 batch-level adjudicator 만** 권위가 있다. * Track B classifier 결과(Critical7 여부)를 입력으로 받아 **CHAIR_HOLD vs AUTO_REMEDIATION_HOLD** 를 확정한다. classifier 결과가 없으면 HOLD_CANDIDATE 로 fail-closed (자동 PASS·자동 수렴 금지). * shared invariant 파손 또는 Critical7 = 전체 CHAIR_HOLD (회장 보고). 그 외 non-Critical HOLD 는 AUTO_REMEDIATION_HOLD 로 자동 수렴 — 회장 보고 0. * independent-ANU verdict 만 authoritative. self-chain PASS 자칭만으로 AUTHORITATIVE_PASS 부여 금지(§5.D fail-closed → HOLD_CANDIDATE). Layer A / NO-CRON: 순수 분류 함수. ZERO cron / dispatch / subprocess / cokacdir / network / 파일 쓰기(엔트리포인트 CLI 의 --output 명시 경로 제외). 회수·발사·등록·remediation 실행 0 — *판정만* 한다. callback owner=독립 ANU key (executor self key 금지·+49 정본) 는 본 모듈이 강제하지 않고 ``anu_v3.callback_owner_validator`` 가 등록 직전 강제한다(중복 구현 금지). """ from __future__ import annotations import argparse import json import sys from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Dict, List, Optional, Sequence import jsonschema ADJUDICATION_SCHEMA = "anu_v3.batch_hold_adjudication.v1" # ── per-track / batch classification taxonomy (회장 verbatim 7종) ──────────── AUTHORITATIVE_PASS = "AUTHORITATIVE_PASS" HOLD_CANDIDATE = "HOLD_CANDIDATE" AUTO_REMEDIATION_HOLD = "AUTO_REMEDIATION_HOLD" CHAIR_HOLD = "CHAIR_HOLD" WAITING_FOR_DEPENDENCY = "WAITING_FOR_DEPENDENCY" NOT_STARTED_BY_DESIGN = "NOT_STARTED_BY_DESIGN" DISPATCH_NOT_RECEIVED = "DISPATCH_NOT_RECEIVED" CLASSIFICATIONS = ( AUTHORITATIVE_PASS, HOLD_CANDIDATE, AUTO_REMEDIATION_HOLD, CHAIR_HOLD, WAITING_FOR_DEPENDENCY, NOT_STARTED_BY_DESIGN, DISPATCH_NOT_RECEIVED, ) # Batch-level verdict. PASS = "PASS" FAIL = "FAIL" HOLD_FOR_CHAIR = "HOLD_FOR_CHAIR" # Only signal an individual collector is permitted to record (§3/§5). COLLECTOR_SIGNAL_HOLD_CANDIDATE = HOLD_CANDIDATE # Independent-ANU collector keys (callback owner = ANU key; executor self-key # forbidden as collector — +49 정본). c119085addb0f8b7 = 독립 ANU key. DEFAULT_ANU_KEYS = ("c119085addb0f8b7",) # Terminal-good classifications that count toward batch all-settled. _SETTLED_GOOD = frozenset({AUTHORITATIVE_PASS, NOT_STARTED_BY_DESIGN}) @dataclass(frozen=True) class TrackHoldInput: """One track's consolidated raw state fed to the adjudicator. Every signal is *evidence*, never a final verdict — the adjudicator re-derives the classification. ``collector_recorded`` carries only what an individual collector is permitted to emit (HOLD_CANDIDATE); any other self-finalization claim is recorded and ignored. """ track_id: str task_id: str dispatch_received: bool = True not_started_by_design: bool = False dependency_unmet: bool = False shared_invariant_breach: bool = False hold_candidate: bool = False # Only "HOLD_CANDIDATE" is honoured; anything else = ignored self-claim. collector_recorded: Optional[str] = None collector_key: str = "" executor_key: str = "" collector_role: str = "" collector_session_is_executor_self: bool = False # Independent-ANU authoritative verdict for this track (PASS/FAIL/HOLD) # — only honoured when independence is provable. authoritative_verdict: Optional[str] = None authoritative_is_independent_anu: bool = False # Track B (critical7_and_codex_high_classifier) output for this track. # None -> classifier result not yet available (fail-closed). classifier_present: bool = False classifier_is_critical7: bool = False classifier_category: str = "" # security|credential|...|coverage|test|... detail: str = "" @dataclass class TrackAdjudication: track_id: str task_id: str classification: str chair_escalation: bool auto_remediation: bool settled: bool reasons: List[str] = field(default_factory=list) @dataclass class BatchHoldAdjudication: schema: str verdict: str # PASS | FAIL | HOLD_FOR_CHAIR batch_classification: str all_settled: bool chair_escalation_required: bool auto_remediation_required: bool critical7_present: bool shared_invariant_breach: bool track_count: int classification_counts: Dict[str, int] tracks: List[TrackAdjudication] reasons: List[str] = field(default_factory=list) def to_dict(self) -> dict: d = asdict(self) return d def to_json(self, indent: int = 2) -> str: return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent) def _independent_anu(t: TrackHoldInput, anu_keys: Sequence[str]) -> bool: """Re-derive independence from owner identity — claim is never trusted. Independent iff: track flags it independent AND collector_key is a configured ANU key AND collector_key != executor_key AND role==ANU AND the session is NOT the executor self-session (§5.D / +49 정본). A bare ``authoritative_is_independent_anu=True`` claim without a provable owner identity is downgraded to self-chain (fail-closed). """ keyset = {k for k in anu_keys if k} owner_proven = ( bool(t.collector_key) and t.collector_key in keyset and t.collector_key != t.executor_key and t.collector_role == "ANU" and not t.collector_session_is_executor_self ) # No owner-identity field supplied at all: a bare # ``authoritative_is_independent_anu=True`` claim cannot prove a non-self # owner -> self-chain quarantine, fail-CLOSED (docstring 일치). if t.collector_key or t.executor_key or t.collector_role: return bool(t.authoritative_is_independent_anu) and owner_proven return False def adjudicate_track( t: TrackHoldInput, anu_keys: Sequence[str] = DEFAULT_ANU_KEYS, ) -> TrackAdjudication: """Re-derive the authoritative classification for a single track. Precedence is fail-closed: a track only reaches AUTHORITATIVE_PASS via a provable independent-ANU PASS; everything ambiguous degrades to HOLD_CANDIDATE rather than silently passing. """ reasons: List[str] = [] # Collector self-finalization guard (§3/§5): individual collector may # only record HOLD_CANDIDATE. Any other claimed final classification is # ignored and re-derived by this batch-level adjudicator. if ( t.collector_recorded is not None and t.collector_recorded != COLLECTOR_SIGNAL_HOLD_CANDIDATE ): reasons.append( f"collector self-finalization {t.collector_recorded!r} IGNORED — " "individual collector may record only HOLD_CANDIDATE; final " "classification is re-derived by the batch-level adjudicator " "(회장 §3/§5)." ) # 1. shared invariant breach -> whole-batch CHAIR_HOLD (회장 §6). if t.shared_invariant_breach: reasons.append( "shared invariant breach detected -> CHAIR_HOLD (회장 §6: " "shared invariant 파손 = 전체 CHAIR_HOLD)." ) return TrackAdjudication( t.track_id, t.task_id, CHAIR_HOLD, chair_escalation=True, auto_remediation=False, settled=False, reasons=reasons, ) # 2. dispatch never received. if not t.dispatch_received: reasons.append( "dispatch not received — track has no executor session; " "DISPATCH_NOT_RECEIVED (NOT a HOLD, NOT chair-escalated)." ) return TrackAdjudication( t.track_id, t.task_id, DISPATCH_NOT_RECEIVED, chair_escalation=False, auto_remediation=False, settled=False, reasons=reasons, ) # 3. not started by design (e.g. event-gated track before its gate). if t.not_started_by_design: reasons.append( "track is NOT_STARTED_BY_DESIGN (event-gated / by-design idle) " "— terminal-good, counts toward all-settled, no chair, no " "remediation." ) return TrackAdjudication( t.track_id, t.task_id, NOT_STARTED_BY_DESIGN, chair_escalation=False, auto_remediation=False, settled=True, reasons=reasons, ) # 4. dependency unmet (logical predecessor durable-success absent). if t.dependency_unmet: reasons.append( "dependency unmet — predecessor durable-success EVENT absent; " "WAITING_FOR_DEPENDENCY (event-driven, NOT fixed-time, NOT " "chair-escalated)." ) return TrackAdjudication( t.track_id, t.task_id, WAITING_FOR_DEPENDENCY, chair_escalation=False, auto_remediation=False, settled=False, reasons=reasons, ) independent = _independent_anu(t, anu_keys) av = (t.authoritative_verdict or "").upper() # 5. HOLD path: collector recorded HOLD_CANDIDATE, or independent-ANU # verdict is FAIL/HOLD. Track B classifier decides CHAIR vs AUTO. hold_path = ( t.hold_candidate or t.collector_recorded == COLLECTOR_SIGNAL_HOLD_CANDIDATE or (independent and av in ("FAIL", "HOLD", "HOLD_FOR_CHAIR")) ) if hold_path: if not t.classifier_present: reasons.append( "HOLD detected but Track B critical7/codex-high classifier " "result ABSENT — cannot finalize CHAIR vs AUTO; remains " "HOLD_CANDIDATE (fail-closed: NO auto-pass, NO auto-converge " "without classifier)." ) return TrackAdjudication( t.track_id, t.task_id, HOLD_CANDIDATE, chair_escalation=False, auto_remediation=False, settled=False, reasons=reasons, ) if t.classifier_is_critical7: reasons.append( f"Track B classifier: Critical7 (category=" f"{t.classifier_category or 'unspecified'!s}) -> CHAIR_HOLD " "(회장 §6: Critical7 = 전체 CHAIR_HOLD, 회장 보고)." ) return TrackAdjudication( t.track_id, t.task_id, CHAIR_HOLD, chair_escalation=True, auto_remediation=False, settled=False, reasons=reasons, ) reasons.append( f"Track B classifier: non-Critical (category=" f"{t.classifier_category or 'unspecified'!s}) -> " "AUTO_REMEDIATION_HOLD — ANU-Codex loop 자동 수렴 (회장 §3/§6: " "회장 보고 0)." ) return TrackAdjudication( t.track_id, t.task_id, AUTO_REMEDIATION_HOLD, chair_escalation=False, auto_remediation=True, settled=False, reasons=reasons, ) # 6. AUTHORITATIVE_PASS only via a provable independent-ANU PASS (§5.D). if av == "PASS": if independent: reasons.append( "independent-ANU authoritative verdict = PASS -> " "AUTHORITATIVE_PASS (회장 §5.D: independent ANU only)." ) return TrackAdjudication( t.track_id, t.task_id, AUTHORITATIVE_PASS, chair_escalation=False, auto_remediation=False, settled=True, reasons=reasons, ) reasons.append( "PASS claimed by self-chain (no provable independent-ANU " "ownership) — confirming PASS from a self-chain verdict alone is " "FORBIDDEN (§5.D); degrades to HOLD_CANDIDATE (fail-closed)." ) return TrackAdjudication( t.track_id, t.task_id, HOLD_CANDIDATE, chair_escalation=False, auto_remediation=False, settled=False, reasons=reasons, ) # 7. default fail-closed: anything unresolved is a HOLD_CANDIDATE for the # individual collector to record — never silently settled. reasons.append( "no provable independent-ANU PASS and no resolved HOLD signal — " "fail-closed default HOLD_CANDIDATE (final classification deferred " "to a later batch-level adjudication once evidence arrives)." ) return TrackAdjudication( t.track_id, t.task_id, HOLD_CANDIDATE, chair_escalation=False, auto_remediation=False, settled=False, reasons=reasons, ) def adjudicate_batch( tracks: Sequence[TrackHoldInput], anu_keys: Sequence[str] = DEFAULT_ANU_KEYS, ) -> BatchHoldAdjudication: """Consolidated batch adjudication over ALL track states. 회장 §6: shared invariant 파손 또는 Critical7 = 전체 CHAIR_HOLD. 그 외 non-Critical HOLD 는 AUTO_REMEDIATION_HOLD 자동 수렴(회장 보고 0). all-settled = 모든 track 이 AUTHORITATIVE_PASS / NOT_STARTED_BY_DESIGN. """ per_track = [adjudicate_track(t, anu_keys) for t in tracks] counts: Dict[str, int] = {c: 0 for c in CLASSIFICATIONS} for ta in per_track: counts[ta.classification] = counts.get(ta.classification, 0) + 1 shared_breach = any( t.shared_invariant_breach for t in tracks ) critical7 = any( ta.classification == CHAIR_HOLD and not shared_breach for ta in per_track ) or any(t.classifier_present and t.classifier_is_critical7 for t in tracks) chair_required = any(ta.chair_escalation for ta in per_track) auto_required = any(ta.auto_remediation for ta in per_track) all_settled = bool(per_track) and all( ta.classification in _SETTLED_GOOD for ta in per_track ) reasons: List[str] = [] if chair_required or critical7 or shared_breach: verdict = HOLD_FOR_CHAIR batch_classification = CHAIR_HOLD if shared_breach: reasons.append( "shared invariant 파손 track 존재 -> 전체 CHAIR_HOLD " "(회장 §6). 회장 보고 필요." ) if critical7: reasons.append( "Critical7 track 존재 -> 전체 CHAIR_HOLD (회장 §6). " "회장 보고 필요. 그 외 non-Critical 은 자동 수렴." ) elif all_settled: verdict = PASS batch_classification = AUTHORITATIVE_PASS reasons.append( "모든 track AUTHORITATIVE_PASS / NOT_STARTED_BY_DESIGN — " "batch all-settled. chair 보고 0." ) elif auto_required: verdict = FAIL # not settled yet, but auto-converging (no chair) batch_classification = AUTO_REMEDIATION_HOLD reasons.append( "non-Critical HOLD 존재 -> AUTO_REMEDIATION_HOLD 자동 수렴 " "(ANU-Codex loop). Critical7 0 · 회장 보고 0 · 진행 트리거는 " "event-driven (fallback/dead-man/fixed-time 아님)." ) else: verdict = FAIL # pending: HOLD_CANDIDATE / waiting / dispatch-absent batch_classification = HOLD_CANDIDATE reasons.append( "미해결 HOLD_CANDIDATE / WAITING_FOR_DEPENDENCY / " "DISPATCH_NOT_RECEIVED 존재 — batch 미정착이나 chair-escalation " "불요(Critical7·invariant 파손 0). 추가 evidence(EVENT) 도착 시 " "재-adjudication." ) return BatchHoldAdjudication( schema=ADJUDICATION_SCHEMA, verdict=verdict, batch_classification=batch_classification, all_settled=all_settled, chair_escalation_required=chair_required, auto_remediation_required=auto_required, critical7_present=critical7, shared_invariant_breach=shared_breach, track_count=len(per_track), classification_counts=counts, tracks=per_track, reasons=reasons, ) # ── real CLI entrypoint (문서-only 금지 · mock-only 금지) ──────────────────── def _track_from_dict(d: dict) -> TrackHoldInput: allowed = TrackHoldInput.__dataclass_fields__.keys() return TrackHoldInput(**{k: v for k, v in d.items() if k in allowed}) def adjudicate_from_payload(payload: dict) -> BatchHoldAdjudication: """Real entrypoint: a {"tracks":[...], "anu_keys":[...]} payload in, a validated BatchHoldAdjudication out. No mocks, no I/O side effects.""" tracks = [_track_from_dict(x) for x in payload.get("tracks", [])] anu_keys = payload.get("anu_keys") or DEFAULT_ANU_KEYS return adjudicate_batch(tracks, anu_keys) def _selfcheck() -> int: """Drive the REAL entrypoint over real cases (no mocks). Non-zero exit on any mismatch — this is the regression backstop callable in-allowlist (Track E owns tests/regression/test_batch_hold_adjudication.py).""" failures: List[str] = [] def expect(name: str, payload: dict, want_verdict: str, want_class: str, **want) -> None: r = adjudicate_from_payload(payload) if r.verdict != want_verdict or r.batch_classification != want_class: failures.append( f"{name}: got verdict={r.verdict}/{r.batch_classification} " f"want {want_verdict}/{want_class}" ) for k, v in want.items(): if getattr(r, k) != v: failures.append(f"{name}: {k}={getattr(r, k)!r} want {v!r}") anu = ["c119085addb0f8b7"] exec_self = "c38fb9955616e24d" # 1. all independent-ANU PASS -> AUTHORITATIVE_PASS / all_settled. expect( "all_pass", {"anu_keys": anu, "tracks": [ {"track_id": "A", "task_id": "task-2610", "authoritative_verdict": "PASS", "authoritative_is_independent_anu": True, "collector_key": "c119085addb0f8b7", "executor_key": exec_self, "collector_role": "ANU"}, ]}, PASS, AUTHORITATIVE_PASS, all_settled=True, chair_escalation_required=False, ) # 2. non-Critical HOLD (Track B says not critical7) -> AUTO_REMEDIATION. expect( "non_critical_auto", {"anu_keys": anu, "tracks": [ {"track_id": "B", "task_id": "task-2611", "hold_candidate": True, "classifier_present": True, "classifier_is_critical7": False, "classifier_category": "test_harness_invariant"}, ]}, FAIL, AUTO_REMEDIATION_HOLD, auto_remediation_required=True, chair_escalation_required=False, critical7_present=False, ) # 3. Critical7 -> CHAIR_HOLD for the whole batch (회장 보고). expect( "critical7_chair", {"anu_keys": anu, "tracks": [ {"track_id": "C", "task_id": "task-2612", "hold_candidate": True, "classifier_present": True, "classifier_is_critical7": True, "classifier_category": "credential"}, {"track_id": "A", "task_id": "task-2610", "authoritative_verdict": "PASS", "authoritative_is_independent_anu": True, "collector_key": "c119085addb0f8b7", "executor_key": exec_self, "collector_role": "ANU"}, ]}, HOLD_FOR_CHAIR, CHAIR_HOLD, chair_escalation_required=True, critical7_present=True, ) # 4. shared invariant breach -> whole batch CHAIR_HOLD. expect( "invariant_chair", {"anu_keys": anu, "tracks": [ {"track_id": "D", "task_id": "task-2613", "shared_invariant_breach": True}, ]}, HOLD_FOR_CHAIR, CHAIR_HOLD, chair_escalation_required=True, shared_invariant_breach=True, ) # 5. HOLD but classifier absent -> HOLD_CANDIDATE (fail-closed). expect( "hold_no_classifier", {"anu_keys": anu, "tracks": [ {"track_id": "B", "task_id": "task-2611", "hold_candidate": True, "classifier_present": False}, ]}, FAIL, HOLD_CANDIDATE, chair_escalation_required=False, auto_remediation_required=False, ) # 6. self-chain PASS claim only -> HOLD_CANDIDATE (§5.D fail-closed). expect( "self_chain_pass_quarantined", {"anu_keys": anu, "tracks": [ {"track_id": "A", "task_id": "task-2610", "authoritative_verdict": "PASS", "authoritative_is_independent_anu": True, "collector_key": exec_self, "executor_key": exec_self, "collector_role": "ANU", "collector_session_is_executor_self": True}, ]}, FAIL, HOLD_CANDIDATE, chair_escalation_required=False, ) # 7. dependency unmet / not-started-by-design / dispatch-not-received. expect( "dep_and_design", {"anu_keys": anu, "tracks": [ {"track_id": "E", "task_id": "task-2614", "dependency_unmet": True}, {"track_id": "F", "task_id": "task-2615", "not_started_by_design": True}, {"track_id": "G", "task_id": "task-2999", "dispatch_received": False}, ]}, FAIL, HOLD_CANDIDATE, chair_escalation_required=False, all_settled=False, ) # 8. collector self-finalization is ignored & re-derived. r8 = adjudicate_from_payload({"anu_keys": anu, "tracks": [ {"track_id": "A", "task_id": "task-2610", "collector_recorded": "AUTHORITATIVE_PASS"}, ]}) if r8.batch_classification != HOLD_CANDIDATE: failures.append( "collector_self_final: claimed AUTHORITATIVE_PASS not ignored " f"-> {r8.batch_classification}" ) # 9. (R1 HIGH#1) batch-wide critical7 flag true but NO per-track chair # signal (inconsistent payload) -> defense-in-depth forces whole-batch # CHAIR_HOLD (회장 §6 fail-safe). expect( "critical7_flag_no_pertrack_chair", {"anu_keys": anu, "tracks": [ {"track_id": "X", "task_id": "task-2610", "classifier_present": True, "classifier_is_critical7": True}, ]}, HOLD_FOR_CHAIR, CHAIR_HOLD, chair_escalation_required=False, critical7_present=True, ) # 10. (R1 HIGH#2) bare authoritative_is_independent_anu=True claim with # NO owner identity field at all -> fail-CLOSED, NOT independent, # degrades to HOLD_CANDIDATE (no AUTHORITATIVE_PASS). expect( "bare_independence_claim_fail_closed", {"anu_keys": anu, "tracks": [ {"track_id": "A", "task_id": "task-2610", "authoritative_verdict": "PASS", "authoritative_is_independent_anu": True}, ]}, FAIL, HOLD_CANDIDATE, chair_escalation_required=False, all_settled=False, ) # 11. (R1 MEDIUM) schema forward-enforces invariant: a valid critical7 # output validates; an invariant-violating output (critical7_present # true but verdict != HOLD_FOR_CHAIR) is REJECTED by the schema. _schema_path = Path(__file__).resolve().parent.parent / \ "schemas" / "batch_hold_adjudication.schema.json" _schema = json.loads(_schema_path.read_text(encoding="utf-8")) _valid_c7 = adjudicate_from_payload({"anu_keys": anu, "tracks": [ {"track_id": "C", "task_id": "task-2612", "hold_candidate": True, "classifier_present": True, "classifier_is_critical7": True, "classifier_category": "credential"}, ]}).to_dict() try: jsonschema.validate(_valid_c7, _schema) except jsonschema.ValidationError as e: failures.append( f"schema_valid_c7: real critical7 output rejected by schema: {e.message}" ) _bad = dict(_valid_c7) _bad["verdict"] = "FAIL" _bad["batch_classification"] = "AUTO_REMEDIATION_HOLD" try: jsonschema.validate(_bad, _schema) failures.append( "schema_reject_invariant_violation: invariant-violating output " "(critical7_present=true, verdict=FAIL) was NOT rejected by schema" ) except jsonschema.ValidationError: pass if failures: for f in failures: sys.stderr.write("SELFCHECK FAIL: " + f + "\n") return 1 sys.stderr.write("SELFCHECK PASS: 11 real-entrypoint cases\n") return 0 def _main(argv: Optional[Sequence[str]] = None) -> int: p = argparse.ArgumentParser( prog="anu_v3.batch_hold_adjudicator", description="batch-level consolidated HOLD adjudicator (Track A).", ) p.add_argument("--input", help="path to a batch-state JSON payload") p.add_argument("--output", help="path to write the adjudication JSON") p.add_argument("--selfcheck", action="store_true", help="run real-entrypoint regression cases") a = p.parse_args(argv) if a.selfcheck: return _selfcheck() if not a.input: p.error("--input or --selfcheck required") payload = json.loads(Path(a.input).read_text(encoding="utf-8")) result = adjudicate_from_payload(payload) out = result.to_json() if a.output: # task-2617 회장승인 BOUNDED: Critical7 arbitrary fs write guard. # 출력경로는 cli_output_path_guard policy 허용 경로(fail-closed)로만. from anu_v3.cli_output_path_guard import atomic_guarded_write atomic_guarded_write(a.output, out + "\n", task_id=None) else: sys.stdout.write(out + "\n") return 0 __all__ = [ "ADJUDICATION_SCHEMA", "AUTHORITATIVE_PASS", "HOLD_CANDIDATE", "AUTO_REMEDIATION_HOLD", "CHAIR_HOLD", "WAITING_FOR_DEPENDENCY", "NOT_STARTED_BY_DESIGN", "DISPATCH_NOT_RECEIVED", "CLASSIFICATIONS", "PASS", "FAIL", "HOLD_FOR_CHAIR", "DEFAULT_ANU_KEYS", "TrackHoldInput", "TrackAdjudication", "BatchHoldAdjudication", "adjudicate_track", "adjudicate_batch", "adjudicate_from_payload", ] if __name__ == "__main__": raise SystemExit(_main())