# -*- coding: utf-8 -*- """dispatch.anu_callback_launch_audit — task-2731 finish-callback 강제 결선. ANCHOR (task-2731 / finish_callback_enforcement_followup_task_packet 축5): - executor 가 callback 안 쏘고 죽어도 OS runner/collector 의 deterministic closeout 경로가 **ANU-owned callback launch 를 코드로 강제**한다(executor 세션 생존 무관 · dev bot prompt 의존 0 · finish-task 끝단 단독 의존 0). - callback = **ANU key 강제, self-key launch 거부**. launch audit jsonl 기록. - result.json 1건당 callback 1회 idempotent(audit jsonl dedupe + runner ledger dedupe + fresh-closeout-only 의 3중 dedupe 중 본 모듈은 audit dedupe 담당). ★ raw key 0: owner_key / ANU key / argv literal 을 audit jsonl 에 **절대 기록하지 않는다**. owner_key_class(ANU/SELF/FOREIGN) 분류 라벨만 담는다. ★ 자가발사 0: 본 모듈은 ANU-owned callback launch **request(argv 데이터)** 를 코드로 강제 빌드하고 audit 를 남길 뿐, 실 cron subprocess 를 직접 발사하지 않는다(``anu_runner_pickup_and_fire`` 재사용 — 그 함수도 surface-only). 실 발사는 권한 있는 ANU 세션이 ACTIVE flag 활성 + ANU key 로 수행한다(activation 별도 승인). ACTIVE=false 기본: ``memory/state/p0b_callback_launch_enabled`` flag 가 "enabled" 가 아니면 **전면 no-op**(audit 미기록 · launch 미빌드). Layer A / NO-CRON: cron register/remove 0, subprocess 직접 0, merge 0, git/gh 0. """ from __future__ import annotations import json import os from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Callable, List, Optional, Sequence # 재사용 모듈 (절대 수정 금지) — self-key 거부/owner 권위 게이트 내장. from dispatch.anu_owned_callback_enforcement import ( # pyright: ignore[reportMissingImports] ANU_KEY, CANONICAL_ROOT, anu_runner_pickup_and_fire, ) from dispatch.callback_owner_enforcer import ( # pyright: ignore[reportMissingImports] DEFAULT_ANU_KEYS, is_anu_key, ) # ── 상수 ───────────────────────────────────────────────────────────────────── CALLBACK_LAUNCH_AUDIT_SCHEMA = "dispatch.anu_callback_launch_audit.v1" # ACTIVE flag (기본 OFF → 전면 no-op). systemd/driver activation 과 독립 gate. CALLBACK_LAUNCH_FLAG_REL = "memory/state/p0b_callback_launch_enabled" CALLBACK_LAUNCH_ENABLED = "enabled" # audit jsonl 기본 경로 DEFAULT_AUDIT_JSONL_REL = "memory/p0b_state/callback_launch_audit.jsonl" # launch verdict LAUNCH_ENFORCED = "LAUNCH_ENFORCED" # ANU-owned callback launch 코드강제 성공(audit 기록) LAUNCH_SELF_KEY_REFUSED = "SELF_KEY_REFUSED" # executor self-key launch 거부(ANU key 강제) LAUNCH_SKIP_DEDUPE = "SKIP_DEDUPE" # (task_id, sha256) 이미 launch 됨 → 재발사 0 LAUNCH_NOOP_DISABLED = "NOOP_DISABLED" # ACTIVE flag OFF → 전면 no-op LAUNCH_NO_RESULT_JSON = "NO_RESULT_JSON" # result.json 부재 LAUNCH_FAIL = "FAIL" # owner enforcement 미통과 등 fail-closed # task-2774+13: normal callback delivery real launcher wiring (gated, default OFF). LAUNCH_REAL_FIRED = "REAL_FIRED" # gated real cokacdir 1-shot fire 성공(cron 등록 확인) LAUNCH_REAL_FIRE_BLOCKED = "REAL_FIRE_BLOCKED" # real_launch_fn 주입됐으나 explicit gate 미충족 → surface-only(fire 0) LAUNCH_REAL_FIRE_FAILED = "REAL_FIRE_FAILED" # real_launch_fn 호출됨 — cron 등록 미확인/예외(fire 시도, 등록 실패) # argv redaction placeholder (raw ANU key 0 — persist/return 용). CALLBACK_ARGV_KEY_PLACEHOLDER = "" # owner_key_class (raw key 아님 — 분류 라벨만) OKC_ANU = "ANU" OKC_SELF = "SELF" OKC_FOREIGN = "FOREIGN" # 성공 launch(surface-only LAUNCH_ENFORCED + gated LAUNCH_REAL_FIRED) 모두 dedupe key. # → 동일 (task_id, sha256) 2회차 pickup 은 SKIP_DEDUPE(real fire single-flight 보존). _AUDIT_DEDUPE_VERDICTS = (LAUNCH_ENFORCED, LAUNCH_REAL_FIRED) # ── CallbackLaunchResult ────────────────────────────────────────────────────── @dataclass class CallbackLaunchResult: """launch_callback_for_closeout 반환값. argv/raw key 미포함(분류 라벨만).""" schema: str verdict: str task_id: str sha256: str owner_key_class: str success: bool executor_session_independent: bool schedule_id: str = "" launch_ts: str = "" audit_path: Optional[str] = None audit_written: bool = False terminal_intent: str = "" reasons: List[str] = field(default_factory=list) # task-2774+13: real launcher 결선 관측값(.ok/.argv contract). argv 는 redacted(raw key 0). argv: Optional[List[str]] = None real_fire_attempted: bool = False @property def ok(self) -> bool: # ENFORCED(신규 강제) + SKIP_DEDUPE(이미 1회 강제됨) 모두 정상 종료. return self.verdict in (LAUNCH_ENFORCED, LAUNCH_SKIP_DEDUPE, LAUNCH_REAL_FIRED) def to_audit_json(self) -> dict: """audit jsonl 한 줄 payload. raw key/argv literal 0 — 분류 라벨만.""" return { "schema": self.schema, "verdict": self.verdict, "task_id": self.task_id, "sha256": self.sha256, "owner_key_class": self.owner_key_class, "launch_ts": self.launch_ts, "schedule_id": self.schedule_id, "success": bool(self.success), "executor_session_independent": bool(self.executor_session_independent), "terminal_intent": self.terminal_intent, "reasons": list(self.reasons), } # ── ACTIVE flag 리더 ────────────────────────────────────────────────────────── def read_callback_launch_enabled( root: str = CANONICAL_ROOT, *, flag_reader: Optional[Callable[[], Optional[str]]] = None ) -> bool: """callback launch ACTIVE flag 첫 줄 trim == "enabled" 일 때만 True. 부재/읽기실패/그 외 값 → False (fail-closed · 전면 no-op). flag_reader: 테스트 주입.""" if flag_reader is not None: try: val = flag_reader() except Exception: # noqa: BLE001 — reader 실패 → disabled(fail-closed) return False if val is None: return False val_str = str(val) return val_str.splitlines()[0].strip() == CALLBACK_LAUNCH_ENABLED if val_str.strip() else False flag_path = os.path.join(root, CALLBACK_LAUNCH_FLAG_REL) try: with open(flag_path, "r", encoding="utf-8") as fh: first = fh.readline() except (OSError, ValueError): return False return first.strip() == CALLBACK_LAUNCH_ENABLED # ── audit dedupe 조회 ───────────────────────────────────────────────────────── def audit_dedupe_hit(audit_path: str, task_id: str, sha256: str) -> bool: """audit jsonl 에 동일 (task_id, sha256) 의 성공 launch 기록 존재 여부. 1 result(=task_id+sha256) → callback 1회 idempotency 의 audit-layer dedupe.""" if not audit_path or not os.path.isfile(audit_path): return False try: with open(audit_path, "r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: entry = json.loads(line) except (ValueError, TypeError): continue # 파싱 실패 라인 skip (fail-safe) if ( isinstance(entry, dict) and entry.get("verdict") in _AUDIT_DEDUPE_VERDICTS and entry.get("task_id") == task_id and entry.get("sha256") == sha256 ): return True except OSError: return False return False def _append_audit(result: CallbackLaunchResult, audit_path: str) -> bool: """CallbackLaunchResult.to_audit_json() 한 줄 append(fsync). 성공 True. 실패는 비치명(fail-safe) — False 반환(launch 자체는 이미 코드강제됨).""" try: os.makedirs(os.path.dirname(audit_path) or ".", exist_ok=True) with open(audit_path, "a", encoding="utf-8") as fh: fh.write(json.dumps(result.to_audit_json(), ensure_ascii=False) + "\n") fh.flush() os.fsync(fh.fileno()) return True except OSError: return False def _classify_owner_key(executor_key: str, anu_keys: Sequence[str]) -> str: """executor_key 의 owner_key_class 라벨(raw key 0). - executor_key 가 ANU key 면 SELF(self-key launch 시도 → 거부 대상). - 비어있거나 ANU key 가 아니면 FOREIGN(executor 자체 key).""" if is_anu_key(executor_key, anu_keys): return OKC_SELF return OKC_FOREIGN def _redact_callback_argv(argv: Optional[Sequence[str]]) -> Optional[List[str]]: """callback argv 에서 raw ANU key 를 placeholder 로 redact(두 형식 모두). - ``--key VALUE`` (공백 구분): 다음 토큰을 placeholder 로 치환(기존 동작 유지). - ``--key=VALUE`` (등호 결합): ``--key=`` 로 치환(신규). 값이 비어있거나(``--key=``) 비정상이어도 raw key 노출 없이 안전 처리. persist/return/log 용 — raw ANU key 0. argv None 이면 None.""" if argv is None: return None out: List[str] = [] redact_next = False for tok in argv: s = str(tok) if redact_next: out.append(CALLBACK_ARGV_KEY_PLACEHOLDER) redact_next = False continue if s == "--key": out.append(s) redact_next = True continue if s.startswith("--key="): # 등호 결합 형식 — 값(비어있거나 비정상 포함)을 통째로 sealed marker 로 치환. out.append("--key=" + CALLBACK_ARGV_KEY_PLACEHOLDER) continue out.append(s) return out def _read_real_fire_outcome(raw: object) -> tuple: """real_launch_fn 반환값(dict 또는 attr 객체) → (cron_registered: bool, schedule_id: str). fail-closed: 필드 부재/None 보수적 기본값. ``build_oneshot_cokacdir_fire_fn`` 의 _fire dict({spawned,cron_registered,callback_evidence,schedule_id}) 형태와 호환.""" if raw is None: return False, "" def _g(key: str): if isinstance(raw, dict): return raw.get(key) return getattr(raw, key, None) schedule_raw = _g("schedule_id") schedule_id = str(schedule_raw).strip() if schedule_raw is not None else "" reg = _g("cron_registered") if reg is None: # 명시값 부재 → 보수적 유추: spawn 했거나 schedule_id 발급됐으면 등록 성립. reg = bool(_g("spawned")) or bool(schedule_id) return bool(reg), schedule_id # ── 공개 API: launch_callback_for_closeout ──────────────────────────────────── def launch_callback_for_closeout( result_json_path: str, *, task_id: str, sha256: str, executor_key: str = "", schedule_id: str = "", terminal_intent: str = "", relay_required: bool = False, root: str = CANONICAL_ROOT, audit_path: Optional[str] = None, flag_reader: Optional[Callable[[], Optional[str]]] = None, clock: Optional[Callable[[], datetime]] = None, anu_key_loader: Optional[Callable[[], Optional[str]]] = None, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), runner_fire_fn: Optional[Callable[..., object]] = None, # ── task-2774+13: normal callback delivery real launcher wiring (default OFF) ── real_launch_fn: Optional[Callable[[Sequence[str]], object]] = None, activation_intent: bool = False, canary_only: bool = False, ) -> CallbackLaunchResult: """collector deterministic closeout 직후, ANU-owned callback launch 를 **코드로 강제**하고 audit jsonl 을 남긴다(executor 세션 생존 무관 · finish-task 끝단 무관). 처리 순서 (fail-closed): 0. ACTIVE flag gate (OFF → 전면 no-op, audit 미기록). 1. result.json 존재 확인 (부재 → NO_RESULT_JSON). 2. self-key 거부: executor_key 가 ANU key 면 SELF → 거부(audit 기록). 3. audit dedupe: (task_id, sha256) 이미 성공 launch → SKIP_DEDUPE(재발사 0). 4. ANU-owned callback launch request 빌드(anu_runner_pickup_and_fire 재사용, self-key 자가발사 금지 내장) → ANU key 강제. 5. audit jsonl append (owner_key_class·launch_ts·schedule_id·success· executor_session_independent — raw key/argv 0). executor_session_independent 는 항상 True: 본 경로는 OS runner/collector 가 호출하며 executor 세션 생명주기와 분리되어 있다(ANCHOR-1). """ _clock = clock or (lambda: datetime.now(timezone.utc)) _audit = audit_path or os.path.join(root, DEFAULT_AUDIT_JSONL_REL) launch_ts = _clock().isoformat() def _result(verdict: str, owner_key_class: str, success: bool, reasons: List[str], *, write_audit: bool, argv: Optional[List[str]] = None, real_fire_attempted: bool = False, schedule_id_override: Optional[str] = None) -> CallbackLaunchResult: res = CallbackLaunchResult( schema=CALLBACK_LAUNCH_AUDIT_SCHEMA, verdict=verdict, task_id=task_id, sha256=sha256, owner_key_class=owner_key_class, success=success, executor_session_independent=True, schedule_id=str(schedule_id_override if schedule_id_override is not None else (schedule_id or "")), launch_ts=launch_ts, audit_path=_audit, terminal_intent=str(terminal_intent or ""), reasons=reasons, argv=argv, real_fire_attempted=real_fire_attempted, ) if write_audit: res.audit_written = _append_audit(res, _audit) return res # ── 0. ACTIVE flag gate ─────────────────────────────────────────────────── if not read_callback_launch_enabled(root, flag_reader=flag_reader): return _result( LAUNCH_NOOP_DISABLED, _classify_owner_key(executor_key, anu_keys), False, ["callback launch ACTIVE flag OFF — 전면 no-op(audit 미기록·launch 미빌드)."], write_audit=False, ) # ── 1. result.json 존재 ─────────────────────────────────────────────────── if not result_json_path or not os.path.isfile(result_json_path): return _result( LAUNCH_NO_RESULT_JSON, OKC_FOREIGN, False, ["result.json 부재 — callback launch 불가(fail-closed)."], write_audit=True, ) # ── 2. self-key 거부 (ANU key 강제) ─────────────────────────────────────── if is_anu_key(executor_key, anu_keys): return _result( LAUNCH_SELF_KEY_REFUSED, OKC_SELF, False, ["executor self-key 로 callback launch 시도 → 거부(ANU key 강제). " "self-collector 패턴 차단(SELF_COLLECTOR_FORBIDDEN)."], write_audit=True, ) # ── 3. audit dedupe (1 result → callback 1회) ───────────────────────────── if audit_dedupe_hit(_audit, task_id, sha256): return _result( LAUNCH_SKIP_DEDUPE, OKC_ANU, True, ["(task_id, sha256) 이미 성공 launch — audit dedupe, callback 재발사 0."], write_audit=False, ) # ── 4. ANU-owned callback launch request 빌드 (self-key 자가발사 금지 내장) ─ _fire = runner_fire_fn or anu_runner_pickup_and_fire _anu_key = ANU_KEY if anu_key_loader is not None: try: loaded = anu_key_loader() except Exception: # noqa: BLE001 — 로더 실패 → 기본 ANU key 상수 fallback loaded = None if loaded: _anu_key = str(loaded) # fail-closed: launch 의 owner_key 는 반드시 독립 ANU key. if not is_anu_key(_anu_key, anu_keys) or (_anu_key == executor_key): return _result( LAUNCH_SELF_KEY_REFUSED, OKC_SELF, False, [f"launch owner_key 가 독립 ANU key 아님 또는 executor self key 와 동일 → " "거부(ANU key 강제)."], write_audit=True, ) try: rr = _fire( result_json_path=result_json_path, executor_key=executor_key, anu_key=_anu_key, anu_keys=tuple(anu_keys), ) except Exception: # noqa: BLE001 — fire 빌드 예외 → fail-closed(audit 기록) # ★ 고정 라벨만 기록: 예외 문자열은 audit jsonl 에 persist 하지 않는다 # (upstream 예외/argv 가 key literal 을 담을 수 있어 raw-key hygiene 위반 위험). return _result( LAUNCH_FAIL, OKC_ANU, False, ["callback launch request 빌드 예외(fail-closed) — 상세는 audit 미기록(raw-key hygiene)."], write_audit=True, ) if not getattr(rr, "ok", False) or getattr(rr, "argv", None) is None: # ★ upstream rr.reasons 는 audit 에 persist 하지 않는다(owner-proof reason 에 # key repr 이 포함될 수 있음 — raw-key hygiene). 고정 라벨만 기록. return _result( LAUNCH_FAIL, OKC_ANU, False, ["ANU-owned callback launch request 빌드 실패(owner enforcement 미통과) — " "상세 reason 은 audit 미기록(raw-key hygiene)."], write_audit=True, ) # ── 5. launch request 빌드 성공. surface-only(기본) vs gated real-fire 분기. ── # rr.argv 는 raw ANU key 를 포함한다 — real_launch_fn 에만 전달하고, persist/return/ # log 에는 절대 쓰지 않는다(redacted_argv 사용). raw key hygiene(task DoD #7). raw_argv = list(getattr(rr, "argv", None) or []) # 위 가드로 argv 非None 보장(getattr: Pyright object 타입 회피). redacted_argv = _redact_callback_argv(raw_argv) path_label = "relay-path" if relay_required else "green-path" # (5-a) default OFF: real_launch_fn 미주입 → surface-only(현행 LAUNCH_ENFORCED, real fire 0). if real_launch_fn is None: return _result( LAUNCH_ENFORCED, OKC_ANU, True, [f"ANU-owned callback launch 코드강제 성공({path_label}·executor 세션 독립). " "실 cron 발사는 권한 있는 ANU 세션이 ANU key 로 수행(자가발사 0)."], write_audit=True, argv=redacted_argv, ) # (5-b) real_launch_fn 주입됐으나 explicit gate(activation_intent + canary_only) 미충족 # → surface-only(REAL_FIRE_BLOCKED, fire 0). owner-proof PASS·governor ALLOW 는 # pickup_once 상류에서 이미 강제됨(미통과 시 callback_launch_fn 자체가 미호출). if not (activation_intent and canary_only): return _result( LAUNCH_REAL_FIRE_BLOCKED, OKC_ANU, True, [f"real_launch_fn 주입됐으나 activation_intent/canary_only 미충족 — surface-only, " f"real fire 0(fail-closed). activation_intent={bool(activation_intent)} " f"canary_only={bool(canary_only)}."], write_audit=True, argv=redacted_argv, ) # (5-c) 전 게이트 충족(ACTIVE flag ON[step0] + activation_intent + canary_only + # real_launch_fn 주입) → ANU-owned callback argv 를 real launcher 로 **정확히 1회** # 발사. single-flight 는 audit dedupe((task_id,sha256), step3)가 보존한다. try: outcome = real_launch_fn(raw_argv) except Exception as exc: # noqa: BLE001 — fire 예외는 type-only fail-closed(raw key/argv/message 미노출) # ★ MED 처리: 예외 type 만 기록(type(exc).__name__). 예외 message 원문·argv·key·env· # path 등 민감정보 섞일 수 있는 내용은 절대 기록하지 않는다(raw-key hygiene). return _result( LAUNCH_REAL_FIRE_FAILED, OKC_ANU, False, [f"real_launch_fn 호출 중 예외(type={type(exc).__name__}) — REAL_FIRE_FAILED " "(type-only·raw key 0·message 원문 미기록). " "재발사는 audit dedupe single-flight 로 차단."], write_audit=True, argv=redacted_argv, real_fire_attempted=True, ) cron_registered, schedule_id_out = _read_real_fire_outcome(outcome) if cron_registered: return _result( LAUNCH_REAL_FIRED, OKC_ANU, True, [f"ANU-owned callback real-fire 1회 성공(cokacdir --cron --once, {path_label}) — " "schedule 등록 확인. single-flight 보존(audit dedupe). raw key 0(redacted argv)."], write_audit=True, argv=redacted_argv, real_fire_attempted=True, schedule_id_override=schedule_id_out or schedule_id, ) return _result( LAUNCH_REAL_FIRE_FAILED, OKC_ANU, False, ["real_launch_fn 호출됨 — cron 등록 미확인(REAL_FIRE_FAILED). real fire 시도·등록 실패. " "고정 라벨(raw key 0). 재시도는 audit dedupe single-flight 적용."], write_audit=True, argv=redacted_argv, real_fire_attempted=True, ) # ── driver/runner 결선용 팩토리 ─────────────────────────────────────────────── def build_callback_launch_fn( root: str = CANONICAL_ROOT, *, audit_path: Optional[str] = None, flag_reader: Optional[Callable[[], Optional[str]]] = None, anu_key_loader: Optional[Callable[[], Optional[str]]] = None, clock: Optional[Callable[[], datetime]] = None, runner_fire_fn: Optional[Callable[..., object]] = None, real_launch_fn: Optional[Callable[[Sequence[str]], object]] = None, activation_intent: bool = False, canary_only: bool = False, ): """pickup_once/driver 에 주입할 callback launch fn 반환(partial). 반환 fn(result_json_path, *, task_id, sha256, executor_key, schedule_id, terminal_intent, relay_required) → CallbackLaunchResult. ACTIVE flag gate 는 fn 내부에서 평가 — flag OFF 면 호출돼도 전면 no-op(audit 0). task-2774+13: real_launch_fn/activation_intent/canary_only 미주입(기본) → surface-only (real fire 0). 주입 시에만 gated real-fire(ACTIVE flag ON + activation_intent + canary_only + real_launch_fn). default OFF 보존(production driver 무주입).""" def _fn(result_json_path, *, task_id, sha256, executor_key="", schedule_id="", terminal_intent="", relay_required=False): return launch_callback_for_closeout( result_json_path, task_id=task_id, sha256=sha256, executor_key=executor_key, schedule_id=schedule_id, terminal_intent=terminal_intent, relay_required=relay_required, root=root, audit_path=audit_path, flag_reader=flag_reader, clock=clock, anu_key_loader=anu_key_loader, runner_fire_fn=runner_fire_fn, real_launch_fn=real_launch_fn, activation_intent=activation_intent, canary_only=canary_only, ) return _fn __all__ = [ "CALLBACK_LAUNCH_AUDIT_SCHEMA", "CALLBACK_LAUNCH_FLAG_REL", "CALLBACK_LAUNCH_ENABLED", "DEFAULT_AUDIT_JSONL_REL", "LAUNCH_ENFORCED", "LAUNCH_SELF_KEY_REFUSED", "LAUNCH_SKIP_DEDUPE", "LAUNCH_NOOP_DISABLED", "LAUNCH_NO_RESULT_JSON", "LAUNCH_FAIL", "LAUNCH_REAL_FIRED", "LAUNCH_REAL_FIRE_BLOCKED", "LAUNCH_REAL_FIRE_FAILED", "CALLBACK_ARGV_KEY_PLACEHOLDER", "OKC_ANU", "OKC_SELF", "OKC_FOREIGN", "CallbackLaunchResult", "read_callback_launch_enabled", "audit_dedupe_hit", "launch_callback_for_closeout", "build_callback_launch_fn", ]