"""v3.6 Runtime Harness — Layer 0: Task MD SHA Contract. chair_authorization_id=CHAIR-AUTH-TASK-2705PLUS1-V36-TASK-MD-SHA-BOOTSTRAP-260528 Implements the 3-sha decision matrix from ``memory/specs/v36_task_md_sha_contract_design_draft_260528.md`` (sections 2, 4, 6). Public API ---------- - ``measure_sha_file(path)`` — read a file and return (sha256, size, content_bytes). - ``classify(...)`` — given three sha values and (optionally) the underlying bytes, return a dict with ``patch_type``, ``content_verbatim_match``, ``continue_allowed``, ``decision_class``, ``reason_code``, ``mismatch_location``. Safe-fail: every public function catches all exceptions and returns a HOLD result with ``reason_code=runtime_internal_error_hold`` — dispatch.py is never affected. """ from __future__ import annotations import os from typing import Any, Dict, Optional, Tuple from .task_md_sha_normalize import ( PATCH_TYPE_DISPATCH_META_SIDECAR, PATCH_TYPE_FORBIDDEN_SEMANTIC_CHANGE, PATCH_TYPE_NO_PATCH, PATCH_TYPE_RETRY_HEADER_PREPEND, PATCH_TYPE_UNKNOWN, PATCH_TYPE_WHITESPACE_NORMALIZATION, VERBATIM_FALSE, VERBATIM_TRUE, VERBATIM_UNVERIFIABLE, compute_sha256, content_verbatim_match, detect_patch_type, ) DECISION_ALLOW = "ALLOW" DECISION_HOLD = "HOLD_FOR_CHAIR" DECISION_DENY = "DENY" CONTINUE_TRUE = "true" CONTINUE_FALSE = "false" CONTINUE_HOLD = "hold" LOCATION_DISPATCH_ENTRY_TO_EXIT = "DISPATCH_ENTRY_TO_EXIT" LOCATION_DISPATCH_EXIT_TO_BOT_READ = "DISPATCH_EXIT_TO_BOT_READ" LOCATION_ANU_PRE_DISPATCH_TO_BOT_READ = "ANU_PRE_DISPATCH_TO_BOT_READ" LOCATION_BOT_READ_TO_BOT_WORK = "BOT_READ_TO_BOT_WORK" LOCATION_UNKNOWN = "UNKNOWN" _ALLOWED_PATCH_TYPES = frozenset({ PATCH_TYPE_NO_PATCH, PATCH_TYPE_DISPATCH_META_SIDECAR, PATCH_TYPE_RETRY_HEADER_PREPEND, PATCH_TYPE_WHITESPACE_NORMALIZATION, }) def measure_sha_file( path: Optional[str], ) -> Tuple[Optional[str], Optional[int], Optional[bytes]]: """Read ``path`` and return (sha256, byte_size, content_bytes).""" try: if not path or not os.path.exists(path): return None, None, None with open(path, "rb") as fh: content = fh.read() return compute_sha256(content), len(content), content except Exception: return None, None, None def _hold(reason: str, mismatch_location: str = LOCATION_UNKNOWN) -> Dict[str, Any]: return { "patch_type": PATCH_TYPE_UNKNOWN, "content_verbatim_match": VERBATIM_UNVERIFIABLE, "continue_allowed": CONTINUE_HOLD, "decision_class": DECISION_HOLD, "reason_code": reason, "mismatch_location": mismatch_location, } def classify( pre_sha: Optional[str], post_sha: Optional[str], observed_sha: Optional[str], pre_content: Optional[bytes] = None, post_content: Optional[bytes] = None, observed_content: Optional[bytes] = None, ) -> Dict[str, Any]: """Compute the layer-0 decision dict per Bootstrap 전략 4–9.""" try: if pre_sha is None and post_sha is None and observed_sha is None: return _hold("all_sha_missing_unverifiable_hold") if pre_sha and post_sha and observed_sha and pre_sha == post_sha == observed_sha: return { "patch_type": PATCH_TYPE_NO_PATCH, "content_verbatim_match": VERBATIM_TRUE, "continue_allowed": CONTINUE_TRUE, "decision_class": DECISION_ALLOW, "reason_code": "all_three_sha_identical_allow", "mismatch_location": LOCATION_UNKNOWN, } if observed_sha is None: return _hold( "executor_observed_sha_missing_hold", LOCATION_BOT_READ_TO_BOT_WORK, ) if post_sha is not None and observed_sha is not None and post_sha != observed_sha: patch_type_post_vs_observed = detect_patch_type(post_content, observed_content) verbatim_post_vs_observed = content_verbatim_match(post_content, observed_content) if ( patch_type_post_vs_observed in _ALLOWED_PATCH_TYPES and verbatim_post_vs_observed == VERBATIM_TRUE ): return { "patch_type": patch_type_post_vs_observed, "content_verbatim_match": VERBATIM_TRUE, "continue_allowed": CONTINUE_TRUE, "decision_class": DECISION_ALLOW, "reason_code": "post_observed_normalize_match_allow", "mismatch_location": LOCATION_DISPATCH_EXIT_TO_BOT_READ, } if patch_type_post_vs_observed == PATCH_TYPE_FORBIDDEN_SEMANTIC_CHANGE: return { "patch_type": PATCH_TYPE_FORBIDDEN_SEMANTIC_CHANGE, "content_verbatim_match": verbatim_post_vs_observed, "continue_allowed": CONTINUE_FALSE, "decision_class": DECISION_DENY, "reason_code": "post_observed_semantic_change_deny", "mismatch_location": LOCATION_DISPATCH_EXIT_TO_BOT_READ, } return _hold( "post_observed_unverifiable_hold", LOCATION_DISPATCH_EXIT_TO_BOT_READ, ) if pre_sha is not None and post_sha is not None and pre_sha != post_sha: patch_type_pre_vs_post = detect_patch_type(pre_content, post_content) verbatim_pre_vs_post = content_verbatim_match(pre_content, post_content) if ( patch_type_pre_vs_post in _ALLOWED_PATCH_TYPES and verbatim_pre_vs_post == VERBATIM_TRUE ): return { "patch_type": patch_type_pre_vs_post, "content_verbatim_match": VERBATIM_TRUE, "continue_allowed": CONTINUE_TRUE, "decision_class": DECISION_ALLOW, "reason_code": "pre_post_normalize_match_allow", "mismatch_location": LOCATION_DISPATCH_ENTRY_TO_EXIT, } if patch_type_pre_vs_post == PATCH_TYPE_FORBIDDEN_SEMANTIC_CHANGE: return { "patch_type": PATCH_TYPE_FORBIDDEN_SEMANTIC_CHANGE, "content_verbatim_match": verbatim_pre_vs_post, "continue_allowed": CONTINUE_FALSE, "decision_class": DECISION_DENY, "reason_code": "pre_post_semantic_change_deny", "mismatch_location": LOCATION_DISPATCH_ENTRY_TO_EXIT, } if pre_content is None or post_content is None: return _hold( "pre_post_content_missing_hold", LOCATION_DISPATCH_ENTRY_TO_EXIT, ) return _hold( "pre_post_unverifiable_hold", LOCATION_DISPATCH_ENTRY_TO_EXIT, ) if post_sha is not None and observed_sha is not None and post_sha == observed_sha: return { "patch_type": PATCH_TYPE_NO_PATCH, "content_verbatim_match": VERBATIM_TRUE, "continue_allowed": CONTINUE_TRUE, "decision_class": DECISION_ALLOW, "reason_code": "post_observed_match_allow", "mismatch_location": LOCATION_UNKNOWN, } return _hold("uncovered_state_hold") except Exception: return _hold("runtime_internal_error_hold")