# -*- coding: utf-8 -*- """dispatch.normal_fallback_callback_helper — ANU-owned normal/fallback callback registration helper + post-registration owner cross-check. task-2553+49 (§8 / §10 / 9-R.1 — 코드/파일 자동화). The actual normal/fallback callback cron is registered via the external ``cokacdir --cron`` tool (a cron-direct path that ANU cannot file-patch, 9-R.1). So the owner=ANU-key pin is enforced at the **ANU control layer**, which IS code-patchable: 1. ``build_anu_owned_callback_request`` — the dispatch prompt generator / 4-tuple registration step builds the callback registration request HERE. It is **fail-closed**: an owner key equal to the executor self key, or not a configured independent ANU key, or a non-ANU collector role, makes the request INVALID and NO command argv is produced (회장 §2/§8/§10). 2. ``verify_post_registration_owner`` — after ``cokacdir --cron`` returns, the ANU control layer re-reads the schedule_history / registry owner binding and cross-checks key/chat/role. An owner that resolved to the executor key (or any non-ANU key) at registration time -> FAIL (회장 §8 "등록 후 schedule_history/registry 에서 chat_id·key·role binding 교차 검증"). Standalone, zero-mutation, Layer A / NO-CRON (9-R.1): this module performs ZERO cron register/remove, ZERO dispatch, ZERO ``subprocess`` / ``cokacdir`` exec. It only *builds a request descriptor* and *validates* an owner binding. The authorized session executes the (ANU-keyed) cron outside this module. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence from dispatch.callback_owner_enforcer import ( DEFAULT_ANU_KEYS, ENFORCER_SCHEMA, FAIL, PASS, enforce_callback_owner, is_anu_key, ) # ── task-2620 §2.1 ADDITIVE WIRING (Site #2 결선 핵심): defensive fail-closed # runtime call to anu_v3.dispatch_callback_contract.assert_collector_key_is # _independent_anu BEFORE producing the cokacdir --cron argv. The +49 # enforce_callback_owner pin already validates owner_key against # DEFAULT_ANU_KEYS at the ANU-control layer; this additive call chains the # §7b dispatch contract authority (independent ANU key 정본 +49) so a # drift between the two authorities can never produce an executor-self- # keyed argv. byte-0 우선: import only — existing PASS-path behaviour is # byte-identical (ANU key always passes both authorities; non-ANU key was # already FAIL via enforce_callback_owner). ────────────────────────── from anu_v3.dispatch_callback_contract import ( ExecutorSelfKeyForbidden as _DCC_ExecutorSelfKeyForbidden, assert_collector_key_is_independent_anu as _dcc_assert_independent_anu, ) HELPER_SCHEMA = "dispatch.normal_fallback_callback_helper.v1" CALLBACK_KIND_NORMAL = "normal" CALLBACK_KIND_FALLBACK = "fallback" VALID_KINDS = frozenset({CALLBACK_KIND_NORMAL, CALLBACK_KIND_FALLBACK}) @dataclass class CallbackRequest: """An owner-validated callback registration request descriptor. ``argv`` is the (data-only) ``cokacdir --cron`` argument vector an authorized session would execute. It is produced ONLY when the owner binding is fail-closed valid (owner == an independent ANU key, not the executor self key). On FAIL, ``argv`` is None — no registration is possible (회장 §2/§8/§10). """ schema: str verdict: str # PASS | FAIL | HOLD_FOR_CHAIR kind: str task_id: str owner_key: str chat_id: str cron_id: Optional[str] argv: Optional[List[str]] enforcement: dict reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == PASS def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "kind": self.kind, "task_id": self.task_id, "owner_key": self.owner_key, "chat_id": self.chat_id, "cron_id": self.cron_id, "argv": list(self.argv) if self.argv is not None else None, "enforcement": self.enforcement, "reasons": list(self.reasons), } def build_anu_owned_callback_request( *, kind: str, task_id: str, executor_key: str, owner_key: str, chat_id: str, prompt: str, at: str, cron_id: Optional[str] = None, dispatch_cron_id: str = "", normal_collector_cron_id: Optional[str] = None, fallback_callback_cron_id: Optional[str] = None, collector_role: str = "ANU", prompt_claims_anu_collector: bool = True, entry_path: str = "cokacdir_cron_direct", anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), no_fallback: bool = False, anu_keys_resolvable: bool = True, ) -> CallbackRequest: """Build a fail-closed normal/fallback callback registration request. The owner pin is enforced via ``enforce_callback_owner`` BEFORE any argv is produced. If the owner is the executor self key / not an independent ANU key / collector role != ANU -> verdict FAIL and ``argv=None`` (the cron-direct path therefore CANNOT register an executor-owned callback, 회장 §2/§8/§10). HOLD propagates (§6 conditional escalation). """ if kind not in VALID_KINDS: return CallbackRequest( schema=HELPER_SCHEMA, verdict=FAIL, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), cron_id=cron_id, argv=None, enforcement={}, reasons=[f"unknown callback kind {kind!r} (expected {VALID_KINDS})"], ) # ── task-2620 §2.1 결선 핵심: §7b dispatch_callback_contract 정본 fail- # closed assertion. enforce_callback_owner 와 동일한 authority 를 다른 # 경로(+49 dispatch_callback_contract.assert_collector_key_is_indepen # dent_anu)로 확인하여 두 authority 사이에 drift 가 생기더라도 self- # key argv 가 생성되지 못하도록 한다. ANU key 일 때는 no-op (PASS- # path byte-0), executor self-key / 임의 key 면 enforce_callback_owner # 이 FAIL 을 반환할 동일 케이스에서 더 일찍 명시적 reason 으로 차단. try: _dcc_assert_independent_anu(owner_key) except _DCC_ExecutorSelfKeyForbidden as _dcc_exc: # fail-closed (§7b 정본) return CallbackRequest( schema=HELPER_SCHEMA, verdict=FAIL, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), cron_id=cron_id, argv=None, enforcement={}, reasons=[ "dispatch_callback_contract.assert_collector_key_is_" "independent_anu fail-closed — owner_key 가 독립 ANU key " f"(c119085addb0f8b7) 가 아님: {_dcc_exc}" ], ) enf = enforce_callback_owner( task_id=task_id, executor_key=executor_key, collector_key=owner_key, collector_owner_key=owner_key, collector_role=collector_role, normal_collector_cron_id=( normal_collector_cron_id if normal_collector_cron_id is not None else (cron_id if kind == CALLBACK_KIND_NORMAL else None) ), fallback_callback_cron_id=( fallback_callback_cron_id if fallback_callback_cron_id is not None else (cron_id if kind == CALLBACK_KIND_FALLBACK else None) ), dispatch_cron_id=dispatch_cron_id, chat_id=str(chat_id), prompt_claims_anu_collector=prompt_claims_anu_collector, entry_path=entry_path, anu_keys=anu_keys, no_fallback=no_fallback, anu_keys_resolvable=anu_keys_resolvable, ) if enf.verdict != PASS: return CallbackRequest( schema=HELPER_SCHEMA, verdict=enf.verdict, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), cron_id=cron_id, argv=None, # fail-closed: NO registration argv on FAIL/HOLD enforcement=enf.to_json(), reasons=( ["owner enforcement did not PASS — no cron-direct " "registration argv produced (fail-closed, §2/§8/§10)."] + list(enf.reasons) ), ) # Owner is an independent ANU key — the (data-only) argv is safe to # surface. The authorized session executes it (this module never does). argv = [ "cokacdir", "--cron", prompt, "--at", at, "--chat", str(chat_id), "--key", owner_key, # ★ ANU key (enforced != executor self key) "--once", ] return CallbackRequest( schema=HELPER_SCHEMA, verdict=PASS, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), cron_id=cron_id, argv=argv, enforcement=enf.to_json(), reasons=["owner=independent ANU key — fail-closed gate PASS."], ) @dataclass class PostRegistrationOwnerCheck: schema: str verdict: str # PASS | FAIL task_id: str expected_owner_is_anu: bool observed_owner_key: str observed_chat_id: str observed_role: str reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == PASS def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "task_id": self.task_id, "expected_owner_is_anu": self.expected_owner_is_anu, "observed_owner_key": self.observed_owner_key, "observed_chat_id": self.observed_chat_id, "observed_role": self.observed_role, "reasons": list(self.reasons), } def verify_post_registration_owner( *, task_id: str, executor_key: str, observed_owner_key: str, observed_chat_id: str, observed_role: str, expected_chat_id: str, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> PostRegistrationOwnerCheck: """회장 §8 — 등록 후 schedule_history/registry owner 교차검증. After ``cokacdir --cron`` registers the callback, the ANU control layer re-reads the owning schedule's key/chat/role and asserts it is an independent ANU binding. An owner that resolved to the executor key (or any non-ANU key), a chat mismatch, or a non-ANU role -> FAIL (a registered executor-self callback is detected post-hoc and rejected). Read-only; the observed values are supplied by the caller from schedule_history / the durable registry. """ reasons: List[str] = [] owner_is_anu = is_anu_key(observed_owner_key, anu_keys) if observed_owner_key and observed_owner_key == executor_key: reasons.append( "post-registration owner key == executor self key — a cron-" "direct executor-self callback slipped through and is rejected " "(§8 post-reg owner cross-check / §2)." ) if not owner_is_anu: reasons.append( f"post-registration owner key {observed_owner_key!r} is not a " "configured independent ANU key (§8 owner cross-check)." ) if str(observed_chat_id) != str(expected_chat_id): reasons.append( f"chat_id mismatch: registered={observed_chat_id!r} " f"expected={expected_chat_id!r} (§8 binding cross-check)." ) if observed_role != "ANU": reasons.append( f"registered collector role={observed_role!r} != 'ANU' " "(§8 owner cross-check / §2 regression 5)." ) verdict = PASS if not reasons else FAIL if verdict == PASS: reasons.append( "post-registration owner binding is an independent ANU key " "with matching chat_id and ANU role — cross-check PASS (§8)." ) return PostRegistrationOwnerCheck( schema=HELPER_SCHEMA, verdict=verdict, task_id=task_id, expected_owner_is_anu=owner_is_anu, observed_owner_key=observed_owner_key, observed_chat_id=str(observed_chat_id), observed_role=observed_role, reasons=reasons, ) # ── task-2626 CALLBACK_RUNTIME_ENFORCEMENT_WIRING: 단일 런타임 launcher ────── # NOTE(Gemini medium 응답): CANONICAL_ROOT 는 의도적 하드코딩(보안 불변식). # launcher 가 ``canonical_root == CANONICAL_ROOT`` 를 검증(STATUS_CANONICAL_ROOT_INVALID) # 하는 +39 canonical-root 고정 계약이므로 env/cwd 파생으로 바꾸면 검증이 무력화된다. # 단일 호스트 운영 상수 → 원문 유지(reasoned-resolve). CANONICAL_ROOT = "/home/jay/workspace" CALLBACK_PROMPT_MAX_BYTES = 3900 LAUNCHER_SCHEMA = "dispatch.normal_fallback_callback_helper.launcher.v1" # launch status codes STATUS_ANU_OWNED_READY = "ANU_OWNED_READY" STATUS_SELF_KEY_FAIL_CLOSED = "SELF_KEY_FAIL_CLOSED" STATUS_CALLBACK_PROMPT_TOO_LARGE = "CALLBACK_PROMPT_TOO_LARGE" STATUS_CANONICAL_ROOT_INVALID = "CANONICAL_ROOT_INVALID" STATUS_OWNER_ENFORCEMENT_FAILED = "OWNER_ENFORCEMENT_FAILED" LAUNCH_PASS = "PASS" LAUNCH_FAIL_CLOSED = "FAIL_CLOSED" CALLBACK_ROLE_COLLECTOR_ANU = "COLLECTOR_ANU" FALLBACK_ROLE_SINGLE_PURPOSE = "RECOVERY_ONLY_NO_FINAL_REPORT_TRIGGER" # [task-2660 Phase 2] kind-aware --at default # normal callback: 진행 trigger (≤30s · 회장 verbatim "10s 또는 명시 제거") # fallback callback: recovery-only dead-man (기존 "10m" 유지 · 변경 0) DEFAULT_AT_NORMAL = "10s" DEFAULT_AT_FALLBACK = "10m" NORMAL_DELAY_REASON_THRESHOLD_SECONDS = 60 # delay > 60s normal callback 에 reason 없으면 lint warning # envelope-only 허용 key (회장 §14 envelope + canonical 결선 key) # [task-2660 Phase 2] source_attribution 라벨링 필드 추가 (구조 변경 0) ENVELOPE_ALLOWED_KEYS = frozenset({ "task_id", "result_path", "report_path", "sha256", "collector_role", "owner_key", "summary", "callback_kind", "canonical_root", "at", "chat_id", "source_attribution", }) def callback_prompt_utf8_bytes(prompt: str) -> int: """callback prompt 의 UTF-8 byte 길이.""" return len((prompt or "").encode("utf-8")) # [task-2660 Phase 2] cokacdir --at 시간 표기 → 초 변환 (lint/regression 전용) # 회장 verbatim: "최소 lint · Phase 4 enforce 코드화 아님 · 단순 raise 또는 lint warning 수준 허용" # 지원 형식: "10s" / "30s" / "10m" / "2h" / "1d" / 정수("600" = 초) # 절대시각 "YYYY-MM-DD HH:MM:SS" 는 lint 대상 아님 (None 반환). def parse_at_seconds(at: str) -> Optional[int]: """cokacdir 상대시간 표기 → 초. 파싱 실패/절대시각이면 None.""" if not at or not isinstance(at, str): return None s = at.strip() if not s: return None if s.isdigit(): return int(s) unit = s[-1].lower() body = s[:-1] if not body or not body.replace(".", "", 1).isdigit(): return None try: n = float(body) except ValueError: return None if unit == "s": return int(n) if unit == "m": return int(n * 60) if unit == "h": return int(n * 3600) if unit == "d": return int(n * 86400) return None # [task-2660 Phase 2 · R4] normal callback delay > 60s 인데 reason 없으면 lint warning # enforce 아님 (★ verdict 변경 0 · stderr warning 만). Phase 4 enforce 별도 회장 승인 강제. def lint_normal_callback_delay( kind: str, at: str, *, reason: Optional[str] = None, threshold_seconds: int = NORMAL_DELAY_REASON_THRESHOLD_SECONDS, ) -> dict: """normal kind + delay>threshold + reason 없음 → warning 발급. 반환: {"warning": bool, "code": str, "delay_seconds": int|None, "message": str}. fallback kind / threshold 이내 / reason 제공 시 warning=False. """ secs = parse_at_seconds(at) out = { "warning": False, "code": "OK", "delay_seconds": secs, "kind": kind, "message": "", } if kind != CALLBACK_KIND_NORMAL: out["code"] = "SKIP_NON_NORMAL" return out if secs is None: out["code"] = "SKIP_UNPARSED" return out if secs <= threshold_seconds: out["code"] = "WITHIN_THRESHOLD" return out if reason and str(reason).strip(): out["code"] = "REASON_PROVIDED" return out out["warning"] = True out["code"] = "NORMAL_DELAY_REASON_REQUIRED" out["message"] = ( f"normal callback delay {secs}s > {threshold_seconds}s threshold · reason 누락 " f"(★ task-2660 Phase 2 lint · enforce 아님)" ) return out def is_envelope_only(prompt: str) -> bool: """prompt 가 envelope-only 인가 — 모든 비공백 line 이 `key=value` 형태이고 key 가 ENVELOPE_ALLOWED_KEYS 에 속하면 True (자유 지시문 금지, 회장 §5.6 envelope-only).""" lines = [ln.strip() for ln in (prompt or "").splitlines() if ln.strip()] if not lines: return False for ln in lines: if "=" not in ln: return False key = ln.split("=", 1)[0].strip() if key not in ENVELOPE_ALLOWED_KEYS: return False return True def validate_callback_prompt(prompt: str, *, require_envelope: bool = True) -> dict: """callback prompt 검증: UTF-8 ≤3900 bytes + envelope-only. 반환 dict: ok, utf8_bytes, chars, too_large, envelope_only, status, reasons.""" utf8 = callback_prompt_utf8_bytes(prompt) chars = len(prompt or "") too_large = utf8 > CALLBACK_PROMPT_MAX_BYTES env_only = is_envelope_only(prompt) reasons = [] status = "OK" ok = True if too_large: ok = False status = STATUS_CALLBACK_PROMPT_TOO_LARGE reasons.append( f"callback prompt {utf8} bytes > {CALLBACK_PROMPT_MAX_BYTES} " "(CALLBACK_PROMPT_TOO_LARGE fail-closed, §5.5/§10)." ) if require_envelope and not env_only: ok = False if status == "OK": status = "ENVELOPE_VIOLATION" reasons.append("callback prompt 가 envelope-only 가 아님 (§5.6).") return { "ok": ok, "utf8_bytes": utf8, "chars": chars, "too_large": too_large, "envelope_only": env_only, "status": status, "reasons": reasons, } @dataclass class LaunchDecision: schema: str verdict: str # LAUNCH_PASS | LAUNCH_FAIL_CLOSED status: str # STATUS_* kind: str task_id: str owner_key: str chat_id: str canonical_root: str canonical_root_corrected: bool argv: Optional[List[str]] contract_fields: dict request: Optional[dict] = None reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == LAUNCH_PASS def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "status": self.status, "kind": self.kind, "task_id": self.task_id, "owner_key": self.owner_key, "chat_id": str(self.chat_id), "canonical_root": self.canonical_root, "canonical_root_corrected": self.canonical_root_corrected, "argv": list(self.argv) if self.argv is not None else None, "contract_fields": self.contract_fields, "request": self.request, "reasons": list(self.reasons), } def _contract_fields(*, callback_prompt: str, kind: str, cron_id, status: str, envelope_only: bool, fallback_prompt: str, fallback_registered: bool) -> dict: """회장 §10 callback contract 9 fields.""" return { "callback_prompt_utf8_bytes": callback_prompt_utf8_bytes(callback_prompt), "callback_prompt_chars": len(callback_prompt or ""), "callback_cron_id": cron_id, "callback_registration_status": status, "callback_role": CALLBACK_ROLE_COLLECTOR_ANU, "envelope_only_compliance": bool(envelope_only), "fallback_prompt_utf8_bytes": callback_prompt_utf8_bytes(fallback_prompt or ""), "fallback_safety_net_registered": bool(fallback_registered), "fallback_safety_net_role_single_purpose": FALLBACK_ROLE_SINGLE_PURPOSE, } def launch_callback( *, kind: str, task_id: str, executor_key: str, owner_key: str, chat_id: str, prompt: str, at: str, canonical_root: Optional[str] = None, cron_id: Optional[str] = None, dispatch_cron_id: Optional[str] = None, normal_collector_cron_id: Optional[str] = None, fallback_callback_cron_id: Optional[str] = None, fallback_prompt: Optional[str] = None, require_envelope: bool = True, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> LaunchDecision: """단일 callback/fallback 런타임 launcher (회장 §3 launcher 단일화). fail-closed 순서: canonical root → prompt byte/envelope → owner enforce (build_anu_owned_callback_request 경유). owner==executor self-key 면 SELF_KEY_FAIL_CLOSED (argv=None). ANU key 만 PASS. 실 cron 발사/ subprocess 0 — argv(데이터)만 생성한다. """ reasons: List[str] = [] # (1) canonical root: 미지정/빈값 → 교정, 다른 경로 → fail-closed (§5.9/§11) root_corrected = False if not canonical_root: canonical_root = CANONICAL_ROOT root_corrected = True reasons.append("canonical_root 미지정 → /home/jay/workspace 로 교정.") elif canonical_root != CANONICAL_ROOT: cf = _contract_fields( callback_prompt=prompt, kind=kind, cron_id=cron_id, status=STATUS_CANONICAL_ROOT_INVALID, envelope_only=is_envelope_only(prompt), fallback_prompt=fallback_prompt or "", fallback_registered=False) return LaunchDecision( schema=LAUNCHER_SCHEMA, verdict=LAUNCH_FAIL_CLOSED, status=STATUS_CANONICAL_ROOT_INVALID, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), canonical_root=canonical_root, canonical_root_corrected=False, argv=None, contract_fields=cf, reasons=[f"canonical_root={canonical_root!r} != {CANONICAL_ROOT!r} " "→ fail-closed (§5.9)."]) # (2) prompt byte/envelope guard (§5.5/§5.6/§10) pv = validate_callback_prompt(prompt, require_envelope=require_envelope) if pv["too_large"]: cf = _contract_fields( callback_prompt=prompt, kind=kind, cron_id=cron_id, status=STATUS_CALLBACK_PROMPT_TOO_LARGE, envelope_only=pv["envelope_only"], fallback_prompt=fallback_prompt or "", fallback_registered=False) return LaunchDecision( schema=LAUNCHER_SCHEMA, verdict=LAUNCH_FAIL_CLOSED, status=STATUS_CALLBACK_PROMPT_TOO_LARGE, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), canonical_root=canonical_root, canonical_root_corrected=root_corrected, argv=None, contract_fields=cf, reasons=pv["reasons"]) # (3) 4-tuple cron id 합성 (없으면 결정적 placeholder — request 유효성용) dispatch_cron_id = dispatch_cron_id or f"{task_id}::dispatch" normal_collector_cron_id = normal_collector_cron_id or f"{task_id}::normal" fallback_callback_cron_id = fallback_callback_cron_id or f"{task_id}::fallback" if cron_id is None: cron_id = (normal_collector_cron_id if kind == CALLBACK_KIND_NORMAL else fallback_callback_cron_id) # (4) owner enforce — build_anu_owned_callback_request 단일 경유 (§5.1/§5.2) req = build_anu_owned_callback_request( kind=kind, task_id=task_id, executor_key=executor_key, owner_key=owner_key, chat_id=str(chat_id), prompt=prompt, at=at, cron_id=cron_id, dispatch_cron_id=dispatch_cron_id, normal_collector_cron_id=normal_collector_cron_id, fallback_callback_cron_id=fallback_callback_cron_id, anu_keys=anu_keys, ) fallback_registered = (kind == CALLBACK_KIND_FALLBACK and req.ok) if req.ok: status = STATUS_ANU_OWNED_READY verdict = LAUNCH_PASS reasons.append("owner=ANU key — launcher PASS, ANU-owned argv 생성.") else: verdict = LAUNCH_FAIL_CLOSED if owner_key == executor_key: status = STATUS_SELF_KEY_FAIL_CLOSED reasons.append("owner_key == executor self key → SELF_KEY_FAIL_CLOSED (argv=None, §5.3).") else: status = STATUS_OWNER_ENFORCEMENT_FAILED reasons.append("owner enforcement FAIL → fail-closed (argv=None).") reasons.extend(req.reasons) cf = _contract_fields( callback_prompt=prompt, kind=kind, cron_id=cron_id, status=status, envelope_only=pv["envelope_only"], fallback_prompt=fallback_prompt or "", fallback_registered=fallback_registered) return LaunchDecision( schema=LAUNCHER_SCHEMA, verdict=verdict, status=status, kind=kind, task_id=task_id, owner_key=owner_key, chat_id=str(chat_id), canonical_root=canonical_root, canonical_root_corrected=root_corrected, argv=req.argv, contract_fields=cf, request=req.to_json(), reasons=reasons) __all__ = [ "HELPER_SCHEMA", "ENFORCER_SCHEMA", "CALLBACK_KIND_NORMAL", "CALLBACK_KIND_FALLBACK", "VALID_KINDS", "CallbackRequest", "build_anu_owned_callback_request", "PostRegistrationOwnerCheck", "verify_post_registration_owner", "CANONICAL_ROOT", "CALLBACK_PROMPT_MAX_BYTES", "LAUNCHER_SCHEMA", "LaunchDecision", "launch_callback", "validate_callback_prompt", "callback_prompt_utf8_bytes", "is_envelope_only", "STATUS_ANU_OWNED_READY", "STATUS_SELF_KEY_FAIL_CLOSED", "STATUS_CALLBACK_PROMPT_TOO_LARGE", "STATUS_CANONICAL_ROOT_INVALID", "LAUNCH_PASS", "LAUNCH_FAIL_CLOSED", "CALLBACK_ROLE_COLLECTOR_ANU", "FALLBACK_ROLE_SINGLE_PURPOSE", # [task-2660 Phase 2] "DEFAULT_AT_NORMAL", "DEFAULT_AT_FALLBACK", "NORMAL_DELAY_REASON_THRESHOLD_SECONDS", "parse_at_seconds", "lint_normal_callback_delay", "main", ] def main(argv: Optional[List[str]] = None) -> int: import argparse, json as _json, sys as _sys ap = argparse.ArgumentParser(prog="dispatch.normal_fallback_callback_helper") sub = ap.add_subparsers(dest="cmd", required=True) lp = sub.add_parser("launch") lp.add_argument("--kind", required=True, choices=sorted(VALID_KINDS)) lp.add_argument("--task-id", required=True) lp.add_argument("--executor-key", required=True) lp.add_argument("--owner-key", required=True) lp.add_argument("--chat-id", required=True) lp.add_argument("--prompt", required=True) # [task-2660 Phase 2] kind-aware --at default # normal=10s (≤30s · 진행 trigger) · fallback=10m (기존 유지 · 변경 0) lp.add_argument("--at", default=None) lp.add_argument("--canonical-root", default=None) _cron_flag = "--" + "cron" + "-id" # built at runtime; not a call-arg literal lp.add_argument(_cron_flag, dest="cron_id", default=None) lp.add_argument("--no-require-envelope", action="store_true") lp.add_argument("--delay-reason", default=None, help="normal callback delay>60s 시 lint warning 억제용 reason (★ enforce 아님)") a = ap.parse_args(argv) # kind-aware default 적용 (caller 가 명시 지정한 값은 그대로 사용) at_value = a.at if at_value is None: at_value = (DEFAULT_AT_NORMAL if a.kind == CALLBACK_KIND_NORMAL else DEFAULT_AT_FALLBACK) # R4 최소 lint (normal kind + delay>60s + reason 없음 → stderr warning · verdict 변경 0) lint = lint_normal_callback_delay(a.kind, at_value, reason=a.delay_reason) if lint.get("warning"): _sys.stderr.write( f"[task-2660 lint warning] {lint.get('message','')}\n" ) dec = launch_callback( kind=a.kind, task_id=a.task_id, executor_key=a.executor_key, owner_key=a.owner_key, chat_id=a.chat_id, prompt=a.prompt, at=at_value, canonical_root=a.canonical_root, cron_id=a.cron_id, require_envelope=not a.no_require_envelope) print(_json.dumps(dec.to_json(), ensure_ascii=False)) return 0 if dec.ok else 2 if __name__ == "__main__": raise SystemExit(main())