refactor(control-plane): decide the stored digest shape in one owner

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(
             {

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
 

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,

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):

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
 

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",

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)

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

loopx/capabilities/manager_context/roundtrip.py

@@ -7,7 +7,6 @@
 from __future__ import annotations
 
 import logging
-import re
 import threading
 from datetime import datetime, timezone, timedelta
 
@@ -19,6 +18,7 @@
 from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
 
 from ...control_plane.collaboration.inbox import needs_conclusion as needs_conclusion
+from ...control_plane.content_digest import BARE_SHA256_PATTERN
 
 PHASES = ("decision", "conclusion")
 DELIVERY_STATUSES = {
@@ -247,7 +247,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

loopx/capabilities/manager_context/tracking.py

@@ -14,6 +14,10 @@
 from ...file_lock import exclusive_file_lock
 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):
@@ -33,7 +37,7 @@ def link(root, registry_path, goal_id, agent_id, request_id, todo_ids, evidence_
         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)
@@ -79,7 +83,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")

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 = (

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
 

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(
             {

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")

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