# -*- coding: utf-8 -*- """anu_v3.batch_dependency_classifier — Track D (task-2613). 목표 (회장 verbatim): batch 전체 context에서 upstream gate 미충족으로 아직 dispatch되지 않은 상태를 정확히 분류한다. 핵심 분리 (★ 혼동 절대 금지): * NOT_STARTED_BY_DESIGN — design-phase gate(EVENT_GATED / AUTO_SEQUENCED, upstream 진행 0)로 dispatch 보류가 SPEC 상 정상. 사고 아님. * WAITING_FOR_DEPENDENCY — 선언된 dependency edge 가 부분 충족(일부 upstream durable-success, 전부는 아님)이라 dispatch 보류. 2608 유형. 사고 아님. * DISPATCH_NOT_RECEIVED — gate 가 *충족*(또는 gate 없음)되고 dispatch 가 fired 됐으나 봇이 미수신한 **사고**. 위 둘과 별개 상태. * ELIGIBLE_NOT_YET_DISPATCHED — gate 충족/없음이나 아직 dispatch fired 전인 정상 전이 구간 (fired 아님 → 미수신 사고 아님). * TERMINAL_PRESENT — result.json 존재. not-started 대상 아님. ANTI-CONFLATION INVARIANT (코드+스키마+regression 강제): I1. verdict==DISPATCH_NOT_RECEIVED ⟹ gate_satisfied ∧ dispatch_fired ∧ ¬receipt I2. ¬gate_satisfied ⟹ verdict ∈ {NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY} (gate 미충족은 dispatch 자체가 SPEC 상 보류 — fired 가 아니므로 봇 미수신 사고가 성립 불가. 전제조건이 상호배타.) I3. 정상 보류/진행 4종(NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY, ELIGIBLE_NOT_YET_DISPATCHED, TERMINAL_PRESENT)은 is_incident=False ∧ is_blocking_for_adjudicator=False. 출력은 Track A ``anu_v3.batch_hold_adjudicator`` 의 입력 (schemas/dependency_wait_state.schema.json). blocking_verdicts 는 DISPATCH_NOT_RECEIVED 단 하나 — adjudicator 는 정상 보류를 HOLD 로 격상하지 않는다(Critical7 만 회장 보고 · non-Critical 은 AUTO_REMEDIATION_HOLD 수렴). NO-CRON / Layer A (9-R): 본 모듈은 ZERO write(호출자가 result 를 기록), ZERO cron register/remove, ZERO dispatch, ZERO merge, ZERO credential. 순수 stdlib 결정 함수 + read-only batch-context 스캔. executor self-* 금지(+49 정본): 본 모듈은 callback/collector/adjudication/ dispatch 를 수행하지 않는다 — 분류만 한다. """ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Sequence CLASSIFIER_SCHEMA = "dependency_wait_state.v1" NOT_STARTED_SCHEMA = "not_started_by_design.v1" CLASSIFIER_VERSION = "1.0.0" CANONICAL_WS_ROOT = Path("/home/jay/workspace") # ── verdict enum ───────────────────────────────────────────────────────────── TERMINAL_PRESENT = "TERMINAL_PRESENT" NOT_STARTED_BY_DESIGN = "NOT_STARTED_BY_DESIGN" WAITING_FOR_DEPENDENCY = "WAITING_FOR_DEPENDENCY" ELIGIBLE_NOT_YET_DISPATCHED = "ELIGIBLE_NOT_YET_DISPATCHED" DISPATCH_NOT_RECEIVED = "DISPATCH_NOT_RECEIVED" # gate kinds GATE_NONE = "NONE" GATE_EVENT = "EVENT_GATED" GATE_AUTOSEQ = "AUTO_SEQUENCED" GATE_DEPENDS = "DEPENDS_ON" _NORMAL_HOLD = frozenset( { TERMINAL_PRESENT, NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY, ELIGIBLE_NOT_YET_DISPATCHED, } ) _BLOCKING = frozenset({DISPATCH_NOT_RECEIVED}) class ConflationError(ValueError): """gate 미충족 정상보류 ↔ DISPATCH_NOT_RECEIVED 혼동 시도 — 산출 거부.""" @dataclass(frozen=True) class TrackContext: """단일 track 의 batch-context 증거 (read-only로 수집된 사실).""" track_id: str task_id: str gate_kind: str = GATE_NONE declared_upstream: List[str] = field(default_factory=list) upstream_durable_success: List[str] = field(default_factory=list) dispatch_fired: bool = False dispatch_receipt: bool = False terminal_result_present: bool = False def upstream_unmet(self) -> List[str]: done = set(self.upstream_durable_success) return [u for u in self.declared_upstream if u not in done] def gate_satisfied(self) -> bool: if self.gate_kind == GATE_NONE and not self.declared_upstream: return True return not self.upstream_unmet() @dataclass(frozen=True) class TrackClassification: track_id: str task_id: str verdict: str gate_kind: str gate_satisfied: bool declared_upstream: List[str] upstream_durable_success: List[str] upstream_unmet: List[str] dispatch_fired: bool dispatch_receipt: bool terminal_result_present: bool is_incident: bool is_blocking_for_adjudicator: bool reasons: List[str] = field(default_factory=list) def to_json(self) -> dict: return { "track_id": self.track_id, "task_id": self.task_id, "verdict": self.verdict, "gate_kind": self.gate_kind, "gate_satisfied": self.gate_satisfied, "declared_upstream": list(self.declared_upstream), "upstream_durable_success": list(self.upstream_durable_success), "upstream_unmet": list(self.upstream_unmet), "dispatch_fired": self.dispatch_fired, "dispatch_receipt": self.dispatch_receipt, "terminal_result_present": self.terminal_result_present, "is_incident": self.is_incident, "is_blocking_for_adjudicator": self.is_blocking_for_adjudicator, "reasons": list(self.reasons), } def as_not_started_record(self, by_design_basis: str) -> dict: """NOT_STARTED_BY_DESIGN/WAITING_FOR_DEPENDENCY → 단일 by-design 기록. schemas/not_started_by_design.schema.json 준수. DISPATCH_NOT_RECEIVED 에는 호출 불가(ConflationError) — 별개 상태이므로 by-design 레코드로 표현 자체가 금지된다.""" if self.verdict not in (NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY): raise ConflationError( f"{self.task_id}: verdict={self.verdict!r} 은 by-design 정상" " 보류가 아님 — not_started_by_design 레코드 생성 금지" " (DISPATCH_NOT_RECEIVED 봇 미수신 사고와 혼동 불가)" ) # MEDIUM hardening: classify_track 우회 직접 생성(_assert_invariants # 미통과) 객체로 정상보류 레코드 직렬화 차단 — verdict 외 분리 # invariant 술어(gate_satisfied False ∧ ¬incident ∧ ¬blocking) 재검증. if not ( self.gate_satisfied is False and self.is_incident is False and self.is_blocking_for_adjudicator is False ): raise ConflationError( f"{self.task_id}: by-design 레코드 재검증 실패 — " f"gate_satisfied={self.gate_satisfied!r} " f"is_incident={self.is_incident!r} " f"is_blocking_for_adjudicator=" f"{self.is_blocking_for_adjudicator!r} (정상 보류는 gate" " 미충족 ∧ ¬incident ∧ ¬blocking 이어야 함; classify_track" " 우회 객체 직렬화 금지 — I2/I3 우회 차단)" ) return { "schema": NOT_STARTED_SCHEMA, "track_id": self.track_id, "task_id": self.task_id, "verdict": self.verdict, "gate_kind": self.gate_kind, "gate_satisfied": False, "is_incident": False, "declared_upstream": list(self.declared_upstream), "upstream_durable_success": list(self.upstream_durable_success), "upstream_unmet": list(self.upstream_unmet), "by_design_basis": by_design_basis, "distinct_from_dispatch_not_received": { "asserted": True, "rationale": ( "gate 미충족 → dispatch 자체가 SPEC 상 보류(fired 아님)." " DISPATCH_NOT_RECEIVED 는 gate 충족·dispatch fired 후 봇이" " 미수신한 사고로 전제조건(gate_satisfied ∧ dispatch_fired)" " 자체가 상호배타 — 동일 레코드로 표현 불가." ), }, "reasons": list(self.reasons), } # ── core decision (deterministic) ──────────────────────────────────────────── def classify_track(ctx: TrackContext) -> TrackClassification: """단일 track 의 not-started 상태 분류. ANTI-CONFLATION INVARIANT 강제.""" reasons: List[str] = [] gate_ok = ctx.gate_satisfied() unmet = ctx.upstream_unmet() if ctx.terminal_result_present: verdict = TERMINAL_PRESENT reasons.append("terminal result.json present — not-started 대상 아님") elif not gate_ok: # I2: gate 미충족 ⟹ 절대 DISPATCH_NOT_RECEIVED 아님. dispatch 자체가 # SPEC 상 보류이므로 봇 미수신 사고 성립 불가(전제조건 상호배타). if ctx.dispatch_fired: # 방어선: gate 미충족인데 dispatch 가 fired 됐다고 들어오면 그것은 # 분류 입력의 모순이지 미수신 사고가 아니다 — 사고로 격상 금지. reasons.append( "WARN: gate 미충족인데 dispatch_fired=True 입력 모순 —" " 미수신 사고로 격상하지 않음(정상 보류 우선, I2 보존)" ) if ctx.gate_kind in (GATE_EVENT,) or ( ctx.gate_kind == GATE_AUTOSEQ and not ctx.upstream_durable_success ): verdict = NOT_STARTED_BY_DESIGN reasons.append( f"design-phase gate={ctx.gate_kind} 미충족(unmet={unmet}) —" " dispatch 보류가 SPEC 상 정상. 사고 아님" ) else: verdict = WAITING_FOR_DEPENDENCY reasons.append( f"선언된 dependency edge 부분 충족" f"(done={ctx.upstream_durable_success}, unmet={unmet}) —" " 2608 유형 event-driven 정상 보류. 사고 아님" ) else: # gate 충족 / gate 없음. if ctx.dispatch_fired and not ctx.dispatch_receipt: verdict = DISPATCH_NOT_RECEIVED reasons.append( "gate 충족 ∧ dispatch fired ∧ 봇 미수신 — 봇 미수신 *사고*." " 정상 gate 보류와 별개 상태(INCIDENT, adjudicator HOLD 후보)" ) elif ctx.dispatch_fired and ctx.dispatch_receipt: verdict = TERMINAL_PRESENT reasons.append( "dispatch fired ∧ 봇 수신 — 실행 진행/완료 경로(미수신 사고 아님)" ) else: verdict = ELIGIBLE_NOT_YET_DISPATCHED reasons.append( "gate 충족이나 아직 dispatch fired 전 — 정상 전이 구간." " fired 아님 → 봇 미수신 사고 아님" ) is_incident = verdict == DISPATCH_NOT_RECEIVED is_blocking = verdict in _BLOCKING tc = TrackClassification( track_id=ctx.track_id, task_id=ctx.task_id, verdict=verdict, gate_kind=ctx.gate_kind, gate_satisfied=gate_ok, declared_upstream=list(ctx.declared_upstream), upstream_durable_success=list(ctx.upstream_durable_success), upstream_unmet=unmet, dispatch_fired=ctx.dispatch_fired, dispatch_receipt=ctx.dispatch_receipt, terminal_result_present=ctx.terminal_result_present, is_incident=is_incident, is_blocking_for_adjudicator=is_blocking, reasons=reasons, ) _assert_invariants(tc) return tc def _assert_invariants(tc: TrackClassification) -> None: """ANTI-CONFLATION INVARIANT 하드 강제. 위반 시 ConflationError.""" # I1 if tc.verdict == DISPATCH_NOT_RECEIVED: if not ( tc.gate_satisfied and tc.dispatch_fired and not tc.dispatch_receipt and tc.is_incident and tc.is_blocking_for_adjudicator ): raise ConflationError( f"I1 위반 {tc.task_id}: DISPATCH_NOT_RECEIVED 는 gate 충족 ∧" " dispatch fired ∧ 봇 미수신 ∧ incident ∧ blocking 일 때만 가능" ) # I2 if not tc.gate_satisfied and tc.verdict not in ( NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY, ): raise ConflationError( f"I2 위반 {tc.task_id}: gate 미충족인데 verdict={tc.verdict!r} —" " gate 미충족은 NOT_STARTED_BY_DESIGN/WAITING_FOR_DEPENDENCY 만" " (봇 미수신 사고와 혼동 금지)" ) # I3 if tc.verdict in _NORMAL_HOLD and ( tc.is_incident or tc.is_blocking_for_adjudicator ): raise ConflationError( f"I3 위반 {tc.task_id}: 정상 보류 {tc.verdict!r} 가 incident/" "blocking 으로 표시됨" ) def classify_batch( contexts: Sequence[TrackContext], batch_id: str ) -> dict: """batch 전체 분류 → dependency_wait_state.v1 (adjudicator 입력).""" rows = [classify_track(c) for c in contexts] by_verdict: Dict[str, int] = {} for r in rows: by_verdict[r.verdict] = by_verdict.get(r.verdict, 0) + 1 incident = sum(1 for r in rows if r.is_incident) blocking = sum(1 for r in rows if r.is_blocking_for_adjudicator) return { "schema": CLASSIFIER_SCHEMA, "classifier_version": CLASSIFIER_VERSION, "generated_at_basis": "batch_context_scan", "batch_id": batch_id, "anti_conflation_invariant_held": True, "adjudicator_input_contract": { "consumer": "anu_v3.batch_hold_adjudicator", "blocking_verdicts": [DISPATCH_NOT_RECEIVED], "non_blocking_verdicts": [ NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY, ELIGIBLE_NOT_YET_DISPATCHED, TERMINAL_PRESENT, ], }, "tracks": [r.to_json() for r in rows], "summary": { "total": len(rows), "by_verdict": by_verdict, "incident_count": incident, "blocking_for_adjudicator_count": blocking, }, "reasons": [ "gate 미충족 정상보류(NOT_STARTED_BY_DESIGN/WAITING_FOR_DEPENDENCY)" " 와 봇 미수신 사고(DISPATCH_NOT_RECEIVED)는 ENUM 차원 분리·" "상호배타 전제조건으로 보존", "blocking_verdicts=[DISPATCH_NOT_RECEIVED] 단 하나 — adjudicator 는" " 정상 보류를 HOLD 로 격상하지 않음", ], } # ── real batch-context scan (실 entrypoint, mock-only 아님) ─────────────────── # durable-success 인식 술어(gate 충족 인식 폭 — 명시 가시화·감사 가능). # 본 집합은 회장 규칙2 'all-settled' 보다 엄격한 fail-closed 보수 부분집합: # 비-durable terminal(status=OK / IMPLEMENTED_PLAN_ONLY 등)은 gate 충족으로 # 인정하지 않는다. 인식 폭은 gate *충족* 방향으로만 작용 — gate 미충족을 # DISPATCH_NOT_RECEIVED 사고로 격상하는 경로는 존재하지 않으므로 분리장벽 # (I2: ¬gate ⟹ 정상보류 · C4: 모순입력 비격상)은 byte-불변. _DURABLE_STATUS = frozenset({"COMPLETED", "DONE", "SETTLED"}) _DURABLE_RESULT_PREFIX = "PASS" _DURABLE_OUTCOME_TOKENS = frozenset({"SETTLED", "ACCEPT"}) def _events_dir(ws_root: Path) -> Path: return ws_root / "memory" / "events" def _has_terminal_result(events: Path, task_id: str) -> bool: return (events / f"{task_id}.result.json").exists() def _durable_success(events: Path, task_id: str) -> bool: """task_id 가 durable-success result 를 가졌는지(gate 충족 인식 술어). 인식 폭(명시·감사 가능): * status ∈ _DURABLE_STATUS ({COMPLETED, DONE, SETTLED}), 또는 * result 가 _DURABLE_RESULT_PREFIX ('PASS') 로 시작, 또는 * outcome 에 _DURABLE_OUTCOME_TOKENS ({SETTLED, ACCEPT}) 중 하나 포함. 회장 규칙2 'all-settled' 보다 엄격한 fail-closed 보수 부분집합. 미인식 upstream → gate 미충족 → 다운스트림 정상 보류(절대 미수신 사고 아님; I2·C4 불변). 인식 폭은 gate 충족 방향만 작용 — 분리장벽 byte-불변.""" rp = events / f"{task_id}.result.json" if not rp.exists(): return False try: d = json.loads(rp.read_text(encoding="utf-8")) except (ValueError, OSError): return False status = str(d.get("status", "")).upper() res = str(d.get("result", "")).upper() outcome = str(d.get("outcome", "")).upper() return ( status in _DURABLE_STATUS or res.startswith(_DURABLE_RESULT_PREFIX) or any(tok in outcome for tok in _DURABLE_OUTCOME_TOKENS) ) def _dispatch_fired(events: Path, task_id: str) -> bool: if (events / f"{task_id}.dispatch-fired.json").exists(): return True for p in events.glob(f"{task_id}*dispatch-fired*.json"): if p.exists(): return True return False def _dispatch_receipt(events: Path, task_id: str) -> bool: """봇 수신 증거: result / collector-result / activation-decision / independent-collector-result 중 하나라도 존재.""" for suffix in ( ".result.json", ".independent-collector-result.json", ".collector-result.json", ".activation-decision.json", "+0.activation-decision.json", ): if (events / f"{task_id}{suffix}").exists(): return True for p in events.glob(f"{task_id}*activation-decision*.json"): if p.exists(): return True return False def _load_preflight_batch(ws_root: Path) -> Optional[dict]: """task-2610 batch-hold-system preflight 에서 6-track 정의를 로드.""" p = _events_dir(ws_root) / ( "task-2610-batch-hold-system-preflight-decision_260519.json" ) if not p.exists(): return None try: return json.loads(p.read_text(encoding="utf-8")) except (ValueError, OSError): return None def build_real_contexts(ws_root: Path = CANONICAL_WS_ROOT) -> List[TrackContext]: """실 workspace(preflight + memory/events) 스캔으로 6-track context 구성. mock 아님 — 실제 디스크의 task-26NN.result/dispatch-fired/activation 증거를 읽어 gate 충족 여부를 사실 기반으로 계산한다.""" events = _events_dir(ws_root) pf = _load_preflight_batch(ws_root) out: List[TrackContext] = [] if not pf: return out task_ids = pf.get("task_ids", {}) # 의존 구조 (preflight verdict verbatim · gate 의미는 *구현 기준* 기술): # A,B,C,D 병렬(gate 없음). # E = AUTO_SEQUENCED on {A,B,C,D}: gate 충족 = 선언 upstream *전부* # durable-success (_durable_success() 인식 술어 — _DURABLE_* 상수). # F = EVENT_GATED on {A,B,C,D,E}: 동일 — gate 충족 = 선언 upstream # 전부 durable-success. # ★ 회장 규칙2 의 'all-settled' 표현 대비 본 구현은 durable-success 만 # gate 충족으로 인정 — all-settled(비-durable terminal 포함)보다 # *엄격한 fail-closed 보수* 선택. 미인식 upstream → 다운스트림은 # 정상 보류(NOT_STARTED_BY_DESIGN/WAITING_FOR_DEPENDENCY)로 남고 # 절대 DISPATCH_NOT_RECEIVED 사고로 격상되지 않는다(I2·C4 불변). A = task_ids.get("TrackA", "task-2610").split()[0] B = task_ids.get("TrackB", "task-2611").split()[0] C = task_ids.get("TrackC", "task-2612").split()[0] D = task_ids.get("TrackD", "task-2613").split()[0] E = task_ids.get("TrackE", "task-2614").split()[0] F = task_ids.get("TrackF", "task-2615").split()[0] spec = [ ("A", A, GATE_NONE, []), ("B", B, GATE_NONE, []), ("C", C, GATE_NONE, []), ("D", D, GATE_NONE, []), ("E", E, GATE_AUTOSEQ, [A, B, C, D]), ("F", F, GATE_EVENT, [A, B, C, D, E]), ] for track_id, tid, gate, ups in spec: ups_done = [u for u in ups if _durable_success(events, u)] out.append( TrackContext( track_id=track_id, task_id=tid, gate_kind=gate, declared_upstream=ups, upstream_durable_success=ups_done, dispatch_fired=_dispatch_fired(events, tid), dispatch_receipt=_dispatch_receipt(events, tid), terminal_result_present=_has_terminal_result(events, tid), ) ) return out # ── embedded regression (canonical cases + anti-conflation invariant) ──────── def regression_cases() -> List[dict]: """결정적 회귀 케이스. mock-only 통과가 아니라 실 entrypoint 의 동작 명세 — 특히 2608 유형 vs DISPATCH_NOT_RECEIVED 분리를 강제한다.""" cases = [] # C1: 2608 유형 — AUTO_SEQ, upstream 부분 충족 → WAITING_FOR_DEPENDENCY c1 = classify_track( TrackContext( "E", "task-2614", GATE_AUTOSEQ, ["task-2610", "task-2611", "task-2612", "task-2613"], ["task-2610", "task-2611"], ) ) assert c1.verdict == WAITING_FOR_DEPENDENCY, c1.verdict assert not c1.is_incident and not c1.is_blocking_for_adjudicator cases.append({"case": "C1_2608_partial_dep", "verdict": c1.verdict, "pass": True}) # C2: design-phase EVENT_GATED 미충족 → NOT_STARTED_BY_DESIGN c2 = classify_track( TrackContext( "F", "task-2615", GATE_EVENT, ["task-2610", "task-2611", "task-2612", "task-2613", "task-2614"], [], ) ) assert c2.verdict == NOT_STARTED_BY_DESIGN, c2.verdict assert not c2.is_incident cases.append({"case": "C2_event_gated_design", "verdict": c2.verdict, "pass": True}) # C3: gate 충족 + dispatch fired + 봇 미수신 → DISPATCH_NOT_RECEIVED (사고) c3 = classify_track( TrackContext( "X", "task-9001", GATE_NONE, [], [], dispatch_fired=True, dispatch_receipt=False, ) ) assert c3.verdict == DISPATCH_NOT_RECEIVED, c3.verdict assert c3.is_incident and c3.is_blocking_for_adjudicator cases.append({"case": "C3_dispatch_not_received_incident", "verdict": c3.verdict, "pass": True}) # C4 (★ anti-conflation): gate 미충족인데 dispatch_fired=True 모순 입력 → # 절대 DISPATCH_NOT_RECEIVED 로 격상 안 함(I2 보존, 정상 보류 우선) c4 = classify_track( TrackContext( "Y", "task-9002", GATE_EVENT, ["task-2610"], [], dispatch_fired=True, dispatch_receipt=False, ) ) assert c4.verdict == NOT_STARTED_BY_DESIGN, c4.verdict assert not c4.is_incident, "gate 미충족은 절대 미수신 사고로 격상 불가" cases.append({"case": "C4_anticonflation_gate_unmet_not_incident", "verdict": c4.verdict, "pass": True}) # C5: as_not_started_record 는 DISPATCH_NOT_RECEIVED 에 호출 불가 conflation_blocked = False try: c3.as_not_started_record("should fail") except ConflationError: conflation_blocked = True assert conflation_blocked, "DISPATCH_NOT_RECEIVED→by-design 레코드 차단 실패" cases.append({"case": "C5_dnr_cannot_be_by_design_record", "pass": True}) # C6: gate 충족 + fired + 수신 → 진행(사고 아님) c6 = classify_track( TrackContext( "Z", "task-9003", GATE_NONE, [], [], dispatch_fired=True, dispatch_receipt=True, ) ) assert c6.verdict == TERMINAL_PRESENT and not c6.is_incident cases.append({"case": "C6_fired_received_not_incident", "verdict": c6.verdict, "pass": True}) # ── AR1 (task-2613+1) additive 회귀: gate 정합·인식폭·hardening ── # C7: gate 의미 정합화 후에도 ¬gate_satisfied ⟹ 정상보류 (I2 byte-불변) c7a = classify_track( TrackContext( "E", "task-2614", GATE_AUTOSEQ, ["task-2610", "task-2611", "task-2612", "task-2613"], ["task-2610"], ) ) c7b = classify_track( TrackContext( "F", "task-2615", GATE_EVENT, ["task-2610", "task-2611", "task-2612", "task-2613", "task-2614"], [], ) ) assert not c7a.gate_satisfied and c7a.verdict in ( NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY), c7a.verdict assert not c7a.is_incident and not c7a.is_blocking_for_adjudicator assert not c7b.gate_satisfied and c7b.verdict == NOT_STARTED_BY_DESIGN assert not c7b.is_incident and not c7b.is_blocking_for_adjudicator cases.append({"case": "C7_gate_align_I2_preserved", "verdict": [c7a.verdict, c7b.verdict], "pass": True}) # C8: 인식 폭 확장 입력에서도 gate 충족+fired+미수신만 # DISPATCH_NOT_RECEIVED · gate 미충족+fired 모순 비격상(C4 불변) c8_dnr = classify_track( TrackContext( "G", "task-9101", GATE_AUTOSEQ, ["u1", "u2"], ["u1", "u2"], dispatch_fired=True, dispatch_receipt=False, ) ) c8_contra = classify_track( TrackContext( "H", "task-9102", GATE_AUTOSEQ, ["u1", "u2"], ["u1"], dispatch_fired=True, dispatch_receipt=False, ) ) assert c8_dnr.verdict == DISPATCH_NOT_RECEIVED, c8_dnr.verdict assert c8_dnr.is_incident and c8_dnr.is_blocking_for_adjudicator assert c8_contra.verdict in ( NOT_STARTED_BY_DESIGN, WAITING_FOR_DEPENDENCY), c8_contra.verdict assert not c8_contra.is_incident, "gate 미충족+fired 모순 비격상(C4)" cases.append({"case": "C8_recognition_breadth_C4_invariant", "verdict": [c8_dnr.verdict, c8_contra.verdict], "pass": True}) # C9: as_not_started_record 가 classify_track 우회 위조 객체 # (gate_satisfied=True / incident=True) 에 ConflationError rogue_gate = TrackClassification( track_id="R", task_id="task-9201", verdict=NOT_STARTED_BY_DESIGN, gate_kind=GATE_EVENT, gate_satisfied=True, declared_upstream=["u1"], upstream_durable_success=[], upstream_unmet=["u1"], dispatch_fired=False, dispatch_receipt=False, terminal_result_present=False, is_incident=False, is_blocking_for_adjudicator=False, reasons=[], ) rogue_incident = TrackClassification( track_id="R", task_id="task-9202", verdict=WAITING_FOR_DEPENDENCY, gate_kind=GATE_AUTOSEQ, gate_satisfied=False, declared_upstream=["u1"], upstream_durable_success=[], upstream_unmet=["u1"], dispatch_fired=False, dispatch_receipt=False, terminal_result_present=False, is_incident=True, is_blocking_for_adjudicator=True, reasons=[], ) blocked_g = blocked_i = False try: rogue_gate.as_not_started_record("rogue") except ConflationError: blocked_g = True try: rogue_incident.as_not_started_record("rogue") except ConflationError: blocked_i = True assert blocked_g, "gate_satisfied=True 우회 by-design 직렬화 차단 실패" assert blocked_i, "incident=True 우회 by-design 직렬화 차단 실패" cases.append({"case": "C9_as_not_started_revalidation", "pass": True}) # C10: 정상보류 출력 schema-valid · DISPATCH_NOT_RECEIVED 가 incident/ # blocking 누락(위반) 시 schema reject (forward-invariant 강화 실증) import importlib try: _js = importlib.import_module("jsonschema") except ImportError as _e: # pragma: no cover raise AssertionError( "C10 requires jsonschema (draft-07 reject 실증)" ) from _e _schema = json.loads( (CANONICAL_WS_ROOT / "schemas" / "dependency_wait_state.schema.json").read_text(encoding="utf-8") ) valid_batch = classify_batch( [TrackContext("E", "task-2614", GATE_AUTOSEQ, ["task-2610", "task-2611"], ["task-2610"])], "c10-batch", ) _js.validate(valid_batch, _schema) bad = json.loads(json.dumps(valid_batch)) bad["tracks"][0]["verdict"] = DISPATCH_NOT_RECEIVED bad["tracks"][0]["gate_satisfied"] = True bad["tracks"][0]["dispatch_fired"] = True bad["tracks"][0]["dispatch_receipt"] = False bad["tracks"][0]["is_incident"] = False bad["tracks"][0]["is_blocking_for_adjudicator"] = False rejected = False try: _js.validate(bad, _schema) except _js.ValidationError: rejected = True assert rejected, "DNR+¬incident 출력을 schema 가 reject 못함" cases.append({"case": "C10_schema_forward_invariant_reject", "pass": True}) return cases def main(argv: Optional[Sequence[str]] = None) -> int: """실 entrypoint. --mode classify|regression|both (default both). classify: 실 workspace(preflight+events) 스캔 → dependency_wait_state.v1. regression: 결정적 회귀(2608 유형 vs DISPATCH_NOT_RECEIVED 분리 강제). both: regression PASS 를 게이트로 실 분류 실행(mock-only 불가).""" import argparse p = argparse.ArgumentParser( description="anu_v3.batch_dependency_classifier (Track D / task-2613)" ) p.add_argument("--ws-root", default=str(CANONICAL_WS_ROOT)) p.add_argument( "--mode", choices=["classify", "regression", "both"], default="both" ) p.add_argument("--batch-id", default="batch-hold-system-2610..2615") p.add_argument("--out", default="") args = p.parse_args(argv) ws = Path(args.ws_root) out: Dict[str, object] = {} if args.mode in ("regression", "both"): cases = regression_cases() out["regression"] = { "all_pass": all(c.get("pass") for c in cases), "cases": cases, "mock_only": False, } if not out["regression"]["all_pass"]: print(json.dumps(out, ensure_ascii=False, indent=2)) return 1 if args.mode in ("classify", "both"): ctxs = build_real_contexts(ws) real = classify_batch(ctxs, args.batch_id) real["real_context"] = { "ws_root": str(ws), "preflight_present": _load_preflight_batch(ws) is not None, "tracks_scanned": len(ctxs), "mock_only": False, } out["classification"] = real text = json.dumps(out, ensure_ascii=False, indent=2) + "\n" if args.out: # 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(args.out, text, task_id=None) print(f"batch_dependency_classifier → {args.out}") else: print(text) return 0 if __name__ == "__main__": # pragma: no cover raise SystemExit(main())