diff --git a/loopx/capabilities/benchmark_toolkit/behavior_finding.py b/loopx/capabilities/benchmark_toolkit/behavior_finding.py index 3669a68a9..db3b96f07 100644 --- a/loopx/capabilities/benchmark_toolkit/behavior_finding.py +++ b/loopx/capabilities/benchmark_toolkit/behavior_finding.py @@ -10,8 +10,8 @@ from collections.abc import Iterable, Mapping from typing import Any +from ...control_plane.content_digest import BARE_SHA256_PATTERN from .study_projection import ( - _DIGEST_RE, _active_envelopes, _bounded_text, _finite_number, @@ -118,7 +118,7 @@ def normalize_benchmark_behavior_finding(payload: Mapping[str, Any]) -> dict[str if basis == "all_available" and sample != population: raise ValueError("all_available requires sample_count == population_count") digest = selection["cohort_digest"] - if not isinstance(digest, str) or not _DIGEST_RE.fullmatch(digest): + if not isinstance(digest, str) or not BARE_SHA256_PATTERN.fullmatch(digest): raise ValueError("cohort_digest must be SHA-256") measures = [] for item in _items(p["measures"], "measures", minimum=0): @@ -165,7 +165,10 @@ def normalize_benchmark_behavior_finding(payload: Mapping[str, Any]) -> dict[str e = _object( item, {"kind", "digest", "label", "relation", "summary"}, "evidence" ) - if not isinstance(e["digest"], str) or not _DIGEST_RE.fullmatch(e["digest"]): + item_digest = e["digest"] + if not isinstance(item_digest, str) or not BARE_SHA256_PATTERN.fullmatch( + item_digest + ): raise ValueError("evidence digest must be SHA-256") evidence.append( { diff --git a/loopx/capabilities/benchmark_toolkit/continuation.py b/loopx/capabilities/benchmark_toolkit/continuation.py index 4de013c66..d6296ecb6 100644 --- a/loopx/capabilities/benchmark_toolkit/continuation.py +++ b/loopx/capabilities/benchmark_toolkit/continuation.py @@ -2,10 +2,10 @@ from __future__ import annotations -import re from collections.abc import Mapping from enum import Enum from typing import Any +from ...control_plane.content_digest import BARE_SHA256_PATTERN BENCHMARK_PUBLIC_PROGRESS_SCHEMA_VERSION = "benchmark_public_progress_v0" BENCHMARK_CONTINUATION_DECISION_SCHEMA_VERSION = "benchmark_continuation_decision_v0" @@ -34,7 +34,7 @@ def _non_negative_int(value: Any, *, field: str) -> int: def _sha256_digest(value: Any, *, field: str) -> str: text = str(value or "").strip().lower() - if not re.fullmatch(r"[0-9a-f]{64}", text): + if not BARE_SHA256_PATTERN.fullmatch(text): raise ValueError(f"{field} must be a lowercase SHA-256 digest") return text diff --git a/loopx/capabilities/benchmark_toolkit/factorial_contrast.py b/loopx/capabilities/benchmark_toolkit/factorial_contrast.py index f4e94458b..96399d1af 100644 --- a/loopx/capabilities/benchmark_toolkit/factorial_contrast.py +++ b/loopx/capabilities/benchmark_toolkit/factorial_contrast.py @@ -10,6 +10,7 @@ BENCHMARK_FOUR_ARM_CONTRACT_SCHEMA_VERSION, BENCHMARK_FOUR_ARM_QUALIFICATION_SCOPE, ) +from ...control_plane.content_digest import BARE_SHA256_PATTERN BENCHMARK_FACTORIAL_CONTRAST_SCHEMA_VERSION = "benchmark_factorial_contrast_v0" @@ -124,7 +125,7 @@ def _normalize_four_arm_design(contract: Mapping[str, Any]) -> dict[str, Any]: if arm_role != expected_role: raise ValueError("four-arm contract role does not match its factor cell") task_goal_sha256 = str(raw_arm.get("task_goal_sha256") or "").strip() - if not re.fullmatch(r"[0-9a-f]{64}", task_goal_sha256): + if not BARE_SHA256_PATTERN.fullmatch(task_goal_sha256): raise ValueError("four-arm task-goal hash must be sha256") arm = { "arm_id": arm_id, diff --git a/loopx/capabilities/benchmark_toolkit/runtime_continuity.py b/loopx/capabilities/benchmark_toolkit/runtime_continuity.py index 177e91efc..4a30b8fad 100644 --- a/loopx/capabilities/benchmark_toolkit/runtime_continuity.py +++ b/loopx/capabilities/benchmark_toolkit/runtime_continuity.py @@ -2,12 +2,12 @@ from __future__ import annotations -import re from enum import Enum from typing import Any +from ...control_plane.content_digest import BARE_SHA256_PATTERN BENCHMARK_RUNTIME_CONTINUITY_SCHEMA_VERSION = "benchmark_runtime_continuity_v0" -_SHA256_DIGEST = re.compile(r"[0-9a-f]{64}\Z") +_SHA256_DIGEST = BARE_SHA256_PATTERN class BenchmarkEventWindowState(str, Enum): diff --git a/loopx/capabilities/benchmark_toolkit/study_projection.py b/loopx/capabilities/benchmark_toolkit/study_projection.py index 9cad9739a..94ae4b1ef 100644 --- a/loopx/capabilities/benchmark_toolkit/study_projection.py +++ b/loopx/capabilities/benchmark_toolkit/study_projection.py @@ -15,6 +15,7 @@ from typing import Any from ...file_lock import exclusive_file_lock +from ...control_plane.content_digest import BARE_SHA256_PATTERN from .experiment_board import ( BENCHMARK_EXPERIMENT_BOARD_ROW_SCHEMA_VERSION, benchmark_experiment_board_row_key, @@ -41,7 +42,6 @@ BENCHMARK_STUDY_DASHBOARD_SCHEMA_VERSION = "benchmark_study_dashboard_v0" _TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:@+-]{0,127}$") -_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") _ARM_ROLES = {"baseline", "control", "treatment", "explore"} _METRIC_ROLES = {"primary", "guardrail", "supporting"} _RECORD_KINDS = { @@ -579,7 +579,7 @@ def normalize_benchmark_upload_envelope( if payload.get("record_id") != rebuilt["record_id"]: raise ValueError("benchmark upload record_id does not match envelope identity") digest = str(payload.get("payload_digest") or "") - if not _DIGEST_RE.fullmatch(digest) or digest != rebuilt["payload_digest"]: + if not BARE_SHA256_PATTERN.fullmatch(digest) or digest != rebuilt["payload_digest"]: raise ValueError("benchmark upload payload digest mismatch") return rebuilt diff --git a/loopx/capabilities/content_ops/item_lifecycle.py b/loopx/capabilities/content_ops/item_lifecycle.py index 262b61937..3653e83b8 100644 --- a/loopx/capabilities/content_ops/item_lifecycle.py +++ b/loopx/capabilities/content_ops/item_lifecycle.py @@ -18,6 +18,7 @@ CONTENT_OPS_QUEUE_PROJECTION_SCHEMA_VERSION, CONTENT_OPS_QUEUE_STATUS_PACKET_SCHEMA_VERSION, ) +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN ALLOWED_ITEM_KINDS = {"article", "post", "profile_update", "reply", "repost"} ALLOWED_ITEM_STATES = { @@ -35,7 +36,7 @@ TERMINAL_STATES = {"readback_verified", "skipped", "superseded"} _TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") -_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_DIGEST_RE = ENVELOPED_SHA256_PATTERN _ITEM_KEYS = { "schema_version", "item_id", diff --git a/loopx/capabilities/issue_fix/outcome_projection.py b/loopx/capabilities/issue_fix/outcome_projection.py index aa76d5b91..d53cf328c 100644 --- a/loopx/capabilities/issue_fix/outcome_projection.py +++ b/loopx/capabilities/issue_fix/outcome_projection.py @@ -6,6 +6,7 @@ from pathlib import Path, PurePosixPath from typing import Any +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN from ...control_plane.runtime.public_safety import public_safe_compact_text from .metadata_preview import normalise_github_issue_link_reference from .pr_lifecycle import BRANCH_REPLAN_MERGE_STATES @@ -31,7 +32,7 @@ DELIVERY_VALIDATION_STATUSES = {"passed", "failed", "partial", "not_run"} DELIVERY_OUTCOME_STATUSES = {"in_progress", "completed", "blocked"} -_REPOSITORY_FINGERPRINT_PATTERN = re.compile(r"sha256:[0-9a-f]{64}") +_REPOSITORY_FINGERPRINT_PATTERN = ENVELOPED_SHA256_PATTERN _COMMIT_OID_PATTERN = re.compile(r"[0-9a-fA-F]{40,64}") _RECOVERY_REF_PATTERN = re.compile( r"refs/(?:heads|remotes|tags)/[A-Za-z0-9][A-Za-z0-9._/-]{0,180}" diff --git a/loopx/capabilities/issue_fix/reviewer_notification.py b/loopx/capabilities/issue_fix/reviewer_notification.py index 076615366..f3d620b71 100644 --- a/loopx/capabilities/issue_fix/reviewer_notification.py +++ b/loopx/capabilities/issue_fix/reviewer_notification.py @@ -15,6 +15,7 @@ reviewer_artifact_notification_gate, reviewer_notification_before_send_gate, ) +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN ISSUE_FIX_REVIEWER_NOTIFICATION_SINKS_INPUT_SCHEMA_VERSION = ( @@ -122,7 +123,7 @@ def reviewer_notification_receipts_from_state( dict.fromkeys( str(value) for value in (values if isinstance(values, list) else []) - if re.fullmatch(r"sha256:[a-f0-9]{64}", str(value)) + if ENVELOPED_SHA256_PATTERN.fullmatch(str(value)) ) ) @@ -140,7 +141,7 @@ def reviewer_notification_queue_from_state( if ( value.get("schema_version") != ISSUE_FIX_REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION - or not re.fullmatch(r"sha256:[a-f0-9]{64}", key) + or not ENVELOPED_SHA256_PATTERN.fullmatch(key) or key in seen ): continue @@ -164,7 +165,7 @@ def reviewer_notification_legacy_queue_from_state( if ( value.get("schema_version") != ISSUE_FIX_REVIEWER_NOTIFICATION_LEGACY_QUEUE_RECEIPT_SCHEMA_VERSION - or not re.fullmatch(r"sha256:[a-f0-9]{64}", key) + or not ENVELOPED_SHA256_PATTERN.fullmatch(key) or key in seen ): continue @@ -183,7 +184,7 @@ def with_reviewer_notification_state( ) for value in sinks_input.get("receipts") or []: text = str(value) - if re.fullmatch(r"sha256:[a-f0-9]{64}", text) and text not in merged_receipts: + if ENVELOPED_SHA256_PATTERN.fullmatch(text) and text not in merged_receipts: merged_receipts.append(text) queue = reviewer_notification_queue_from_state( @@ -517,7 +518,7 @@ def validate_issue_fix_reviewer_notification_sinks_result( errors.append(f"sink result {field} must be false") receipts = packet.get("receipts") if not isinstance(receipts, list) or any( - not re.fullmatch(r"sha256:[a-f0-9]{64}", str(value)) + not ENVELOPED_SHA256_PATTERN.fullmatch(str(value)) for value in (receipts if isinstance(receipts, list) else []) ): errors.append("receipts must contain only stable sha256 keys") @@ -535,7 +536,7 @@ def validate_issue_fix_reviewer_notification_sinks_result( if ( receipt.get("schema_version") != ISSUE_FIX_REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION - or not re.fullmatch(r"sha256:[a-f0-9]{64}", key) + or not ENVELOPED_SHA256_PATTERN.fullmatch(key) or key in queued_keys or receipt.get("status") != "queued" or not public_safe_compact_text(receipt.get("sink_kind"), limit=50) @@ -693,7 +694,7 @@ def build_issue_fix_reviewer_notification_sinks_result( receipts = { str(value) for value in (raw_receipts if isinstance(raw_receipts, list) else []) - if re.fullmatch(r"sha256:[a-f0-9]{64}", str(value)) + if ENVELOPED_SHA256_PATTERN.fullmatch(str(value)) } semantic_history_pr_refs = { public_safe_compact_text(value, limit=300) diff --git a/loopx/capabilities/machine_configuration/store.py b/loopx/capabilities/machine_configuration/store.py index c804a91ce..c9bee293e 100644 --- a/loopx/capabilities/machine_configuration/store.py +++ b/loopx/capabilities/machine_configuration/store.py @@ -21,6 +21,7 @@ normalize_machine_configuration, project_machine_configuration, ) +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN MACHINE_CONFIGURATION_UPDATE_PLAN_SCHEMA = "machine_configuration_update_plan_v0" @@ -346,8 +347,8 @@ def _read_transaction(runtime_root: Path, transaction_id: str) -> dict[str, Any] raise ValueError("machine-configuration transaction receipt is invalid") receipt["receipt_revision"] = receipt_revision applied_revision = str(receipt.get("applied_revision") or "") - if applied_revision != _MISSING_REVISION and not re.fullmatch( - r"sha256:[0-9a-f]{64}", applied_revision + if applied_revision != _MISSING_REVISION and not ( + ENVELOPED_SHA256_PATTERN.fullmatch(applied_revision) ): raise ValueError("machine-configuration transaction revision is invalid") return receipt diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 7a098fd28..97caa4a52 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -36,6 +36,7 @@ _request_lock, needs_conclusion as needs_conclusion, ) +from ...control_plane.content_digest import BARE_SHA256_PATTERN PHASES = ("decision", "conclusion") DELIVERY_STATUSES = { @@ -310,7 +311,7 @@ def project_chat_return_deliveries(root, session_id, messages): if route.get("session_id") != session_id: continue request_id = str(route.get("request_id") or "") - if path.stem != request_id or not re.fullmatch(r"[a-f0-9]{64}", request_id): + if path.stem != request_id or not BARE_SHA256_PATTERN.fullmatch(request_id): continue route_message_ids = { "handoff." + _hash([request_id, phase]) for phase in PHASES diff --git a/loopx/capabilities/manager_context/tracking.py b/loopx/capabilities/manager_context/tracking.py index 002a82e83..d80af1b73 100644 --- a/loopx/capabilities/manager_context/tracking.py +++ b/loopx/capabilities/manager_context/tracking.py @@ -21,6 +21,10 @@ ) from ...todos import list_goal_todos from ...chat_manager_details import _text +from ...control_plane.content_digest import ( + BARE_SHA256_PATTERN, + ENVELOPED_SHA256_PATTERN, +) def _core_todos(registry_path, root, goal_id): @@ -50,7 +54,7 @@ def link( raise ValueError("too many context links") if any(not re.fullmatch(r"todo_[a-f0-9]{12}", x) for x in todo_ids): raise ValueError("invalid Core Todo id") - if any(not re.fullmatch(r"sha256:[a-f0-9]{64}", x) for x in evidence_ids): + if any(not ENVELOPED_SHA256_PATTERN.fullmatch(x) for x in evidence_ids): raise ValueError("evidence references must be opaque SHA256 identifiers") if todo_ids: rows = _core_todos(registry_path, root, goal_id) @@ -118,7 +122,7 @@ def query( limit=8, ): """External callers see only requests from their exact audience, never raw text.""" - if request_id is not None and not re.fullmatch(r"[a-f0-9]{64}", request_id): + if request_id is not None and not BARE_SHA256_PATTERN.fullmatch(request_id): raise ValueError("invalid context request id") if not owner_scope and not channel_id: raise ValueError("handoff audience required") diff --git a/loopx/capabilities/periodic_report/adapters.py b/loopx/capabilities/periodic_report/adapters.py index ff87b28b0..2888f67ce 100644 --- a/loopx/capabilities/periodic_report/adapters.py +++ b/loopx/capabilities/periodic_report/adapters.py @@ -15,6 +15,7 @@ _SINK_STATUSES, _SOURCE_STATUSES, ) +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN SOURCE_RESULT_SCHEMA = "periodic_report_source_result_v0" @@ -756,7 +757,7 @@ def _normalize_artifact_result( document_digest = _text( artifact.get("document_digest"), "artifact.document_digest", maximum=80 ) - if not re.fullmatch(r"sha256:[0-9a-f]{64}", document_digest): + if not ENVELOPED_SHA256_PATTERN.fullmatch(document_digest): raise ValueError("artifact.document_digest must use sha256") if expected_document is not None: expected_document_digest = ( diff --git a/loopx/capabilities/periodic_report/archive.py b/loopx/capabilities/periodic_report/archive.py index aeb9d2940..0ebf2d188 100644 --- a/loopx/capabilities/periodic_report/archive.py +++ b/loopx/capabilities/periodic_report/archive.py @@ -14,6 +14,7 @@ from typing import Any from urllib.parse import unquote, urlsplit +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN from .adapters import ARTIFACT_SCHEMA, DOCUMENT_SCHEMA from .core import _normalize_trigger_receipt, _reject_raw_keys @@ -24,7 +25,6 @@ MEMORY_REFERENCE_SCHEMA = "periodic_report_memory_reference_v0" _TOKEN_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$") -_SHA256_RE = re.compile(r"^sha256:[0-9a-f]{64}$") ArchiveReadback = Callable[[str], Mapping[str, Any]] @@ -68,7 +68,7 @@ def _token(value: object, label: str) -> str: def _sha256(value: object, label: str) -> str: digest = _text(value, label, maximum=80) - if not _SHA256_RE.fullmatch(digest): + if not ENVELOPED_SHA256_PATTERN.fullmatch(digest): raise ValueError(f"{label} must use sha256") return digest diff --git a/loopx/capabilities/periodic_report/bindings.py b/loopx/capabilities/periodic_report/bindings.py index 82bbee9f9..b9e34e09e 100644 --- a/loopx/capabilities/periodic_report/bindings.py +++ b/loopx/capabilities/periodic_report/bindings.py @@ -16,6 +16,7 @@ _SINK_ROLES, _SINK_STATUSES, ) +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN GENERATION_BUNDLE_SCHEMA = "periodic_report_generation_bundle_v0" GENERATION_RECEIPT_SCHEMA = "periodic_report_generation_receipt_v0" @@ -192,7 +193,7 @@ def _generation_receipt(raw: object) -> dict[str, Any]: document_digest = _text( receipt.get("document_digest"), "document_digest", maximum=80 ) - if not re.fullmatch(r"sha256:[0-9a-f]{64}", document_digest): + if not ENVELOPED_SHA256_PATTERN.fullmatch(document_digest): raise ValueError("generation_receipt.document_digest must use sha256") artifacts = _sequence(receipt.get("artifact_receipts"), "artifact_receipts") if not artifacts: @@ -209,7 +210,7 @@ def _generation_receipt(raw: object) -> dict[str, Any]: content_digest = _text( artifact.get("content_digest"), f"{label}.content_digest", maximum=80 ) - if not re.fullmatch(r"sha256:[0-9a-f]{64}", content_digest): + if not ENVELOPED_SHA256_PATTERN.fullmatch(content_digest): raise ValueError(f"{label}.content_digest must use sha256") normalized_artifacts.append( { diff --git a/loopx/capabilities/periodic_report/cadence_journal.py b/loopx/capabilities/periodic_report/cadence_journal.py index 629b8648a..26bb1bf36 100644 --- a/loopx/capabilities/periodic_report/cadence_journal.py +++ b/loopx/capabilities/periodic_report/cadence_journal.py @@ -16,6 +16,7 @@ from ...file_lock import LockAcquisitionPolicy, exclusive_file_lock from ...registry import atomic_write_json from .cadence import report_cadence_window +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN CADENCE_WINDOW_SCHEMA = "periodic_report_cadence_window_v0" JOURNAL_SCHEMA = "periodic_report_cadence_journal_v0" @@ -47,7 +48,7 @@ def validate_cadence_window(raw: object) -> dict[str, Any]: for key in ("goal_id", "agent_id") ): raise ValueError("cadence window identity is invalid") - if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(value["subscription_revision"])): + if not ENVELOPED_SHA256_PATTERN.fullmatch(str(value["subscription_revision"])): raise ValueError("cadence subscription revision is invalid") if not isinstance(value["profile_ref"], Mapping) or not isinstance(value["trigger_policy"], Mapping): raise ValueError("cadence profile facts are invalid") diff --git a/loopx/capabilities/periodic_report/incremental.py b/loopx/capabilities/periodic_report/incremental.py index c17a1c659..0fec0820f 100644 --- a/loopx/capabilities/periodic_report/incremental.py +++ b/loopx/capabilities/periodic_report/incremental.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN from ...file_lock import LockAcquisitionPolicy, exclusive_file_lock from ...registry import atomic_write_json, read_json @@ -17,7 +18,6 @@ INCREMENTAL_BASELINE_SCHEMA = "periodic_report_incremental_baseline_v0" _IDENTITY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$") -_SHA256_RE = re.compile(r"^sha256:[0-9a-f]{64}$") def _canonical_digest(value: object) -> str: @@ -60,7 +60,7 @@ def _timestamp(value: object, label: str) -> str: def _digest(value: object, label: str) -> str: digest = _required_text(value, label, maximum=80) - if not _SHA256_RE.fullmatch(digest): + if not ENVELOPED_SHA256_PATTERN.fullmatch(digest): raise ValueError(f"{label} must use sha256") return digest diff --git a/loopx/capabilities/periodic_report/machine_defaults.py b/loopx/capabilities/periodic_report/machine_defaults.py index 28bd3a90e..31f801222 100644 --- a/loopx/capabilities/periodic_report/machine_defaults.py +++ b/loopx/capabilities/periodic_report/machine_defaults.py @@ -2,7 +2,6 @@ import hashlib import json -import re from collections.abc import Mapping from datetime import datetime, timezone from typing import Any @@ -10,6 +9,7 @@ from .cadence import normalize_report_cadence +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN from ...control_plane.todos.contract import normalize_todo_claimed_by from ..configuration_ui import resolve_capability_configuration from ..machine_configuration.contract import ( @@ -31,7 +31,6 @@ SUBSCRIPTION_ERROR_SCHEMA = "periodic_report_subscription_error_v0" _INHERITANCE_MODE = "live_machine_default" -_REVISION_RE = re.compile(r"^sha256:[0-9a-f]{64}$") class PeriodicReportSubscriptionConfigurationError(ValueError): @@ -400,7 +399,7 @@ def normalize_periodic_report_delivery_authority(raw: object) -> dict[str, Any]: authority.get("effective_revision"), "delivery_authority.effective_revision", ) - if not _REVISION_RE.fullmatch(effective_revision): + if not ENVELOPED_SHA256_PATTERN.fullmatch(effective_revision): raise ValueError("delivery_authority.effective_revision is invalid") return { "schema_version": DELIVERY_AUTHORITY_SCHEMA, diff --git a/loopx/capabilities/progress_review/receipt.py b/loopx/capabilities/progress_review/receipt.py index d538a4e2b..336abe35e 100644 --- a/loopx/capabilities/progress_review/receipt.py +++ b/loopx/capabilities/progress_review/receipt.py @@ -16,6 +16,7 @@ import re import tempfile from typing import Any +from ...control_plane.content_digest import BARE_SHA256_PATTERN PROGRESS_REVIEW_RECEIPT_SCHEMA_VERSION = "progress_review_receipt_v0" PROGRESS_REVIEW_RECEIPT_STATUSES: tuple[str, ...] = ( @@ -39,7 +40,6 @@ PROGRESS_REVIEW_PENDING_REASON = "pending_evaluation" MAX_RECEIPT_BYTES = 65536 MAX_LOADED_RECEIPTS = 256 -_HEX64 = re.compile(r"^[a-f0-9]{64}$") _TEXT_LIMIT = 200 @@ -71,7 +71,7 @@ def _text(value: Any, *, field: str, required: bool = True) -> str | None: def _hex64(value: Any, *, field: str) -> str: text = _text(value, field=field) - if text is None or not _HEX64.fullmatch(text): + if text is None or not BARE_SHA256_PATTERN.fullmatch(text): raise ValueError(f"receipt.{field} must be a sha256 hex digest") return text diff --git a/loopx/chat_action_normalization.py b/loopx/chat_action_normalization.py index f137d0c73..94f42e828 100644 --- a/loopx/chat_action_normalization.py +++ b/loopx/chat_action_normalization.py @@ -7,12 +7,12 @@ from typing import Any, Mapping from .agent_registry import registered_agent_ids_for_goal +from .control_plane.content_digest import BARE_SHA256_PATTERN from .control_plane.runtime.time import now_utc, parse_timestamp, utc_isoformat from .control_plane.todos.contract import require_supported_todo_resume_when from .registry import registry_goals -_SHA256 = re.compile(r"^[0-9a-f]{64}$") _AUTHORITY_PRINCIPAL = re.compile(r"^[a-z][a-z0-9._-]{0,30}:[A-Za-z0-9._:-]{1,200}$") @@ -65,7 +65,7 @@ def _normalize( if not isinstance(payload, Mapping): raise ValueError("operation payload must be an object") payload_digest = str(values.get("payload_digest") or "").strip() - if not _SHA256.fullmatch(payload_digest): + if not BARE_SHA256_PATTERN.fullmatch(payload_digest): raise ValueError("operation payload_digest must be lowercase SHA-256") if _digest(payload) != payload_digest: raise ValueError("operation payload_digest does not match payload") diff --git a/loopx/configuration_transaction.py b/loopx/configuration_transaction.py index 842408e37..254d9bdab 100644 --- a/loopx/configuration_transaction.py +++ b/loopx/configuration_transaction.py @@ -2,14 +2,13 @@ import hashlib import json -import re from collections.abc import Mapping from copy import deepcopy from typing import Any +from .control_plane.content_digest import ENVELOPED_SHA256_PATTERN CONFIGURATION_REVISION_MISSING = "absent" -_REVISION_RE = re.compile(r"^sha256:[0-9a-f]{64}$") def configuration_payload_revision(value: object) -> str: @@ -51,8 +50,9 @@ def goal_capability_configuration_revision( def _validated_revision(value: str, *, label: str) -> str: revision = str(value or "").strip() - if revision != CONFIGURATION_REVISION_MISSING and not _REVISION_RE.fullmatch( - revision + if ( + revision != CONFIGURATION_REVISION_MISSING + and not ENVELOPED_SHA256_PATTERN.fullmatch(revision) ): raise ValueError(f"{label} must be absent or a sha256 revision") return revision diff --git a/loopx/control_plane/collaboration/delegation_inventory.py b/loopx/control_plane/collaboration/delegation_inventory.py index 193f13cf2..66bd66a9f 100644 --- a/loopx/control_plane/collaboration/delegation_inventory.py +++ b/loopx/control_plane/collaboration/delegation_inventory.py @@ -2,12 +2,12 @@ from __future__ import annotations import heapq -import re from typing import TYPE_CHECKING from ..effect_runtime import EffectRuntimeRemoteError, effect_runtime_result from .inbox import _read from .peers import _goal, require_operation_id +from ..content_digest import BARE_SHA256_PATTERN if TYPE_CHECKING: from ...collaboration_mcp import Delegations @@ -26,7 +26,7 @@ def addresses(): for path in entries: if path.suffix != ".json": continue - if not re.fullmatch(r"[a-f0-9]{64}", path.stem): + if not BARE_SHA256_PATTERN.fullmatch(path.stem): raise ValueError("unexpected delegation record address; reconcile inventory storage") if query["cursor"] is None or path.stem > query["cursor"]: yield path.stem diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py index 797c9658c..97b037710 100644 --- a/loopx/control_plane/collaboration/inbox.py +++ b/loopx/control_plane/collaboration/inbox.py @@ -16,6 +16,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any from ...file_lock import exclusive_file_lock +from ..content_digest import BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN if TYPE_CHECKING: from .goal_instance_scope import CollaborationGoalScope @@ -162,7 +163,7 @@ def pending( for path in paths: if path.suffix != ".json": continue - if not re.fullmatch(r"[a-f0-9]{64}", path.stem): + if not BARE_SHA256_PATTERN.fullmatch(path.stem): raise ValueError("invalid context request filename") if path.stem <= after: continue @@ -248,7 +249,7 @@ def acknowledge( reason, scope=goal_scope, ) - if not re.fullmatch(r"[a-f0-9]{64}", request_id): + if not BARE_SHA256_PATTERN.fullmatch(request_id): raise ValueError("invalid context request id") target = _record_identity(goal_id, agent_id, scope) entry = _entry( @@ -288,7 +289,7 @@ def _now(): def _entry(root, goal_id, agent_id, request_id, *, scope=None): - if not isinstance(request_id, str) or not re.fullmatch(r"[a-f0-9]{64}", request_id): + if not isinstance(request_id, str) or not BARE_SHA256_PATTERN.fullmatch(request_id): raise ValueError("invalid context request id") target = _target(goal_id, agent_id, scope) identity = _record_identity(goal_id, agent_id, scope) @@ -372,15 +373,15 @@ def _receipt(root, lane, row): raise ValueError("invalid read receipt") if lane == "links": for key, pattern in [ - ("todo_ids", r"todo_[a-f0-9]{12}"), - ("evidence_ids", r"sha256:[a-f0-9]{64}"), + ("todo_ids", re.compile(r"todo_[a-f0-9]{12}")), + ("evidence_ids", ENVELOPED_SHA256_PATTERN), ]: refs = value.get(key) if ( not isinstance(refs, list) or len(refs) > 16 or any( - not isinstance(ref, str) or not re.fullmatch(pattern, ref) + not isinstance(ref, str) or not pattern.fullmatch(ref) for ref in refs ) ): diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py index 6ee921e59..4516c3cb8 100644 --- a/loopx/control_plane/collaboration/peers.py +++ b/loopx/control_plane/collaboration/peers.py @@ -33,6 +33,7 @@ from ...agent_registry import registered_agent_ids_for_goal from ...thread_agent_binding import resolve_thread_agent_binding from ..projects.registry_codec import load_project_registry +from ..content_digest import BARE_SHA256_PATTERN PEER_INSTRUCTION = ( "This is a peer's request for help or independent review, not an owner instruction. " @@ -477,7 +478,7 @@ def consume_return( def _request_id(value): - if not isinstance(value, str) or not re.fullmatch(r"[a-f0-9]{64}", value): + if not isinstance(value, str) or not BARE_SHA256_PATTERN.fullmatch(value): raise ValueError("invalid context request id") return value diff --git a/loopx/control_plane/content_digest.py b/loopx/control_plane/content_digest.py new file mode 100644 index 000000000..4eb81f56c --- /dev/null +++ b/loopx/control_plane/content_digest.py @@ -0,0 +1,20 @@ +"""One owner for the two shapes a stored SHA-256 digest can take. + +A digest reaches a record in one of two envelopes: the bare 64 lowercase hex +characters, or that same hex behind the ``sha256:`` prefix that the +periodic-report and content-ops writers concatenate. Before this module the +decision was compiled independently in eighteen modules under three spellings. + +Patterns that merely contain a hex digest inside a larger grammar (a +``cadence_…`` identifier, a journal filename, a ``40|64`` Git object id, a +compound cursor) answer a different question and stay with the surface that +owns that grammar. Producers that build the envelope by hand are the other half +of this decision and are deliberately unchanged here. +""" + +from __future__ import annotations + +import re + +ENVELOPED_SHA256_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") +BARE_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") diff --git a/loopx/control_plane/coordination/local_authority_shadow_outbox.py b/loopx/control_plane/coordination/local_authority_shadow_outbox.py index 40e5eebd9..6b6c8b244 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_outbox.py +++ b/loopx/control_plane/coordination/local_authority_shadow_outbox.py @@ -42,6 +42,7 @@ LOCAL_AUTHORITY_SHADOW_READ_REQUEST_SCHEMA, LOCAL_AUTHORITY_SHADOW_READ_RESULT_SCHEMA, ) +from ..content_digest import ENVELOPED_SHA256_PATTERN from .shadow_management import ( read_shadow_capture_binding, shadow_maintenance_lock_target, @@ -260,7 +261,6 @@ def _index_entry_files( _WRITER_RUNTIMES = frozenset({WRITER_RUNTIME_PYTHON, WRITER_RUNTIME_TYPESCRIPT}) _SOURCE_KINDS = frozenset({SOURCE_MARKDOWN, SOURCE_STATE_EVENT_LOG, SOURCE_TASK_LEASE}) -_DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") def _load_prepared_record( @@ -299,7 +299,7 @@ def _load_prepared_record( and bool(writer.get("write_class")) and source.get("kind") in _SOURCE_KINDS and isinstance(root_digest, str) - and _DIGEST_PATTERN.match(root_digest) is not None + and ENVELOPED_SHA256_PATTERN.match(root_digest) is not None and isinstance(lineage_id, str) and bool(lineage_id) and source_ref is not None @@ -508,7 +508,8 @@ def invalid() -> OutboxError: or ( digest is not None and ( - not isinstance(digest, str) or _DIGEST_PATTERN.fullmatch(digest) is None + not isinstance(digest, str) + or ENVELOPED_SHA256_PATTERN.fullmatch(digest) is None ) ) or any( diff --git a/loopx/control_plane/coordination/shadow_management.py b/loopx/control_plane/coordination/shadow_management.py index 445b66d6e..e803f42d7 100644 --- a/loopx/control_plane/coordination/shadow_management.py +++ b/loopx/control_plane/coordination/shadow_management.py @@ -17,9 +17,10 @@ SHADOW_MANAGEMENT_MANIFEST_SCHEMA, SHADOW_MANAGEMENT_STATE_SCHEMA, ) from .local_authority_shadow_projection import sha256_digest +from ..content_digest import ENVELOPED_SHA256_PATTERN SHADOW_CAPTURE_PROFILE = "file_outbox_v1" -_DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z") +_DIGEST = ENVELOPED_SHA256_PATTERN _STATE_KEYS = { "schema_version", "goal_id", "source_root_digest", "status", "binding", "operation", "previous_operation_id", "result", diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index ed9ec1831..2c256983c 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -21,6 +21,7 @@ from typing import IO, Any from ..file_lock import process_is_alive +from .content_digest import BARE_SHA256_PATTERN EFFECT_RUNTIME_REQUEST_SCHEMA_VERSION = "loopx_effect_runtime_request_v0" EFFECT_RUNTIME_RESPONSE_SCHEMA_VERSION = "loopx_effect_runtime_response_v1" @@ -662,7 +663,7 @@ def _read_local_snapshot_response( size, digest = ref.get("byte_count"), ref.get("sha256") if (not isinstance(size, int) or isinstance(size, bool) or size <= 0 or size > MAX_LOCAL_SNAPSHOT_BYTES or not isinstance(digest, str) - or re.fullmatch(r"[a-f0-9]{64}", digest) is None): + or BARE_SHA256_PATTERN.fullmatch(digest) is None): raise ValueError("invalid local snapshot reference") descriptor = os.open(sink, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(descriptor, "rb") as file: diff --git a/loopx/control_plane/goals/activation_service.py b/loopx/control_plane/goals/activation_service.py index 99eab4e74..edaec5db1 100644 --- a/loopx/control_plane/goals/activation_service.py +++ b/loopx/control_plane/goals/activation_service.py @@ -5,7 +5,6 @@ from enum import Enum import hashlib from pathlib import Path -import re from typing import Any from ..projects.registry_codec import project_registry_transaction @@ -23,6 +22,7 @@ goal_activation_state, normalize_goal_activation_state, ) +from ..content_digest import BARE_SHA256_PATTERN from .configure_goal_service import resolve_configure_goal_sync_target @@ -34,7 +34,6 @@ GOAL_ACTIVATION_AUTHORITY_ROUTE_SCHEMA_VERSION = ( "loopx_goal_activation_authority_route_v1" ) -_SHA256 = re.compile(r"^[a-f0-9]{64}$") class GoalActivationAuthorityRouteMode(str, Enum): @@ -285,7 +284,7 @@ def set_goal_activation_state( source_bytes = source_registry.read_bytes() source_goal = _goal(load_registry(source_registry), normalized_goal_id) normalized_fingerprint = str(expected_state_fingerprint or "").strip() or None - if normalized_fingerprint is not None and not _SHA256.fullmatch( + if normalized_fingerprint is not None and not BARE_SHA256_PATTERN.fullmatch( normalized_fingerprint ): raise ValueError("expected state fingerprint must be a SHA-256 digest") diff --git a/loopx/control_plane/goals/deletion_service.py b/loopx/control_plane/goals/deletion_service.py index 8de6cc02d..2aee5d89e 100644 --- a/loopx/control_plane/goals/deletion_service.py +++ b/loopx/control_plane/goals/deletion_service.py @@ -37,6 +37,7 @@ _source_and_target, _source_status, ) +from ..content_digest import BARE_SHA256_PATTERN GOAL_DELETION_SCHEMA_VERSION = "loopx_goal_deletion_v1" @@ -46,7 +47,6 @@ ) GOAL_DELETION_RECOVERY_SCHEMA_VERSION = "loopx_goal_deletion_recovery_v1" _OPAQUE_ID = re.compile(r"^[A-Za-z0-9._:-]{1,200}$") -_SHA256 = re.compile(r"^[a-f0-9]{64}$") _MAX_RECOVERY_RECEIPT_BYTES = 64 * 1024 @@ -111,9 +111,9 @@ def _normalize_source_basis(value: Mapping[str, Any] | None) -> dict[str, str] | route_mode = str(value.get("route_mode") or "") if schema_version != GOAL_DELETION_SOURCE_BASIS_SCHEMA_VERSION: raise ValueError("expected source basis has an unsupported schema version") - if not _SHA256.fullmatch(source_identity): + if not BARE_SHA256_PATTERN.fullmatch(source_identity): raise ValueError("expected source identity must be a SHA-256 digest") - if not _SHA256.fullmatch(source_content_sha256): + if not BARE_SHA256_PATTERN.fullmatch(source_content_sha256): raise ValueError("expected source content digest must be a SHA-256 digest") if route_mode not in {mode.value for mode in GoalActivationAuthorityRouteMode}: raise ValueError("expected source route mode is unsupported") @@ -335,7 +335,7 @@ def _validated_recovery_record( if value.get("goal_id") != goal_id: raise ValueError("Goal deletion recovery goal identity does not match") state_fingerprint = str(value.get("state_fingerprint") or "") - if not _SHA256.fullmatch(state_fingerprint): + if not BARE_SHA256_PATTERN.fullmatch(state_fingerprint): raise ValueError("Goal deletion recovery state fingerprint is invalid") source_basis_value = value.get("source_basis") if not isinstance(source_basis_value, Mapping): @@ -390,7 +390,7 @@ def _validated_recovery_record( field="snapshot backup path", ) preimage_sha256 = str(snapshot_value.get("preimage_sha256") or "") - if not _SHA256.fullmatch(preimage_sha256): + if not BARE_SHA256_PATTERN.fullmatch(preimage_sha256): raise ValueError("Goal deletion recovery preimage digest is invalid") if backup_path.parent != registry_path.parent or not ( backup_path.name.startswith(f"{registry_path.name}.goal-delete-") @@ -1180,7 +1180,7 @@ def delete_stopped_goal( normalized_goal_id = _require_opaque_id(goal_id, field="goal_id") requested_registry = Path(registry_path) normalized_fingerprint = str(expected_state_fingerprint or "").strip() or None - if normalized_fingerprint is not None and not _SHA256.fullmatch( + if normalized_fingerprint is not None and not BARE_SHA256_PATTERN.fullmatch( normalized_fingerprint ): raise ValueError("expected state fingerprint must be a SHA-256 digest") diff --git a/loopx/control_plane/goals/goal_amendment_proposal.py b/loopx/control_plane/goals/goal_amendment_proposal.py index 29f28cad0..bb40ff6f1 100644 --- a/loopx/control_plane/goals/goal_amendment_proposal.py +++ b/loopx/control_plane/goals/goal_amendment_proposal.py @@ -57,7 +57,6 @@ from __future__ import annotations import json -import re from collections.abc import Mapping from pathlib import Path from typing import Any @@ -82,6 +81,7 @@ autonomous_replan_is_required, autonomous_replan_scope_decision, ) +from ..content_digest import ENVELOPED_SHA256_PATTERN from .shared_goal_work_source import read_shared_goal_work_source from .shared_goal_alignment import ( DEFAULT_REGISTRY_RELATIVE_PATH, @@ -110,7 +110,6 @@ ) AMENDMENT_PROPOSAL_JOURNAL_DIRNAME = "amendment-proposals" AMENDMENT_PROPOSAL_JOURNAL_BASENAME = "journal.jsonl" -_SHA256_DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") def amendment_proposal_journal_path( @@ -441,7 +440,7 @@ def _check_admission_shape( != str(proposal.get("proposal_id") or "").strip().lower() or admission.get("base_revision_basis") != str(proposal.get("base_revision_basis") or "").strip() - or not _SHA256_DIGEST_PATTERN.fullmatch( + or not ENVELOPED_SHA256_PATTERN.fullmatch( str(admission.get("proposal_digest") or "") ) ): diff --git a/loopx/control_plane/projects/registry_codec.py b/loopx/control_plane/projects/registry_codec.py index a7899c319..cfa40b37b 100644 --- a/loopx/control_plane/projects/registry_codec.py +++ b/loopx/control_plane/projects/registry_codec.py @@ -9,10 +9,10 @@ import json import os from pathlib import Path -import re import tempfile from typing import Any, TypeVar +from ..content_digest import ENVELOPED_SHA256_PATTERN from ...file_lock import exclusive_cross_runtime_file_lock from ...paths import GLOBAL_REGISTRY_FILENAME @@ -27,7 +27,6 @@ "minimum_writer_protocol", "payload_sha256", } -_SHA256_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$") T = TypeVar("T") @@ -146,7 +145,7 @@ def _decode_document(raw_bytes: bytes) -> _ProjectRegistryDocument: "strict project registry minimum_writer_protocol must be nonempty" ) digest = header["payload_sha256"] - if not isinstance(digest, str) or not _SHA256_PATTERN.fullmatch(digest): + if not isinstance(digest, str) or not ENVELOPED_SHA256_PATTERN.fullmatch(digest): raise ProjectRegistryError( "strict project registry payload_sha256 is malformed" ) diff --git a/loopx/control_plane/testing/release_commit_qualification.py b/loopx/control_plane/testing/release_commit_qualification.py index 994369fad..e85d7c567 100644 --- a/loopx/control_plane/testing/release_commit_qualification.py +++ b/loopx/control_plane/testing/release_commit_qualification.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import Any, Callable, Mapping +from ..content_digest import ENVELOPED_SHA256_PATTERN from ..runtime.public_safety import public_safe_compact_text from .actual_default_model_behavior_portfolio import ( ACTUAL_DEFAULT_MODEL_BEHAVIOR_CONTRAST_COUNT, @@ -45,7 +46,6 @@ } _HEX_ID_RE = re.compile(r"^[0-9a-f]{40}(?:[0-9a-f]{24})?$") -_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") _VERSION_RE = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+(?:[A-Za-z0-9.+-]*)?$") _TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:@/+\-]{0,159}$") @@ -91,7 +91,7 @@ def _hex_id(value: Any, *, field: str) -> str: def _digest(value: Any, *, field: str) -> str: text = str(value or "").strip().lower() - if not _DIGEST_RE.fullmatch(text): + if not ENVELOPED_SHA256_PATTERN.fullmatch(text): raise ValueError(f"{field} must be a sha256 digest") return text diff --git a/loopx/control_plane/todos/completion_result.py b/loopx/control_plane/todos/completion_result.py index a0c31878b..59abfe1de 100644 --- a/loopx/control_plane/todos/completion_result.py +++ b/loopx/control_plane/todos/completion_result.py @@ -14,10 +14,11 @@ import tempfile from pathlib import Path from typing import Any +from ..content_digest import BARE_SHA256_PATTERN MAX_RESULT_BYTES = 128_000 _CONTENT_TYPES = {".json": "application/json", ".md": "text/markdown", ".txt": "text/plain"} -_DIGEST = re.compile(r"[a-f0-9]{64}\Z") +_DIGEST = BARE_SHA256_PATTERN def _object_path(runtime_root: Path, goal_id: str, digest: str) -> Path: diff --git a/loopx/control_plane/todos/completion_transaction.py b/loopx/control_plane/todos/completion_transaction.py index 3f550fa7d..221eb0c82 100644 --- a/loopx/control_plane/todos/completion_transaction.py +++ b/loopx/control_plane/todos/completion_transaction.py @@ -2,12 +2,12 @@ from __future__ import annotations -import re from collections.abc import Mapping from typing import Any from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from .contract import normalize_todo_id_list +from ..content_digest import BARE_SHA256_PATTERN TODO_COMPLETION_TRANSACTION_REQUEST_SCHEMA = "loopx_todo_completion_transaction_v0" @@ -289,10 +289,10 @@ def _valid_receipt(value: Any) -> bool: value.get("validation_declaration_sha256") is None or ( isinstance(value.get("validation_declaration_sha256"), str) - and re.fullmatch( - r"[a-f0-9]{64}", - value.get("validation_declaration_sha256"), - ) is not None + and BARE_SHA256_PATTERN.fullmatch( + value.get("validation_declaration_sha256") + ) + is not None ) ) ) @@ -404,10 +404,10 @@ def _valid_execute_validation_result(result: Mapping[str, Any]) -> bool: effect.get("validation_declaration_sha256") is None or ( isinstance(effect.get("validation_declaration_sha256"), str) - and re.fullmatch( - r"[a-f0-9]{64}", - effect.get("validation_declaration_sha256"), - ) is not None + and BARE_SHA256_PATTERN.fullmatch( + effect.get("validation_declaration_sha256") + ) + is not None ) ) ) diff --git a/loopx/control_plane/todos/completion_validation.py b/loopx/control_plane/todos/completion_validation.py index 85ac24563..1284b9697 100644 --- a/loopx/control_plane/todos/completion_validation.py +++ b/loopx/control_plane/todos/completion_validation.py @@ -1,6 +1,5 @@ from __future__ import annotations -import re import subprocess from collections.abc import Mapping from json import loads as json_loads @@ -36,6 +35,7 @@ read_completion_validation_declaration, ) from .contract import TODO_STATUS_DONE, normalize_todo_status +from ..content_digest import BARE_SHA256_PATTERN # Kept safely under the 30s outer CLI/MCP subprocess budget so a timed-out # validation still produces a typed receipt before the outer call is killed. @@ -383,7 +383,7 @@ def run_declared_completion_validation_effect( declaration_digest = effect.get("validation_declaration_sha256") if declaration_digest is not None and ( not isinstance(declaration_digest, str) - or not re.fullmatch(r"[a-f0-9]{64}", declaration_digest) + or not BARE_SHA256_PATTERN.fullmatch(declaration_digest) ): raise ValueError( "validation_effect.validation_declaration_sha256 must be a SHA-256 digest" diff --git a/loopx/control_plane/todos/completion_validation_store.py b/loopx/control_plane/todos/completion_validation_store.py index 9b76e2fca..8a8681c83 100644 --- a/loopx/control_plane/todos/completion_validation_store.py +++ b/loopx/control_plane/todos/completion_validation_store.py @@ -15,6 +15,7 @@ completion_validation_declaration, completion_validation_declaration_sha256, ) +from ..content_digest import BARE_SHA256_PATTERN DECLARATION_SCHEMA_VERSION = "loopx_todo_completion_validation_declaration_v0" @@ -101,7 +102,7 @@ def prepare_completion_validation_declaration( def _read_prepared_declaration(path: Path, goal_id: str, digest: str) -> dict[str, Any] | None: - if not re.fullmatch(r"[a-f0-9]{64}", digest): + if not BARE_SHA256_PATTERN.fullmatch(digest): raise ValueError("canonical validation digest must be SHA-256") try: value = read_json(path.parent / "blobs" / f"{digest}.json") diff --git a/loopx/control_plane/todos/machine_section_projection.py b/loopx/control_plane/todos/machine_section_projection.py index bf167d20e..9a28f3b01 100644 --- a/loopx/control_plane/todos/machine_section_projection.py +++ b/loopx/control_plane/todos/machine_section_projection.py @@ -51,6 +51,7 @@ todo_marker_for_status, ) from .todo_summary import canonical_todo_read_record, todo_priority_parts, normalize_todo_text +from ..content_digest import BARE_SHA256_PATTERN TODO_SECTION_PROJECTION_SCHEMA_VERSION = "loopx_todo_section_projection_v0" @@ -459,7 +460,7 @@ def render_canonical_todo_sections( f"Todo {todo_id!r} has a validation digest without authority" ) continue - if not isinstance(digest, str) or not re.fullmatch(r"[a-f0-9]{64}", digest): + if not isinstance(digest, str) or not BARE_SHA256_PATTERN.fullmatch(digest): raise TodoSectionProjectionError( f"Todo {todo_id!r} requires validation but omits its declaration digest" ) diff --git a/loopx/control_plane/work_items/governed_transition_proposal.py b/loopx/control_plane/work_items/governed_transition_proposal.py index 6fca748ce..82f9731ac 100644 --- a/loopx/control_plane/work_items/governed_transition_proposal.py +++ b/loopx/control_plane/work_items/governed_transition_proposal.py @@ -19,6 +19,7 @@ ) from ..coordination.coordination_state_contract_generated import COORDINATION_STATE_CONTRACT from ..runtime.public_safety import validate_public_safe_value +from ..content_digest import ENVELOPED_SHA256_PATTERN from ..todos.contract import ( TODO_STATUS_DONE, TODO_STATUS_OPEN, @@ -57,7 +58,6 @@ } _LANE_TODO_ID_LIMIT = 8 _LANE_TODO_ID = re.compile(r"^todo_[A-Za-z0-9]{1,40}$") -_INTENT_BASIS = re.compile(r"^sha256:[0-9a-f]{64}$") _LANE_SETTLEMENT_FIELDS = { "lane_id", "agent_id", @@ -178,7 +178,7 @@ def validate_governed_transition_receipts( intent_basis = receipt.get("intent_basis") if intent_basis is not None and ( not isinstance(intent_basis, str) - or not _INTENT_BASIS.fullmatch(intent_basis) + or not ENVELOPED_SHA256_PATTERN.fullmatch(intent_basis) ): raise ValueError( "governed transition proposal receipt intent_basis is invalid" diff --git a/loopx/control_plane/work_items/progress_review_policy.py b/loopx/control_plane/work_items/progress_review_policy.py index b5c5f5a88..4448a5ccb 100644 --- a/loopx/control_plane/work_items/progress_review_policy.py +++ b/loopx/control_plane/work_items/progress_review_policy.py @@ -11,8 +11,8 @@ from __future__ import annotations from collections.abc import Mapping -import re from typing import Any +from ..content_digest import BARE_SHA256_PATTERN PROGRESS_REVIEW_POLICY_SCHEMA_VERSION = "progress_review_policy_v0" PROGRESS_REVIEW_MODES: tuple[str, ...] = ("off", "shadow", "assist") @@ -22,7 +22,6 @@ PROGRESS_REVIEW_DEFAULT_DRIFT_THRESHOLD = 2 PROGRESS_REVIEW_MIN_DRIFT_THRESHOLD = 2 PROGRESS_REVIEW_MAX_DRIFT_THRESHOLD = 20 -_HEX64 = re.compile(r"^[a-f0-9]{64}$") def normalize_progress_review_mode(value: Any) -> str: @@ -70,7 +69,7 @@ def normalize_progress_review_contract_revision(value: Any) -> str | None: # An explicit empty value clears a pin at the change layer; the # effective policy reads it back as "no pin". return "" - if not _HEX64.fullmatch(text): + if not BARE_SHA256_PATTERN.fullmatch(text): raise ValueError( "progress_review.contract_revision must be a sha256 hex digest" ) diff --git a/loopx/control_plane/work_items/task_lease.py b/loopx/control_plane/work_items/task_lease.py index bc585f8a1..6dae0c4d8 100644 --- a/loopx/control_plane/work_items/task_lease.py +++ b/loopx/control_plane/work_items/task_lease.py @@ -44,6 +44,7 @@ require_expected_version as require_expected_version, write_lease as write_lease, ) +from ..content_digest import BARE_SHA256_PATTERN DEFAULT_TASK_LEASE_TTL_SECONDS = 45 * 60 MAX_TASK_LEASE_TTL_SECONDS = 24 * 60 * 60 @@ -147,7 +148,7 @@ def _native_fence_payload( operation_id = raw.get("fence_operation_id") if operation_id is not None and ( not isinstance(operation_id, str) - or not re.fullmatch(r"[a-f0-9]{64}", operation_id) + or not BARE_SHA256_PATTERN.fullmatch(operation_id) ): raise TaskLeaseError( f"native task-lease {operation} result has an invalid fence operation id", diff --git a/loopx/domain_packs/issue_fix.py b/loopx/domain_packs/issue_fix.py index 2a847f37f..193e4b640 100644 --- a/loopx/domain_packs/issue_fix.py +++ b/loopx/domain_packs/issue_fix.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any +from ..control_plane.content_digest import ENVELOPED_SHA256_PATTERN from ..domain_state import default_domain_state_file_path, upsert_domain_state_jsonl from ..file_lock import exclusive_file_lock @@ -16,7 +17,7 @@ ISSUE_FIX_FEASIBILITY_LEDGER_FILENAME = "feasibility.jsonl" ISSUE_FIX_CANDIDATE_PREFLIGHT_LEDGER_FILENAME = "candidate-preflight.jsonl" ISSUE_FIX_REPOSITORY_SNAPSHOT_LEDGER_FILENAME = "repository-snapshots.jsonl" -REVIEWER_NOTIFICATION_RECEIPT_PATTERN = re.compile(r"sha256:[a-f0-9]{64}") +REVIEWER_NOTIFICATION_RECEIPT_PATTERN = ENVELOPED_SHA256_PATTERN REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION = ( "issue_fix_reviewer_notification_queue_receipt_v1" ) diff --git a/loopx/extensions/lark/document_comment_provider.py b/loopx/extensions/lark/document_comment_provider.py index 670354edf..3d08f0db6 100644 --- a/loopx/extensions/lark/document_comment_provider.py +++ b/loopx/extensions/lark/document_comment_provider.py @@ -22,6 +22,7 @@ from typing import Any from urllib.parse import urlsplit +from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN from ...file_lock import exclusive_file_lock from ..external_connector_provider import ( build_external_connector_permission_requirement, @@ -46,7 +47,7 @@ SAFE_TOKEN_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,199}") SAFE_PROFILE_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}") -IDEMPOTENCY_KEY_PATTERN = re.compile(r"sha256:[0-9a-f]{64}") +IDEMPOTENCY_KEY_PATTERN = ENVELOPED_SHA256_PATTERN CURSOR_PREFIX = "lark-comment-v0." REPLY_CHAIN_PREFIX = "lark-reply-v0." MAX_PROVIDER_PAGES = 20 diff --git a/loopx/extensions/openviking_semantic_preference/history_export.py b/loopx/extensions/openviking_semantic_preference/history_export.py index e37476d9c..4c00829fd 100644 --- a/loopx/extensions/openviking_semantic_preference/history_export.py +++ b/loopx/extensions/openviking_semantic_preference/history_export.py @@ -29,6 +29,7 @@ from pathlib import Path from typing import Any +from ...control_plane.content_digest import BARE_SHA256_PATTERN from ...control_plane.runtime.public_safety import public_safe_compact_text from ...history import collect_history, validate_goal_id_path_segment @@ -40,7 +41,6 @@ _MANIFEST_FILE = ".loopx-history-conclusion-export.json" _MANIFEST_SCHEMA_VERSION = "loopx_history_conclusion_export_manifest_v0" _SLUG_RE = re.compile(r"[^A-Za-z0-9._-]+") -_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") # Bounded ISO-8601-ish timestamp: digits, T/space, colon, dot, +/- and Z only. _TIMESTAMP_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}[T ][0-9:.+\-]{1,20}Z?$") _PUBLIC_GOAL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,120}$") @@ -162,7 +162,7 @@ def _read_owned_manifest(path: Path, *, goal_id: str) -> dict[str, str]: not name or Path(name).name != name or not name.endswith(".md") - or not _SHA256_RE.fullmatch(digest) + or not BARE_SHA256_PATTERN.fullmatch(digest) or name in owned ): raise RuntimeError("history export manifest contains an invalid file entry") diff --git a/loopx/extensions/presentation.py b/loopx/extensions/presentation.py index 4d8bb701a..508b8de0c 100644 --- a/loopx/extensions/presentation.py +++ b/loopx/extensions/presentation.py @@ -23,6 +23,7 @@ run_standalone_extension, ) from .process_runtime import run_capped_process +from ..control_plane.content_digest import BARE_SHA256_PATTERN from .readiness import ( CORE_VIEW_VALIDATORS, ResolvedRuntimeEntrypoint, @@ -44,7 +45,6 @@ _ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") _ANCHOR_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)*$") -_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") _MARKUP_RE = re.compile(r"<[^>]*>|javascript:", re.IGNORECASE) _LOCAL_PATH_RE = re.compile( r"(?:^|[\s(])(?:~[/\\]|/+(?:Users|home|tmp|private|var|etc|opt)/|" @@ -313,7 +313,7 @@ def _evidence_reference(value: Any, *, context: str) -> str: def _sha256(value: Any, *, context: str) -> str: text = _plain_text(value, context=context, max_length=64) - if not _SHA256_RE.fullmatch(text): + if not BARE_SHA256_PATTERN.fullmatch(text): raise ValueError(f"{context} must be a lowercase SHA-256") return text diff --git a/loopx/presentation/chat_bundle.py b/loopx/presentation/chat_bundle.py index c8dd04c97..a6c11f6f1 100644 --- a/loopx/presentation/chat_bundle.py +++ b/loopx/presentation/chat_bundle.py @@ -6,6 +6,11 @@ import json import re from pathlib import Path, PurePosixPath +# Absolute by necessity: setup.py and the two build scripts exec this module by file +# path, so there is no package context for a relative import. Each of those loaders puts +# the checkout root on sys.path before exec'ing it, which is what keeps a source build +# working when LoopX is not installed. +from loopx.control_plane.content_digest import BARE_SHA256_PATTERN MANIFEST = "bundle-manifest.json" CHAT_BUNDLE_SCHEMA_VERSION = "loopx_chat_bundle_v1" @@ -108,7 +113,7 @@ def validate_bundle(bundle: Path, *, source_root: Path | None = None) -> dict: if ( not safe_relative(name) or not isinstance(expected, str) - or not re.fullmatch(r"[0-9a-f]{64}", expected) + or not BARE_SHA256_PATTERN.fullmatch(expected) ): raise ValueError("invalid bundle witness") path = bundle / name diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 05cb12cb8..ec992d73e 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -255,7 +255,7 @@ }, { "site": "loopx/capabilities/manager_context/roundtrip.py::.drain::codec_read:load_project_registry#1", - "line": 865, + "line": 866, "column": 22, "kind": "codec_read", "api": "load_project_registry", @@ -487,7 +487,7 @@ }, { "site": "loopx/cli.py::.main::codec_read:load_project_registry#1", - "line": 846, + "line": 835, "column": 17, "kind": "codec_read", "api": "load_project_registry", @@ -967,7 +967,7 @@ }, { "site": "loopx/control_plane/collaboration/peers.py::._goal::codec_read:load_project_registry#1", - "line": 51, + "line": 52, "column": 22, "kind": "codec_read", "api": "load_project_registry", @@ -1039,7 +1039,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::._projected_source_identity::codec_read:load_registry#1", - "line": 91, + "line": 90, "column": 33, "kind": "codec_read", "api": "load_registry", @@ -1047,7 +1047,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::._readback::codec_read:load_registry#1", - "line": 231, + "line": 230, "column": 15, "kind": "codec_read", "api": "load_registry", @@ -1055,7 +1055,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::._readback::codec_read:load_registry#2", - "line": 234, + "line": 233, "column": 15, "kind": "codec_read", "api": "load_registry", @@ -1063,7 +1063,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::._source_and_target::codec_read:load_registry#1", - "line": 172, + "line": 171, "column": 25, "kind": "codec_read", "api": "load_registry", @@ -1071,7 +1071,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::._source_status::codec_read:load_registry#1", - "line": 156, + "line": 155, "column": 26, "kind": "codec_read", "api": "load_registry", @@ -1079,7 +1079,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::.set_goal_activation_state::codec_read:load_registry#1", - "line": 286, + "line": 285, "column": 25, "kind": "codec_read", "api": "load_registry", @@ -1087,7 +1087,7 @@ }, { "site": "loopx/control_plane/goals/activation_service.py::.set_goal_activation_state::codec_transaction:project_registry_transaction#1", - "line": 381, + "line": 380, "column": 10, "kind": "codec_transaction", "api": "project_registry_transaction", @@ -1303,7 +1303,7 @@ }, { "site": "loopx/control_plane/goals/goal_amendment_proposal.py::.admit_goal_amendment_proposal::codec_read:load_registry#1", - "line": 199, + "line": 198, "column": 28, "kind": "codec_read", "api": "load_registry", @@ -1423,7 +1423,7 @@ }, { "site": "loopx/control_plane/projects/registry_codec.py::.add_project_registry_backend::codec_write:mutate_project_registry#1", - "line": 572, + "line": 571, "column": 12, "kind": "codec_write", "api": "mutate_project_registry", @@ -1431,7 +1431,7 @@ }, { "site": "loopx/control_plane/projects/registry_codec.py::.decode_registry_snapshot::codec_read:decode_project_registry#1", - "line": 207, + "line": 206, "column": 15, "kind": "codec_read", "api": "decode_project_registry", @@ -1439,7 +1439,7 @@ }, { "site": "loopx/control_plane/projects/registry_codec.py::.load_registry::codec_read:decode_registry_snapshot#1", - "line": 221, + "line": 220, "column": 12, "kind": "codec_read", "api": "decode_registry_snapshot", @@ -1447,7 +1447,7 @@ }, { "site": "loopx/control_plane/projects/registry_codec.py::.mutate_project_registry::codec_transaction:project_registry_transaction#1", - "line": 544, + "line": 543, "column": 10, "kind": "codec_transaction", "api": "project_registry_transaction", @@ -1575,7 +1575,7 @@ }, { "site": "loopx/control_plane/work_items/task_lease.py::.runtime_root_from_registry::codec_read:load_registry#1", - "line": 563, + "line": 564, "column": 16, "kind": "codec_read", "api": "load_registry", diff --git a/scripts/chat_bundle.py b/scripts/chat_bundle.py index e10210439..ea4926289 100644 --- a/scripts/chat_bundle.py +++ b/scripts/chat_bundle.py @@ -11,10 +11,15 @@ from pathlib import Path import shutil import subprocess +import sys import tempfile import zipfile ROOT = Path(__file__).resolve().parents[1] +# The contract asks the shared digest owner what a stored digest looks like, so loading +# it by file path needs the checkout importable: a source build has no installed LoopX. +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) spec = importlib.util.spec_from_file_location( "chat_bundle_contract", ROOT / "loopx/presentation/chat_bundle.py" ) diff --git a/scripts/desktop_runtime_bundle.py b/scripts/desktop_runtime_bundle.py index 512676c53..5ba50cf35 100644 --- a/scripts/desktop_runtime_bundle.py +++ b/scripts/desktop_runtime_bundle.py @@ -65,6 +65,10 @@ def build(root: Path) -> None: raise RuntimeError( "frontend build belongs to another source revision; rebuild before desktop packaging" ) + # The contract delegates its digest shape to the shared owner; this build runs from a + # checkout, where LoopX may not be installed, so the root has to be importable first. + if str(root) not in sys.path: + sys.path.insert(0, str(root)) spec = importlib.util.spec_from_file_location( "chat_contract", root / "loopx/presentation/chat_bundle.py" ) diff --git a/setup.py b/setup.py index b573b0727..cd5ed709b 100644 --- a/setup.py +++ b/setup.py @@ -3,6 +3,7 @@ from pathlib import Path import importlib.util import shutil +import sys from setuptools import setup from setuptools.command.build_py import build_py from setuptools.command.sdist import sdist @@ -10,6 +11,10 @@ def verify_frontend(): root = Path(__file__).parent + # Same reason as scripts/chat_bundle.py: the contract delegates its digest shape to + # the shared owner, and a source distribution is verified before LoopX is installed. + if str(root) not in sys.path: + sys.path.insert(0, str(root)) spec = importlib.util.spec_from_file_location( "chat_bundle_contract", root / "loopx/presentation/chat_bundle.py" ) diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py new file mode 100644 index 000000000..8304b7e08 --- /dev/null +++ b/tests/architecture/test_content_digest_single_owner.py @@ -0,0 +1,1565 @@ +"""One owner decides what a stored SHA-256 looks like, and this keeps it that way. + +Four layers, and none of them substitutes for another: + +1. a **stated-shape scan**: no module outside the owner may state either whole-value + shape, judged by the value a piece of source denotes; +2. an **ownership manifest**: the modules that delegate the decision are pinned, they must + name a canonical export, and every name they import has to be used; +3. a **readability rule**: a module that asks the owner may not also hand `re` a pattern + this file cannot read, because that is where a shape could hide; +4. **behavioural cases** that enter through each surface's own reader, the only layer that + can notice a surface wired to the wrong envelope. + +Layer 1 judges values rather than spellings because the first two rounds of this pull +request, and of its TypeScript twin, each found a spelling a spelling rule had to miss: an +unanchored `re.fullmatch` argument, an anchored literal, a constant built with `+`, a name +bound elsewhere in the same file, a `from re import fullmatch` import and a pattern parked +in a tuple all state the same decision. A rule of the form "a literal that looks like +`^...$`" only ever closes the spellings someone has already thought of. + +Two residues are named instead of claimed. A shape that reaches a matcher from data this +file cannot read (a configuration value, a provider payload) in a module that never imports +the owner stays invisible to static scanning, and a whole-value decision written without +any pattern at all (`len(text) == 64 and text in HEXDIGITS`) is not something a pattern +scan can see. Both are recorded as follow-up work here, not reported as forbidden. + +Only whole-value shapes are owned. A hex digest inside a larger grammar (a `cadence_...` +identifier, a journal filename, a `40|64` Git object id, a compound cursor) answers that +grammar's question and stays with the surface that owns it, and producers that concatenate +`"sha256:"` by hand are the other half of the decision and are deliberately unchanged. +""" + +from __future__ import annotations + +import ast +import importlib +import json +import re +from pathlib import Path +from typing import Any + +import pytest + +from loopx.control_plane import content_digest +from loopx.control_plane.content_digest import ( + BARE_SHA256_PATTERN, + ENVELOPED_SHA256_PATTERN, +) + +PACKAGE_ROOT = Path(__file__).resolve().parents[2] / "loopx" +OWNER_MODULE = "loopx/control_plane/content_digest.py" +OWNER_DOTTED = "loopx.control_plane.content_digest" +CANONICAL_EXPORTS = ("BARE_SHA256_PATTERN", "ENVELOPED_SHA256_PATTERN") + +# The two spellings of the same ten digits and six letters. Order inside a character class +# carries no meaning, so both denote one decision and neither is a second owner. +HEX64_CLASSES = ("[0-9a-f]", "[a-f0-9]") +ENVELOPE = "sha256:" +LEADING = ("^", r"\A") +TRAILING = ("$", r"\Z", r"\z") + +HEX64 = "a" * 64 +MIXED_HEX64 = "0123456789abcdef" * 4 +ENVELOPED = f"{ENVELOPE}{HEX64}" +TODO_ID = f"todo_{'a' * 12}" + +ENVELOPED_ACCEPTS = (ENVELOPED, f"{ENVELOPE}{MIXED_HEX64}") +ENVELOPED_REJECTS = ( + HEX64, # the envelope is part of the stored shape + f"{ENVELOPE}{HEX64[:-1]}", + f"{ENVELOPE}{HEX64}0", + f"{ENVELOPE}{HEX64.upper()}", + f"{ENVELOPE}{'z' * 64}", + f"prefix-{ENVELOPE}{HEX64}", + f"{ENVELOPE}{HEX64} trailing", + f"{ENVELOPE}{HEX64}\n", + "", +) +BARE_ACCEPTS = (HEX64, MIXED_HEX64, "f" * 64) +BARE_REJECTS = ( + ENVELOPED, + HEX64[:-1], + f"{HEX64}0", + HEX64.upper(), + "z" * 64, + f"x{HEX64}", + f"{HEX64} ", + f"{HEX64}\n", + "", + f"{ENVELOPE}{HEX64[:-1]}", +) + +# Probes that stress the one difference between the spellings this branch collapses: `$` +# also matches in front of a trailing newline while `re.fullmatch` demands the end of the +# string, so a value ending in "\n" is the case that could have divided them. +PARITY_PROBES = ( + HEX64, + MIXED_HEX64, + "f" * 64, + HEX64.upper(), + HEX64[:-1], + f"{HEX64}0", + f"{HEX64}\n", + f"{HEX64}\n\n", + f"\n{HEX64}", + f"{HEX64} ", + f" {HEX64}", + "", + "z" * 64, + ENVELOPED, + f"{ENVELOPE}{MIXED_HEX64}", + f"{ENVELOPE}{HEX64.upper()}", + f"{ENVELOPE}{HEX64}\n", + f"{ENVELOPE}{HEX64[:-1]}", + f"{ENVELOPE}{HEX64}0", + ENVELOPED.upper(), + HEX64 * 2, +) + +# Each shape that used to be stated at a call site, to be invoked the way that site invoked +# it. Every pair has to agree with the owner on every probe above, which is the evidence +# that reading one decision out of one module is not a behaviour change. +RETIRED_SPELLINGS = ( + ("bare, unanchored, through re.fullmatch", "[a-f0-9]{64}", BARE_SHA256_PATTERN), + ("bare, anchored, compiled then .fullmatch", "^[0-9a-f]{64}$", BARE_SHA256_PATTERN), + (r"bare, closed with \Z, compiled", "[0-9a-f]{64}\\Z", BARE_SHA256_PATTERN), + ( + "enveloped, unanchored, through re.fullmatch", + "sha256:[0-9a-f]{64}", + ENVELOPED_SHA256_PATTERN, + ), + ( + "enveloped, anchored, compiled", + "^sha256:[0-9a-f]{64}$", + ENVELOPED_SHA256_PATTERN, + ), + ( + "enveloped, [a-f0-9] class order", + "sha256:[a-f0-9]{64}", + ENVELOPED_SHA256_PATTERN, + ), +) + +# Recorded instead of absorbed, each with a reason that stands on the field contract rather +# than on which other pull request happens to be open. An entry that stops being true fails +# its own test below instead of ageing quietly. +DEFERRED_WHOLE_VALUE_SITES: dict[str, str] = { + "loopx/capabilities/manager_context/inspection.py": ( + "published as a JSON-schema `pattern` string, so it is schema data handed to a " + "validator rather than a matcher this owner may replace; the schema would have to " + "move for a verdict-for-verdict substitution to be safe" + ), +} + +# Pinned by review rather than derived: a module that starts asking the owner has to be +# added here, which turns widening the fan-out into a visible event instead of a quiet one. +CONSUMER_MODULES = ( + "loopx.capabilities.benchmark_toolkit.behavior_finding", + "loopx.capabilities.benchmark_toolkit.continuation", + "loopx.capabilities.benchmark_toolkit.factorial_contrast", + "loopx.capabilities.benchmark_toolkit.runtime_continuity", + "loopx.capabilities.benchmark_toolkit.study_projection", + "loopx.capabilities.content_ops.item_lifecycle", + "loopx.capabilities.issue_fix.outcome_projection", + "loopx.capabilities.issue_fix.reviewer_notification", + "loopx.capabilities.machine_configuration.store", + "loopx.capabilities.manager_context.roundtrip", + "loopx.capabilities.manager_context.tracking", + "loopx.capabilities.periodic_report.adapters", + "loopx.capabilities.periodic_report.archive", + "loopx.capabilities.periodic_report.bindings", + "loopx.capabilities.periodic_report.cadence_journal", + "loopx.capabilities.periodic_report.incremental", + "loopx.capabilities.periodic_report.machine_defaults", + "loopx.capabilities.progress_review.receipt", + "loopx.chat_action_normalization", + "loopx.configuration_transaction", + "loopx.control_plane.collaboration.delegation_inventory", + "loopx.control_plane.collaboration.inbox", + "loopx.control_plane.collaboration.peers", + "loopx.control_plane.coordination.local_authority_shadow_outbox", + "loopx.control_plane.coordination.shadow_management", + "loopx.control_plane.effect_runtime", + "loopx.control_plane.goals.activation_service", + "loopx.control_plane.goals.deletion_service", + "loopx.control_plane.goals.goal_amendment_proposal", + "loopx.control_plane.projects.registry_codec", + "loopx.control_plane.testing.release_commit_qualification", + "loopx.control_plane.todos.completion_result", + "loopx.control_plane.todos.completion_transaction", + "loopx.control_plane.todos.completion_validation", + "loopx.control_plane.todos.completion_validation_store", + "loopx.control_plane.todos.machine_section_projection", + "loopx.control_plane.work_items.governed_transition_proposal", + "loopx.control_plane.work_items.progress_review_policy", + "loopx.control_plane.work_items.task_lease", + "loopx.domain_packs.issue_fix", + "loopx.extensions.lark.document_comment_provider", + "loopx.extensions.openviking_semantic_preference.history_export", + "loopx.extensions.presentation", + "loopx.presentation.chat_bundle", +) + +# A consumer that hands `re` a pattern this file cannot fold. Pinned empty: folding reads +# every construction in every consumer today, so adding an entry has to be argued. +UNREADABLE_CONSUMER_CONSTRUCTIONS: dict[str, str] = {} + +# Reading `.pattern` off the owner and recompiling it states the decision a second time +# while showing the scan no shape at all. Pinned empty; widening it is a review event. +OWNER_TEXT_READS: dict[str, str] = {} + +RE_CONSTRUCTIONS = frozenset( + { + "compile", + "fullmatch", + "match", + "search", + "sub", + "subn", + "split", + "findall", + "finditer", + } +) +NOT_A_PATTERN_ARGUMENT = frozenset({"escape"}) +_SCOPE_ATTRIBUTE = "_digest_owner_scope" +MAX_FOLD_DEPTH = 12 + + +class _Binding: + """One name in one lexical scope, and whether its value can be read.""" + + __slots__ = ("value", "scope", "writes") + + def __init__(self) -> None: + self.value: ast.expr | None = None + self.scope: _Scope | None = None + self.writes = 0 + + @property + def foldable(self) -> bool: + # A name written twice, or bound by a parameter, an import, an unpacking target, a + # loop target, a `with` target, an exception name or a `global`/`nonlocal` + # statement, does not have one value to read. The answer then is "no value", which + # sends the site to the declaration layer instead of inventing a fold that could + # hide one owner behind another owner's wrong value. + return self.writes == 1 and self.value is not None and self.scope is not None + + +class _Scope: + __slots__ = ("parent", "names") + + def __init__(self, parent: _Scope | None) -> None: + self.parent = parent + self.names: dict[str, _Binding] = {} + + def binding(self, name: str) -> _Binding | None: + """The binding Python would resolve `name` to: the first scope that has one. + + Stopping at the first hit is the point. A flat table keyed by the name's text lets + a later function's local of the same name overwrite an earlier one, and that is + exactly how the third round of review on the TypeScript twin made a second owner + invisible while every guard stayed green. + """ + + scope: _Scope | None = self + while scope is not None: + found = scope.names.get(name) + if found is not None: + return found + scope = scope.parent + return None + + def declare(self, name: str, value: ast.expr | None, scope: _Scope | None) -> None: + found = self.names.setdefault(name, _Binding()) + found.writes += 1 + found.value = value + found.scope = scope + + def block(self, name: str) -> None: + """Record a name with no readable value, so nothing folds through it.""" + + found = self.names.setdefault(name, _Binding()) + found.writes += 1 + found.value = None + found.scope = None + + +def _argument_names(arguments: ast.arguments) -> list[str]: + names = [argument.arg for argument in arguments.posonlyargs] + names += [argument.arg for argument in arguments.args] + names += [argument.arg for argument in arguments.kwonlyargs] + if arguments.vararg is not None: + names.append(arguments.vararg.arg) + if arguments.kwarg is not None: + names.append(arguments.kwarg.arg) + return names + + +def _target_names(target: ast.expr) -> list[str]: + if isinstance(target, ast.Name): + return [target.id] + if isinstance(target, (ast.Tuple, ast.List)): + names: list[str] = [] + for element in target.elts: + names.extend(_target_names(element)) + return names + if isinstance(target, ast.Starred): + return _target_names(target.value) + return [] + + +def _collect_scopes(tree: ast.AST) -> _Scope: + """Record every name against the scope that declared it, in one traversal. + + Functions, lambdas, classes and comprehensions open a scope the way the compiler does; + a class body is kept in the lookup chain even though Python resolves nested functions + past it, because that only ever folds *more*, and every value it could fold to is a + literal this file reads by value anyway. Statements that bind something which can + change between runs are blocked rather than guessed at. + """ + + module = _Scope(None) + + def visit(node: ast.AST, scope: _Scope) -> None: + setattr(node, _SCOPE_ATTRIBUTE, scope) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + for decorator in node.decorator_list: + visit(decorator, scope) + for default in [*node.args.defaults, *node.args.kw_defaults]: + if default is not None: + visit(default, scope) + for argument in [ + *node.args.posonlyargs, + *node.args.args, + *node.args.kwonlyargs, + ]: + if argument.annotation is not None: + visit(argument.annotation, scope) + if node.returns is not None: + visit(node.returns, scope) + inner = _Scope(scope) + for name in _argument_names(node.args): + inner.declare(name, None, None) + for statement in node.body: + visit(statement, inner) + return + if isinstance(node, ast.Lambda): + for default in [*node.args.defaults, *node.args.kw_defaults]: + if default is not None: + visit(default, scope) + inner = _Scope(scope) + for name in _argument_names(node.args): + inner.declare(name, None, None) + visit(node.body, inner) + return + if isinstance(node, ast.ClassDef): + for base in [*node.bases, *node.keywords, *node.decorator_list]: + visit(base, scope) + inner = _Scope(scope) + for statement in node.body: + visit(statement, inner) + return + if isinstance( + node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp) + ): + inner = _Scope(scope) + for index, comprehension in enumerate(node.generators): + for name in _target_names(comprehension.target): + inner.block(name) + visit(comprehension.target, inner) + # The first iterable is evaluated where the comprehension is written. + visit(comprehension.iter, scope if index == 0 else inner) + for condition in comprehension.ifs: + visit(condition, inner) + if isinstance(node, ast.DictComp): + visit(node.key, inner) + visit(node.value, inner) + else: + visit(node.elt, inner) + return + if isinstance(node, (ast.For, ast.AsyncFor)): + for name in _target_names(node.target): + scope.block(name) + visit(node.target, scope) + visit(node.iter, scope) + for statement in [*node.body, *node.orelse]: + visit(statement, scope) + return + if isinstance(node, (ast.With, ast.AsyncWith)): + for item in node.items: + visit(item.context_expr, scope) + if item.optional_vars is not None: + for name in _target_names(item.optional_vars): + scope.block(name) + visit(item.optional_vars, scope) + for statement in node.body: + visit(statement, scope) + return + if isinstance(node, ast.ExceptHandler): + if node.name is not None: + scope.block(node.name) + if node.type is not None: + visit(node.type, scope) + for statement in node.body: + visit(statement, scope) + return + if isinstance(node, (ast.Global, ast.Nonlocal)): + for name in node.names: + scope.block(name) + return + if isinstance(node, ast.Import): + for alias in node.names: + scope.block(alias.asname or alias.name.split(".")[0]) + return + if isinstance(node, ast.ImportFrom): + for alias in node.names: + scope.block(alias.asname or alias.name) + return + if isinstance(node, ast.Delete): + for target in node.targets: + for name in _target_names(target): + scope.block(name) + visit(target, scope) + return + if isinstance(node, ast.AugAssign): + if isinstance(node.target, ast.Name): + scope.declare(node.target.id, None, None) + visit(node.target, scope) + visit(node.value, scope) + return + if isinstance(node, ast.AnnAssign): + if isinstance(node.target, ast.Name): + if node.value is None: + scope.block(node.target.id) + else: + scope.declare(node.target.id, node.value, scope) + visit(node.target, scope) + if node.annotation is not None: + visit(node.annotation, scope) + if node.value is not None: + visit(node.value, scope) + return + if isinstance(node, ast.NamedExpr): + if isinstance(node.target, ast.Name): + scope.declare(node.target.id, None, None) + visit(node.target, scope) + visit(node.value, scope) + return + if isinstance(node, ast.Assign): + single = len(node.targets) == 1 and isinstance(node.targets[0], ast.Name) + for target in node.targets: + if isinstance(target, ast.Name): + if single: + scope.declare(target.id, node.value, scope) + else: + scope.block(target.id) + else: + for name in _target_names(target): + scope.block(name) + visit(target, scope) + visit(node.value, scope) + return + if isinstance(node, ast.MatchAs) and node.name is not None: + scope.block(node.name) + for child in ast.iter_child_nodes(node): + visit(child, scope) + + for statement in tree.body: + visit(statement, module) + return module + + +def _scope_of(node: ast.AST, fallback: _Scope) -> _Scope: + return getattr(node, _SCOPE_ATTRIBUTE, fallback) + + +def _fold_text(node: ast.expr | None, scope: _Scope, depth: int = 0) -> str | None: + """The text an expression denotes, or None when it does not denote exactly one. + + Reads through `+` concatenation, same-file constant bindings, single-element lists and + f-strings that carry no substitution. A bytes literal is decoded, because a pattern + written as bytes is still a pattern stated in the source. Anything else answers + "unknown", which is a different claim from "not a digest shape". + """ + + if node is None or depth > MAX_FOLD_DEPTH: + return None + if isinstance(node, ast.Constant): + if isinstance(node.value, str): + return node.value + if isinstance(node.value, bytes): + try: + return node.value.decode("ascii") + except UnicodeDecodeError: + return None + return None + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = _fold_text(node.left, scope, depth + 1) + right = _fold_text(node.right, scope, depth + 1) + return left + right if left is not None and right is not None else None + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for value in node.values: + if isinstance(value, ast.Constant) and isinstance(value.value, str): + parts.append(value.value) + else: + return None + return "".join(parts) + if isinstance(node, ast.List) and len(node.elts) == 1: + return _fold_text(node.elts[0], scope, depth + 1) + if isinstance(node, ast.Name): + binding = scope.binding(node.id) + if binding is None or not binding.foldable: + return None + # The initializer is read in the scope that declared it, not in the scope that is + # asking, which is what keeps a nested local from answering for an outer name. + return _fold_text(binding.value, binding.scope, depth + 1) # type: ignore[arg-type] + return None + + +def _whole_value_shape(text: Any) -> str | None: + """Classify a value as a whole-value digest shape, ignoring how it is anchored. + + Anchoring belongs to the call rather than to the value: `re.fullmatch` gives + whole-string semantics to a literal carrying no anchors, and `\\Z` is `$` for that + purpose. Anything with more grammar - a prefix of its own, an alternation, a suffix, a + wider class - answers a different question and is left with the surface that wrote it. + """ + + if not isinstance(text, str) or "{64}" not in text: + return None + body = text + if body.startswith(LEADING): + body = body[1:] + for tail in TRAILING: + if body.endswith(tail): + body = body[: -len(tail)] + break + enveloped = body.startswith(ENVELOPE) + remainder = body[len(ENVELOPE) :] if enveloped else body + for character_class in HEX64_CLASSES: + if remainder == f"{character_class}{{64}}": + return "enveloped" if enveloped else "bare" + return None + + +def _owner_import_map(tree: ast.AST) -> dict[str, str | None]: + """local name -> canonical export name, over imports of the owner module only. + + `None` marks a binding this file cannot attribute: a wildcard import, the module object + itself, or some other attribute taken off the owner. + """ + + imported: dict[str, str | None] = {} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom): + module = node.module or "" + if module != "content_digest" and not module.endswith( + "control_plane.content_digest" + ): + continue + for alias in node.names: + bound = alias.asname or alias.name + imported[bound] = ( + alias.name if alias.name in CANONICAL_EXPORTS else None + ) + elif isinstance(node, ast.Import): + for alias in node.names: + if alias.name.endswith("control_plane.content_digest"): + imported[alias.asname or "content_digest"] = None + return imported + + +def _loaded_names(tree: ast.AST) -> set[str]: + """Names this module reads, with the import clauses themselves excluded. + + This is what catches a consumer that keeps `BARE_SHA256_PATTERN` imported while + checking the field with something else: the import leaves the name in the module + dictionary, so an identity check on the attribute cannot see the substitution. + """ + + loaded: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Import, ast.ImportFrom)): + continue + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load): + loaded.add(node.id) + elif isinstance(node, ast.Attribute): + base = node + while isinstance(base, ast.Attribute): + base = base.value + if isinstance(base, ast.Name): + loaded.add(base.id) + return loaded + + +def _shape_statements(tree: ast.AST, root: _Scope) -> list[dict[str, Any]]: + """Every whole-value digest shape this module states, by value, not by spelling.""" + + found: dict[tuple[int, str], dict[str, Any]] = {} + for node in ast.walk(tree): + if isinstance(node, ast.Constant) and isinstance(node.value, (str, bytes)): + text = _fold_text(node, root) + shape = _whole_value_shape(text) + if shape: + found[(node.lineno, text or "")] = { + "line": node.lineno, + "value": text, + "shape": shape, + "how": "literal", + } + continue + if isinstance(node, (ast.BinOp, ast.JoinedStr, ast.List, ast.Name)): + value = _fold_text(node, _scope_of(node, root)) + shape = _whole_value_shape(value) + if shape: + found[(node.lineno, value or "")] = { + "line": node.lineno, + "value": value, + "shape": shape, + "how": f"folded from {type(node).__name__}", + } + return sorted(found.values(), key=lambda site: (site["line"], str(site["value"]))) + + +def _re_names_bound(tree: ast.AST) -> tuple[set[str], set[str]]: + """(names bound to the `re` module, names bound to one of its pattern functions).""" + + modules: set[str] = set() + functions: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "re" or alias.name.endswith(".re"): + modules.add(alias.asname or "re") + elif isinstance(node, ast.ImportFrom): + if node.module == "re" or (node.module or "").endswith(".re"): + for alias in node.names: + if alias.name in RE_CONSTRUCTIONS: + functions.add(alias.asname or alias.name) + return modules, functions + + +def _regex_constructions(tree: ast.AST, root: _Scope) -> list[dict[str, Any]]: + """Every regex construction, and the value it is handed when that can be read. + + Constructions only: `re.(...)`, a bare `(...)` bound by `from re import + `, and a `getattr(re, ...)` hop. A method call on a compiled pattern + (`PATTERN.fullmatch(value)`) selects with a pattern that already exists, so counting + those would sweep in every field read in the package. + """ + + modules, functions = _re_names_bound(tree) + found: list[dict[str, Any]] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not node.args: + continue + callable_node = node.func + construction: str | None = None + if isinstance(callable_node, ast.Attribute): + base = callable_node.value + if ( + callable_node.attr in RE_CONSTRUCTIONS + and isinstance(base, ast.Name) + and base.id in modules + ): + construction = callable_node.attr + elif isinstance(base, ast.Call): + inner = base.func + hop = base.args[0] if base.args else None + if ( + isinstance(inner, ast.Name) + and inner.id == "getattr" + and isinstance(hop, ast.Name) + and hop.id in modules + ): + construction = "getattr" + elif isinstance(callable_node, ast.Name) and callable_node.id in functions: + construction = callable_node.id + if construction is None or construction in NOT_A_PATTERN_ARGUMENT: + continue + pattern = node.args[0] + value = _fold_text(pattern, _scope_of(pattern, root)) + found.append( + { + "line": node.lineno, + "construction": construction, + "value": value, + "shape": _whole_value_shape(value), + "readable": value is not None, + "argument": type(pattern).__name__, + } + ) + return found + + +def _owner_text_reads(tree: ast.AST) -> list[int]: + """Lines reading `.pattern` / `.source` / `.flags` off a name bound to the owner.""" + + imported = _owner_import_map(tree) + lines = [] + for node in ast.walk(tree): + if ( + isinstance(node, ast.Attribute) + and node.attr in {"pattern", "source", "flags"} + and isinstance(node.value, ast.Name) + and node.value.id in imported + ): + lines.append(node.lineno) + return lines + + +def _analyse(source: str, name: str = "") -> dict[str, Any]: + tree = ast.parse(source, filename=name) + root = _collect_scopes(tree) + return { + "tree": tree, + "root": root, + "statements": _shape_statements(tree, root), + "constructions": _regex_constructions(tree, root), + "owner_imports": _owner_import_map(tree), + "owner_text_reads": _owner_text_reads(tree), + "loaded": _loaded_names(tree), + } + + +_REPO_SCAN: list[dict[str, Any]] | None = None + + +def _repo_scan() -> list[dict[str, Any]]: + """One pass over every module under `loopx/`, shared by all four layers. + + 1,167 modules parse and analyse in about four seconds, so this runs once per test + session rather than once per layer, and no layer is given a different view of the tree. + """ + + global _REPO_SCAN + if _REPO_SCAN is None: + entries: list[dict[str, Any]] = [] + for path in sorted( + candidate + for candidate in PACKAGE_ROOT.rglob("*.py") + if "__pycache__" not in candidate.parts + ): + source = path.read_text(encoding="utf-8") + entry = _analyse(source, str(path)) + entry["path"] = path + entry["source"] = source + entry["relative"] = f"loopx/{path.relative_to(PACKAGE_ROOT).as_posix()}" + parts = list(path.relative_to(PACKAGE_ROOT.parent).parts) + parts[-1] = parts[-1][: -len(".py")] + entry["dotted"] = ".".join(parts) + entries.append(entry) + _REPO_SCAN = entries + return _REPO_SCAN + + +def _module_of(dotted: str) -> dict[str, Any]: + for entry in _repo_scan(): + if entry["dotted"] == dotted: + return entry + raise AssertionError(f"{dotted} is no longer a module under loopx/") + + +def _runtime_whole_value_patterns(module: Any) -> list[tuple[str, str]]: + """Patterns a loaded module holds that state a whole-value digest shape. + + Module attributes, and the values one container level deep, because the shape that + hides best is parked in a tuple or a dict rather than bound to an obvious name. + Identity is the requirement, not equality: a copy carrying the same text is a second + owner, and so is one assembled at run time from pieces this file could not fold. + """ + + owned = {id(BARE_SHA256_PATTERN), id(ENVELOPED_SHA256_PATTERN)} + found: list[tuple[str, str]] = [] + + def consider(label: str, value: Any) -> None: + if isinstance(value, re.Pattern) and id(value) not in owned: + shape = _whole_value_shape(value.pattern) + if shape: + found.append((label, f"{shape}: {value.pattern!r}")) + + for attribute, value in list(vars(module).items()): + if attribute.startswith("__"): + continue + consider(attribute, value) + if isinstance(value, (list, tuple, set, frozenset)): + for index, element in enumerate(value): + consider(f"{attribute}[{index}]", element) + elif isinstance(value, dict): + for key, element in value.items(): + consider(f"{attribute}[{key!r}]", element) + return found + + +# --- layer 1: no second statement of the shape ----------------------------------------- + + +def test_only_the_owner_module_states_a_whole_value_digest_shape() -> None: + offenders = {} + for entry in _repo_scan(): + relative = entry["relative"] + if relative in DEFERRED_WHOLE_VALUE_SITES or relative == OWNER_MODULE: + continue + if entry["statements"]: + offenders[relative] = entry["statements"] + assert not offenders, f"second owner(s) of the digest shape: {offenders}" + + +def test_owner_module_defines_each_shape_exactly_once() -> None: + statements = _module_of(OWNER_DOTTED)["statements"] + assert sorted({item["shape"] for item in statements}) == ["bare", "enveloped"], ( + statements + ) + values = [item["value"] for item in statements] + assert len(values) == len(set(values)), f"one shape stated twice: {statements}" + + +def test_deferred_sites_are_still_the_ones_this_branch_recorded() -> None: + for relative, reason in DEFERRED_WHOLE_VALUE_SITES.items(): + assert reason, relative + entry = next( + (item for item in _repo_scan() if item["relative"] == relative), None + ) + assert entry is not None, f"{relative} moved or vanished; update the allowlist" + assert entry["statements"], f"{relative} no longer restates the shape" + + +# --- layer 1 self-check: the scan is judged by value, never by spelling ----------------- + +# Each entry is (label, source, expectation). `stated` has to be reported as a shape this +# module states; `other-question` has to be left alone; the rest name the layer that catches +# a shape which is not written as a plain anchored literal. +BYPASS_CORPUS = ( + ( + "anchored literal, the only spelling the first scan knew", + 'import re\n\n\nP = re.compile(r"^[0-9a-f]{64}$")\n', + "stated", + ), + ( + "unanchored literal through re.fullmatch, reviewer round one", + 'import re\n\n\ndef check(value):\n return re.fullmatch(r"[0-9a-f]{64}", value)\n', + "stated", + ), + ( + r"closed with \Z instead of $", + 'import re\n\n\nP = re.compile(r"[0-9a-f]{64}\\Z")\n', + "stated", + ), + ( + "the other character-class order", + 'import re\n\n\nP = re.compile(r"^[a-f0-9]{64}$")\n', + "stated", + ), + ( + "two literals joined by +", + 'import re\n\n\nP = re.compile("^[0-9a-f]" + "{64}$")\n', + "stated", + ), + ( + "the marker itself split across two literals", + 'import re\n\n\nP = re.compile("^[0-9a-f]{" + "64}$")\n', + "stated", + ), + ( + "envelope and body joined from three pieces", + 'import re\n\n\nP = re.compile("sha256:" + r"[0-9a-f]{64}" + "$")\n', + "stated", + ), + ( + "same-file constant, unanchored, reached through re.fullmatch", + 'import re\n\n\nHEAD = r"[0-9a-f]{64}"\n\n\ndef check(value):\n' + " return re.fullmatch(HEAD, value)\n", + "stated", + ), + ( + "same-file constant built by + and compiled through the name", + 'import re\n\n\nSHAPE = "^" + "[0-9a-f]" + "{64}" + "$"\nP = re.compile(SHAPE)\n', + "stated", + ), + ( + "f-string carrying no substitution", + 'import re\n\n\nP = re.compile(f"^[0-9a-f]{{64}}$")\n', + "stated", + ), + ( + "pattern written as bytes", + "import re\n\n\nP = re.compile(rb'^[0-9a-f]{64}$')\n", + "stated", + ), + ( + "from re import fullmatch, no module prefix", + "from re import fullmatch\n\n\ndef check(value):\n" + ' return fullmatch(r"[0-9a-f]{64}", value)\n', + "stated", + ), + ( + "import re as rx, an aliased module", + 'import re as rx\n\n\nP = rx.compile(r"sha256:[0-9a-f]{64}")\n', + "stated", + ), + ( + "getattr(re, ...) hop", + 'import re\n\n\nP = getattr(re, "compile")(r"sha256:[0-9a-f]{64}")\n', + "stated", + ), + ( + "pattern parked in a tuple and used through the loop variable", + 'import re\n\n\nfor key, pattern in [("evidence", r"sha256:[a-f0-9]{64}")]:\n' + " re.fullmatch(pattern, key)\n", + "stated", + ), + ( + "pattern inside a dict of field rules", + 'import re\n\n\nRULES = {"digest": r"^[0-9a-f]{64}$"}\n', + "stated", + ), + ( + "handle compiled from a foldable name, matched later", + 'import re\n\n\nSHAPE = r"[a-f0-9]{64}"\nP = re.compile(SHAPE)\n\n\ndef check(value):' + "\n return P.fullmatch(value)\n", + "stated", + ), + ( + "the owner's text under IGNORECASE states it a second time, so it must be said", + "import re\n" + "from loopx.control_plane.content_digest import BARE_SHA256_PATTERN\n\n" + 'P = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE)\n', + "stated", + ), + ( + "compound id that merely contains a hex64", + 'import re\n\n\nP = re.compile(r"cadence_[0-9a-f]{64}")\n', + "other-question", + ), + ( + "git object id alternation", + 'import re\n\n\nP = re.compile(r"[0-9a-f]{40}|[0-9a-f]{64}")\n', + "other-question", + ), + ( + "two hex64 fields inside one cursor", + 'import re\n\n\nP = re.compile(r"1:[0-9a-f]{64}:[0-9a-f]{64}")\n', + "other-question", + ), + ( + "journal filename grammar", + 'import re\n\n\nP = re.compile(r"prq_[0-9a-f]{64}\\.json$")\n', + "other-question", + ), + ( + "a wider class that also accepts uppercase is its own policy", + 'import re\n\n\nP = re.compile(r"^[0-9a-fA-F]{64}$")\n', + "other-question", + ), + ( + "inline case-insensitive group is its own policy", + 'import re\n\n\nP = re.compile(r"(?i)^[0-9a-f]{64}$")\n', + "other-question", + ), + ( + "twelve-hex identifier, not a digest", + 'import re\n\n\nP = re.compile(r"todo_[a-f0-9]{12}")\n', + "other-question", + ), + ( + "a consumer that asks the owner states nothing itself", + "from loopx.control_plane.content_digest import BARE_SHA256_PATTERN\n\n\n" + "def check(value):\n return BARE_SHA256_PATTERN.fullmatch(value)\n", + "consumer", + ), + ( + "a consumer that aliases the owner object", + "from loopx.control_plane.content_digest import ENVELOPED_SHA256_PATTERN\n\n\n" + "RECEIPT_PATTERN = ENVELOPED_SHA256_PATTERN\n", + "consumer", + ), + ( + "a consumer that rebuilds the owner's text", + "import re\nfrom loopx.control_plane.content_digest import BARE_SHA256_PATTERN\n\n\n" + "P = re.compile(BARE_SHA256_PATTERN.pattern)\n", + "text-read", + ), + ( + "a consumer that keeps the import and checks something else", + "import re\nfrom loopx.control_plane.content_digest import BARE_SHA256_PATTERN\n\n\n" + "def check(value):\n return len(value) == 64 and value.isalnum()\n", + "unused-import", + ), + ( + "a pattern that arrives from data the scan cannot read", + "import re\nfrom loopx.control_plane.content_digest import BARE_SHA256_PATTERN\n\n\n" + "def check(value, shape):\n return re.fullmatch(shape, value)\n", + "unreadable", + ), +) + + +@pytest.mark.parametrize( + "label,source,expectation", + BYPASS_CORPUS, + ids=[entry[0] for entry in BYPASS_CORPUS], +) +def test_the_scan_judges_the_value_and_not_the_spelling( + label: str, source: str, expectation: str +) -> None: + analysis = _analyse(source, "fixture") + unfoldable = [item for item in analysis["constructions"] if not item["readable"]] + + if expectation == "stated": + assert analysis["statements"], ( + f"{label}: a second owner in this spelling escaped" + ) + assert {item["shape"] for item in analysis["statements"]} <= { + "bare", + "enveloped", + } + elif expectation == "other-question": + assert not analysis["statements"], f"{label}: a different question was absorbed" + elif expectation == "consumer": + assert not analysis["statements"], ( + f"{label}: asking the owner read as stating one" + ) + assert not unfoldable, f"{label}: a consumer was left unreadable" + assert not analysis["owner_text_reads"], ( + f"{label}: owner text read off the object" + ) + elif expectation == "text-read": + assert len(analysis["owner_text_reads"]) == 1, analysis["owner_text_reads"] + assert not analysis["statements"], ( + "expected the static scan alone to miss this one" + ) + elif expectation == "unused-import": + assert analysis["owner_imports"] == { + "BARE_SHA256_PATTERN": "BARE_SHA256_PATTERN" + }, analysis["owner_imports"] + assert "BARE_SHA256_PATTERN" not in analysis["loaded"], ( + f"{label}: the dangling import looked used" + ) + elif expectation == "unreadable": + assert not analysis["statements"], ( + f"{label}: expected nothing for the scan to read" + ) + assert unfoldable, f"{label}: a pattern from data passed as readable" + else: # pragma: no cover - protects the matrix against a mistyped expectation + raise AssertionError(f"unknown expectation {expectation!r} for {label!r}") + + +def test_a_later_local_of_the_same_name_does_not_answer_for_an_outer_fold() -> None: + """The defect the third round of review found on the TypeScript twin, in Python. + + `first()` asks for a module constant; `second()` binds a local of the same name to + something that is not a shape. Read through a flat, text-keyed table, the later + binding wins and the digest shape disappears from the scan. Read through scopes, the + call site in `first()` still resolves to the module value, and it has to be reported. + """ + + source = ( + "import re\n" + "\n" + 'HEAD = "[0-9a-f]"\n' + 'SHAPE = "^" + HEAD + "{64}$"\n' + "\n" + "\n" + "def first(value):\n" + " return re.fullmatch(SHAPE, value)\n" + "\n" + "\n" + "def second():\n" + ' SHAPE = "a prose label"\n' + " return SHAPE\n" + ) + statements = _analyse(source)["statements"] + assert statements, "a shape reached only through a name escaped the scan entirely" + assert any(item["how"] == "folded from Name" for item in statements), ( + f"the call site did not fold to the outer binding: {statements}" + ) + assert {item["value"] for item in statements} == {"^[0-9a-f]{64}$"}, statements + + +def test_a_local_shape_does_not_leak_out_to_the_module_scope() -> None: + """The other direction: an inner binding must not answer for an outer name. + + `uses_it` refers to a `SHAPE` the module never declares, so the scan must not find the + shape that `makes_it` built inside its own frame at the outer call site - and must + still report the shape `makes_it` states. + """ + + source = ( + "import re\n" + "\n" + "\n" + "def makes_it():\n" + ' SHAPE = "^" + "[0-9a-f]" + "{64}$"\n' + " return SHAPE\n" + "\n" + "\n" + "def uses_it(value):\n" + " return re.fullmatch(SHAPE, value)\n" + ) + statements = _analyse(source)["statements"] + # Both reported sites are inside `makes_it`: the binding it wrote, and the read of that + # binding. Nothing at the `uses_it` line, whose `SHAPE` the module never declares. + assert [item["line"] for item in statements] == [5, 6], statements + assert [item["how"] for item in statements] == [ + "folded from BinOp", + "folded from Name", + ] + assert all(item["shape"] == "bare" for item in statements), statements + + +def test_the_bypass_matrix_names_every_layer_it_relies_on() -> None: + expectations = {entry[2] for entry in BYPASS_CORPUS} + assert expectations == { + "stated", + "other-question", + "consumer", + "text-read", + "unused-import", + "unreadable", + }, expectations + labels = [entry[0] for entry in BYPASS_CORPUS] + assert len(labels) == len(set(labels)), ( + "two fixtures share a label; test ids would collide" + ) + + +# --- layer 2: who may depend on the owner, and how -------------------------------------- + + +def test_the_consumer_manifest_is_exactly_the_pinned_set() -> None: + derived = {entry["dotted"] for entry in _repo_scan() if entry["owner_imports"]} + pinned = set(CONSUMER_MODULES) + assert derived == pinned, ( + f"imports the owner but is not pinned: {sorted(derived - pinned)}; " + f"pinned but imports nothing from it any more: {sorted(pinned - derived)}" + ) + + +def test_consumers_name_the_canonical_exports_and_use_them() -> None: + unnamed = {} + dangling = {} + for entry in _repo_scan(): + imported = entry["owner_imports"] + if not imported: + continue + for local, canonical in imported.items(): + if canonical is None: + unnamed.setdefault(entry["dotted"], []).append(local) + elif local not in entry["loaded"]: + dangling.setdefault(entry["dotted"], []).append(local) + assert not unnamed, f"wildcard or module-object import of the owner: {unnamed}" + assert not dangling, f"imported from the owner but never referenced: {dangling}" + + +def test_every_consumer_holds_the_owner_object_not_an_equal_copy() -> None: + for dotted in CONSUMER_MODULES: + module = importlib.import_module(dotted) + copies = _runtime_whole_value_patterns(module) + assert not copies, ( + f"{dotted} holds a copy of a shape it should borrow: {copies}" + ) + + +def test_no_module_rebuilds_the_shape_from_the_owner_text() -> None: + offenders = {} + for entry in _repo_scan(): + if entry["relative"] == OWNER_MODULE or entry["relative"] in OWNER_TEXT_READS: + continue + if entry["owner_text_reads"]: + offenders[entry["relative"]] = entry["owner_text_reads"] + assert not offenders, f"owner text read off the object and recompiled: {offenders}" + + +# --- layer 3: a consumer has to stay readable -------------------------------------------- + + +def test_no_consumer_holds_a_regex_this_file_cannot_read() -> None: + offenders = {} + for dotted in CONSUMER_MODULES: + unreadable = [ + item for item in _module_of(dotted)["constructions"] if not item["readable"] + ] + if unreadable: + offenders[dotted] = unreadable + for dotted in UNREADABLE_CONSUMER_CONSTRUCTIONS: + offenders.pop(dotted, None) + assert not offenders, ( + "a consumer handed `re` a pattern this scan cannot fold; fold it back into a " + f"literal or record it with the question it answers: {offenders}" + ) + + +def test_every_declared_unreadable_consumer_site_is_still_there() -> None: + for dotted, reason in UNREADABLE_CONSUMER_CONSTRUCTIONS.items(): + assert reason, dotted + unreadable = [ + item for item in _module_of(dotted)["constructions"] if not item["readable"] + ] + assert unreadable, f"{dotted} has no unreadable construction left to declare" + + +# --- layer 4: behaviour, per spelling and per surface ------------------------------------- + + +@pytest.mark.parametrize("value", ENVELOPED_ACCEPTS) +def test_enveloped_pattern_accepts_a_prefixed_digest(value: str) -> None: + assert ENVELOPED_SHA256_PATTERN.fullmatch(value) is not None + + +@pytest.mark.parametrize("value", ENVELOPED_REJECTS) +def test_enveloped_pattern_rejects_every_other_shape(value: object) -> None: + assert ENVELOPED_SHA256_PATTERN.fullmatch(value) is None + + +@pytest.mark.parametrize("value", BARE_ACCEPTS) +def test_bare_pattern_accepts_lowercase_hex(value: str) -> None: + assert BARE_SHA256_PATTERN.fullmatch(value) is not None + + +@pytest.mark.parametrize("value", BARE_REJECTS) +def test_bare_pattern_rejects_everything_else(value: object) -> None: + assert BARE_SHA256_PATTERN.fullmatch(value) is None + + +@pytest.mark.parametrize( + "label,spelling,owner", + RETIRED_SPELLINGS, + ids=[entry[0] for entry in RETIRED_SPELLINGS], +) +@pytest.mark.parametrize("value", PARITY_PROBES) +def test_each_retired_spelling_and_the_owner_split_the_same_strings( + label: str, spelling: str, owner: re.Pattern[str], value: str +) -> None: + """A migration changes behaviour only if some probe divides the two. + + Each retired spelling is invoked the way its site invoked it - `re.fullmatch` on the + text for the sites that used the module function, `.fullmatch` on a compiled pattern for + the sites that held a handle - and both are compared with the owner object. The probe set + includes the trailing newline, which is where `$` and `\\Z` could have disagreed. + """ + + assert bool(re.fullmatch(spelling, value)) is bool(owner.fullmatch(value)), ( + label, + value, + ) + assert bool(re.compile(spelling).fullmatch(value)) is bool( + owner.fullmatch(value) + ), (label, value) + + +def test_the_owner_is_a_leaf_and_exports_only_the_two_shapes() -> None: + public = {name for name in vars(content_digest) if not name.startswith("_")} + assert public == {"annotations", "re", *CANONICAL_EXPORTS}, public + patterns = { + name + for name, value in vars(content_digest).items() + if isinstance(value, re.Pattern) + } + assert patterns == set(CANONICAL_EXPORTS), patterns + + +def test_periodic_report_generation_receipt_checks_both_digest_fields() -> None: + """The site a reviewer named: two digest fields, checked by one reader. + + Both are the enveloped shape, and swapping either for the bare one has to be visible + here rather than only in the text of the module, because an envelope swap leaves the + owner object in place and therefore no trace for the value scan. + """ + + from loopx.capabilities.periodic_report import bindings + + def artifact(content: str, document: str) -> dict[str, str]: + return { + "artifact_id": "art-1", + "renderer_id": "renderer-1", + "renderer_kind": "markdown", + "artifact_ref": "report://document/1", + "content_digest": content, + "document_digest": document, + } + + def receipt(document: str, content: str | None = None) -> dict[str, Any]: + normalized = artifact(content or document, document) + return { + "schema_version": bindings.GENERATION_RECEIPT_SCHEMA, + "status": "succeeded", + "generation_id": bindings._identity( + {"document_digest": document, "artifacts": [normalized]}, + prefix="report_generation", + ), + "document_digest": document, + "artifact_receipts": [normalized], + "artifact_count": 1, + "provider_required": False, + "external_writes_performed": False, + } + + assert ( + bindings._generation_receipt(receipt(ENVELOPED))["document_digest"] == ENVELOPED + ) + with pytest.raises(ValueError, match="document_digest must use sha256"): + bindings._generation_receipt(receipt(HEX64)) + with pytest.raises(ValueError, match="content_digest must use sha256"): + bindings._generation_receipt(receipt(ENVELOPED, HEX64)) + + +def test_schema_string_site_is_the_same_question_as_the_owner_bare_shape() -> None: + statements = _module_of("loopx.capabilities.manager_context.inspection")[ + "statements" + ] + assert statements, "the recorded schema site no longer states the shape" + for statement in statements: + assert statement["shape"] == "bare" + declared = re.compile(statement["value"]) + for value in PARITY_PROBES: + assert bool(declared.fullmatch(value)) is bool( + BARE_SHA256_PATTERN.fullmatch(value) + ), value + + +# --- per-surface wiring: every case enters through that surface's own reader ------------- + + +def test_periodic_report_archive_requires_the_envelope() -> None: + from loopx.capabilities.periodic_report import archive + + assert archive._sha256(ENVELOPED, "revision") == ENVELOPED + with pytest.raises(ValueError, match="must use sha256"): + archive._sha256(HEX64, "revision") + + +def test_periodic_report_incremental_requires_the_envelope() -> None: + from loopx.capabilities.periodic_report import incremental + + assert incremental._digest(ENVELOPED, "fact_fingerprint") == ENVELOPED + with pytest.raises(ValueError, match="must use sha256"): + incremental._digest(HEX64, "fact_fingerprint") + + +def test_progress_review_receipt_rejects_the_envelope_it_never_stored() -> None: + from loopx.capabilities.progress_review import receipt + + assert receipt._hex64(HEX64, field="basis_digest") == HEX64 + with pytest.raises(ValueError, match="must be a sha256 hex digest"): + receipt._hex64(ENVELOPED, field="basis_digest") + + +def test_presentation_extension_keeps_its_own_message_and_bare_shape() -> None: + from loopx.extensions import presentation + + assert presentation._sha256(HEX64, context="artifact") == HEX64 + with pytest.raises(ValueError, match="must be a lowercase SHA-256"): + presentation._sha256("z" * 64, context="artifact") + # This surface also caps the field at 64 characters, so an enveloped digest never + # reaches the shape check here. That limit is the surface's own policy and is left + # alone; it is why the rejected probe above is same-length. + with pytest.raises(ValueError, match="at most 64 characters"): + presentation._sha256(ENVELOPED, context="artifact") + + +def test_release_commit_qualification_normalises_before_the_owner_check() -> None: + from loopx.control_plane.testing import release_commit_qualification + + assert release_commit_qualification._digest(ENVELOPED, field="commit") == ENVELOPED + with pytest.raises(ValueError, match="must be a sha256 digest"): + release_commit_qualification._digest(HEX64, field="commit") + + +def test_configuration_revision_allows_the_absent_sentinel() -> None: + from loopx.configuration_transaction import _validated_revision + + assert _validated_revision("absent", label="revision") == "absent" + assert _validated_revision(ENVELOPED, label="revision") == ENVELOPED + with pytest.raises(ValueError, match="must be absent or a sha256 revision"): + _validated_revision(HEX64, label="revision") + + +def test_progress_review_policy_keeps_clearing_and_null_as_distinct_values() -> None: + from loopx.control_plane.work_items.progress_review_policy import ( + normalize_progress_review_contract_revision, + ) + + assert normalize_progress_review_contract_revision(None) is None + assert normalize_progress_review_contract_revision("") == "" + assert ( + normalize_progress_review_contract_revision(HEX64) == HEX64 + ) # positive control + with pytest.raises(ValueError, match="must be a sha256 hex digest"): + normalize_progress_review_contract_revision(ENVELOPED) + + +def test_deletion_source_basis_checks_both_digest_fields() -> None: + from loopx.control_plane.goals.deletion_service import ( + GOAL_DELETION_SOURCE_BASIS_SCHEMA_VERSION, + _normalize_source_basis, + ) + + basis: dict[str, Any] = { + "schema_version": GOAL_DELETION_SOURCE_BASIS_SCHEMA_VERSION, + "source_identity": HEX64, + "source_content_sha256": MIXED_HEX64, + "route_mode": "source_to_global", + } + assert _normalize_source_basis(basis)["source_content_sha256"] == MIXED_HEX64 + with pytest.raises(ValueError, match="source identity"): + _normalize_source_basis({**basis, "source_identity": ENVELOPED}) + with pytest.raises(ValueError, match="content digest"): + _normalize_source_basis({**basis, "source_content_sha256": HEX64.upper()}) + + +def test_periodic_report_delivery_authority_checks_its_effective_revision() -> None: + from loopx.capabilities.periodic_report.machine_defaults import ( + DELIVERY_AUTHORITY_SCHEMA, + normalize_periodic_report_delivery_authority, + ) + + authority = { + "schema_version": DELIVERY_AUTHORITY_SCHEMA, + "kind": "enabled_periodic_report_subscription", + "goal_id": "goal-1", + "source": "machine_default", + "effective_revision": ENVELOPED, + "route_ref": "route-1", + } + assert ( + normalize_periodic_report_delivery_authority(authority)["effective_revision"] + == ENVELOPED + ) + with pytest.raises(ValueError, match="effective_revision is invalid"): + normalize_periodic_report_delivery_authority( + {**authority, "effective_revision": HEX64} + ) + + +def test_governed_transition_receipt_checks_the_intent_basis_field_only() -> None: + from loopx.control_plane.work_items.governed_transition_proposal import ( + GOVERNED_TRANSITION_RECEIPT_SCHEMA_VERSION as RECEIPT_SCHEMA, + validate_governed_transition_receipts, + ) + + receipt = { + "schema_version": RECEIPT_SCHEMA, + "kind": "continuous_monitor_upsert", + "status": "committed", + "proposal_id": "prop-1", + "proposal_digest": ENVELOPED, + "action": "upsert", + "todo_id": "todo-1", + "monitor_key": "monitor-1", + "target_key": "target-1", + "intent_basis": ENVELOPED, + } + assert ( + validate_governed_transition_receipts([receipt])[0]["intent_basis"] == ENVELOPED + ) + with pytest.raises(ValueError, match="intent_basis is invalid"): + validate_governed_transition_receipts([{**receipt, "intent_basis": HEX64}]) + + +def test_shadow_outbox_cursor_keeps_its_envelope_and_its_absent_option() -> None: + from loopx.control_plane.coordination.local_authority_shadow_outbox import ( + DRAIN_CURSOR_SCHEMA, + OutboxError, + decode_cursor, + ) + from loopx.control_plane.coordination.local_authority_shadow_projection import ( + PARTITIONS, + ) + + def cursor(digest: object) -> dict[str, Any]: + return { + "schema_version": DRAIN_CURSOR_SCHEMA, + "partition": PARTITIONS[0], + "last_seq": 1, + "last_entry_id": f"local-shadow-tx-{HEX64}", + "last_partition_digest": digest, + "last_cursor": "cursor-opaque", + "last_provider_revision": "revision-opaque", + "updated_at": "2026-09-28T00:00:00+00:00", + } + + partition = PARTITIONS[0] + assert decode_cursor(cursor(ENVELOPED), partition=partition)["last_seq"] == 1 + assert ( + decode_cursor(cursor(None), partition=partition) is not None + ) # an unbound cursor is legal on this surface + with pytest.raises(OutboxError, match="cursor binding"): + decode_cursor(cursor(HEX64), partition=partition) + + +# --- the four surfaces this round moved, entered through their own reader ---------------- + + +def test_inbox_linked_references_keep_their_two_shapes(tmp_path: Path) -> None: + """The site whose shape reached `re.fullmatch` through a loop variable, not a literal. + + Both fields of a links receipt are checked by the same loop, so this case is also the + evidence that `todo_` identifiers kept their own twelve-character shape while the + evidence ids moved onto the owner's envelope. + """ + + from loopx.control_plane.collaboration import inbox + + row = {"request_id": HEX64} + + def read(value: dict[str, Any]) -> tuple[dict, str | None]: + path = inbox._root(tmp_path) / "links" / f"{row['request_id']}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value), encoding="utf-8") + return inbox._receipt(tmp_path, "links", row) + + accepted = { + "request_id": HEX64, + "todo_ids": [TODO_ID], + "evidence_ids": [ENVELOPED], + } + assert read(accepted) == (accepted, None) + # The todo shape stayed where it was, so an uppercase one is still not a todo id. + assert read({**accepted, "todo_ids": [TODO_ID.upper()]})[1] == ( + "links_unreadable_or_conflicting" + ) + # A bare digest is no longer accepted as evidence: this field carries the envelope. + assert read({**accepted, "evidence_ids": [HEX64]}) == ( + {}, + "links_unreadable_or_conflicting", + ) + assert read({**accepted, "todo_ids": [HEX64]}) == ( + {}, + "links_unreadable_or_conflicting", + ) + + +def test_outcome_projection_fingerprint_requires_the_envelope() -> None: + from loopx.capabilities.issue_fix import outcome_projection + + pattern = outcome_projection._REPOSITORY_FINGERPRINT_PATTERN + assert pattern.fullmatch(ENVELOPED) is not None + assert pattern.fullmatch(HEX64) is None + assert pattern is ENVELOPED_SHA256_PATTERN, "the surface kept a copy, not the owner" + + +def test_reviewer_notification_receipt_requires_the_envelope() -> None: + from loopx.domain_packs import issue_fix + + pattern = issue_fix.REVIEWER_NOTIFICATION_RECEIPT_PATTERN + assert pattern.fullmatch(ENVELOPED) is not None + assert pattern.fullmatch(f"{ENVELOPE}{HEX64.upper()}") is None + assert pattern is ENVELOPED_SHA256_PATTERN, "the domain pack restated the envelope" + + +def test_lark_idempotency_key_requires_the_envelope() -> None: + from loopx.extensions.lark import document_comment_provider + + pattern = document_comment_provider.IDEMPOTENCY_KEY_PATTERN + assert pattern.fullmatch(ENVELOPED) is not None + assert pattern.fullmatch(HEX64) is None + assert pattern is ENVELOPED_SHA256_PATTERN, "the provider kept its own copy" + + +def test_chat_bundle_source_digest_uses_the_owner_bare_shape() -> None: + from loopx.presentation import chat_bundle + + assert chat_bundle.BARE_SHA256_PATTERN is BARE_SHA256_PATTERN diff --git a/tests/presentation/test_chat_bundle_source_bootstrap.py b/tests/presentation/test_chat_bundle_source_bootstrap.py new file mode 100644 index 000000000..6926f2c41 --- /dev/null +++ b/tests/presentation/test_chat_bundle_source_bootstrap.py @@ -0,0 +1,184 @@ +"""A source checkout has to be able to build its own frontend without LoopX installed. + +`loopx/presentation/chat_bundle.py` asks the digest owner what a stored SHA-256 looks +like, and three places load that module by file path: `scripts/chat_bundle.py`, +`scripts/desktop_runtime_bundle.py` and `setup.py` (whose `build_py`/`sdist` hooks verify +the bundle for a non-editable install). A file-path load gives the module no package +context, so its import of the owner only resolves if the loader made the checkout +importable first - which is exactly what a checkout that has never been `pip install`ed +does not otherwise have. + +Two halves, both needed: + +* the real entry point is run under an interpreter with `site` disabled, so an installed + LoopX cannot rescue it, and the reported failure has to be the documented "bundle not + built yet" state rather than an import error; +* a de-bootstrapped copy of the same two files is run the same way and has to fail with + the import error. Without that control this file cannot tell a guard from a tautology. +""" + +from __future__ import annotations + +import ast +import os +from pathlib import Path +import subprocess +import sys + +ROOT = Path(__file__).resolve().parents[2] +CONTRACT = "loopx/presentation/chat_bundle.py" +IMPORT_ERROR = "No module named" + + +def _loads_contract(node: ast.AST) -> bool: + """Is this `spec_from_file_location(...)` call pointing at the chat bundle contract?""" + + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "spec_from_file_location" + and CONTRACT in ast.dump(node) + ) + + +def _loader_files() -> list[Path]: + """Every script or packaging file that execs the contract by file path.""" + + candidates = [ROOT / "setup.py", ROOT / "scripts"] + found: list[Path] = [] + for candidate in candidates: + paths = [candidate] if candidate.is_file() else sorted(candidate.rglob("*.py")) + for path in paths: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and _loads_contract(node): + found.append(path) + break + return found + + +def test_every_loader_of_the_contract_makes_the_checkout_importable() -> None: + """A new file-path load of the contract has to bring the root onto sys.path too. + + Checked structurally: the module-level `sys.path` insertion must be present in the + loader, because the contract no longer carries its own copy of the digest shape. + """ + + loaders = _loader_files() + assert len(loaders) >= 3, ( + f"expected the three known loaders of {CONTRACT}, found {[str(p) for p in loaders]}" + ) + missing = [] + for path in loaders: + source = path.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(path)) + assigns = [ + node + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "insert" + and isinstance(node.func.value, ast.Attribute) + and node.func.value.attr == "path" + ] + if not assigns: + missing.append(str(path.relative_to(ROOT))) + assert not missing, ( + f"loader(s) of the contract that never touch sys.path: {missing}" + ) + + +def _run_uninstalled( + script: Path, *arguments: str, cwd: Path +) -> subprocess.CompletedProcess: + """Run a script with `site` disabled, so an installed LoopX cannot answer for the tree.""" + + # `-I` ignores PYTHONPATH and user site, `-S` hides site-packages; together they are + # what stops an installed LoopX from rescuing a checkout-local loader. The environment + # is inherited minus PYTHONPATH, because on Windows a stripped environment loses + # SystemRoot and the interpreter then fails for a reason this test is not about. + environment = { + key: value for key, value in os.environ.items() if key != "PYTHONPATH" + } + return subprocess.run( + [sys.executable, "-I", "-S", str(script), *arguments], + cwd=cwd, + env=environment, + capture_output=True, + text=True, + timeout=180, + check=False, + ) + + +def test_the_real_entry_point_starts_without_an_installed_loopx(tmp_path: Path) -> None: + """`--help` must succeed, and `verify` must fail for the bundle, not for the import. + + On a checkout that has never built the frontend there is no bundle to verify, so the + expected outcome for `verify` is the documented missing-bundle error. The point of this + case is the *absence* of an import error, which is the regression this branch fixed. + """ + + help_result = _run_uninstalled( + ROOT / "scripts" / "chat_bundle.py", "--help", cwd=tmp_path + ) + assert help_result.returncode == 0, help_result.stderr + assert IMPORT_ERROR not in help_result.stderr + help_result.stdout + assert "build" in help_result.stdout + + verify_result = _run_uninstalled( + ROOT / "scripts" / "chat_bundle.py", "verify", cwd=tmp_path + ) + combined = verify_result.stdout + verify_result.stderr + assert IMPORT_ERROR not in combined, combined + assert "bundle-manifest.json" in combined or verify_result.returncode == 0, combined + + +def test_the_bootstrap_is_load_bearing(tmp_path: Path) -> None: + """Remove those two lines and the same command must die at the import instead. + + Without this control the tests above would pass in any environment that happens to have + LoopX installed, and would prove nothing about a bare source checkout. + """ + + sandbox = tmp_path / "checkout" + (sandbox / "scripts").mkdir(parents=True, exist_ok=True) + for relative in ( + "loopx/__init__.py", + "loopx/presentation/__init__.py", + "loopx/control_plane/__init__.py", + ): + target = sandbox / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text("", encoding="utf-8") + for relative in ( + "loopx/presentation/chat_bundle.py", + "loopx/control_plane/content_digest.py", + ): + source = ROOT / relative + destination = sandbox / relative + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(source.read_text(encoding="utf-8"), encoding="utf-8") + + script = (ROOT / "scripts" / "chat_bundle.py").read_text(encoding="utf-8") + with_bootstrap = sandbox / "scripts" / "chat_bundle_with_bootstrap.py" + without_bootstrap = sandbox / "scripts" / "chat_bundle_without_bootstrap.py" + with_bootstrap.write_text(script, encoding="utf-8") + + dropped = ( + "if str(ROOT) not in sys.path:", + "sys.path.insert(0, str(ROOT))", + ) + stripped = "".join( + line for line in script.splitlines(keepends=True) if line.strip() not in dropped + ) + assert "sys.path.insert" not in stripped + assert len(stripped.splitlines()) == len(script.splitlines()) - 2, stripped + without_bootstrap.write_text(stripped, encoding="utf-8") + + kept = _run_uninstalled(with_bootstrap, "--help", cwd=tmp_path) + assert kept.returncode == 0, kept.stderr + removed = _run_uninstalled(without_bootstrap, "--help", cwd=tmp_path) + assert removed.returncode != 0 + assert IMPORT_ERROR in removed.stderr, removed.stderr + assert "loopx" in removed.stderr