# -*- coding: utf-8 -*- """utils.callback_collector_helper_integration — task-2646 helper 통합 wiring (v2). task-2644+1 ANU_CALLBACK_COLLECTOR_CONTROL_PLANE_CLEAN_REPLACEMENT spec (read-only 참조): memory/specs/system_anu_callback_collector_control_plane_spec_260524.md task md: memory/tasks/task-2644+1.md (sha256 b79d6f150d3f44cc0971824b94b62b348d452a5aa905d98ed5a8614d49e7e5dd) ★ 본 module 은 v2 control-plane 산출물들이 task-2646 helper 3종을 - utils.callback_registration.register_callback / verify_actual_owner - utils.callback_authority_validator.validate_authority - utils.callback_source_cross_checker.cross_check_sources 를 단일 진입점으로 사용하도록 강제하는 wiring layer. task md 11 필수 원칙 (1:1): 4. task-2646 helper/authority validator 기준 필수 5. ANU key actual owner 검증 필수 (등록 직후 actual schedule owner key == c119085addb0f8b7) 6. self-key callback 이면 즉시 FAIL (SELF_COLLECTOR_FORBIDDEN) 7. 4 source cross-check 필수 (schedule_history + cron-history + envelope + result artifact) 10. registration helper bypass 시 fail-closed helper 부재 시 fail-closed: HELPER_INTEGRATION_BYPASS state 로 강제 종료. """ from __future__ import annotations import importlib import importlib.util import sys from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Sequence SCHEMA = "utils.callback_collector_helper_integration.v1" ANU_KEY = "c119085addb0f8b7" DEFAULT_ANU_KEYS: frozenset = frozenset({ANU_KEY}) HELPER_REGISTRATION = "utils.callback_registration" HELPER_AUTHORITY = "utils.callback_authority_validator" HELPER_CROSS_CHECK = "utils.callback_source_cross_checker" # Bound module aliases registered into sys.modules under a non-conflicting # namespace so loaders inside the helpers continue to resolve siblings. _ALIAS_REGISTRATION = "_task_2646_helpers.callback_registration" _ALIAS_AUTHORITY = "_task_2646_helpers.callback_authority_validator" _ALIAS_CROSS_CHECK = "_task_2646_helpers.callback_source_cross_checker" REQUIRED_HELPERS = (HELPER_REGISTRATION, HELPER_AUTHORITY, HELPER_CROSS_CHECK) INTEGRATION_OK = "HELPER_INTEGRATION_OK" INTEGRATION_BYPASS = "HELPER_INTEGRATION_BYPASS" SELF_COLLECTOR_FORBIDDEN = "SELF_COLLECTOR_FORBIDDEN" # task-2646 worktree fallback path — only used when main lacks the helpers. _FALLBACK_WORKTREE = Path("/home/jay/workspace/.worktrees/task-2646-dev3") _HELPER_FILES = { HELPER_REGISTRATION: ("callback_registration.py", _ALIAS_REGISTRATION), HELPER_AUTHORITY: ("callback_authority_validator.py", _ALIAS_AUTHORITY), HELPER_CROSS_CHECK: ("callback_source_cross_checker.py", _ALIAS_CROSS_CHECK), } @dataclass class IntegrationStatus: schema: str available: bool state: str helpers_loaded: List[str] helpers_missing: List[str] reasons: List[str] = field(default_factory=list) fallback_path_used: Optional[str] = None def to_json(self) -> Dict[str, Any]: return { "schema": self.schema, "available": self.available, "state": self.state, "helpers_loaded": list(self.helpers_loaded), "helpers_missing": list(self.helpers_missing), "reasons": list(self.reasons), "fallback_path_used": self.fallback_path_used, } def _load_from_file(alias: str, file_path: Path): """Load a helper module by absolute file path under an alias namespace. Returns the loaded module or None on failure. Registers under both the alias and the canonical helper name so callers using either work. """ if alias in sys.modules: return sys.modules[alias] spec = importlib.util.spec_from_file_location(alias, str(file_path)) if spec is None or spec.loader is None: return None module = importlib.util.module_from_spec(spec) sys.modules[alias] = module try: spec.loader.exec_module(module) except Exception: sys.modules.pop(alias, None) raise return module def _try_import_helpers() -> IntegrationStatus: loaded: List[str] = [] missing: List[str] = [] reasons: List[str] = [] fallback_used: Optional[str] = None for name in REQUIRED_HELPERS: try: importlib.import_module(name) loaded.append(name) except Exception as exc: # noqa: BLE001 — record reason, continue missing.append(name) reasons.append(f"primary import {name!r} failed: {exc}") if missing and _FALLBACK_WORKTREE.is_dir(): fallback_str = str(_FALLBACK_WORKTREE) still_missing: List[str] = [] for name in list(missing): filename, alias = _HELPER_FILES[name] helper_file = _FALLBACK_WORKTREE / "utils" / filename if not helper_file.is_file(): still_missing.append(name) reasons.append(f"fallback file {helper_file} missing") continue try: module = _load_from_file(alias, helper_file) if module is None: still_missing.append(name) reasons.append(f"fallback spec_from_file_location({helper_file}) returned None") continue # Bind canonical name too so callers using import_module(name) work. sys.modules[name] = module loaded.append(name) fallback_used = fallback_str except Exception as exc: # noqa: BLE001 still_missing.append(name) reasons.append(f"fallback import {name!r} from file failed: {exc}") missing = still_missing if missing: return IntegrationStatus( schema=SCHEMA, available=False, state=INTEGRATION_BYPASS, helpers_loaded=loaded, helpers_missing=missing, reasons=reasons + [ "HELPER_INTEGRATION_BYPASS: required task-2646 helpers are " "unavailable; registration must be considered fail-closed " "(task md ANCHOR-2 violation)." ], fallback_path_used=fallback_used, ) return IntegrationStatus( schema=SCHEMA, available=True, state=INTEGRATION_OK, helpers_loaded=loaded, helpers_missing=[], reasons=reasons, fallback_path_used=fallback_used, ) def integration_status() -> IntegrationStatus: """Public probe used by hooks/adjudicator/runner v2.""" return _try_import_helpers() def require_helpers() -> IntegrationStatus: """Eagerly import helpers. Returns IntegrationStatus; raises only if truly unavailable (callers handle BYPASS as fail-closed).""" status = _try_import_helpers() return status def _get_modules(): status = require_helpers() if not status.available: return None, status reg = importlib.import_module(HELPER_REGISTRATION) auth = importlib.import_module(HELPER_AUTHORITY) cross = importlib.import_module(HELPER_CROSS_CHECK) return (reg, auth, cross), status def register_normal_callback( *, task_id: str, executor_key: str, chat_id: str, prompt: str, at: str, canonical_root: str = "/home/jay/workspace", anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), require_envelope: bool = True, dispatch_path: bool = True, direct_cron_path: bool = False, ) -> Dict[str, Any]: """ANU normal callback 등록 (helper integration 단일 진입점). Caller MUST pass executor_key == 본 봇의 self key. owner_key 는 항상 ANU_KEY 로 강제됨 (self-key 발사 금지). """ modules, status = _get_modules() if modules is None: return { "schema": SCHEMA, "verdict": "FAIL", "state": INTEGRATION_BYPASS, "integration_status": status.to_json(), "reasons": status.reasons, } reg = modules[0] if executor_key == ANU_KEY: return { "schema": SCHEMA, "verdict": "FAIL", "state": SELF_COLLECTOR_FORBIDDEN, "integration_status": status.to_json(), "reasons": [ "executor_key == ANU_KEY: collector self-key registration is " "forbidden by task md 원칙 6 (SELF_COLLECTOR_FORBIDDEN)." ], } result = reg.register_callback( kind="normal", task_id=task_id, executor_key=executor_key, owner_key=ANU_KEY, chat_id=str(chat_id), prompt=prompt, at=at, canonical_root=canonical_root, anu_keys=anu_keys, require_envelope=require_envelope, dispatch_path=dispatch_path, direct_cron_path=direct_cron_path, ) payload = result.to_json() payload["integration_status"] = status.to_json() return payload def verify_owner( *, registration_payload: Dict[str, Any], observed_owner_key: str, observed_chat_id: str, observed_role: str = "ANU", expected_chat_id: Optional[str] = None, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> Dict[str, Any]: """register_normal_callback 의 결과를 입력으로 받아 actual owner 검증. 회장 원칙 5: 등록 직후 actual schedule owner key 가 ANU_KEY 인지 검증. """ modules, status = _get_modules() if modules is None: return { "schema": SCHEMA, "verdict": "FAIL", "state": INTEGRATION_BYPASS, "integration_status": status.to_json(), "reasons": status.reasons, } reg = modules[0] rr = reg.RegistrationResult( schema=registration_payload.get("schema", "utils.callback_registration.v1"), verdict=registration_payload.get("verdict", "PASS"), state=registration_payload.get("state", reg.STATE_DISPATCH_SUBMITTED_UNVERIFIED), kind=registration_payload.get("kind", "normal"), task_id=registration_payload.get("task_id", ""), owner_key=registration_payload.get("owner_key", ANU_KEY), chat_id=str(registration_payload.get("chat_id", "")), prompt_utf8_bytes=int(registration_payload.get("prompt_utf8_bytes", 0)), prompt_byte_classification=registration_payload.get( "prompt_byte_classification", "OK_TARGET" ), argv=registration_payload.get("argv"), launch_decision=registration_payload.get("launch_decision"), authority_marker=registration_payload.get("authority_marker"), reasons=list(registration_payload.get("reasons", [])), registered_at_iso=registration_payload.get("registered_at_iso"), ) verified = reg.verify_actual_owner( registration_result=rr, observed_owner_key=observed_owner_key, observed_chat_id=observed_chat_id, observed_role=observed_role, anu_keys=anu_keys, expected_chat_id=expected_chat_id, ) payload = verified.to_json() payload["integration_status"] = status.to_json() return payload def validate_callback_authority( *, envelope_collector_key: str, actual_owner_key: str, executor_key: str, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> Dict[str, Any]: modules, status = _get_modules() if modules is None: return { "schema": SCHEMA, "verdict": "FAIL", "state": INTEGRATION_BYPASS, "integration_status": status.to_json(), "reasons": status.reasons, } auth = modules[1] verdict = auth.validate_authority( envelope_collector_key=envelope_collector_key, actual_owner_key=actual_owner_key, executor_key=executor_key, anu_keys=anu_keys, ) payload = verdict.to_json() payload["integration_status"] = status.to_json() return payload def cross_check_four_sources( *, cron_id: str, schedule_history_records: List[Dict[str, Any]], cron_history_records: Dict[str, List[Dict[str, Any]]], envelope: Optional[Dict[str, Any]], result_artifact: Optional[Dict[str, Any]], cron_list_present: Optional[bool] = None, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> Dict[str, Any]: modules, status = _get_modules() if modules is None: return { "schema": SCHEMA, "verdict": "FAIL", "state": INTEGRATION_BYPASS, "integration_status": status.to_json(), "reasons": status.reasons, } cross = modules[2] result = cross.cross_check_sources( cron_id=cron_id, schedule_history_records=schedule_history_records, cron_history_records=cron_history_records, envelope=envelope, result_artifact=result_artifact, cron_list_present=cron_list_present, anu_keys=anu_keys, ) payload = result.to_json() payload["integration_status"] = status.to_json() return payload def validate_spawn_callback_contract( *, envelope: Optional[Dict[str, Any]], executor_key: str, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), ) -> Dict[str, Any]: """SessionStart hook 직후 collector 가 호출하는 single contract check. Combines: helper availability + envelope presence + self-key forbidden gate + envelope.collector_key authority pre-check (envelope-only — actual owner verification still requires verify_owner after schedule lookup). """ status = integration_status() reasons: List[str] = list(status.reasons) if not status.available: return { "schema": SCHEMA, "verdict": "FAIL", "state": INTEGRATION_BYPASS, "integration_status": status.to_json(), "reasons": reasons, } if envelope is None: reasons.append( "envelope is None — collector 의 7 단계 중 1 단계 (envelope parse) " "불가; SAFE_DEGRADED_MODE 또는 HOLD_FOR_CHAIR 강제 (task md 원칙 11)." ) return { "schema": SCHEMA, "verdict": "FAIL", "state": "ENVELOPE_MISSING", "integration_status": status.to_json(), "reasons": reasons, } envelope_owner = envelope.get("owner_key") or envelope.get("collector_key") if envelope_owner == executor_key: reasons.append( f"envelope owner_key={envelope_owner!r} == executor_key — " "SELF_COLLECTOR_FORBIDDEN (task md 원칙 6)." ) return { "schema": SCHEMA, "verdict": "FAIL", "state": SELF_COLLECTOR_FORBIDDEN, "integration_status": status.to_json(), "reasons": reasons, } if envelope_owner not in set(anu_keys): reasons.append( f"envelope owner_key={envelope_owner!r} ∉ anu_keys={list(anu_keys)} " "— authority pre-check FAIL (verify_owner still required post-lookup)." ) return { "schema": SCHEMA, "verdict": "FAIL", "state": "ENVELOPE_OWNER_NOT_ANU", "integration_status": status.to_json(), "reasons": reasons, } return { "schema": SCHEMA, "verdict": "PASS", "state": "ENVELOPE_PRECHECK_PASS", "integration_status": status.to_json(), "envelope_owner_key": envelope_owner, "reasons": reasons + [ "envelope owner_key ∈ anu_keys AND != executor_key; " "proceed to schedule lookup + verify_owner." ], } def selftest() -> Dict[str, Any]: failures: List[str] = [] status = integration_status() if not status.available: failures.append( f"helper integration unavailable: missing={status.helpers_missing} " f"reasons={status.reasons}" ) return { "schema": SCHEMA, "ok": False, "failures": failures, "tests_run": 1, "integration_status": status.to_json(), } # 1: spawn contract — None envelope → ENVELOPE_MISSING r1 = validate_spawn_callback_contract(envelope=None, executor_key="abc") if r1["verdict"] != "FAIL" or r1["state"] != "ENVELOPE_MISSING": failures.append(f"spawn contract None envelope: got {r1}") # 2: spawn contract — self-key envelope → SELF_COLLECTOR_FORBIDDEN r2 = validate_spawn_callback_contract( envelope={"owner_key": "self-exec-key"}, executor_key="self-exec-key", ) if r2["state"] != SELF_COLLECTOR_FORBIDDEN: failures.append(f"self-key envelope: got {r2}") # 3: spawn contract — ANU envelope with non-self executor → PASS r3 = validate_spawn_callback_contract( envelope={"owner_key": ANU_KEY}, executor_key="other-key", ) if r3["verdict"] != "PASS": failures.append(f"ANU envelope passthrough: got {r3}") # 4: register_normal_callback — self-key forbidden r4 = register_normal_callback( task_id="t-self", executor_key=ANU_KEY, chat_id="0", prompt="x", at="10m", require_envelope=False, ) if r4["state"] != SELF_COLLECTOR_FORBIDDEN: failures.append(f"register self-key: got {r4}") # 5: validate_authority — ANU/ANU verified r5 = validate_callback_authority( envelope_collector_key=ANU_KEY, actual_owner_key=ANU_KEY, executor_key="other-key", ) if r5.get("verdict") != "PASS": failures.append(f"authority ANU/ANU: got {r5}") # 6: cross_check — all 4 absent + cron_list False → CALLBACK_MISSING r6 = cross_check_four_sources( cron_id="zzz", schedule_history_records=[], cron_history_records={}, envelope=None, result_artifact=None, cron_list_present=False, ) if r6.get("state") != "CALLBACK_MISSING": failures.append(f"cross_check missing: got {r6}") return { "schema": SCHEMA, "ok": len(failures) == 0, "failures": failures, "tests_run": 6, "integration_status": status.to_json(), } # ── task-2680 4-source validator wiring ───────────────────────────────────── # 회장 수정 목표 3 (actual owner key 검증 collector gate) + # 수정 목표 4 (self-key callback NON_AUTHORITATIVE_SELF_COLLECTOR 자동 분류) + # 수정 목표 5 (ANU independent reverify flow 강제) # # 본 wiring 은 utils.callback_authority_4source_validator 를 collector spawn # 시점에 호출하여 외부 cokacdir CLI + schedule_history 의 actual owner key 를 # 4-source 교차검증한다. envelope text 의 owner_key 가 ANU 라고 명기되어도 # actual cron owner 가 self-key 이면 즉시 NON_AUTHORITATIVE_SELF_COLLECTOR # 분류 + ANU independent reverify request 자동 emit. # task-2680 추가 classification (utils.callback_authority_4source_validator.py 와 1:1) NON_AUTHORITATIVE_SELF_COLLECTOR = "NON_AUTHORITATIVE_SELF_COLLECTOR" NON_AUTHORITATIVE_KEY_DRIFT = "NON_AUTHORITATIVE_KEY_DRIFT" UNDETERMINED_HISTORY_GAP = "UNDETERMINED_HISTORY_GAP" PROMPT_DRIFT = "PROMPT_DRIFT" ANU_AUTHORITATIVE = "ANU_AUTHORITATIVE" def classify_collector_authority_4source( *, schedule_id: str, executor_key: str, task_id: str, anu_key: str = ANU_KEY, chat_id: str = "6937032012", cokacdir_runner=None, ) -> Dict[str, Any]: """4-source 교차검증 게이트 (task-2680 수정 목표 3 + 4). collector 가 spawn 직후 호출. 외부 state (schedule_history 파일 + cokacdir --cron-history) 를 query 해 actual cron owner 가 ANU key 인지 검증한다. self-key 면 즉시 NON_AUTHORITATIVE_SELF_COLLECTOR 분류 → caller 가 collector 자가심사 차단 + ANU reverify dispatch. Returns dict with: classification, is_authoritative, requires_anu_reverify, reverify_request (or None), evidence, escalation_hint. """ try: from utils.callback_authority_4source_validator import ( classify_collector_authority, ) except ImportError as exc: return { "schema": SCHEMA, "verdict": "FAIL", "state": INTEGRATION_BYPASS, "classification": UNDETERMINED_HISTORY_GAP, "is_authoritative": False, "requires_anu_reverify": True, "reasons": [ f"utils.callback_authority_4source_validator import failed: {exc}" " — 4-source gate unavailable, treating as UNDETERMINED." ], "evidence": None, "reverify_request": None, } classification = classify_collector_authority( schedule_id=schedule_id, executor_key=executor_key, task_id=task_id, anu_key=anu_key, chat_id=chat_id, cokacdir_runner=cokacdir_runner, ) payload = classification.to_json() payload["schema"] = SCHEMA payload["verdict"] = "PASS" if classification.is_authoritative else "FAIL" return payload def emit_anu_independent_reverify_request( *, schedule_id: str, executor_key: str, task_id: str, chair_authorization_id: str, anu_key: str = ANU_KEY, chat_id: str = "6937032012", cokacdir_runner=None, ) -> Dict[str, Any]: """4-source 게이트 결과가 reverify trigger 면 reverify request 를 build. Returns dict with: classification, reverify_request (or None), action ("REVERIFY_REQUESTED" | "NO_REVERIFY_NEEDED" | "FAIL"). NOTE: 본 함수는 actual dispatch 0 — pure request builder. caller (ANU dispatch layer 또는 chair escalation path) 가 reverify_request 를 받아 실제 ANU dispatch 또는 extract_followup.send_anu_notify 발송한다. """ try: from utils.callback_authority_4source_validator import ( classify_collector_authority, build_anu_independent_reverify_request, ) except ImportError as exc: return { "schema": SCHEMA, "action": "FAIL", "reasons": [ f"utils.callback_authority_4source_validator import failed: {exc}" ], "reverify_request": None, "classification": None, } classification = classify_collector_authority( schedule_id=schedule_id, executor_key=executor_key, task_id=task_id, anu_key=anu_key, chat_id=chat_id, cokacdir_runner=cokacdir_runner, ) if not classification.requires_anu_reverify: return { "schema": SCHEMA, "action": "NO_REVERIFY_NEEDED", "classification": classification.to_json(), "reverify_request": None, } request = build_anu_independent_reverify_request( classification=classification, chair_authorization_id=chair_authorization_id, ) return { "schema": SCHEMA, "action": "REVERIFY_REQUESTED", "classification": classification.to_json(), "reverify_request": request.to_json() if request else None, } __all__ = [ "SCHEMA", "ANU_KEY", "DEFAULT_ANU_KEYS", "INTEGRATION_OK", "INTEGRATION_BYPASS", "SELF_COLLECTOR_FORBIDDEN", # task-2680 4-source classification enums "ANU_AUTHORITATIVE", "NON_AUTHORITATIVE_SELF_COLLECTOR", "NON_AUTHORITATIVE_KEY_DRIFT", "UNDETERMINED_HISTORY_GAP", "PROMPT_DRIFT", "IntegrationStatus", "integration_status", "require_helpers", "register_normal_callback", "verify_owner", "validate_callback_authority", "cross_check_four_sources", "validate_spawn_callback_contract", "classify_collector_authority_4source", "emit_anu_independent_reverify_request", "selftest", ] if __name__ == "__main__": import json as _json print(_json.dumps(selftest(), ensure_ascii=False, indent=2))