# -*- 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(), } __all__ = [ "SCHEMA", "ANU_KEY", "DEFAULT_ANU_KEYS", "INTEGRATION_OK", "INTEGRATION_BYPASS", "SELF_COLLECTOR_FORBIDDEN", "IntegrationStatus", "integration_status", "require_helpers", "register_normal_callback", "verify_owner", "validate_callback_authority", "cross_check_four_sources", "validate_spawn_callback_contract", "selftest", ] if __name__ == "__main__": import json as _json print(_json.dumps(selftest(), ensure_ascii=False, indent=2))